GitNexus/gitnexus/test/integration/parse-impl-env-reads.test.ts
Gergő Magyar 8f006bd759
perf(parse): batch cache packs into one dispatch round (#3196)
* perf(parse): batch cache packs into one dispatch round

`WorkerPool.dispatch` is a barrier, so dispatching one parse-cache pack at a
time leaves most slots idle for every round-trip. Packs are keyed by
`(language, hash(path) % 128)`, so the byte budget rarely binds: this repo
produces 1285 packs where the budget alone needs 16, and 549 of those hold a
single file. In a real analyze, 76 of 221 dispatched chunks carried one file
and cost 15.3s — 20% of the parse phase for 3.4% of the files.

Chunks now accumulate into a round bounded by `GITNEXUS_PARSE_ROUND_BYTES` of
cache-missing source (default: the chunk byte budget) and go out through a new
`WorkerPool.dispatchGroups`. Jobs are still cut at pack boundaries, so each job
carries exactly one `chunkHash` and every result stays attributable to the pack
whose cache key owns it. Cache hits ride along as round entries, and rounds
drain in `chunkIdx` order, so deferred aggregation stays deterministic.

Cold `analyze --index-only` on this repo (2234 parseable files, 16 workers):
110.3s -> 70.5s total, parse phase 74.0s -> 40.5s, 221 dispatches -> 15.
Graph output is unchanged: 51,286 nodes / 163,092 edges / 2106 clusters /
759 flows in both arms. Peak main-thread RSS 3372MB -> 3487MB (+3.4%).

`dispatchGroups` also claims the pool synchronously and rejects a concurrent
call. Two overlapping dispatches hand the same slots out twice and both stall;
the first version of this change did exactly that, and the only symptom was
every worker idle-timing out ~10s later with no indication of the cause.

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

* fix(parse): bound what an open round holds, not just what it dispatches

Follow-up to the review of #3196. Three reviewers independently found the
same defect: `roundMissBytes` was the only in-loop close condition, but cache
HITS were queued into the same round without contributing to it. A warm
re-analyze misses nothing, so no round ever closed and every chunk's source
plus its cached worker output stayed resident until the tail drain — the
#2649 heap failure shape on a large repo.

- Hit entries now carry a file COUNT, not the file array, so a replayed chunk
  never pins its source text. `applyChunkResults` only ever read `.length`.
- Track `roundBufferedBytes` across hits and misses and close on either cap.
  Verified on a warm run: with the cap, draining starts as soon as 2MB is
  buffered; without it all 221 merges land in the final 10% of the phase.
- Warm progress no longer freezes at the phase floor. `filesParsedSoFar` only
  advances at drain, so a new `queuedFilesSoFar` feeds the progress events
  while `filesParsedSoFar` stays the merge-accurate throughput number.
- A throw from `drainRound` used to unwind straight to `terminate()` while the
  next round's workers were still busy — the #2432 mid-N-API abort hazard.
  Settle the in-flight round first, then propagate.
- `dispatchGroups` returns one array per group; assert that length instead of
  `?? []`, which turned a contract break into a silently empty chunk.
- Collapse `PendingWorkerChunk` into the `miss` RoundEntry it duplicated.
- Repair two stale doc comments: `dispatch`'s JSDoc had been orphaned onto
  `dispatchGroups`, and `dispatchChunkParse` still described chunk overlap
  that now lives in parse-impl's round machinery.
- New test: a round mixing a cache hit and a cache miss. `drainRound` walks
  entries in chunkIdx order but pulls results on a separate cursor, and no
  existing test put both kinds in one round with content assertions.

Cold analyze unchanged: 71.3s, 15 rounds, 51,286 nodes / 163,092 edges.

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>
2026-09-06 18:27:09 +01:00

183 lines
7 KiB
TypeScript

/**
* U14 (F7 architectural from PR #1693 review) — Function-scope env reads
* in parse-impl.
*
* Pre-U14, `CHUNK_BYTE_BUDGET` was a module-load IIFE constant that
* captured `GITNEXUS_CHUNK_BYTE_BUDGET` once and froze the value for
* the module's lifetime. That defeated `PipelineOptions.chunkByteBudget`
* (silently no-op'd because the body read the frozen constant) AND
* forced tests to use `vi.resetModules` to vary the chunk layout (see
* the U7 deferred-extraction test and the U6 multi-chunk integration
* test for examples of the workaround).
*
* After U14:
* - Option present -> option wins (per-call, no env / no vi.resetModules)
* - Option absent -> env wins (back-compat)
* - Both absent -> built-in 2 MB default
*
* This file pins all three resolution branches, plus the behavioral
* invariant the workaround was masking: two back-to-back runs in the
* same vitest worker process can use DIFFERENT `chunkByteBudget` values
* and observe DIFFERENT chunking on the same fixture WITHOUT needing
* `vi.resetModules` between them.
*/
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { runChunkedParseAndResolve } from '../../src/core/ingestion/pipeline-phases/parse-impl.js';
import { createKnowledgeGraph } from '../../src/core/graph/graph.js';
import { PARSE_CACHE_VERSION, parseCacheBucketId } from '../../src/storage/parse-cache.js';
const ORIGINAL_BUDGET = process.env.GITNEXUS_CHUNK_BYTE_BUDGET;
type Fixture = Record<string, string>;
function makeRepo(fixture: Fixture): string {
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'parse-impl-env-reads-'));
for (const [name, content] of Object.entries(fixture)) {
fs.writeFileSync(path.join(dir, name), content);
}
return dir;
}
function scanned(repo: string, files: string[]) {
return files.map((rel) => ({
path: rel,
size: fs.statSync(path.join(repo, rel)).size,
}));
}
/**
* Read the chunk count out of the progress stream. parse-impl reports progress
* as "Parsing chunk X/Y" for a single chunk and "Parsing chunks X-Z/Y" when a
* dispatch round batches several — so the DENOMINATOR, not the number of
* distinct messages, is the count of packs the loop produced. Reading `Y`
* keeps this independent of how chunks are grouped into rounds while still
* exercising the real budget-resolution path inside
* `runChunkedParseAndResolve`, rather than re-deriving packs in the test.
*/
async function countChunksFromProgress(
repoPath: string,
files: string[],
options?: { chunkByteBudget?: number },
): Promise<number> {
const scan = scanned(repoPath, files);
const graph = createKnowledgeGraph();
const totals = new Set<number>();
await runChunkedParseAndResolve(
graph,
scan,
files,
files.length,
repoPath,
Date.now(),
(p) => {
if (typeof p.message !== 'string') return;
const m = /Parsing chunks? \d+(?:-\d+)?\/(\d+)/.exec(p.message);
if (m !== null) totals.add(Number(m[1]));
},
// Chunk count is byte-budget-driven and emitted before the pool runs, so it
// is independent of worker vs sequential. Sequential parsing was removed, so
// this runs through the (auto-sized, dist-backed) worker pool — hence the
// integration tier.
{ ...options },
);
// Every message in a run carries the same denominator; more than one value
// would mean the loop changed its chunk count mid-run.
expect(totals.size).toBeLessThanOrEqual(1);
return totals.values().next().value ?? 0;
}
describe('parse-impl chunkByteBudget resolution (U14 / F7)', () => {
let repoPath = '';
beforeEach(() => {
repoPath = makeRepo({
'a.ts': 'export const A = 1;\n',
'b.ts': 'export const B = 2;\n',
'c.ts': 'export const C = 3;\n',
});
});
afterEach(() => {
if (repoPath && fs.existsSync(repoPath)) {
fs.rmSync(repoPath, { recursive: true, force: true });
}
if (ORIGINAL_BUDGET === undefined) {
delete process.env.GITNEXUS_CHUNK_BYTE_BUDGET;
} else {
process.env.GITNEXUS_CHUNK_BYTE_BUDGET = ORIGINAL_BUDGET;
}
});
it('option-first: PipelineOptions.chunkByteBudget overrides the env var', async () => {
// Force the env to a HUGE value that would normally collapse the
// fixture to a single chunk; pass a SMALL option that produces 3
// chunks. If the option wins, we observe 3 chunks; if env wins, 1.
process.env.GITNEXUS_CHUNK_BYTE_BUDGET = String(10 * 1024 * 1024);
const chunks = await countChunksFromProgress(repoPath, ['a.ts', 'b.ts', 'c.ts'], {
chunkByteBudget: 8,
});
expect(chunks).toBe(3);
});
it('env-fallback: GITNEXUS_CHUNK_BYTE_BUDGET is honored when the option is absent', async () => {
process.env.GITNEXUS_CHUNK_BYTE_BUDGET = '8';
const chunks = await countChunksFromProgress(repoPath, ['a.ts', 'b.ts', 'c.ts']);
expect(chunks).toBe(3);
});
it('default-fallback: 2 MB budget packs by bucket, not one sequential mega-chunk', async () => {
delete process.env.GITNEXUS_CHUNK_BYTE_BUDGET;
const files = ['a.ts', 'b.ts', 'c.ts'];
const expectedBuckets = new Set(files.map((f) => `typescript\0${parseCacheBucketId(f)}`));
const chunks = await countChunksFromProgress(repoPath, files);
expect(chunks).toBe(expectedBuckets.size);
});
it('per-call: two back-to-back runs with different option values observe their own values, not the previous call', async () => {
// The behavioral invariant U14 restores: a long-running host
// (eval-server, MCP daemon) calling runChunkedParseAndResolve twice
// with different chunkByteBudget values gets the value it passed,
// not whatever the first call set (pre-U14, the module-load IIFE
// froze the value at import — the option was a silent no-op).
const files = ['a.ts', 'b.ts', 'c.ts'];
delete process.env.GITNEXUS_CHUNK_BYTE_BUDGET;
const small = await countChunksFromProgress(repoPath, files, {
chunkByteBudget: 8,
});
const large = await countChunksFromProgress(repoPath, files, {
chunkByteBudget: 10 * 1024 * 1024,
});
expect(small).toBe(3);
const expectedBuckets = new Set(files.map((f) => `typescript\0${parseCacheBucketId(f)}`));
expect(large).toBe(expectedBuckets.size);
});
it('workerPoolSize 1 vs 2 produce the same cache keys when budget is unset (#3088)', async () => {
delete process.env.GITNEXUS_CHUNK_BYTE_BUDGET;
const files = ['a.ts', 'b.ts', 'c.ts'];
const keysForPool = async (workerPoolSize: number): Promise<string[]> => {
const parseCache = {
version: PARSE_CACHE_VERSION,
entries: new Map(),
usedKeys: new Set<string>(),
};
const graph = createKnowledgeGraph();
await runChunkedParseAndResolve(
graph,
scanned(repoPath, files),
files,
files.length,
repoPath,
Date.now(),
() => {},
{ workerPoolSize, parseCache },
);
return [...parseCache.usedKeys].sort();
};
expect(await keysForPool(1)).toEqual(await keysForPool(2));
});
});