perf(synth): shard near-duplicate signing across the resolver pool

__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,15 +173,39 @@ 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);
     expect(snapshot()).toEqual(before);
 
     // 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);
+
+    // 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);
   });
 

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));
   }
@@ -655,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
@@ -827,7 +833,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 +857,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 +1866,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,20 +2262,21 @@ 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<void> {
     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; }
         },
         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 +2296,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<void>; backpressure(): Promise<void> | null } | null
+    walValve?: { stop(): void; start(): void; drain(): Promise<void>; backpressure(): Promise<void> | null } | null,
+    /** Run the near-duplicate refresh inside the resolver's pool lifetime. */
+    withNearDuplicates = false
   ): Promise<ResolutionResult> {
     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

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<NearDupSignature[]>;
+
 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();
@@ -172,14 +194,47 @@ export function refreshNearDuplicates(deps: NearDuplicateDeps): { signed: number
   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<string, Set<number>>();
-  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) });
+  // 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<string, Set<number> | 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.
+    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 });
+      }
+    }
+    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)}`);
+      }
+    }
+  }
+  if (!signedOffThread) {
+    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.
@@ -229,7 +284,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<string, Set<number>>): { kept: number; rescored: number } {
+function pairNearDuplicates(deps: NearDuplicateDeps, freshSets?: ReadonlyMap<string, Set<number> | 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
@@ -271,7 +326,8 @@ function pairNearDuplicates(deps: NearDuplicateDeps, freshSets?: ReadonlyMap<str
       const node = queries.getNodeById(row.nodeId);
       // A set signed this refresh is already in memory; only unchanged bodies
       // pay the tokenize pass.
-      const cached = node ? freshSets?.get(node.id) : undefined;
+      const cachedRaw = node ? freshSets?.get(node.id) : undefined;
+      const cached = cachedRaw instanceof Set ? cachedRaw : cachedRaw ? new Set(cachedRaw) : undefined;
       const text = !cached && node ? lines(node.filePath)?.slice(node.startLine - 1, node.endLine).join('\n') : undefined;
       hit = { nodeId: row.nodeId, sig: asSignature(row.sig), set: cached ?? (text ? shingles(text, node!.name) : null) };
       sigs.set(rowid, hit);

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> | void;
       bulkEdgeLoad?: { begin: () => void; end: () => void | Promise<void> };
       /** 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.