diff --git a/__tests__/near-duplicates.test.ts b/__tests__/near-duplicates.test.ts index 26ad13b61..b2e1c6421 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,31 @@ 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); + + // 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); }); diff --git a/src/codegraph.ts b/src/codegraph.ts index d7c0a5312..b6208a150 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)); } @@ -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,13 +2262,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 +2276,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 +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; 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 e1021f0f7..a11864c6d 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(); @@ -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>(); - 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 | 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>): { 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 @@ -271,7 +326,8 @@ function pairNearDuplicates(deps: NearDuplicateDeps, freshSets?: ReadonlyMap 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 f3fb01853..60c17f11d 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 0079df8db..dfdcd6022 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 e702df964..be4e04968 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; }