diff --git a/__tests__/wal-deferral.test.ts b/__tests__/wal-deferral.test.ts index 13b7555..5ac3340 100644 --- a/__tests__/wal-deferral.test.ts +++ b/__tests__/wal-deferral.test.ts @@ -294,3 +294,69 @@ describe('sync WAL deferral end-to-end (#1248)', () => { } }); }); + +describe('resolution-phase WAL backpressure plumbing (§7a.1)', () => { + // The valve's timer-driven passive checkpoints stay perpetually partial + // against the resolver pool's continuous reads, so during resolution the + // writer-side backpressure() hook is the ONLY mechanism that can complete + // a backfill and let the WAL wrap — a kernel-scale run without it grew a + // 22GB WAL on a 4.6GB DB. These pin that the batch loop (a) calls the hook + // at the pool-idle boundary and (b) actually parks on a returned promise. + + async function seedPendingRefs(cg: CodeGraph): Promise { + const raw = (cg as unknown as { db: DatabaseConnection }).db.getDb(); + const node = raw.prepare("SELECT id, file_path FROM nodes WHERE kind = 'function' LIMIT 1").get() as + | { id: string; file_path: string } + | undefined; + expect(node).toBeDefined(); + const ins = raw.prepare( + "INSERT INTO unresolved_refs (from_node_id, reference_name, reference_kind, line, col, file_path, language, status) VALUES (?, ?, 'calls', 1, 0, ?, 'typescript', 'pending')" + ); + ins.run(node!.id, 'helper0', node!.file_path); + ins.run(node!.id, 'helper1', node!.file_path); + } + + it('calls the backpressure hook once per settled batch', async () => { + writeFixtureProject(); + const cg = CodeGraph.initSync(tmpDir); + await cg.indexAll(); + await seedPendingRefs(cg); + + let calls = 0; + const result = await cg.resolveReferencesBatched(undefined, undefined, () => { + calls++; + return null; // under the hard cap — loop must proceed without waiting + }); + expect(result.stats.total).toBeGreaterThan(0); + expect(calls).toBeGreaterThanOrEqual(1); + await cg.close(); + }); + + it('parks the batch loop on a backpressure promise until it resolves', async () => { + writeFixtureProject(); + const cg = CodeGraph.initSync(tmpDir); + await cg.indexAll(); + await seedPendingRefs(cg); + + let release!: () => void; + const gate = new Promise((r) => { release = r; }); + let hookHit = false; + const done = cg + .resolveReferencesBatched(undefined, undefined, () => { + if (hookHit) return null; // park only on the first boundary + hookHit = true; + return gate; + }) + .then(() => true); + + // Give the loop ample turns: it must reach the hook and then be parked. + for (let i = 0; i < 50; i++) await new Promise((r) => setImmediate(r)); + expect(hookHit).toBe(true); + const settledEarly = await Promise.race([done, Promise.resolve(false)]); + expect(settledEarly).toBe(false); // still parked on the gate + + release(); + expect(await done).toBe(true); + await cg.close(); + }); +}); diff --git a/src/index.ts b/src/index.ts index ac36c61..70f2fe2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -591,7 +591,8 @@ export class CodeGraph { current: done, total: totalPasses, }); - } + }, + walValve ? () => walValve!.backpressure() : undefined ); if (process.env.CODEGRAPH_SYNTH_TIMINGS) console.error(`[phase-timing] resolution: ${Date.now() - tResolve}ms`); @@ -1145,7 +1146,13 @@ export class CodeGraph { */ async resolveReferencesBatched( onProgress?: (current: number, total: number) => void, - onSynthesisProgress?: (done: number, total: number) => void + onSynthesisProgress?: (done: number, total: number) => void, + // The WAL valve's writer-side backstop, threaded into the batch loop's + // pool-idle boundaries. Without it the valve's only lever during + // resolution is timer-driven passive checkpoints, which the pool's + // continuous reads keep perpetually partial — the WAL then accretes the + // whole phase's write volume (22GB on a 4.6GB DB at kernel scale). + backpressure?: () => Promise | null ): Promise { return this.resolver.resolveAndPersistBatched(onProgress, undefined, onSynthesisProgress, { dbPath: this.db.getPath(), @@ -1158,6 +1165,7 @@ export class CodeGraph { begin: () => this.db.beginBulkEdgeLoad(), end: () => this.db.endBulkEdgeLoad(), }, + backpressure, }); } diff --git a/src/resolution/callback-synthesizer.ts b/src/resolution/callback-synthesizer.ts index 28d4080..143681c 100644 --- a/src/resolution/callback-synthesizer.ts +++ b/src/resolution/callback-synthesizer.ts @@ -3553,7 +3553,11 @@ export async function synthesizeCallbackEdges( // A live resolver pool to fan the independent passes across (structural type // so this file never imports the pool — resolver-worker imports THIS file). // Null/omitted → the sequential path, byte-identical to the pool path. - pool?: { runSynthPass(name: string): Promise<{ edges: Edge[]; ms: number }> } | null + pool?: { runSynthPass(name: string): Promise<{ edges: Edge[]; ms: number }> } | null, + // WAL-valve writer backstop (WalCheckpointValve.backpressure), called at + // pool-idle points in the edge-insert loops below — the passes themselves + // only read; every write in this function happens with the pool idle. + backpressure?: () => Promise | null ): Promise { // Each sub-pass below is a whole-graph scan, and there are ~30 of them, all // running synchronously on the indexer's main thread. Their AGGREGATE can run @@ -3618,10 +3622,18 @@ export async function synthesizeCallbackEdges( // otherwise orphaned from the struct, and goImplementsEdges (next) derives a // struct's method set from its `contains` edges — so without this it would // under-count the interfaces a cross-file struct satisfies. (#583) + // Writer-side WAL backstop for the insert loops here (see the param doc): + // one fstat when under the valve's hard cap, a parked full backfill past it. + const foldIfOver = async (): Promise => { + const bp = backpressure?.(); + if (bp) await bp; + }; + const goMethodContains = has('go') ? await goCrossFileMethodContainsEdges(queries, yieldToLoop) : NONE; for (let i = 0; i < goMethodContains.length; i += 2000) { queries.insertEdges(goMethodContains.slice(i, i + 2000)); await yieldToLoop(); + await foldIfOver(); } await yieldToLoop(); __mark('goMethodContains'); @@ -3633,6 +3645,7 @@ export async function synthesizeCallbackEdges( for (let i = 0; i < goImpl.length; i += 2000) { queries.insertEdges(goImpl.slice(i, i + 2000)); await yieldToLoop(); + await foldIfOver(); } await yieldToLoop(); __mark('goImplements'); @@ -3720,6 +3733,7 @@ export async function synthesizeCallbackEdges( for (let i = 0; i < merged.length; i += 2000) { queries.insertEdges(merged.slice(i, i + 2000)); await yieldToLoop(); + await foldIfOver(); } __mark('insertMergedEdges'); return merged.length + goImpl.length + goMethodContains.length; diff --git a/src/resolution/index.ts b/src/resolution/index.ts index a7154ab..8326b8f 100644 --- a/src/resolution/index.ts +++ b/src/resolution/index.ts @@ -1338,8 +1338,17 @@ export class ReferenceResolver { // Sequential fallback on any pool failure. CODEGRAPH_NO_PARALLEL_RESOLVE=1 // disables entirely. bulkEdgeLoad hooks (when provided) bracket the batch // loop with drop/recreate of the non-unique edge indexes on big runs — - // see DatabaseConnection.beginBulkEdgeLoad. - parallel?: { dbPath: string; bulkEdgeLoad?: { begin: () => void; end: () => void | Promise } } + // see DatabaseConnection.beginBulkEdgeLoad. backpressure (when provided) + // is the WAL valve's writer-side backstop (WalCheckpointValve.backpressure): + // called at pool-idle boundaries so a full backfill can actually complete — + // the valve's timer-driven passive passes stay perpetually partial against + // the pool's continuous reads, which is how a kernel-scale resolution grew + // a 22GB WAL on a 4.6GB DB (migration plan §7a.1). + parallel?: { + dbPath: string; + bulkEdgeLoad?: { begin: () => void; end: () => void | Promise }; + backpressure?: () => Promise | null; + } ): Promise { // Resolution runs on the indexer's MAIN thread, and the #850 liveness // watchdog SIGKILLs a process whose event loop stalls past its window (60s @@ -1476,6 +1485,15 @@ export class ReferenceResolver { const result = await settleBatch(inFlight, batch); if (process.env.CODEGRAPH_SYNTH_TIMINGS) console.error(`[pool-timing] batch ${inFlight.mode}: ${batch.length} refs in ${Date.now() - tBatch}ms`); + // WAL-valve backstop at the ONE pool-idle boundary of the double-buffer + // (this batch settled, the next not yet fanned out): past the hard cap + // the writer parks for a full backfill here, where the pool's readers + // are all between statements — so the backfill completes, readers + // re-enter at SQLite's backfilled mark, and the next persist commit + // WRAPS the WAL instead of growing it. No-op (one fstat) under the cap. + const bp = parallel?.backpressure?.(); + if (bp) await bp; + // Persist in bounded sub-transactions with yields between: a whole // batch's edge insert / keyed deletes are otherwise one solid // synchronous span each on a multi-GB index, sitting BETWEEN the @@ -1585,6 +1603,11 @@ export class ReferenceResolver { const tIdx = Date.now(); await parallel!.bulkEdgeLoad!.end(); if (process.env.CODEGRAPH_SYNTH_TIMINGS) console.error(`[phase-timing] edge-index-recreate: ${Date.now() - tIdx}ms`); + // The recreate just wrote every non-unique edge index into the WAL + // (multi-GB at kernel scale) with the pool idle — fold before the + // synthesis passes pin readers against it for minutes. + const bp = parallel?.backpressure?.(); + if (bp) await bp; } } @@ -1601,7 +1624,8 @@ export class ReferenceResolver { this.queries, this.context, onSynthesisProgress, - pool + pool, + parallel?.backpressure ); } catch { // synthesis is additive and optional; ignore failures