scriptoriumadmin/GitNexus
1
0
Fork 0
mirror of https://github.com/abhigyanpatwari/GitNexus.git synced 2026-09-10 22:43:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

perf(parse): tighten the dispatch-round memory bound and unclamp the worker-pool override (#3200)

Browse source 
* docs(parse): record why dispatchGroups is a required interface member

Review finding #10 argued dispatchGroups should be optional to match
`getQuarantinedPaths?` / `getStats?`. Those are compatibility accommodation
for WorkerPool shapes that predate them, not a convention for new members;
optional here would force a `?.` plus an unreachable fallback at the single
production call site. Documenting the decision so the next reader does not
re-litigate it from the neighbouring optional markers.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
(cherry picked from commit addaab647377f3c4553f752fa3ca1388bcb9ca81)

* refactor(parse): simplify round accounting and dispatch setup

Simplification pass over the dispatch-rounds change. Behavior preserved:
identical graph on a full analyze (51,286 nodes / 163,092 edges).

- Drop `roundMissBytes`. `roundBufferedBytes` counts the same bytes plus the
  cache hits, so it is always the greater of the two and the first disjunct of
  the close condition could never fire on its own. One counter, one reset, one
  check.
- Measure round bytes with `Buffer.byteLength(content, 'utf8')` instead of
  `String.length`. UTF-16 code units undercount non-ASCII source by up to 3x,
  so the cap meant to bound main-thread retention was letting a CJK-heavy repo
  hold well past its nominal budget. Matches `estimateItemBytes` in the pool.
- Reset the durable ParsedFile directories for a round's chunks concurrently.
  Each targets its own chunk-hash directory, and running them serially put N
  round trips of fs work on the critical path the round exists to shorten.
  The try/catch stays inside the mapped callback, so one failure still
  degrades that chunk alone.
- Skip the quarantine filter entirely when nothing is quarantined, which is
  every run without a worker death. It was an identity copy of every group.
- `dispatchChunkParseRound` takes `DispatchGroup<...>` rather than re-declaring
  that shape inline; the type was already imported and used in its body.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
(cherry picked from commit 527d5b6e0ca8ae7bbc6a414c5ac7e27fd85e9995)

* refactor(parse): count round misses with the same idiom startRound uses

`drainRound` hand-rolled a reduce to count 'miss' entries while `startRound`,
one function above, filters the same predicate over the same union. Same
integer, one idiom.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
(cherry picked from commit 7eaa193b0cb5fa515844f36ae1401d6fb2fed7b8)

* fix(parse): honor GITNEXUS_WORKER_POOL_SIZE above the auto sizing cap

The auto pool size is bounded by source bytes so a tiny repo does not spawn a
full idle pool. That bound was also clamping the operator's env override,
because the env value is read inside `resolveAutoPoolSize()` and the result
went through `Math.min(..., workProportionalCap)`.

`DEFAULT_POOL_SIZE_CAP`'s own comment offers `GITNEXUS_WORKER_POOL_SIZE` and
`--workers <N>` as equivalent escape hatches for operators on bigger machines.
They were not. Measured on a 30MB corpus, where the byte-derived cap is 16:

  --workers 24                  -> pool: 24/24 active
  GITNEXUS_WORKER_POOL_SIZE=24  -> pool: 16/16 active   (silently ignored)

Both are deliberate operator input, so both now bypass the work-proportional
cap, which goes back to bounding only the auto default. After the fix, on the
same corpus, with identical graph output (51,286 nodes / 163,092 edges):

  GITNEXUS_WORKER_POOL_SIZE=24  -> pool: 24/24 active
  GITNEXUS_WORKER_POOL_SIZE=4   -> pool: 4/4 active
  unset                         -> pool: 16/16 active

Verified by hand against the pool's own throughput log; not covered by an
automated regression test, since the pool size is only observable through
that log line and not through the progress stream a test can read.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
(cherry picked from commit 17ed08608c878079b2927da25cfd39c1608a02a2)

* fix(parse): bound the durable-reset fan-out and pin the pool-size override

Review follow-ups on #3200.

The round's durable ParsedFile directory resets went out as one unbounded
`Promise.all` — one recursive rm + mkdir per miss chunk, all at once. A round
can hold hundreds of small packs, and those resets compete for descriptors with
the chunk prefetch this loop already has in flight. `readFileContents` degrades
a losing read SILENTLY by documented contract, so a dropped file would vanish
from the chunk, from the graph, and from the chunk hash — shipping a narrowed
index with exit 0. Now routed through `mapConcurrent` at the same width the file
reads use, which keeps the pipelining win and caps in-flight descriptors.

An operator's pool size is now also bounded by the number of files there are to
parse, so `GITNEXUS_WORKER_POOL_SIZE=100000` on a five-file repo cannot become
the literal thread count. This applies to `--workers` and the env var alike, so
the parity the previous commit established is intact. It does NOT shrink an
incremental re-analyze: `totalParseable` counts every parseable file in the
scan, not the changed ones.

Adds the regression test a reviewer asked for. The existing coverage
(`worker-pool-resilience` calling `resolveAutoPoolSize` directly,
`analyze-worker-pool-size` mocking `runFullAnalysis`) never reaches
`runChunkedParseAndResolve`'s `effectivePoolSize`, so both stayed green through
a revert of the fix. The new test drives the real parse phase with a worker
double that writes a per-`threadId` marker, and counts them: verified it fails
on the reverted line with `expected [ 'worker-1' ] to have a length of 3 but
got 1`, and passes on HEAD.

Also corrects the `GITNEXUS_PARSE_ROUND_BYTES` docstring, which still described
the cache-miss counter deleted two commits ago.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(parse): skip caching a chunk with a stale durable generation; warn on over-subscription

Closes the two findings left open by the review of #3200.

When `prepareDurableParsedFileChunk` fails, the previous generation's shards
are still on disk, so a later warm hit would union them with the new ones. The
chunk is now recorded and its parse-cache write skipped -- the same posture
`finalizeWorkerChunk` already takes for a quarantined chunk, and for the same
reason: do not cache what we cannot vouch for. The next run re-dispatches into
a directory it can actually clear. Bounding the reset fan-out removed the
correlated trigger; this closes the individual case.

Pool size over-subscription now warns rather than caps. Silently capping is
precisely what the override exists to prevent, so an operator's number is still
honored -- 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. The warning names
the host's usable core count, so it is a hardware fact rather than an invented
threshold. `resolveHostParallelism` is extracted from `resolveAutoPoolSize`
rather than re-deriving the cgroup-aware fallback at the new call site.

Tests: the stale-generation skip is pinned by a new case asserting nothing is
written under any key; verified it fails without the guard with
`expected 1 to be +0`. 60 unit and 49 integration tests pass across the
affected suites.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* test(parse): guard dispatch-round cadence with a bench, not a wall-clock budget

Round boundaries are deliberately invisible to graph output — batching that
changed output would be a bug — so nothing in the repo could see the #3196 win
regress. It would have come back as a silent ~1.5x on every cold analyze. Two
earlier attempts to pin it as a unit test failed for that exact reason: one
scraped a logger line the progress stream does not carry, the other asserted
graph content that is identical either way.

Extracts the round-close fold into `createRoundBudget`, so the decision is a
shared unit the bench measures rather than a copy that drifts. The parse loop
is streaming and cannot know chunk sizes up front, so an accumulator is the
honest shape — not a planner.

Four deterministic arms, one ratio, no millisecond gate:
- layout_fingerprint — pack membership. Every cache key derives from it, so
  drift needs a SCHEMA_BUMP, never a lone re-baseline.
- packs / single_file_packs — the FLOOR. `rounds` only asserts something while
  the corpus over-splits (774 packs where the byte budget needs 5). This is
  bench/import-target's lesson, where four heap arms read 0 B and passed every
  ceiling: a ceiling says "not too big", nothing said "still measuring".
- rounds — the regression signal, both directions.
- cjk_rounds vs ascii_rounds — pins UTF-8 byte accounting. The two corpora
  share a UTF-16 length and differ only in encoded size, so String.length
  collapses them to equal. This is the arm no unit test could be.
- pack_scaling_ratio — (t_4n/t_n)/4, min-of-15. A ratio because wall-clock is
  runner-speed-dependent and this repo has the scar: callable-value-flow's ms
  gate failed twice at 2.07 and 1.975 against 1.9 with correct code, on a
  sub-11ms measurement.

Every arm verified to fail before being recorded: close-every-chunk reads 774
rounds, disabling the close reads 1, reverting roundFileBytes to String.length
takes cjk_rounds 8 -> 3, and shrinking the corpus trips the shape floor.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs(bench): record the analyze phase breakdown and the rejected optimizations

Where analyze time actually goes, measured while landing #3194/#3196/#3200,
plus the two optimizations that looked compelling and were measured away.

The headline is that the parse work is done: a one-file-edit re-analyze is
36.5s, of which parse is 2.8s (8%). scopeResolution is 40% and the unlogged
graph emit + FTS rebuild is 49% — neither is incremental, and the ~18s sits
outside the phase runner so every phase log is blind to it.

Also records the trap that invalidated an earlier measurement: a non-git
corpus never records a schema fingerprint, so every run is a forced rebuild
and any "warm" number taken that way is fiction.

Rejected, with numbers: more workers (16/20/24 land inside run-to-run spread)
and bundling the worker entry (~250ms on a normal filesystem; the 8.6s that
motivated it was a 9p-mount artifact).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Gergő Magyar 2026-09-07 13:53:31 +01:00 • committed by GitHub
parent 1c1cbf111e
commit f48bf81256
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 765 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
.github/workflows/ci-tests.yml vendored
View file

@ -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

126
gitnexus/bench/analyze-phase-breakdown.md Normal file
View file

@ -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: <name> (<ms>)` 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.

31
gitnexus/bench/parse-dispatch-rounds/baselines.json Normal file
View file

@ -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."
}

279
gitnexus/bench/parse-dispatch-rounds/measure.mjs Normal file
View file

@ -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.');
}

5
gitnexus/src/core/ingestion/parsing-processor.ts
View file

@ -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<DispatchGroup<{ path: string; content: string }>>,
workerPool: WorkerPool,
onFileProgress?: FileProgressCallback,
): Promise<ParseWorkerResult[][]> => {

138
gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts
View file

@ -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 <N>` 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<string>();
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<void> => {
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<void> => {
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

63
gitnexus/src/core/ingestion/pipeline-phases/parse-round-budget.ts Normal file
View file

@ -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;
},
};
};

51
gitnexus/src/core/ingestion/workers/worker-pool.ts
View file

@ -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<TInput, TResult>(
groups: readonly DispatchGroup<TInput>[],
@ -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 <N>`.
* `GITNEXUS_WORKER_POOL_SIZE` or `--workers <N>`; 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,

18
gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts
View file

@ -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',

94
gitnexus/test/unit/parse-impl-worker-lazy-cache.test.ts
View file

@ -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';