From 94d675d6cabb4a5c1d4806e5ef8208e3f08345b1 Mon Sep 17 00:00:00 2001 From: bompus Date: Sat, 31 Oct 2026 16:33:28 -0600 Subject: [PATCH 1/4] probe: shard near-duplicate signing across the resolver pool --- __tests__/near-duplicates.test.ts | 23 ++++++++++-- src/codegraph.ts | 30 ++++++++++++---- src/graph/near-duplicates.ts | 56 ++++++++++++++++++++++++++---- src/resolution/index.ts | 15 ++++++++ src/resolution/resolver-pool.ts | 36 +++++++++++++++++++ src/resolution/resolver-worker.ts | 28 +++++++++++++++ src/resolution/tier-synthesizer.ts | 4 +++ 7 files changed, 175 insertions(+), 17 deletions(-) diff --git a/__tests__/near-duplicates.test.ts b/__tests__/near-duplicates.test.ts index 26ad13b614..a200cc3af8 100644 --- a/__tests__/near-duplicates.test.ts +++ b/__tests__/near-duplicates.test.ts @@ -11,7 +11,7 @@ import * as os from 'os'; import CodeGraph from '../src/index'; import { execFileSync } from 'child_process'; import { ToolHandler } from '../src/mcp/tools'; -import { forEachBandBucket, refreshNearDuplicates, signature, similarity } from '../src/graph/near-duplicates'; +import { forEachBandBucket, refreshNearDuplicates, shingles, signature, signatureFromSet, similarity } from '../src/graph/near-duplicates'; const urlOf = (name: string, label: string) => [ `export function ${name}(input: string | URL): string {`, @@ -173,7 +173,7 @@ describe('near-duplicates in the index', () => { // dsUrlOf changes: its two pairs are re-scored, the points pair is not read. db.prepare("UPDATE nodes SET updated_at = updated_at + 1 WHERE name = 'dsUrlOf'").run(); - const r = refreshNearDuplicates(deps); + const r = await refreshNearDuplicates(deps); expect(r.signed).toBe(1); expect(r.rescored).toBe(2); expect(reads.some((p) => p.includes('points'))).toBe(false); @@ -181,7 +181,24 @@ describe('near-duplicates in the index', () => { // Every body re-signed: a full recompute stores the same pairs. db.prepare("UPDATE nodes SET updated_at = updated_at + 1 WHERE kind = 'function'").run(); - refreshNearDuplicates(deps); + await refreshNearDuplicates(deps); + expect(snapshot()).toEqual(before); + + // The off-thread signer path stores the same signatures and pairs as the + // in-process loop (the pool worker computes shingles/signatureFromSet). + db.prepare("UPDATE nodes SET updated_at = updated_at + 1 WHERE kind = 'function'").run(); + const sign = async (bodies: Array<{ id: string; name: string; filePath: string; startLine: number; endLine: number; updatedAt: number }>) => + bodies.map((b) => { + const set = shingles(deps.readFile(b.filePath)!.split(/\r?\n/).slice(b.startLine - 1, b.endLine).join('\n'), b.name); + const sig = set ? signatureFromSet(set) : null; + return { + id: b.id, + sig: sig ? new Uint8Array(sig.buffer, sig.byteOffset, sig.byteLength).slice() : new Uint8Array(0), + set: set ? Uint32Array.from(set) : new Uint32Array(0), + }; + }); + const viaSigner = await refreshNearDuplicates({ ...deps, sign }); + expect(viaSigner.signed).toBeGreaterThan(0); expect(snapshot()).toEqual(before); }); diff --git a/src/codegraph.ts b/src/codegraph.ts index d7c0a53122..80c12aa08d 100644 --- a/src/codegraph.ts +++ b/src/codegraph.ts @@ -58,7 +58,7 @@ import type { FileWatcher, WatchOptions, PendingFile } from './sync'; import { EXTRACTION_VERSION } from './extraction/extraction-version'; import { isGeneratedFile } from './extraction/generated-detection'; import { hasGrammarLoadFailure, isFileLevelOnlyLanguage } from './extraction/grammars'; -import { refreshNearDuplicates } from './graph/near-duplicates'; +import { refreshNearDuplicates, type NearDupSigner } from './graph/near-duplicates'; import { getCodeGraphDir } from './directory'; import { deriveProjectNameTokens } from './search/query-utils'; import { findSeedSource, copySeedIndex, type SeedSource } from './sync/worktree-seed'; @@ -189,6 +189,8 @@ export class CodeGraph { // method call away from the constructor body. private lazyOrchestrator?: ExtractionOrchestrator; private lazyResolver?: ReferenceResolver; + /** Set by the resolver's post-synthesis near-dup hook so indexAll does not repeat it. */ + private nearDuplicatesRan = false; private get orchestrator(): ExtractionOrchestrator { return (this.lazyOrchestrator ??= new (extraction().ExtractionOrchestrator)(this.projectRoot, this.queries)); } @@ -827,7 +829,8 @@ export class CodeGraph { total: totalPasses, }); }, - walValve + walValve, + true ); if (process.env.CODEGRAPH_SYNTH_TIMINGS) console.error(`[phase-timing] resolution: ${Date.now() - tResolve}ms`); @@ -850,7 +853,7 @@ export class CodeGraph { await this.resolver.resolveDeferredThisMemberRefs(); if (process.env.CODEGRAPH_SYNTH_TIMINGS) console.error(`[synth-timing] deferredThisMember: ${Date.now() - tDeferred}ms`); } - if (result.success) this.refreshNearDuplicates(); + if (result.success && !this.nearDuplicatesRan) await this.refreshNearDuplicates(); // Refresh planner stats + checkpoint the WAL after bulk writes. // Off-thread (worker connection): on a multi-GB index this is minutes @@ -1859,7 +1862,7 @@ export class CodeGraph { throw new Error('Reconciliation did not complete synthesized-edge refresh.'); } if (filesChanged || result.filesRemoved > 0) { - this.refreshNearDuplicates(result.changedFilePaths, result.filesRemoved > 0); + await this.refreshNearDuplicates(result.changedFilePaths, result.filesRemoved > 0); } // Refresh planner stats + checkpoint the WAL after bulk writes. @@ -2255,13 +2258,13 @@ export class CodeGraph { * Only changed bodies are re-signed, so a sync that touched no function costs * one query. Never fails an index: the pairs are an annotation. */ - private refreshNearDuplicates(changedFiles?: readonly string[], force = false): void { + private async refreshNearDuplicates(changedFiles?: readonly string[], force = false, sign?: NearDupSigner): Promise { const t = Date.now(); try { const deprioritized = this.queries.getDeprioritizedPathMatcher(); // The index's content-based generated flag, not just the path check. const generated = this.queries.generatedPredicateFor(this.queries.getAllFiles().map((f) => f.path)); - const r = refreshNearDuplicates({ + const r = await refreshNearDuplicates({ queries: this.queries, readFile: (p) => { try { return fs.readFileSync(path.join(this.projectRoot, p), 'utf-8'); } catch { return null; } @@ -2269,6 +2272,7 @@ export class CodeGraph { isExcluded: (p) => generated(p) || isGeneratedFile(p) || (deprioritized?.(p) ?? false), changedFiles, force, + sign, }); if (process.env.CODEGRAPH_SYNTH_TIMINGS) console.error(`[synth-timing] nearDuplicates: ${Date.now() - t}ms (${r.signed} signed, ${r.rescored} pairs re-scored, ${r.pairs ?? 'unchanged'} pairs)`); } catch (error) { @@ -2288,10 +2292,22 @@ export class CodeGraph { // kernel conn is live on the db (the rusqlite build's wal-index locks // can't see node:sqlite's — an off-thread checkpoint could rewrite the // -shm header under a writer commit). Structural so tests can fake it. - walValve?: { stop(): void; start(): void; drain(): Promise; backpressure(): Promise | null } | null + walValve?: { stop(): void; start(): void; drain(): Promise; backpressure(): Promise | null } | null, + /** Run the near-duplicate refresh inside the resolver's pool lifetime. */ + withNearDuplicates = false ): Promise { return this.resolver.resolveAndPersistBatched(onProgress, undefined, onSynthesisProgress, { dbPath: this.db.getPath(), + nearDuplicates: withNearDuplicates + ? (pool) => { + const sign: NearDupSigner | undefined = pool + ? (bodies) => pool.signNearDuplicates(bodies, this.projectRoot) + : undefined; + return this.refreshNearDuplicates(undefined, false, sign).finally(() => { + this.nearDuplicatesRan = true; + }); + } + : undefined, // Bulk-edge-load hooks: on big runs the resolver drops the non-unique // edge indexes for the batch loop and recreates them before synthesis // (which reads kind-keyed). Concurrent readers (a daemon serving this diff --git a/src/graph/near-duplicates.ts b/src/graph/near-duplicates.ts index e1021f0f72..943f058065 100644 --- a/src/graph/near-duplicates.ts +++ b/src/graph/near-duplicates.ts @@ -128,10 +128,32 @@ function asSignature(bytes: Uint8Array): Uint32Array { return new Uint32Array(bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength)); } +/** One body a sharded signer computes a signature for. */ +export interface NearDupBody { + id: string; + name: string; + filePath: string; + startLine: number; + endLine: number; + updatedAt: number; +} + +/** A signed body: the MinHash bytes plus its shingle set (empty when unreadable). */ +export interface NearDupSignature { + id: string; + sig: Uint8Array; + set: Uint32Array; +} + +/** Off-thread signer for a batch of bodies (the resolver pool in a full index). */ +export type NearDupSigner = (bodies: NearDupBody[]) => Promise; + export interface NearDuplicateDeps { queries: QueryBuilder; /** The file's text, or null when it cannot be read. */ readFile(filePath: string): string | null; + /** When set, compute the signatures off-thread instead of reading files here. */ + sign?: NearDupSigner; /** Generated or configured-away files, left out like tests and vendored code. */ isExcluded?(filePath: string): boolean; /** @@ -152,7 +174,7 @@ export interface NearDuplicateDeps { * the pairs. Returns how many signatures were written and pairs stored * (`pairs` is null when nothing changed and the stored pairs stand). */ -export function refreshNearDuplicates(deps: NearDuplicateDeps): { signed: number; pairs: number | null; rescored: number } { +export async function refreshNearDuplicates(deps: NearDuplicateDeps): Promise<{ signed: number; pairs: number | null; rescored: number }> { const { queries } = deps; const excluded = (p: string) => isTestFile(p) || VENDORED.test(p) || (deps.isExcluded?.(p) ?? false); const tSign = Date.now(); @@ -173,13 +195,33 @@ export function refreshNearDuplicates(deps: NearDuplicateDeps): { signed: number // The shingle sets just computed, keyed by node id: scoring reuses them // instead of re-tokenizing every body that lands in a candidate pair. const freshSets = new Map>(); - for (const [filePath, bodies] of stale) { - const lines = deps.readFile(filePath)?.split(/\r?\n/); - for (const b of bodies) { - const set = lines ? shingles(lines.slice(b.startLine - 1, b.endLine).join('\n'), b.name) : null; - const sig = set ? signatureFromSet(set) : null; + if (deps.sign) { + // The bodies are grouped by file for the in-process path; the signer takes + // them as one batch and the worker/pool groups what it can. + const bodies: NearDupBody[] = []; + for (const [filePath, list] of stale) { + for (const b of list) { + bodies.push({ id: b.id, name: b.name, filePath, startLine: b.startLine, endLine: b.endLine, updatedAt: b.updatedAt }); + } + } + const signed = bodies.length > 0 ? await deps.sign(bodies) : []; + const byId = new Map(bodies.map((b) => [b.id, b])); + for (const s of signed) { + const b = byId.get(s.id); + if (!b) continue; + const set = s.set.length > 0 ? new Set(s.set) : null; if (set) freshSets.set(b.id, set); - rows.push({ nodeId: b.id, nodeUpdatedAt: b.updatedAt, sig: sig ? new Uint8Array(sig.buffer) : new Uint8Array(0) }); + rows.push({ nodeId: b.id, nodeUpdatedAt: b.updatedAt, sig: s.sig.length > 0 ? s.sig : new Uint8Array(0) }); + } + } else { + for (const [filePath, bodies] of stale) { + const lines = deps.readFile(filePath)?.split(/\r?\n/); + for (const b of bodies) { + const set = lines ? shingles(lines.slice(b.startLine - 1, b.endLine).join('\n'), b.name) : null; + const sig = set ? signatureFromSet(set) : null; + if (set) freshSets.set(b.id, set); + rows.push({ nodeId: b.id, nodeUpdatedAt: b.updatedAt, sig: sig ? new Uint8Array(sig.buffer) : new Uint8Array(0) }); + } } } // A re-signed body's cached pair scores are stale; a dropped body's are moot. diff --git a/src/resolution/index.ts b/src/resolution/index.ts index 52caca2d93..13eacefb80 100644 --- a/src/resolution/index.ts +++ b/src/resolution/index.ts @@ -2280,6 +2280,9 @@ export class ReferenceResolver { // a 22GB WAL on a 4.6GB DB (migration plan §7a.1). parallel?: { dbPath: string; + /** Full index: sign near-duplicate bodies after synthesis, while the + * pool is still alive; null pool means the in-process path. */ + nearDuplicates?: (pool: ResolverPool | null) => Promise | void; bulkEdgeLoad?: { begin: () => void; end: () => void | Promise }; /** unresolved_refs index window for the batched loop — the loop only * reads the status index + PK; dropping the sync-path ref indexes cuts @@ -3051,6 +3054,18 @@ export class ReferenceResolver { // synthesis is additive and optional; ignore failures } if (process.env.CODEGRAPH_SYNTH_TIMINGS) console.error(`[phase-timing] callback-synthesis: ${Date.now() - tSynth}ms`); + // Full index: sign near-duplicate bodies now, while the pool is still + // open and idle. The callback falls back to the in-process path when no + // pool engaged; it never fails the index. + if (parallel?.nearDuplicates) { + try { + await parallel.nearDuplicates(pool); + } catch (err) { + logDebug('Near-duplicate refresh failed; the index stands', { + error: err instanceof Error ? err.message : String(err), + }); + } + } } finally { if (pool) await pool.destroy().catch(() => undefined); // The workers' kernel-reader snapshot is per-run scratch — remove it. diff --git a/src/resolution/resolver-pool.ts b/src/resolution/resolver-pool.ts index f3fb01853b..60c17f11d9 100644 --- a/src/resolution/resolver-pool.ts +++ b/src/resolution/resolver-pool.ts @@ -14,6 +14,7 @@ import * as fs from 'fs'; import * as path from 'path'; import * as os from 'os'; import type { Edge, UnresolvedReference } from '../types'; +import type { NearDupBody, NearDupSignature } from '../graph/near-duplicates'; import type { KernelResolveStats, ResolvedRef, UnresolvedRef } from './types'; import { memoryBudgetBytes } from './memory-budget'; import { terminateOnceStarted, workerStarted } from '../worker-teardown'; @@ -124,6 +125,7 @@ export class ResolverPool { private nextId = 0; private waiters = new Map void; reject: (e: Error) => void }>(); private synthWaiters = new Map void; reject: (e: Error) => void }>(); + private signWaiters = new Map void; reject: (e: Error) => void }>(); private recycleWaiters = new Map void; done: () => void }>(); // Dispatch hold for the WAL backpressure window: in-flight chunks finish // and release their read marks, no new ones start, so the valve's @@ -318,6 +320,11 @@ export class ResolverPool { const waiter = this.synthWaiters.get(msg.id); this.synthWaiters.delete(msg.id); waiter?.resolve({ edges: msg.edges ?? [], ms: msg.ms ?? 0, skips: msg.skips }); + } else if (msg.type === 'near-dup-signed' && msg.id !== undefined) { + pw.busy--; + const waiter = this.signWaiters.get(msg.id); + this.signWaiters.delete(msg.id); + waiter?.resolve((msg as { results?: NearDupSignature[] }).results ?? []); } else if (msg.type === 'recycling' && msg.id !== undefined) { this.recycleWaiters.get(msg.id)?.arm(); } else if (msg.type === 'recycled' && msg.id !== undefined) { @@ -336,6 +343,10 @@ export class ResolverPool { const waiter = this.synthWaiters.get(msg.id)!; this.synthWaiters.delete(msg.id); waiter.reject(err); + } else if (msg.id !== undefined && this.signWaiters.has(msg.id)) { + const waiter = this.signWaiters.get(msg.id)!; + this.signWaiters.delete(msg.id); + waiter.reject(err); } else { this.fail(err); } @@ -363,6 +374,8 @@ export class ResolverPool { this.pendingChunks = []; for (const [, waiter] of this.synthWaiters) waiter.reject(this.failed); this.synthWaiters.clear(); + for (const [, waiter] of this.signWaiters) waiter.reject(this.failed); + this.signWaiters.clear(); // Pending recycles resolve rather than reject: their per-call timeout // owns rejection, and the recycle caller checks this.failed next round. for (const [, w] of this.recycleWaiters) w.done(); @@ -481,6 +494,29 @@ export class ResolverPool { }); } + /** + * Sign near-duplicate bodies across the idle pool (the full index calls + * this after synthesis, before the pool is destroyed). Workers read the + * bodies from disk and return the MinHash bytes plus the shingle set the + * pair pass reuses; the main thread writes the rows. + */ + async signNearDuplicates(bodies: NearDupBody[], projectRoot: string): Promise { + if (this.failed) throw this.failed; + if (bodies.length === 0) return []; + const lanes = Math.min(this.workers.length, 8, Math.max(1, Math.ceil(bodies.length / 256))); + const chunks: NearDupBody[][] = Array.from({ length: lanes }, () => []); + const per = Math.ceil(bodies.length / lanes); + for (let i = 0; i < bodies.length; i++) chunks[Math.min(lanes - 1, Math.floor(i / per))]!.push(bodies[i]!); + const parts = await Promise.all(chunks.filter((c) => c.length > 0).map((chunk) => new Promise((resolve, reject) => { + const id = this.nextId++; + const pw = this.workers.reduce((a, b) => (b.busy < a.busy ? b : a)); + pw.busy++; + this.signWaiters.set(id, { resolve, reject }); + pw.worker.postMessage({ type: 'near-dup-sign', id, bodies: chunk, projectRoot }); + }))); + return parts.flat(); + } + /** * Ask every worker to close and reopen its read-only connection, and wait * for all acks. MUST be called only at the pool-idle boundary (all fanned diff --git a/src/resolution/resolver-worker.ts b/src/resolution/resolver-worker.ts index 0079df8dbd..dfdcd6022a 100644 --- a/src/resolution/resolver-worker.ts +++ b/src/resolution/resolver-worker.ts @@ -21,6 +21,9 @@ try { } catch { /* cache is best-effort */ } import { parentPort, workerData } from 'worker_threads'; +import * as fs from 'fs'; +import * as path from 'path'; +import { shingles, signatureFromSet, type NearDupBody, type NearDupSignature } from '../graph/near-duplicates'; import { SynthSkips, SYNTH_SKIPS_VERSION } from './synth-skips'; import { createDatabase, SqliteDatabase } from '../db/sqlite-adapter'; import { QueryBuilder } from '../db/queries'; @@ -52,6 +55,7 @@ type InMessage = | { type: 'recycle'; id: number; kernel?: { dbPath: string; generation: string } } | { type: 'resolve'; id: number; refs: UnresolvedReference[] } | { type: 'synth'; id: number; pass: string; shard?: { index: number; count: number; from?: number; to?: number } } + | { type: 'near-dup-sign'; id: number; bodies: NearDupBody[]; projectRoot: string } | { type: 'close' }; let dbPath: string | null = null; @@ -128,6 +132,30 @@ port.on('message', (msg: InMessage) => { port.postMessage({ type: 'result', id: msg.id, ...out }); break; } + case 'near-dup-sign': { + const byFile = new Map(); + for (const b of msg.bodies) { + const list = byFile.get(b.filePath) ?? []; + list.push(b); + byFile.set(b.filePath, list); + } + const results: NearDupSignature[] = []; + for (const [file, list] of byFile) { + let lines: string[] | null = null; + try { lines = fs.readFileSync(path.join(msg.projectRoot, file), 'utf-8').split(/\r?\n/); } catch { lines = null; } + for (const b of list) { + const set = lines ? shingles(lines.slice(b.startLine - 1, b.endLine).join('\n'), b.name) : null; + const sig = set ? signatureFromSet(set) : null; + results.push({ + id: b.id, + sig: sig ? new Uint8Array(sig.buffer, sig.byteOffset, sig.byteLength).slice() : new Uint8Array(0), + set: set ? Uint32Array.from(set) : new Uint32Array(0), + }); + } + } + port.postMessage({ type: 'near-dup-signed', id: msg.id, results }); + break; + } case 'synth': { // Run one synthesis pass against this worker's read-only connection. // Passes only READ (graph + source via the resolver's context); their diff --git a/src/resolution/tier-synthesizer.ts b/src/resolution/tier-synthesizer.ts index e702df9649..be4e04968e 100644 --- a/src/resolution/tier-synthesizer.ts +++ b/src/resolution/tier-synthesizer.ts @@ -1324,6 +1324,7 @@ export async function crossTierEdges(ctx: ResolutionContext, onYield: MaybeYield const cache = new Map(); let scanned = 0; + const tCollect = Date.now(); for (const file of ctx.getAllFiles()) { if (!JS_FILE.test(file) || isTestPath(file) || isGeneratedFile(file)) continue; if ((++scanned & 63) === 0) await onYield(); @@ -1354,6 +1355,8 @@ export async function crossTierEdges(ctx: ResolutionContext, onYield: MaybeYield } } + const collectMs = Date.now() - tCollect; + const tMatch = Date.now(); const edges: Edge[] = []; const seen = new Set(); for (const site of httpSites) { @@ -1392,5 +1395,6 @@ export async function crossTierEdges(ctx: ResolutionContext, onYield: MaybeYield await onYield(); pairQueue(producers, consumers, edges, seen); pairEvents(dispatches, handlers, edges, seen); + if (process.env.CODEGRAPH_SYNTH_TIMINGS) console.error(`[synth-timing] tierEdges sub: collect=${collectMs}ms match=${Date.now() - tMatch}ms`); return edges; } From 2928488fcf0e0986afbf53e18879cc8e6a137948 Mon Sep 17 00:00:00 2001 From: bompus Date: Sat, 31 Oct 2026 17:59:57 -0600 Subject: [PATCH 2/4] probe: store signed shingle arrays and build Sets only for pair bodies --- src/graph/near-duplicates.ts | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/src/graph/near-duplicates.ts b/src/graph/near-duplicates.ts index 943f058065..a9a72561aa 100644 --- a/src/graph/near-duplicates.ts +++ b/src/graph/near-duplicates.ts @@ -194,7 +194,10 @@ export async function refreshNearDuplicates(deps: NearDuplicateDeps): Promise<{ const rows: Array<{ nodeId: string; nodeUpdatedAt: number; sig: Uint8Array }> = []; // The shingle sets just computed, keyed by node id: scoring reuses them // instead of re-tokenizing every body that lands in a candidate pair. - const freshSets = new Map>(); + // The in-process path stores Sets; the off-thread signer stores its + // transferred Uint32Arrays and the pair pass builds a Set only for a body + // that actually lands in a candidate pair (36k Sets was most of the handoff). + const freshSets = new Map | Uint32Array>(); if (deps.sign) { // The bodies are grouped by file for the in-process path; the signer takes // them as one batch and the worker/pool groups what it can. @@ -209,8 +212,7 @@ export async function refreshNearDuplicates(deps: NearDuplicateDeps): Promise<{ for (const s of signed) { const b = byId.get(s.id); if (!b) continue; - const set = s.set.length > 0 ? new Set(s.set) : null; - if (set) freshSets.set(b.id, set); + if (s.set.length > 0) freshSets.set(b.id, s.set); rows.push({ nodeId: b.id, nodeUpdatedAt: b.updatedAt, sig: s.sig.length > 0 ? s.sig : new Uint8Array(0) }); } } else { @@ -271,7 +273,7 @@ export function forEachBandBucket( } /** LSH over the stored signatures, one band at a time so memory stays linear in the bodies. */ -function pairNearDuplicates(deps: NearDuplicateDeps, freshSets?: ReadonlyMap>): { kept: number; rescored: number } { +function pairNearDuplicates(deps: NearDuplicateDeps, freshSets?: ReadonlyMap | Uint32Array>): { kept: number; rescored: number } { const { queries } = deps; // Every signature in one scan (rowid order), then each band's buckets from // the bytes in memory: sixteen SQL scans with a per-row substr and base64 @@ -313,7 +315,8 @@ function pairNearDuplicates(deps: NearDuplicateDeps, freshSets?: ReadonlyMap Date: Sat, 31 Oct 2026 19:14:27 -0600 Subject: [PATCH 3/4] fix(graph): fall back to in-process signing when the nearDup signer fails Review of #609: a rejected pool/worker signing task left the stale bodies unsigned and the pairs stale. Catch the rejection and run the same in-process loop the no-pool path uses, logging under SYNTH_TIMINGS. The new test forces a rejecting signer and asserts the rows and pairs match the in-process result. --- __tests__/near-duplicates.test.ts | 7 +++++++ src/graph/near-duplicates.ts | 27 +++++++++++++++++++-------- 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/__tests__/near-duplicates.test.ts b/__tests__/near-duplicates.test.ts index a200cc3af8..b2e1c6421e 100644 --- a/__tests__/near-duplicates.test.ts +++ b/__tests__/near-duplicates.test.ts @@ -200,6 +200,13 @@ describe('near-duplicates in the index', () => { const viaSigner = await refreshNearDuplicates({ ...deps, sign }); expect(viaSigner.signed).toBeGreaterThan(0); expect(snapshot()).toEqual(before); + + // A worker/pool rejection falls back to the in-process loop; the rows and + // pairs must be the same, not stale or empty. + db.prepare("UPDATE nodes SET updated_at = updated_at + 1 WHERE kind = 'function'").run(); + const viaFallback = await refreshNearDuplicates({ ...deps, sign: async () => { throw new Error('pool down'); } }); + expect(viaFallback.signed).toBeGreaterThan(0); + expect(snapshot()).toEqual(before); }); it('a git-scoped sync that edits and removes files stores the same pairs as a fresh index', async () => { diff --git a/src/graph/near-duplicates.ts b/src/graph/near-duplicates.ts index a9a72561aa..a11864c6d5 100644 --- a/src/graph/near-duplicates.ts +++ b/src/graph/near-duplicates.ts @@ -198,6 +198,7 @@ export async function refreshNearDuplicates(deps: NearDuplicateDeps): Promise<{ // transferred Uint32Arrays and the pair pass builds a Set only for a body // that actually lands in a candidate pair (36k Sets was most of the handoff). const freshSets = new Map | Uint32Array>(); + let signedOffThread = false; if (deps.sign) { // The bodies are grouped by file for the in-process path; the signer takes // them as one batch and the worker/pool groups what it can. @@ -207,15 +208,25 @@ export async function refreshNearDuplicates(deps: NearDuplicateDeps): Promise<{ bodies.push({ id: b.id, name: b.name, filePath, startLine: b.startLine, endLine: b.endLine, updatedAt: b.updatedAt }); } } - const signed = bodies.length > 0 ? await deps.sign(bodies) : []; - const byId = new Map(bodies.map((b) => [b.id, b])); - for (const s of signed) { - const b = byId.get(s.id); - if (!b) continue; - if (s.set.length > 0) freshSets.set(b.id, s.set); - rows.push({ nodeId: b.id, nodeUpdatedAt: b.updatedAt, sig: s.sig.length > 0 ? s.sig : new Uint8Array(0) }); + try { + const signed = bodies.length > 0 ? await deps.sign(bodies) : []; + const byId = new Map(bodies.map((b) => [b.id, b])); + for (const s of signed) { + const b = byId.get(s.id); + if (!b) continue; + if (s.set.length > 0) freshSets.set(b.id, s.set); + rows.push({ nodeId: b.id, nodeUpdatedAt: b.updatedAt, sig: s.sig.length > 0 ? s.sig : new Uint8Array(0) }); + } + signedOffThread = true; + } catch (err) { + // A worker/pool failure must not leave the signatures stale: sign the + // same bodies in-process below, exactly as a run without a pool does. + if (process.env.CODEGRAPH_SYNTH_TIMINGS) { + console.error(`[synth-timing] nearDup signer failed, signing in-process: ${err instanceof Error ? err.message : String(err)}`); + } } - } else { + } + if (!signedOffThread) { for (const [filePath, bodies] of stale) { const lines = deps.readFile(filePath)?.split(/\r?\n/); for (const b of bodies) { From 352fe4d6685898a1168dbf03f057c89fe3b7d7af Mon Sep 17 00:00:00 2001 From: bompus Date: Sat, 31 Oct 2026 19:26:45 -0600 Subject: [PATCH 4/4] fix(graph): reset the near-duplicate ran flag for each full index Review of #609: the post-synthesis hook set the instance field true and a later indexAll on the same CodeGraph instance skipped the fallback even when the hook did not run (no refs, no pool, resolver skipped). Reset it at the start of each full index. --- src/codegraph.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/codegraph.ts b/src/codegraph.ts index 80c12aa08d..b6208a1504 100644 --- a/src/codegraph.ts +++ b/src/codegraph.ts @@ -657,6 +657,10 @@ export class CodeGraph { return { success: false, filesIndexed: 0, filesSkipped: 0, filesErrored: 0, nodesCreated: 0, edgesCreated: 0, errors: [{ message: 'Could not acquire file lock - another process may be indexing', severity: 'error' as const }], durationMs: 0 }; } const startedAt = Date.now(); + // Each full index owns its near-duplicate pass: the post-synthesis hook + // sets this, and a later index on the same instance must run it again + // even when that hook did not (no refs, no pool, resolver skipped). + this.nearDuplicatesRan = false; // Defer WAL auto-checkpointing for the whole bulk run (#1231): the // default 1000-page interval re-writes hot pages into the main DB file // over and over — ~95% of all disk I/O during a bulk index, and a