diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml index a0d4abcad..7b469948c 100644 --- a/.github/workflows/ci-tests.yml +++ b/.github/workflows/ci-tests.yml @@ -587,6 +587,18 @@ jobs: run: node --import tsx bench/finalize-reexport/measure.mjs --check working-directory: gitnexus + - name: Parse dispatch-round cadence guards (#3194, #3196) + if: ${{ !cancelled() }} + # Build-free: asserts parse-cache pack membership is unchanged + # (fingerprint — every cache key derives from it), that a fixed corpus + # still batches into a fixed number of dispatch rounds, and that the + # round budget counts UTF-8 bytes rather than UTF-16 code units. Round + # boundaries are deliberately invisible to graph output, so no test can + # see these regress. Rationale and history: see the header of + # bench/parse-dispatch-rounds/measure.mjs. + run: node --import tsx bench/parse-dispatch-rounds/measure.mjs --check + working-directory: gitnexus + - name: C++ qualified-namespace resolution guards (#2788) if: ${{ !cancelled() }} # Build-free: asserts resolveCppQualifiedNamespaceMember resolves an diff --git a/gitnexus/bench/analyze-phase-breakdown.md b/gitnexus/bench/analyze-phase-breakdown.md new file mode 100644 index 000000000..2a3a0a221 --- /dev/null +++ b/gitnexus/bench/analyze-phase-breakdown.md @@ -0,0 +1,126 @@ +# Analyze phase breakdown — where the time actually goes + +Measured 2026-09-06/07 while landing #3194, #3196 and #3200. Records what the +`analyze` pipeline costs per phase, and — as importantly — the optimizations +that were measured and **rejected**, so the next person does not re-derive them. + +Unlike `parse-throughput.md` (a synthetic-fixture scaffold), these numbers come +from a real repo corpus. They are not a CI gate; the gate for the parse +dispatch path is `bench/parse-dispatch-rounds/`. + +--- + +## Method, and one trap that invalidates everything + +**The corpus must be a git repository.** A non-git checkout cannot record a +schema fingerprint, so the tool forces a full rebuild on every run and prints: + +> `non-git repositories never record a schema fingerprint, so this run rebuilds regardless` + +A "warm" run measured that way is a forced cold rebuild wearing a warm label. A +37.7s figure was recorded that way during this work and was meaningless. On a +real git repo an unchanged re-analyze short-circuits to `Already up to date`. + +Corpus: this repository, `git archive` of HEAD into a scratch dir, then +`git init && git add -A && git commit`. 5350 paths, 2234 parseable, ~30MB. +16 workers, `dist` on a local overlay filesystem (see "Filesystem" below). +Phase numbers come from the `✓ Phase: ()` lines under +`NODE_ENV=development`. + +--- + +## Cold analyze + +| | before #3194 | after #3196 | +| ---------------- | -----------: | ----------: | +| total | 110.3s | **63.6s** | +| parse | 74.0s | 36.0s | +| scopeResolution | 17.0s | 16.0s | +| all other phases | 1.2s | 1.2s | +| dispatches | 221 | 15 | + +Graph output identical throughout: 51,286 nodes / 163,092 edges / 2106 clusters +/ 759 flows. + +## Re-analyze after a one-file edit — the developer loop + +One file changed out of 5350, on a git repo. **36.5s total.** + +| phase | ms | share | +| ----------------------------------- | ------: | ----: | +| scopeResolution | 14,717 | 40% | +| unlogged — graph emit + FTS rebuild | ~18,000 | 49% | +| parse | 2,832 | 8% | +| all other phases | ~1,200 | 3% | + +The parse cache works: it replays 2231 of 2232 chunks. **Everything the parse +work optimized is that 8%.** The other 92% is not incremental at all — the run +banner says so directly: _"Rebuilt the graph and FTS while reusing cached +parser output."_ + +Note the ~18s sits **outside the phase runner**, so every `✓ Phase` line is +blind to it. `phasesSum` and the log span agree exactly; the gap is wall-clock +before the first phase and after the last. + +## scopeResolution is memory-traffic bound, not algorithmic + +`--cpu-prof` of a one-file-edit run, top main-thread self time: + +``` +4294ms (garbage collector) +1552ms v8.deserialize + 756ms crypto update + 728ms runScopeResolution + 620ms scope-resolution/pipeline/reconcile-ownership + 380ms v8-sidecar walk + 323ms internString +``` + +then a tail of passes at 200–750ms (`emitReceiverBoundCalls`, +`emitCallableValueFlow`, `buildGraphNodeLookup`, `resolveReferenceSites`). + +No dominant hot function, nothing quadratic. The cost is rehydrating every +file's `ParsedFile` from the durable `.v8` shards into main-thread memory and +re-running every pass over them. + +**So the win is not caching resolution output** — that stores more of exactly +what is already the memory problem, and #2649 (large-repo OOM) is the standing +constraint. The win is skipping rehydration and re-resolution for files whose +inputs provably did not change, as a streaming/bounded design. + +--- + +## Measured and rejected + +Recorded because each cost real time to establish and each looks attractive +from the armchair. + +**More workers buys nothing.** Isolated-harness wall time at 16 / 20 / 24 +workers: 44.1s / 44.8s / 43.3s — a 1.5s spread against a 3.7s within-size +spread. A full-analyze sweep appeared to show 20 beating 16 by 6.3s; it was +noise, and the sweep was invalid anyway because `GITNEXUS_WORKER_POOL_SIZE` was +silently clamped at the time (fixed in #3200). `DEFAULT_POOL_SIZE_CAP = 16` +stands. + +**Bundling the worker entry buys ~250ms.** Worker boot profiled at 12.2s per +worker, of which `getPackageScopeConfig` 5.86s + `internalModuleStat` 3.14s + +`lstat`/`open` ~2.2s — ESM module resolution, not native grammars (all 11 +`tree-sitter-*` imports together are 33ms) and not V8 compile (53ms; +`NODE_COMPILE_CACHE` gives zero gain). An esbuild bundle takes 16-worker boot +8.6s → 0.37s. + +**That was a filesystem artifact.** On a normal overlay filesystem the same +boot is 515ms stock vs 263ms bundled — 0.2% of a 110s analyze. The 8.6s only +reproduces with the repo on a 9p mount (WSL2 `D:\`). Dropped. + +Caveat carried by every number here: `dist` on a 9p mount costs ~3.5s of a 73s +run (73.0s vs 69.6s on overlay). Measure on a local filesystem. + +--- + +## Open + +The ~18s of graph emit + FTS rebuild is **unprofiled**. It is the largest +single share of the edit loop and nothing is known about it beyond the banner. +Profile it before proposing anything — two optimizations in this document +looked compelling until measured. diff --git a/gitnexus/bench/parse-dispatch-rounds/baselines.json b/gitnexus/bench/parse-dispatch-rounds/baselines.json new file mode 100644 index 000000000..febd79edd --- /dev/null +++ b/gitnexus/bench/parse-dispatch-rounds/baselines.json @@ -0,0 +1,31 @@ +{ + "_what": "Baselines for bench/parse-dispatch-rounds/measure.mjs --check. Guards parse-cache pack layout and dispatch-round cadence. Neither is visible in graph output — batching that changed output would be a bug — so nothing else in the repo can see these regress. Four of the five arms are deterministic; only pack_scaling_ratio is a timing signal.", + + "_triage": "READ THIS BEFORE RE-RUNNING. layout_fingerprint, packs, single_file_packs, rounds, cjk_rounds and ascii_rounds are DETERMINISTIC: a re-run never changes them, and none may be re-baselined to make CI green. pack_scaling_ratio is the only timing arm; runner contention dominates it, so re-run on an idle machine before investigating and read the reported `reps` first. If exactly one arm fails and it is that one, suspect the machine.", + + "layout_fingerprint": "cc875fd264498964b463aef55cec0166d57468a092303e94f1ed7f09fe141a44", + "_layout_fingerprint_note": "sha256 over the sorted pack membership — which files share a pack, and their order within it. Every parse-cache key derives from a pack's file set, so a change here invalidates every cached chunk for every user. This is a CORRECTNESS gate: drift needs a SCHEMA_BUMP in src/storage/parse-cache.ts alongside a new fingerprint, never a lone re-baseline.", + + "packs": 774, + "single_file_packs": 251, + "_shape_note": "THE FLOOR. Without these two, every arm below is a ceiling over nothing. `rounds` only asserts something while the corpus OVER-SPLITS — 774 packs where the byte budget alone needs 5, 251 of them holding a single file. Shrink the corpus until packing stops over-splitting and rounds still reads 5 and still passes, asserting a property the corpus no longer has. bench/import-target learned this the hard way: four heap arms read 0 B and passed every ceiling, because a ceiling says 'not too big' and nothing said 'still measuring something'.", + + "rounds": 5, + "_rounds_note": "Exact round count for the fixed corpus at the 2MB budget, folded through the production accumulator in pipeline-phases/parse-round-budget.ts. HIGHER (toward packs=774) means dispatch went back to one barrier per cache pack — the #3196 regression, measured at ~1.5x on a cold analyze with no visible symptom. LOWER (toward 1) means the close condition stopped firing, so an open round retains the whole repo until the tail drain (#2649 heap shape). Both directions verified to fail this arm before it was recorded: forcing close-every-chunk reads 774, disabling the close reads 1.", + + "cjk_rounds": 8, + "ascii_rounds": 3, + "_encoding_note": "The round budget bounds what the MAIN THREAD HOLDS, so it must count UTF-8 bytes. String.length returns UTF-16 code units: a CJK character is one unit but three UTF-8 bytes, so reverting the unit would let a CJK-heavy repo hold ~3x its nominal budget before draining. The two corpora are constructed to have IDENTICAL UTF-16 length and differ only in encoded size, so under String.length both close 3 rounds and the arm collapses. Verified: reverting roundFileBytes to content.length takes cjk_rounds 8 -> 3. This is the arm that pins the change no unit test could — round cadence changes no graph output, so a test asserting output passes either way.", + + "pack_scaling_budget": 1.6, + "_pack_scaling_note": "(t_4n / t_n) / 4 for packParseCacheChunks; ~1.0 is linear. A RATIO rather than a millisecond ceiling, deliberately: wall-clock is runner-speed-dependent, and this repo has already been bitten by a fixed ms budget — bench/callable-value-flow's widening_overhead gate failed twice on a shared runner at 2.07 and 1.975 against a 1.9 budget while the code was correct, on a sub-11ms measurement. A ratio divides the machine out. Measured over 5 runs on a NON-idle box: 0.940, 0.946, 0.977, 0.998, 1.085 (peak-to-peak 1.154). Budget is 1.6, i.e. 1.47x the measured maximum — this file's siblings use ~1.5x on ratios. It catches packParseCacheChunks going superlinear (it sorts within each bucket, so a global sort or a nested scan lands here) and is not tight enough to police drift. min-of-15 estimator, matching bench/import-target's finding that N=5 tripped its own budget ~1 run in 20 while N=15 held every language inside a 1.13-1.26x swing.", + + "_measured": { + "pack_scaling_ratio": 1.085, + "pack_scaling_ratio_samples": [0.94, 0.946, 0.977, 0.998, 1.085], + "small_ms": 1.91, + "large_ms_4x": 7.46, + "reps": 15 + }, + "_measured_note": "Maxima over 5 runs on a box that was NOT idle, so the ratio spread is an upper bound on its real noise. small_ms/large_ms_4x are recorded for context only — nothing gates on them, because an absolute millisecond is exactly the gate this file avoids." +} diff --git a/gitnexus/bench/parse-dispatch-rounds/measure.mjs b/gitnexus/bench/parse-dispatch-rounds/measure.mjs new file mode 100644 index 000000000..6543b561e --- /dev/null +++ b/gitnexus/bench/parse-dispatch-rounds/measure.mjs @@ -0,0 +1,279 @@ +/** + * Build-free bench for parse-cache pack layout and dispatch-round cadence. + * + * WHY THIS EXISTS. `WorkerPool.dispatch` is a barrier: it resolves only once + * every job it created has committed. Packs are keyed `(language, + * sha256(path) % 128)`, so the byte budget almost never binds and most packs + * land far below the pool size — this repo produced 1285 packs where the + * budget alone needed 16, and 549 held a single file. Dispatching one pack at + * a time therefore left most workers idle for every round-trip. #3194 fixed + * fan-out WITHIN a pack; #3196 batched packs into bounded rounds and took a + * cold analyze from 110.3s to 70.5s (221 dispatches -> 15). + * + * Nothing guarded that. Round boundaries are deliberately invisible to the + * graph — batching that changed output would be a bug — so no test can see the + * regression, and it would come back as a silent 1.5x on every cold analyze. + * Two earlier attempts to pin this as a unit test failed for exactly that + * reason: one scraped a logger line the progress stream does not carry, the + * other asserted graph content that is identical either way. + * + * FOUR ARMS, and only the last is a timing arm: + * + * - `rounds` — EXACT. The regression signal. A fixed corpus and budget must + * produce a fixed number of rounds. Per-pack dispatch coming back sends this + * to `packs`; a broken close condition sends it to 1. + * + * - `cjk_rounds` vs `ascii_rounds` — EXACT. The round budget bounds what the + * MAIN THREAD HOLDS, so it must count UTF-8 bytes. `String.length` returns + * UTF-16 code units: a CJK character is one unit but three UTF-8 bytes, so + * reverting the unit would let a CJK-heavy repo hold ~3x its nominal budget + * before draining — the #2649 heap-failure shape. The two corpora are + * identical in UTF-16 length and differ only in encoded size, so under + * `String.length` they would close the SAME number of rounds. Only a UTF-8 + * count separates them. + * + * - `packs` / `single_file_packs` — EXACT, and they are the FLOOR. `rounds` + * only asserts something while the corpus over-splits (774 packs where the + * byte budget alone needs 5). Shrink the corpus past that and `rounds` still + * reads 5 and still passes, gating a property the corpus no longer has. + * bench/import-target learned this when four heap arms read 0 B and passed. + * + * - `pack_scaling_ratio` — the only timing arm, and a RATIO not a millisecond + * ceiling. (t_4n/t_n)/4 divides the machine out; ~1.0 is linear. A fixed ms + * budget on a shared runner is a coin flip, and this repo has the scar: + * bench/callable-value-flow's gate failed twice at 2.07 and 1.975 against a + * 1.9 budget with correct code, on a sub-11ms measurement. Catches + * `packParseCacheChunks` going superlinear; not tight enough to police drift. + * + * Usage: + * node --import tsx bench/parse-dispatch-rounds/measure.mjs # report + * node --import tsx bench/parse-dispatch-rounds/measure.mjs --check # CI gate + */ +import { performance } from 'node:perf_hooks'; +import { createHash } from 'node:crypto'; +import { readFileSync } from 'node:fs'; +import { packParseCacheChunks } from '../../src/storage/parse-cache.js'; +import { createRoundBudget } from '../../src/core/ingestion/pipeline-phases/parse-round-budget.js'; + +const baselines = JSON.parse( + new URL('./baselines.json', import.meta.url).pathname + ? readFileSync(new URL('./baselines.json', import.meta.url), 'utf8') + : '{}', +); + +/** Matches DEFAULT_CHUNK_BYTE_BUDGET / the round budget's default in parse-impl.ts. */ +const BUDGET = 2 * 1024 * 1024; + +/** + * A repo shaped like a real one: many languages, so `(language, bucket)` + * packing over-splits well past what the byte budget alone would need. Sizes + * are deliberately uneven — a uniform corpus hides an off-by-one in the fold. + */ +function mixedCorpus(scale = 1) { + const langs = [ + ['ts', 900], + ['py', 400], + ['java', 260], + ['go', 240], + ['rb', 120], + ['rs', 180], + ['php', 90], + ['cs', 140], + ]; + const files = []; + for (const [ext, count] of langs) { + for (let i = 0; i < count * scale; i++) { + files.push({ + path: `src/${ext}/mod${i}.${ext}`, + // 400B - 8KB, varying by index so packs are not uniform. + size: 400 + ((i * 977) % 7700), + language: ext, + }); + } + } + return files; +} + +/** Feed chunks through the real accumulator and count the rounds it closes. */ +function roundsFor(chunks, contentsByPath, budgetBytes) { + const budget = createRoundBudget(budgetBytes); + let rounds = 0; + for (const chunk of chunks) { + if (budget.addChunk(chunk.map((p) => contentsByPath.get(p)))) rounds++; + } + // The tail drain closes a partially-filled round when anything is left. + if (budget.bufferedBytes > 0) rounds++; + return rounds; +} + +/** + * Two corpora with IDENTICAL UTF-16 length and different UTF-8 size. Under + * `String.length` both close the same number of rounds; under UTF-8 the CJK + * one closes strictly more. + */ +function encodingCorpora() { + // 1 UTF-16 unit / 3 UTF-8 bytes each, vs 1 unit / 1 byte each. + const cjkLine = '説'.repeat(240); + const asciiLine = 'a'.repeat(240); + const count = 260; + const files = Array.from({ length: count }, (_, i) => ({ + path: `src/enc/mod${i}.ts`, + size: 240, + language: 'ts', + })); + const chunks = packParseCacheChunks(files, BUDGET); + const cjk = new Map(files.map((f) => [f.path, cjkLine])); + const ascii = new Map(files.map((f) => [f.path, asciiLine])); + // A budget small enough that both corpora close several rounds. + const encBudget = 24 * 1024; + return { + utf16Length: cjkLine.length === asciiLine.length, + cjkRounds: roundsFor(chunks, cjk, encBudget), + asciiRounds: roundsFor(chunks, ascii, encBudget), + }; +} + +/** + * Min-of-N estimator. `fastest` rather than a mean because the minimum is the + * least contaminated sample on a shared runner — the same choice, and the same + * reason, as bench/import-target's `fastest()`. + */ +function fastest(fn, reps) { + fn(); // warm + let best = Infinity; + for (let r = 0; r < reps; r++) { + const t0 = performance.now(); + fn(); + best = Math.min(best, performance.now() - t0); + } + return best; +} + +const REPS = 15; + +const corpus = mixedCorpus(); +const corpus4x = mixedCorpus(4); + +const packs = packParseCacheChunks(corpus, BUDGET); +// A RATIO, not a millisecond ceiling. Wall-clock is runner-speed-dependent and +// a fixed ms budget on a shared runner is a coin flip — this file's sibling +// benches record exactly that failure. (t_4n / t_n) / 4 divides the machine +// out: ~1.0 is linear, and packParseCacheChunks going superlinear (it sorts +// within each bucket) shows up here regardless of how fast the box is. +const smallMs = fastest(() => packParseCacheChunks(corpus, BUDGET), REPS); +const largeMs = fastest(() => packParseCacheChunks(corpus4x, BUDGET), REPS); +const packScaling = largeMs / smallMs / 4; + +const contents = new Map(corpus.map((f) => [f.path, 'x'.repeat(f.size)])); +const rounds = roundsFor(packs, contents, BUDGET); + +const enc = encodingCorpora(); + +/** + * Order-independent hash of the pack layout: which files share a pack, and in + * what order within it. Catches a packing change that leaves the counts intact + * but moves files between packs — which would silently change every cache key. + */ +const layoutFingerprint = createHash('sha256') + .update( + packs + .map((chunk) => chunk.join(',')) + .sort() + .join('\n'), + ) + .digest('hex'); + +const singleFilePacks = packs.filter((c) => c.length === 1).length; +const totalBytes = corpus.reduce((sum, f) => sum + f.size, 0); +const budgetFloor = Math.ceil(totalBytes / BUDGET); + +console.log(`files : ${corpus.length}`); +console.log( + `packs : ${packs.length} (expect ${baselines.packs}; byte budget alone needs ${budgetFloor})`, +); +console.log(`rounds : ${rounds} (expect ${baselines.rounds})`); +console.log(`single_file_packs : ${singleFilePacks} (expect ${baselines.single_file_packs})`); +console.log(`cjk_rounds : ${enc.cjkRounds} (UTF-8 bytes)`); +console.log(`ascii_rounds : ${enc.asciiRounds} (same UTF-16 length)`); +console.log(`layout_fingerprint : ${layoutFingerprint.slice(0, 16)}`); +console.log( + `pack_scaling_ratio : ${packScaling.toFixed(3)} (budget <= ${baselines.pack_scaling_budget}; ~1.0 is linear)`, +); +console.log( + `reps : ${REPS} small ${smallMs.toFixed(2)}ms / 4x ${largeMs.toFixed(2)}ms`, +); + +if (process.argv.includes('--check')) { + let failed = false; + + if (layoutFingerprint !== baselines.layout_fingerprint) { + failed = true; + console.error( + `\nFAIL layout_fingerprint: ${layoutFingerprint}\n` + + ` expected ${baselines.layout_fingerprint}\n` + + ` Pack membership moved. Every parse-cache key is derived from a pack's\n` + + ` file set, so this invalidates every cached chunk for every user. If the\n` + + ` change is intended, it needs a SCHEMA_BUMP in src/storage/parse-cache.ts\n` + + ` alongside a new fingerprint here — never re-baseline it alone.`, + ); + } + + if (rounds !== baselines.rounds) { + failed = true; + console.error( + `\nFAIL rounds: ${rounds}, expected exactly ${baselines.rounds}.\n` + + ` HIGHER (toward packs=${packs.length}) means rounds stopped batching and\n` + + ` dispatch went back to one barrier per cache pack — the #3196 regression,\n` + + ` worth ~1.5x on a cold analyze with no visible symptom.\n` + + ` LOWER (toward 1) means the close condition stopped firing, so an open\n` + + ` round retains the whole repo until the tail drain (#2649 heap shape).\n` + + ` Check createRoundBudget in pipeline-phases/parse-round-budget.ts.`, + ); + } + + if (!enc.utf16Length) { + failed = true; + console.error( + `\nFAIL encoding arm is broken: its two corpora no longer share a UTF-16 length.`, + ); + } else if (enc.cjkRounds <= enc.asciiRounds) { + failed = true; + console.error( + `\nFAIL cjk_rounds ${enc.cjkRounds} <= ascii_rounds ${enc.asciiRounds}.\n` + + ` These corpora have identical UTF-16 length and differ only in encoded\n` + + ` size, so equal round counts mean the budget is counting String.length\n` + + ` again instead of Buffer.byteLength. A CJK-heavy repo would then hold\n` + + ` ~3x its nominal budget on the main thread before draining.\n` + + ` See roundFileBytes in pipeline-phases/parse-round-budget.ts.`, + ); + } + + // SHAPE — the floor. Without it every arm below is a ceiling over nothing: + // shrink the corpus until packing stops over-splitting and `rounds` still + // reads 5 and still passes, asserting a property the corpus no longer has. + if (packs.length !== baselines.packs || singleFilePacks !== baselines.single_file_packs) { + failed = true; + console.error( + `\nFAIL shape: packs ${packs.length} (expected ${baselines.packs}), ` + + `single_file_packs ${singleFilePacks} (expected ${baselines.single_file_packs}).\n` + + ` The corpus must stay one that OVER-SPLITS — ${packs.length} packs where the\n` + + ` byte budget alone needs ${budgetFloor}. That gap is the entire reason rounds\n` + + ` exist, so if it closes, the rounds arm below asserts nothing.`, + ); + } + + if (packScaling > baselines.pack_scaling_budget) { + failed = true; + console.error( + `\nFAIL pack_scaling_ratio: ${packScaling.toFixed(3)} exceeds ` + + `${baselines.pack_scaling_budget} (~1.0 is linear).\n` + + ` packParseCacheChunks grew superlinearly in file count — it sorts within\n` + + ` each bucket, so a global sort or a nested scan lands here.\n` + + ` This is the ONLY timing arm in this file: re-run on an idle machine\n` + + ` before investigating, and check \`reps\` in the report first.`, + ); + } + + if (failed) process.exit(1); + console.log('\nOK — within budget.'); +} diff --git a/gitnexus/src/core/ingestion/parsing-processor.ts b/gitnexus/src/core/ingestion/parsing-processor.ts index 9a1f39233..4798c1f9f 100644 --- a/gitnexus/src/core/ingestion/parsing-processor.ts +++ b/gitnexus/src/core/ingestion/parsing-processor.ts @@ -253,10 +253,7 @@ export const dispatchChunkParse = async ( * owns it. */ export const dispatchChunkParseRound = async ( - groups: ReadonlyArray<{ - items: { path: string; content: string }[]; - chunkHash?: string; - }>, + groups: ReadonlyArray>, workerPool: WorkerPool, onFileProgress?: FileProgressCallback, ): Promise => { diff --git a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts index b4ec9a5b8..18542637d 100644 --- a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts +++ b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts @@ -69,6 +69,8 @@ import { createWorkerPool, workerPoolDisabledByEnv, resolveAutoPoolSize, + envWorkerPoolSize, + resolveHostParallelism, WorkerPoolInitializationError, WorkerPoolDisabledError, } from '../workers/worker-pool.js'; @@ -115,6 +117,8 @@ import { import { isDebugHeapEnabled, logHeapProbe } from '../utils/heap-probe.js'; import { logger } from '../../logger.js'; +import { mapConcurrent } from '../../../lib/utils.js'; +import { createRoundBudget } from './parse-round-budget.js'; // ── Constants ────────────────────────────────────────────────────────────── /** @@ -213,11 +217,18 @@ const CHUNK_BYTES_PER_WORKER = DEFAULT_CHUNK_BYTE_BUDGET; */ const TARGET_JOBS_PER_WORKER = 3; +/** + * Concurrent durable ParsedFile directory resets per round. Matches the file + * reader's `READ_CONCURRENCY`, because both compete for the same descriptors. + */ +const DURABLE_RESET_CONCURRENCY = 32; + /** Floor for a derived sub-batch so jobs don't shrink to per-file IPC churn. */ const MIN_SUB_BATCH_BYTES = 256 * 1024; /** - * Source bytes of cache-missing chunks allowed in flight in one pool round. + * Source bytes an open round may HOLD — cache hits and misses alike — before + * it is dispatched and drained. * * A `dispatch` is a barrier, so one round-trip per cache pack leaves most slots * idle: packs are keyed by `(language, hash(path) % 128)` and routinely land far @@ -579,12 +590,42 @@ export async function runChunkedParseAndResolve( // cores-based auto size is capped by source bytes / CHUNK_BYTES_PER_WORKER // so a tiny repo does not spawn a full idle pool. Cache pack membership // is independent of this number (#3088). - const explicitPoolSize = options?.workerPoolSize; + // `--workers ` and `GITNEXUS_WORKER_POOL_SIZE` are both deliberate + // operator input, so both bypass the work-proportional cap below. Only the + // env path used to be clamped by it, which made the documented escape hatch + // silently do nothing: on a 30MB repo the cap resolves to 16, so an operator + // asking for 24 still got 16 with no warning, while `--workers 24` got 24. + const explicitPoolSize = options?.workerPoolSize ?? envWorkerPoolSize(); + // Cores-based auto size, bounded by source bytes so a tiny repo does not + // spawn a full idle pool. const workProportionalCap = Math.max(1, Math.ceil(totalBytes / CHUNK_BYTES_PER_WORKER)); + // An operator's number is honored, but never exceeds the number of files + // there are to parse — `GITNEXUS_WORKER_POOL_SIZE=100000` on a five-file repo + // should not become the literal thread count. This bounds `--workers` and the + // env var identically, keeping the parity above intact. Note it does NOT + // shrink an incremental re-analyze: `totalParseable` counts every parseable + // file in the scan, not the changed ones, so a warm run of a large repo still + // spawns the full requested pool. const effectivePoolSize = explicitPoolSize && explicitPoolSize > 0 - ? explicitPoolSize + ? Math.min(explicitPoolSize, Math.max(1, totalParseable)) : Math.min(resolveAutoPoolSize(), workProportionalCap); + // Deliberate over-subscription is the operator's call, so this warns rather + // than caps — silently capping is what the override exists to stop. But an + // exported `GITNEXUS_WORKER_POOL_SIZE` applies to EVERY analyze in a + // long-lived caller (watch auto-sync, the MCP server), including small + // incremental ones, and that is easy to set once and forget. + if (explicitPoolSize && explicitPoolSize > 0) { + const hostParallelism = resolveHostParallelism(); + if (effectivePoolSize > hostParallelism) { + logger.warn( + { requested: explicitPoolSize, spawning: effectivePoolSize, hostParallelism }, + `Worker pool size ${effectivePoolSize} exceeds this host's ${hostParallelism} usable core(s); ` + + `parsing is CPU-bound, so the extra workers add memory pressure without throughput. ` + + `This applies to every analyze while the override is set.`, + ); + } + } // Cache packs: stable (language, hash(path) mod 128) buckets, then the // per-call byte budget inside each bucket (#3088). Pool size is used only // for worker count and sub-batch fan-out, not membership. @@ -861,20 +902,31 @@ export async function runChunkedParseAndResolve( readonly chunkStartMs: number | null; }; + /** + * Chunk hashes whose durable ParsedFile directory could not be reset. The + * old generation's shards are still on disk, so a warm hit would union + * stale shards with the new ones. Treated exactly like a quarantined chunk: + * skip the parse-cache write so the next run re-dispatches into a clean + * directory rather than trusting a generation we could not clear. + */ + const durablePrepareFailures = new Set(); + const roundByteBudget = resolveParseRoundByteBudget(options); let roundEntries: RoundEntry[] = []; - let roundMissBytes = 0; /** * Bytes an open round is HOLDING, counting hits as well as misses. * - * `roundMissBytes` alone bounds only what the workers are asked to do, so a - * warm run — where nothing misses — would never reach the close condition - * and would buffer every chunk's cached output until the tail drain. That - * is the #2649 heap failure on a large repo. Closing on either cap keeps a - * hits-only run draining at the same cadence as a cold one; `startRound` - * already supports a round with no misses. + * Counting only the cache-MISSING bytes would bound just what the workers + * are asked to do, so a warm run — where nothing misses — would never reach + * the close condition and would buffer every chunk's cached output until + * the tail drain. That is the #2649 heap failure on a large repo. Counting + * both keeps a hits-only run draining at the same cadence as a cold one; + * `startRound` already supports a round with no misses. + * + * Measured in UTF-8 bytes, matching `estimateItemBytes` in the worker pool, + * so the cap means the same thing here as it does for a job's payload. */ - let roundBufferedBytes = 0; + const roundBudget = createRoundBudget(roundByteBudget); /** * Files QUEUED into rounds so far. `filesParsedSoFar` only advances when a * round drains, so it is the right number for the throughput log but would @@ -998,7 +1050,14 @@ export async function runChunkedParseAndResolve( if (parseCache && p.chunkHash && rawResults.length > 0) { const quarantineSet = new Set(workerPool?.getQuarantinedPaths?.() ?? []); const chunkHadQuarantine = p.chunkFiles.some((f) => quarantineSet.has(f.path)); - if (chunkHadQuarantine) { + const durableGenerationStale = durablePrepareFailures.has(p.chunkHash); + if (durableGenerationStale) { + logger.warn( + { chunkHash: p.chunkHash.slice(0, 8) }, + 'parse-cache SKIP: durable generation for this chunk could not be reset, ' + + 'so its shards may be stale; next run will re-dispatch it', + ); + } else if (chunkHadQuarantine) { if (isDev) { const quarantinedInChunk = p.chunkFiles.filter((f) => quarantineSet.has(f.path)).length; logger.info( @@ -1033,21 +1092,39 @@ export async function runChunkedParseAndResolve( if (misses.length === 0) { return { entries, results: Promise.resolve([]) }; } - for (const miss of misses) { - if (durableParsedFileDir !== undefined && miss.chunkHash !== null) { + // Each chunk resets its own directory, so these are independent and run + // concurrently: serially they would sit on the critical path this round + // exists to shorten, with the pool idle and the previous round's merge + // waiting, once per miss. + // + // BOUNDED, though. A round can hold hundreds of small packs, and each + // reset is a recursive rm + mkdir. Firing all of them at once competes + // for descriptors with the chunk prefetch this loop already has in + // flight, and `readFileContents` degrades a losing read SILENTLY by + // contract — a dropped file would vanish from the chunk, from the graph, + // and from the chunk hash, shipping a narrowed index with exit 0. Same + // helper and width the file reads use. + await mapConcurrent( + misses, + async (miss) => { + if (durableParsedFileDir === undefined || miss.chunkHash === null) return; try { await prepareDurableParsedFileChunk(durableParsedFileDir, miss.chunkHash); } catch (err) { // The durable store is an optimization — degrade like the restore // path does instead of failing the analyze. Workers recreate the // directory on write, so at worst the old generation lingers. + // Caught per chunk so one failure cannot abort the others. + durablePrepareFailures.add(miss.chunkHash); logger.warn( { err, chunkHash: miss.chunkHash.slice(0, 8) }, - 'parsedfile-cache: could not reset durable chunk generation; continuing', + 'parsedfile-cache: could not reset durable chunk generation; ' + + 'continuing without caching this chunk', ); } - } - } + }, + { concurrency: DURABLE_RESET_CONCURRENCY }, + ); const roundFiles = misses.reduce((sum, miss) => sum + miss.chunkFiles.length, 0); const firstIdx = misses[0].chunkIdx; const lastIdx = misses[misses.length - 1].chunkIdx; @@ -1105,10 +1182,7 @@ export async function runChunkedParseAndResolve( missResults: ParseWorkerResult[][]; }): Promise => { const missResults = round.missResults; - const missCount = round.entries.reduce( - (sum, entry) => sum + (entry.kind === 'miss' ? 1 : 0), - 0, - ); + const missCount = round.entries.filter((entry) => entry.kind === 'miss').length; // `dispatchGroups` returns one array per input group. If that contract // ever breaks, every later entry in this round would silently merge the // wrong chunk's results and skip its cache write, with a clean exit. @@ -1151,8 +1225,7 @@ export async function runChunkedParseAndResolve( const closeRound = async (): Promise => { const started = await startRound(roundEntries); roundEntries = []; - roundMissBytes = 0; - roundBufferedBytes = 0; + roundBudget.reset(); const previous = pendingRound; pendingRound = null; if (previous) { @@ -1275,6 +1348,8 @@ export async function runChunkedParseAndResolve( durableExpectedPaths !== undefined && (await durableChunkHasShards(parsedFileStorePath, chunkHash, durableExpectedPaths)); + // Set by whichever branch queues this chunk; drives the close below. + let roundIsFull = false; if (cachedRaw && cachedRaw.length > 0 && (durableHit || parsedFileStorePath === undefined)) { // Cache hit: replay cached worker output. Finalize any parked worker // chunk FIRST so deferred aggregation stays in chunk order, then merge @@ -1312,7 +1387,7 @@ export async function runChunkedParseAndResolve( chunkStartMs, cachedRaw, }); - for (const file of chunkFiles) roundBufferedBytes += file.content.length; + roundIsFull = roundBudget.addChunk(chunkFiles.map((file) => file.content)); queuedFilesSoFar += chunkFiles.length; } else { // Cache miss: queue for the round's single dispatch; the raw results @@ -1320,19 +1395,14 @@ export async function runChunkedParseAndResolve( chunkCacheMisses++; reparsedFileCount += chunkFiles.length; roundEntries.push({ kind: 'miss', chunkIdx, chunkHash, chunkFiles, chunkStartMs }); - for (const file of chunkFiles) { - roundMissBytes += file.content.length; - roundBufferedBytes += file.content.length; - } + roundIsFull = roundBudget.addChunk(chunkFiles.map((file) => file.content)); queuedFilesSoFar += chunkFiles.length; } - // Close on EITHER cap. `roundMissBytes` sizes the worker round; - // `roundBufferedBytes` bounds what the main thread is holding, which is - // the only cap a warm run can ever reach. - if (roundMissBytes >= roundByteBudget || roundBufferedBytes >= roundByteBudget) { - await closeRound(); - } + // One cap, on what the main thread is holding. That bounds the worker + // round too, since a round's dispatched bytes are a subset of its + // buffered bytes. + if (roundIsFull) await closeRound(); // (Per-chunk aggregation + parse-cache write + throughput log now run in // `applyChunkResults` / `finalizeWorkerChunk` — see the merge-pipelining diff --git a/gitnexus/src/core/ingestion/pipeline-phases/parse-round-budget.ts b/gitnexus/src/core/ingestion/pipeline-phases/parse-round-budget.ts new file mode 100644 index 000000000..9ab85b715 --- /dev/null +++ b/gitnexus/src/core/ingestion/pipeline-phases/parse-round-budget.ts @@ -0,0 +1,63 @@ +/** + * The fold that decides when an open dispatch round closes. + * + * Extracted so the decision is a shared, inspectable unit rather than four + * loose statements inside `runChunkedParseAndResolve`. The parse loop is + * STREAMING — it reads chunk contents lazily, so it cannot know every chunk's + * size up front and cannot "plan" rounds ahead. That makes an accumulator, not + * a planner, the honest shape: feed it each chunk as it is queued and it tells + * you whether the round is now full. + * + * Being a real unit is what makes round cadence observable. Round boundaries + * are otherwise invisible from outside the parse phase: they change no graph + * output (that is the point of batching) and surface only in a log line, which + * is why `bench/parse-dispatch-rounds` measures this directly rather than + * inferring cadence from a full analyze. + */ + +/** Bytes a file contributes to the open round's retained total. */ +export const roundFileBytes = (content: string): number => Buffer.byteLength(content, 'utf8'); + +export interface RoundBudget { + /** + * Add one queued chunk's files. Returns true when the round is now full and + * the caller should close it. Closing resets the accumulator. + */ + addChunk(contents: readonly string[]): boolean; + /** Bytes currently held by the open round. */ + readonly bufferedBytes: number; + /** Reset without closing — used when the caller closes for another reason. */ + reset(): void; +} + +/** + * `budgetBytes` bounds what the main thread HOLDS, counting cache hits as well + * as misses. Counting only cache-missing bytes would bound just the work sent + * to workers, so a warm run — where nothing misses — would never reach the + * close condition and would buffer every chunk's cached output until the tail + * drain. That is the #2649 heap failure on a large repo. + * + * Measured in UTF-8 bytes, matching `estimateItemBytes` in the worker pool. + * `String.length` would return UTF-16 code units, undercounting non-ASCII + * source by up to 3x and letting a CJK-heavy repo hold well past its nominal + * budget before draining. + */ +export const createRoundBudget = (budgetBytes: number): RoundBudget => { + let bufferedBytes = 0; + return { + addChunk(contents) { + for (const content of contents) bufferedBytes += roundFileBytes(content); + if (bufferedBytes >= budgetBytes) { + bufferedBytes = 0; + return true; + } + return false; + }, + get bufferedBytes() { + return bufferedBytes; + }, + reset() { + bufferedBytes = 0; + }, + }; +}; diff --git a/gitnexus/src/core/ingestion/workers/worker-pool.ts b/gitnexus/src/core/ingestion/workers/worker-pool.ts index e5186c6e1..b972d7326 100644 --- a/gitnexus/src/core/ingestion/workers/worker-pool.ts +++ b/gitnexus/src/core/ingestion/workers/worker-pool.ts @@ -125,6 +125,12 @@ export interface WorkerPool { * * Returns one result array per input group, in input order. A group whose * items were all quarantined yields an empty array. + * + * Required, not optional. `getQuarantinedPaths?` and `getStats?` below are + * marked optional as a compatibility accommodation for `WorkerPool` shapes + * that predate them — not as a convention for new members. Making this one + * optional would force a `?.` plus a fallback branch at its only production + * call site, and that branch could never run. */ dispatchGroups( groups: readonly DispatchGroup[], @@ -469,7 +475,9 @@ const DEFAULT_WORKER_READY_TIMEOUT_MS = 5_000; * extraction / structured-clone overhead, and the marginal worker adds * memory pressure (tree-sitter state + sub-batch buffer) without much * throughput gain. Operators on bigger machines override via - * `GITNEXUS_WORKER_POOL_SIZE` or `--workers `. + * `GITNEXUS_WORKER_POOL_SIZE` or `--workers `; both are deliberate + * operator input and bypass the work-proportional sizing in `parse-impl`, + * which only bounds the AUTO default. */ const DEFAULT_POOL_SIZE_CAP = 16; @@ -647,7 +655,7 @@ export function resolveWorkerPoolOptions( * GITNEXUS_WORKER_POOL_SIZE=`) is an accident, not a request for zero workers; * only a literal `0` disables the pool. */ -function envWorkerPoolSize(): number | undefined { +export function envWorkerPoolSize(): number | undefined { const raw = process.env.GITNEXUS_WORKER_POOL_SIZE; if (raw === undefined || raw.trim() === '') return undefined; return nonNegativeInteger(raw); @@ -691,9 +699,19 @@ export function resolveAutoPoolSize(): number { // pool cap exists to prevent. Falls back to os.cpus().length on // older Node versions. Mirrors `capabilities.ts:85` // (`defaultEmbeddingThreads`). - const cores = - typeof os.availableParallelism === 'function' ? os.availableParallelism() : os.cpus().length; - return Math.min(DEFAULT_POOL_SIZE_CAP, Math.max(1, cores - 1)); + return Math.min(DEFAULT_POOL_SIZE_CAP, Math.max(1, resolveHostParallelism() - 1)); +} + +/** + * Usable parallelism for this process. Prefers `os.availableParallelism` so + * cgroup CPU limits are honored, falling back to `os.cpus().length` on older + * Node. Exported so callers that size work against the host (rather than + * against the pool default) do not re-derive the fallback. + */ +export function resolveHostParallelism(): number { + return typeof os.availableParallelism === 'function' + ? os.availableParallelism() + : os.cpus().length; } /** @@ -1383,15 +1401,20 @@ export const createWorkerPool = ( // Layer 3: filter out quarantined paths so a known-bad file never reaches // a worker again this pool lifetime. The caller queries // `getQuarantinedPaths` after dispatch to route filtered items. - const dispatchableGroups = groups.map((group) => { - const items: TInput[] = []; - for (const item of group.items) { - const path = itemPath(item); - if (path !== undefined && quarantine.has(path)) continue; - items.push(item); - } - return { items, chunkHash: group.chunkHash }; - }); + // Quarantine is empty on every run that has not had a worker die, so the + // filter below would be an identity copy of every group's items. Skip it. + const dispatchableGroups = + quarantine.size === 0 + ? groups + : groups.map((group) => { + const items: TInput[] = []; + for (const item of group.items) { + const path = itemPath(item); + if (path !== undefined && quarantine.has(path)) continue; + items.push(item); + } + return { items, chunkHash: group.chunkHash }; + }); const dispatchableCount = dispatchableGroups.reduce( (sum, group) => sum + group.items.length, 0, diff --git a/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts b/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts index 410d0c048..c01baaca2 100644 --- a/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts +++ b/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts @@ -383,6 +383,24 @@ describe('parse-impl warm-cache ParsedFile coverage (#2038)', () => { } }); + it('does not cache a chunk whose durable generation could not be reset', async () => { + // The reset failed, so the previous generation's shards are still on disk. + // Caching this chunk would let a future warm hit union those stale shards + // with the new ones. Same posture as a quarantined chunk: leave it uncached + // so the next run re-dispatches into a directory it can actually clear. + const f = writeFile('src/stale-generation.ts', 'export function stale() { return 1; }\n'); + const cache = newCache(); + prepareOverride.impl = () => Promise.reject(new Error('EACCES: simulated cache failure')); + try { + await expect(run(cache, [f])).resolves.toBeDefined(); + } finally { + prepareOverride.impl = undefined; + } + + // Nothing was written under any key -- neither on disk nor in memory. + expect(cache.onDiskKeys.size + cache.entries.size).toBe(0); + }); + it('retains worker ParsedFiles when the main-thread run-store write fails', async () => { const f = writeFile( 'src/persist-fallback.ts', diff --git a/gitnexus/test/unit/parse-impl-worker-lazy-cache.test.ts b/gitnexus/test/unit/parse-impl-worker-lazy-cache.test.ts index d35629a75..216170a2b 100644 --- a/gitnexus/test/unit/parse-impl-worker-lazy-cache.test.ts +++ b/gitnexus/test/unit/parse-impl-worker-lazy-cache.test.ts @@ -103,6 +103,50 @@ parentPort.on('message', (msg) => { ); }; +/** + * Like `writeResultWorker`, but each spawned instance writes its OWN marker + * keyed by `threadId`. The shared single-marker workers above can only prove + * "at least one worker started"; counting files in `markerDir` gives the actual + * pool size the parse phase asked `createWorkerPool` for, which is the only + * thing that distinguishes a clamped pool from an honored override. + */ +const writeSpawnCountingWorker = (workerPath: string, markerDir: string): void => { + fs.writeFileSync( + workerPath, + ` +const fs = require('node:fs'); +const path = require('node:path'); +const { parentPort, threadId } = require('node:worker_threads'); +fs.mkdirSync(${JSON.stringify(markerDir)}, { recursive: true }); +fs.writeFileSync(path.join(${JSON.stringify(markerDir)}, 'worker-' + threadId), 'spawned'); +parentPort.postMessage({ type: 'ready' }); +const accumulated = { + nodes: [], relationships: [], symbols: [], imports: [], calls: [], assignments: [], heritage: [], + routes: [], fetchCalls: [], fetchWrapperDefs: [], decoratorRoutes: [], routerIncludes: [], routerImports: [], toolDefs: [], ormQueries: [], constructorBindings: [], + fileScopeBindings: [], parsedFiles: [], skippedLanguages: {}, fileCount: 0, +}; +parentPort.on('message', (msg) => { + if (msg && msg.type === 'sub-batch') { + for (const file of msg.files) { + const filePath = file.path; + const name = filePath.split('/').pop().replace(/\.ts$/, ''); + accumulated.nodes.push({ + id: 'Function:' + filePath + ':' + name, + label: 'Function', + properties: { name, filePath, startLine: 1, endLine: 1, language: 'typescript' }, + }); + accumulated.fileCount++; + } + parentPort.postMessage({ type: 'progress', filesProcessed: accumulated.fileCount }); + parentPort.postMessage({ type: 'sub-batch-done' }); + return; + } + if (msg && msg.type === 'flush') parentPort.postMessage({ type: 'result', data: accumulated }); +}); +`, + ); +}; + const writeExitBeforeReadyWorker = (workerPath: string): void => { fs.writeFileSync(workerPath, `process.exit(1);\n`); }; @@ -240,6 +284,56 @@ describe('parse-impl worker pool lazy startup', () => { expect(Array.from(graph.nodes.values()).some((n) => n.properties.name === 'fatal')).toBe(false); }); + it('honors GITNEXUS_WORKER_POOL_SIZE above the work-proportional cap', async () => { + // The auto pool size is bounded by source bytes so a tiny repo does not + // spawn a full idle pool. That bound must apply to the AUTO default only — + // it used to clamp the operator's env override too, so an operator asking + // for more workers silently got the byte-derived number while `--workers` + // was honored. + // + // This asserts the pool the PARSE PHASE actually builds, not the resolver in + // isolation: `resolveAutoPoolSize()` already honored the env var before the + // fix, so a test at that level stays green through a revert. + const saved = process.env.GITNEXUS_WORKER_POOL_SIZE; + process.env.GITNEXUS_WORKER_POOL_SIZE = '3'; + try { + // Four tiny files: total bytes are far under one CHUNK_BYTES_PER_WORKER so + // the work-proportional cap is 1, while the parseable count stays above the + // requested 3 (the pool never exceeds the number of files to parse). + const rels = ['src/a.ts', 'src/b.ts', 'src/c.ts', 'src/d.ts']; + const scanned = rels.map((rel) => { + const full = path.join(repoDir, rel); + fs.mkdirSync(path.dirname(full), { recursive: true }); + fs.writeFileSync(full, `export function ${path.basename(rel, '.ts')}() { return 1; }\n`); + return { path: rel, size: fs.statSync(full).size }; + }); + + const markerDir = path.join(tempDir, 'pool-size-markers'); + const workerPath = path.join(tempDir, 'pool-size-worker.js'); + writeSpawnCountingWorker(workerPath, markerDir); + + const result = await runChunkedParseAndResolve( + createKnowledgeGraph(), + scanned, + rels, + rels.length, + repoDir, + Date.now(), + () => {}, + // No `workerPoolSize`: the env var is the only override in play. + { workerUrlForTest: pathToFileURL(workerPath) }, + ); + + expect(result.usedWorkerPool).toBe(true); + // 3, not the byte-derived 1. Exactly this assertion fails on the clamped + // parent commit, which is what makes it a regression test for the fix. + expect(fs.readdirSync(markerDir)).toHaveLength(3); + } finally { + if (saved === undefined) delete process.env.GITNEXUS_WORKER_POOL_SIZE; + else process.env.GITNEXUS_WORKER_POOL_SIZE = saved; + } + }); + it('throws when GITNEXUS_WORKER_POOL_SIZE=0 and no --workers flag (sequential parsing removed)', async () => { const saved = process.env.GITNEXUS_WORKER_POOL_SIZE; process.env.GITNEXUS_WORKER_POOL_SIZE = '0';