diff --git a/CHANGELOG.md b/CHANGELOG.md index d3ba154..ad62dbc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### New Features +- Reference resolution now runs in parallel on large projects. When a project has enough pending references to make it worthwhile (roughly 150k+, typical for big Java/Kotlin/Spring codebases), resolution fans out across worker threads while results are applied in the exact order the single-threaded path would have used — the graph comes out byte-for-byte identical, about twice as fast end-to-end on a 4,000-file Java project in our testing. Small projects keep the single-threaded path automatically (the fan-out costs more than it saves there). Set `CODEGRAPH_NO_PARALLEL_RESOLVE=1` to disable, or `CODEGRAPH_PARALLEL_RESOLVE_MIN=` to tune when it engages. +- Indexing is significantly faster — a fresh `codegraph init` on a medium TypeScript project takes about a third less wall-clock time, with the same graph produced byte-for-byte. The gains come from batching database writes, storing files on a dedicated writer thread, memoizing repeated import-resolution lookups, skipping per-row search-index maintenance during the bulk build (rebuilt once at the end), and — on completely fresh databases only — deferring disk durability until the index completes, since an interrupted first index is simply re-run. Set `CODEGRAPH_NO_FAST_INIT=1` to keep full crash-durability during the initial build, or `CODEGRAPH_NO_STORE_WORKER=1` to store on the main thread. - `codegraph install` and `codegraph upgrade` now offer CodeGraph Pro beta access after finishing — answer yes, type your email, and you join the same waitlist as the getcodegraph.com homepage form. Strictly opt-in and asked at most once per machine total: nothing is sent unless you say yes and enter an email, either answer is remembered so no later install or upgrade ever re-asks, and non-interactive runs (`--yes`, scripts, CI) never see the question. - Every release is now cryptographically verifiable: npm packages publish with npm provenance (the "Provenance" badge on npmjs.com, proving each version was built by this repository's release workflow from a specific commit), and the GitHub Release bundles carry signed build attestations you can check with `gh attestation verify -R colbymchenry/codegraph`. diff --git a/src/bin/codegraph.ts b/src/bin/codegraph.ts index 69d9856..1275d1f 100644 --- a/src/bin/codegraph.ts +++ b/src/bin/codegraph.ts @@ -28,6 +28,15 @@ // otherwise blinds the PPID watchdog forever (#1185) — see early-ppid.ts. import '../mcp/early-ppid'; +// Persist V8 compile artifacts across runs (Node ≥22.8). Every invocation — +// and every worker thread, which re-requires the whole extraction module +// graph — skips recompiling unchanged sources. Worth hundreds of ms of +// worker-boot latency per bulk index; harmless no-op when unavailable. +try { + // eslint-disable-next-line @typescript-eslint/no-require-imports + (require('node:module') as { enableCompileCache?: () => void }).enableCompileCache?.(); +} catch { /* cache is best-effort */ } + import { Command } from 'commander'; import * as path from 'path'; import * as fs from 'fs'; diff --git a/src/db/index.ts b/src/db/index.ts index 0cd2823..79e5d3f 100644 --- a/src/db/index.ts +++ b/src/db/index.ts @@ -112,9 +112,78 @@ export class DatabaseConnection { runMigrations(db, currentVersion); } + // Self-heal a bulk-load window that never closed (crash between + // beginBulkNodeLoad and endBulkNodeLoad): the FTS triggers are missing and + // nodes_fts is stale. Rebuild + recreate so search stays in sync. + conn.healBulkNodeLoad(); + return conn; } + /** + * FTS maintenance triggers dropped/recreated around a bulk load. + * Names must match schema.sql. + */ + private static readonly FTS_TRIGGER_NAMES = ['nodes_ai', 'nodes_ad', 'nodes_au'] as const; + + /** + * Enter bulk-load mode: drop the per-row FTS sync triggers so mass node + * inserts skip per-row tokenization. MUST be paired with endBulkNodeLoad() + * (use try/finally); a crash inside the window is healed on the next open(). + * The window is DB-wide (triggers are schema objects), which is safe because + * endBulkNodeLoad() rebuilds nodes_fts from the nodes table wholesale — any + * row written by anyone during the window is captured by the rebuild. + */ + beginBulkNodeLoad(): void { + for (const t of DatabaseConnection.FTS_TRIGGER_NAMES) { + this.db.exec(`DROP TRIGGER IF EXISTS ${t}`); + } + } + + /** + * Leave bulk-load mode: rebuild the whole FTS index from the nodes table in + * one pass (far cheaper than per-row trigger firings), then recreate the + * triggers by re-running schema.sql (idempotent — everything in it is + * IF NOT EXISTS). + */ + endBulkNodeLoad(): void { + this.db.exec(`INSERT INTO nodes_fts(nodes_fts) VALUES('rebuild')`); + this.recreateFtsTriggers(); + } + + /** Recreate the FTS triggers + rebuild if a bulk-load window never closed. */ + private healBulkNodeLoad(): void { + const row = this.db + .prepare( + `SELECT count(*) AS c FROM sqlite_master WHERE type = 'trigger' AND name IN ('nodes_ai','nodes_ad','nodes_au')` + ) + .get() as { c: number } | undefined; + if ((row?.c ?? 0) >= DatabaseConnection.FTS_TRIGGER_NAMES.length) return; + this.endBulkNodeLoad(); + } + + /** + * Recreate the FTS sync triggers from schema.sql — extracted from the file + * rather than duplicated here so the DDL cannot drift from the schema. + * (Re-execing the whole schema is not an option: it contains data INSERTs + * that are not idempotent, e.g. schema_versions.) + */ + private recreateFtsTriggers(): void { + const schemaPath = path.join(__dirname, 'schema.sql'); + const schema = fs.readFileSync(schemaPath, 'utf-8'); + const triggerDdls = schema.match( + /CREATE TRIGGER IF NOT EXISTS nodes_a[idu]\b[\s\S]*?END;/g + ); + if (!triggerDdls || triggerDdls.length !== DatabaseConnection.FTS_TRIGGER_NAMES.length) { + throw new Error( + `schema.sql: expected ${DatabaseConnection.FTS_TRIGGER_NAMES.length} nodes FTS triggers, found ${triggerDdls?.length ?? 0}` + ); + } + for (const ddl of triggerDdls) { + this.db.exec(ddl); + } + } + /** * Get the underlying database instance */ diff --git a/src/db/queries.ts b/src/db/queries.ts index 15f2611..7370a09 100644 --- a/src/db/queries.ts +++ b/src/db/queries.ts @@ -245,6 +245,46 @@ export class QueryBuilder { private segmentedNames: Set = new Set(); private static readonly MAX_SEGMENTED_NAMES = 65536; + // Multi-row INSERT statements, cached per (statement kind × row count). The + // bulk write path decomposes N rows into a few fixed batch sizes so each + // size's statement is prepared once and reused — one .run() binds a whole + // chunk instead of one row, which is where the per-call overhead lives. + // Row order within and across chunks is the input order, so rowid assignment + // (and therefore resolution's insertion-order disambiguation) is identical + // to the one-row-per-run path. + private batchStmts: Map = new Map(); + private static readonly BATCH_SIZES: readonly number[] = [128, 32, 8, 1]; + + /** + * Run `rows` through a multi-row `INSERT` built as `head + (tuple,)*n`, + * decomposed greedily into the cached batch sizes. Preserves row order. + */ + private runBatched(kind: string, head: string, tuple: string, rows: unknown[][]): void { + if (rows.length === 0) return; + let i = 0; + for (const size of QueryBuilder.BATCH_SIZES) { + while (rows.length - i >= size) { + const key = `${kind}:${size}`; + let stmt = this.batchStmts.get(key); + if (!stmt) { + stmt = this.db.prepare(head + new Array(size).fill(tuple).join(',')); + this.batchStmts.set(key, stmt); + } + if (size === 1) { + stmt.run(...rows[i]!); + } else { + const params: unknown[] = []; + for (let r = 0; r < size; r++) { + const row = rows[i + r]!; + for (let c = 0; c < row.length; c++) params.push(row[c]); + } + stmt.run(...params); + } + i += size; + } + } + } + constructor(db: SqliteDatabase) { this.db = db; } @@ -351,17 +391,14 @@ export class QueryBuilder { /** Write `name`'s segments into name_segment_vocab (idempotent). */ private insertNameSegments(name: string): void { - if (this.segmentedNames.has(name)) return; - if (this.segmentedNames.size >= QueryBuilder.MAX_SEGMENTED_NAMES) this.segmentedNames.clear(); - this.segmentedNames.add(name); - if (!this.stmts.insertNameSegment) { - this.stmts.insertNameSegment = this.db.prepare( - 'INSERT OR IGNORE INTO name_segment_vocab (segment, name) VALUES (?, ?)', - ); - } - for (const segment of splitIdentifierSegments(name)) { - this.stmts.insertNameSegment.run(segment, name); - } + const rows: unknown[][] = []; + this.collectNameSegmentRows(name, rows); + this.runBatched( + 'insertNameSegments', + 'INSERT OR IGNORE INTO name_segment_vocab (segment, name) VALUES ', + '(?,?)', + rows + ); } /** @@ -369,12 +406,124 @@ export class QueryBuilder { */ insertNodes(nodes: Node[]): void { this.db.transaction(() => { + // Bulk path: same semantics as insertNode() per row (validation, cache + // invalidation, segment vocab), but bound as multi-row INSERTs — the + // per-.run() call overhead dominates the store phase on full indexes. + const rows: unknown[][] = []; + const segmentRows: unknown[][] = []; for (const node of nodes) { - this.insertNode(node); + if (!node.id || !node.kind || !node.name || !node.filePath || !node.language) { + console.error('[CodeGraph] Skipping node with missing required fields:', { + id: node.id, + kind: node.kind, + name: node.name, + filePath: node.filePath, + language: node.language, + }); + continue; + } + this.nodeCache.delete(node.id); + rows.push([ + node.id, + node.kind, + node.name, + node.qualifiedName ?? node.name, + node.filePath, + node.language, + node.startLine ?? 0, + node.endLine ?? 0, + node.startColumn ?? 0, + node.endColumn ?? 0, + node.docstring ?? null, + node.signature ?? null, + node.visibility ?? null, + node.isExported ? 1 : 0, + node.isAsync ? 1 : 0, + node.isStatic ? 1 : 0, + node.isAbstract ? 1 : 0, + node.decorators ? JSON.stringify(node.decorators) : null, + node.typeParameters ? JSON.stringify(node.typeParameters) : null, + node.returnType ?? null, + node.updatedAt ?? Date.now(), + ]); + if (this.isSegmentableKind(node.kind)) this.collectNameSegmentRows(node.name, segmentRows); } + this.runBatched( + 'insertNodes', + `INSERT OR REPLACE INTO nodes ( + id, kind, name, qualified_name, file_path, language, + start_line, end_line, start_column, end_column, + docstring, signature, visibility, + is_exported, is_async, is_static, is_abstract, + decorators, type_parameters, return_type, updated_at + ) VALUES `, + '(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)', + rows + ); + this.runBatched( + 'insertNameSegments', + 'INSERT OR IGNORE INTO name_segment_vocab (segment, name) VALUES ', + '(?,?)', + segmentRows + ); })(); } + /** + * Store one file's whole extraction bundle — nodes, edges, unresolved refs, + * and the file record — in a SINGLE transaction. The bulk-index path calls + * this once per file instead of opening one transaction per table (#1015 + * file-order commit discipline is unchanged: callers still invoke it in file + * order, and row order within is input order). + * + * Edges MUST already be endpoint-filtered by the caller (the store path + * filters to the file's own inserted node ids), so the per-file existence + * SELECT that insertEdges() pays is skipped here. + */ + storeFileBundle(bundle: { + nodes: Node[]; + edges: Edge[]; + refs: UnresolvedReference[]; + file: FileRecord; + }): void { + this.db.transaction(() => { + this.insertNodes(bundle.nodes); + if (bundle.edges.length > 0) { + const rows: unknown[][] = []; + for (const edge of bundle.edges) { + rows.push([ + edge.source, + edge.target, + edge.kind, + edge.metadata ? JSON.stringify(edge.metadata) : null, + edge.line ?? null, + edge.column ?? null, + edge.provenance ?? null, + ]); + } + this.runBatched( + 'insertEdges', + 'INSERT OR IGNORE INTO edges (source, target, kind, metadata, line, col, provenance) VALUES ', + '(?,?,?,?,?,?,?)', + rows + ); + } + if (bundle.refs.length > 0) this.insertUnresolvedRefsBatch(bundle.refs); + this.upsertFile(bundle.file); + })(); + } + + /** + * Collect (segment, name) rows for a name, honouring the same session-dedupe + * semantics as insertNameSegments(). Shared by the bulk write paths. + */ + private collectNameSegmentRows(name: string, out: unknown[][]): void { + if (this.segmentedNames.has(name)) return; + if (this.segmentedNames.size >= QueryBuilder.MAX_SEGMENTED_NAMES) this.segmentedNames.clear(); + this.segmentedNames.add(name); + for (const segment of splitIdentifierSegments(name)) out.push([segment, name]); + } + /** * Update an existing node */ @@ -510,7 +659,14 @@ export class QueryBuilder { /** Insert segments for a batch of names in one transaction (vocab heal path). */ insertNameSegmentsBatch(names: string[]): void { this.db.transaction(() => { - for (const name of names) this.insertNameSegments(name); + const rows: unknown[][] = []; + for (const name of names) this.collectNameSegmentRows(name, rows); + this.runBatched( + 'insertNameSegments', + 'INSERT OR IGNORE INTO name_segment_vocab (segment, name) VALUES ', + '(?,?)', + rows + ); })(); } @@ -1500,12 +1656,27 @@ export class QueryBuilder { } const existingNodeIds = this.getExistingNodeIds([...endpointIds]); + const rows: unknown[][] = []; for (const edge of edges) { if (!existingNodeIds.has(edge.source) || !existingNodeIds.has(edge.target)) { continue; } - this.insertEdge(edge); + rows.push([ + edge.source, + edge.target, + edge.kind, + edge.metadata ? JSON.stringify(edge.metadata) : null, + edge.line ?? null, + edge.column ?? null, + edge.provenance ?? null, + ]); } + this.runBatched( + 'insertEdges', + 'INSERT OR IGNORE INTO edges (source, target, kind, metadata, line, col, provenance) VALUES ', + '(?,?,?,?,?,?,?)', + rows + ); })(); } @@ -1789,9 +1960,25 @@ export class QueryBuilder { insertUnresolvedRefsBatch(refs: UnresolvedReference[]): void { if (refs.length === 0) return; const insert = this.db.transaction(() => { + const rows: unknown[][] = []; for (const ref of refs) { - this.insertUnresolvedRef(ref); + rows.push([ + ref.fromNodeId, + ref.referenceName, + ref.referenceKind, + ref.line, + ref.column, + ref.candidates ? JSON.stringify(ref.candidates) : null, + ref.filePath ?? '', + ref.language ?? 'unknown', + ]); } + this.runBatched( + 'insertUnresolvedRefs', + 'INSERT INTO unresolved_refs (from_node_id, reference_name, reference_kind, line, col, candidates, file_path, language) VALUES ', + '(?,?,?,?,?,?,?,?)', + rows + ); }); insert(); } diff --git a/src/db/sqlite-adapter.ts b/src/db/sqlite-adapter.ts index ab0376c..2ca02e5 100644 --- a/src/db/sqlite-adapter.ts +++ b/src/db/sqlite-adapter.ts @@ -49,11 +49,12 @@ export type SqliteBackend = 'node-sqlite'; */ class NodeSqliteAdapter implements SqliteDatabase { private _db: any; + private _txDepth = 0; - constructor(dbPath: string) { + constructor(dbPath: string, opts?: { readOnly?: boolean }) { // eslint-disable-next-line @typescript-eslint/no-require-imports const { DatabaseSync } = require('node:sqlite'); - this._db = new DatabaseSync(dbPath); + this._db = opts?.readOnly ? new DatabaseSync(dbPath, { readOnly: true }) : new DatabaseSync(dbPath); } get open(): boolean { @@ -108,13 +109,29 @@ class NodeSqliteAdapter implements SqliteDatabase { transaction(fn: (...args: any[]) => T): (...args: any[]) => T { return (...args: any[]) => { + // Nested call (a transaction()-wrapped helper invoked from inside another + // transaction): run the body directly inside the enclosing transaction. + // BEGIN would throw "cannot start a transaction within a transaction", + // so no existing caller ever relied on nested rollback granularity — + // flattening is behavior-preserving and free. + if (this._txDepth > 0) { + this._txDepth++; + try { + return fn(...args); + } finally { + this._txDepth--; + } + } this._db.exec('BEGIN'); + this._txDepth = 1; try { const result = fn(...args); this._db.exec('COMMIT'); + this._txDepth = 0; return result; } catch (error) { this._db.exec('ROLLBACK'); + this._txDepth = 0; throw error; } }; @@ -134,9 +151,9 @@ class NodeSqliteAdapter implements SqliteDatabase { * report it per-instance — MCP can open multiple project DBs in one process, so * a process-global would race. */ -export function createDatabase(dbPath: string): { db: SqliteDatabase; backend: SqliteBackend } { +export function createDatabase(dbPath: string, opts?: { readOnly?: boolean }): { db: SqliteDatabase; backend: SqliteBackend } { try { - return { db: new NodeSqliteAdapter(dbPath), backend: 'node-sqlite' }; + return { db: new NodeSqliteAdapter(dbPath, opts), backend: 'node-sqlite' }; } catch (error) { const msg = error instanceof Error ? error.message : String(error); throw new Error( diff --git a/src/extraction/index.ts b/src/extraction/index.ts index fe80df6..f2e7e04 100644 --- a/src/extraction/index.ts +++ b/src/extraction/index.ts @@ -15,6 +15,7 @@ import { FileRecord, ExtractionResult, ExtractionError, + Node, Edge, UnresolvedReference, ReferenceKind, @@ -22,6 +23,7 @@ import { import { QueryBuilder } from '../db/queries'; import { extractFromSource } from './tree-sitter'; import { ParseWorkerPool, resolveParsePoolSize, resolveParseTimeoutMs } from './parse-pool'; +import { StoreWriter, StoreBundle } from './store-writer'; import { detectLanguage, isSourceFile, isLanguageSupported, isFileLevelOnlyLanguage, initGrammars, loadGrammarsForLanguages, readGrammarWasmBytes } from './grammars'; import { loadExtensionOverrides, loadIncludeIgnoredPatterns, loadExcludePatterns, loadIncludePatterns } from '../project-config'; import { isCodeGraphDataDir } from '../directory'; @@ -1493,7 +1495,13 @@ export class ExtractionOrchestrator { // null in the normal case, or a promise to await (at this safe, // between-transactions boundary) when the WAL has outrun the off-thread // checkpointer past its hard cap. See db/wal-valve.ts. - walBackpressure?: () => Promise | null + walBackpressure?: () => Promise | null, + // Fresh-DB store offload (perf): when set, per-file store bundles are + // applied by a dedicated writer thread instead of the main thread. Only + // passed for a COMPLETELY fresh database, where the main thread performs + // no reads/writes during the parse loop, so one writer applying bundles + // in file order preserves the #1015 determinism exactly. + storeWriterOpts?: { dbPath: string; fastInit: boolean } | null ): Promise { await initGrammars(); const startTime = Date.now(); @@ -1604,11 +1612,35 @@ export class ExtractionOrchestrator { grammarBuffers, }); log(`Parse worker pool: ${poolSize} worker(s)`); + // Bulk index: every core will be needed — spawn the whole pool now so + // worker boot overlaps the first read batches instead of trickling in + // behind queue-pressure growth. + pool.prewarm(); } else { // In-process fallback: load grammars locally and parse on the main thread. await loadGrammarsForLanguages(neededLanguages); } + // Dedicated store writer thread (fresh DB only — see the parameter doc). + // Same availability rule as the parse pool: needs the compiled worker + // (absent when running from source in tests → main-thread fallback). + const storeWorkerPath = path.join(__dirname, 'store-worker.js'); + let storeWriter: StoreWriter | null = null; + if ( + storeWriterOpts && + process.env.CODEGRAPH_NO_STORE_WORKER !== '1' && + fs.existsSync(storeWorkerPath) + ) { + // Deliberately NOT awaiting ready(): worker_threads delivers messages in + // order, so bundles posted while the worker is still booting queue + // behind 'open'. A boot failure surfaces at the first drain() — same + // propagation point as a store error. + storeWriter = new StoreWriter(storeWorkerPath, storeWriterOpts.dbPath, storeWriterOpts.fastInit); + log('Store writer thread active'); + } + /** Queue-depth bound for un-acked bundles (bundles hold whole node/edge arrays). */ + const STORE_WRITER_WINDOW = 64; + /** * Parse one file: on the pool when available (the promise REJECTS on a worker * crash/timeout — the caller records it and the retry pass re-attempts), or @@ -1655,10 +1687,17 @@ export class ExtractionOrchestrator { const bp = walBackpressure?.(); if (bp) await bp; - // Store in database on main thread (SQLite is not thread-safe) + // Store: on the writer thread when active (fresh DB — bundles applied + // in the same file order this chain dispatches them), else on the main + // thread (SQLite connections are per-thread). if (result.nodes.length > 0 || result.errors.length === 0) { const language = detectLanguage(filePath, content, overrides); - await this.storeExtractionResult(filePath, content, language, stats, result, commitYield); + if (storeWriter) { + storeWriter.send(this.buildFreshStoreBundle(filePath, content, language, stats, result)); + await storeWriter.waitBelow(STORE_WRITER_WINDOW); + } else { + await this.storeExtractionResult(filePath, content, language, stats, result, commitYield); + } } if (result.errors.length > 0) { @@ -1835,10 +1874,25 @@ export class ExtractionOrchestrator { if (!aborted) { await Promise.all(inFlight); await flushOrdered(); - if (flushError) throw flushError; + if (flushError) { + if (storeWriter) await storeWriter.close(); + throw flushError; + } + // All bundles are posted; wait for the writer to apply them, then close + // its connection BEFORE any main-thread DB work below (retry pass, + // resolution) so exactly one connection writes at a time. + if (storeWriter) { + try { + await storeWriter.drain(); + } finally { + await storeWriter.close(); + storeWriter = null; + } + } } if (signal?.aborted || aborted) { + if (storeWriter) await storeWriter.close(); if (pool) await pool.destroy(); return { success: false, @@ -2203,6 +2257,49 @@ export class ExtractionOrchestrator { // This prevents FK violations when edges reference nodes that would // be silently skipped by insertNode() (see issue #42). const validNodes = result.nodes.filter((n) => n.id && n.kind && n.name && n.filePath && n.language); + const insertedIds = new Set(validNodes.map((n) => n.id)); + const validEdges = result.edges.filter( + (e) => insertedIds.has(e.source) && insertedIds.has(e.target) + ); + const validRefs = result.unresolvedReferences + .filter((ref) => insertedIds.has(ref.fromNodeId)) + .map((ref) => ({ + ...ref, + filePath: ref.filePath ?? filePath, + language: ref.language ?? language, + })); + + // Fast path for the common case (everything fits one chunk): the whole + // file — nodes, edges, refs, file record — lands in ONE transaction with + // no event-loop yields in between. Giant generated files keep the chunked + // + yielding path below so the #850 watchdog heartbeat stays serviced. + const fitsOneChunk = + validNodes.length <= STORE_CHUNK && + validEdges.length <= STORE_CHUNK && + validRefs.length <= STORE_CHUNK; + if (fitsOneChunk) { + // Snapshot/re-resolution of cross-file incoming edges (below) still runs + // for the sync path; on a fresh bulk index crossFileIncomingEdges is []. + this.queries.storeFileBundle({ + nodes: validNodes, + edges: validEdges, + refs: validRefs, + file: { + path: filePath, + contentHash, + language, + size: stats.size, + modifiedAt: stats.mtimeMs, + indexedAt: Date.now(), + nodeCount: result.nodes.length, + errors: result.errors.length > 0 ? result.errors : undefined, + }, + }); + if (crossFileIncomingEdges.length > 0) { + this.reattachCrossFileEdges(crossFileIncomingEdges, validNodes); + } + return; + } // Insert nodes (chunked — see STORE_CHUNK above) for (let i = 0; i < validNodes.length; i += STORE_CHUNK) { @@ -2211,11 +2308,7 @@ export class ExtractionOrchestrator { } // Filter edges to only reference nodes that were actually inserted - if (result.edges.length > 0) { - const insertedIds = new Set(validNodes.map((n) => n.id)); - const validEdges = result.edges.filter( - (e) => insertedIds.has(e.source) && insertedIds.has(e.target) - ); + if (validEdges.length > 0) { for (let i = 0; i < validEdges.length; i += STORE_CHUNK) { this.queries.insertEdges(validEdges.slice(i, i + STORE_CHUNK)); await onYield?.(); @@ -2241,43 +2334,13 @@ export class ExtractionOrchestrator { // a ref from the target's plain name would strip receiver/qualifier // context and risk a rebind a full re-index would never make. if (crossFileIncomingEdges.length > 0) { - const newNodesByKindName = new Map(); - for (const n of validNodes) { - newNodesByKindName.set(`${n.kind}\0${n.name}`, n.id); - } - const reinserted: Edge[] = []; - const resurrected: UnresolvedReference[] = []; - for (const e of crossFileIncomingEdges) { - const newTargetId = newNodesByKindName.get(`${e.targetKind}\0${e.targetName}`); - if (newTargetId) { - reinserted.push({ source: e.source, target: newTargetId, kind: e.kind, metadata: e.metadata, line: e.line, column: e.column, provenance: e.provenance }); - } else { - const ref = resurrectRefFromDroppedEdge(e); - if (ref) resurrected.push(ref); - } - } - if (reinserted.length > 0) { - this.queries.insertEdges(reinserted); - } - if (resurrected.length > 0) { - this.queries.insertUnresolvedRefsBatch(resurrected); - } + this.reattachCrossFileEdges(crossFileIncomingEdges, validNodes); } // Insert unresolved references in batch with denormalized filePath/language - if (result.unresolvedReferences.length > 0) { - const insertedIds = new Set(validNodes.map((n) => n.id)); - const refsWithContext = result.unresolvedReferences - .filter((ref) => insertedIds.has(ref.fromNodeId)) - .map((ref) => ({ - ...ref, - filePath: ref.filePath ?? filePath, - language: ref.language ?? language, - })); - for (let i = 0; i < refsWithContext.length; i += STORE_CHUNK) { - this.queries.insertUnresolvedRefsBatch(refsWithContext.slice(i, i + STORE_CHUNK)); - await onYield?.(); - } + for (let i = 0; i < validRefs.length; i += STORE_CHUNK) { + this.queries.insertUnresolvedRefsBatch(validRefs.slice(i, i + STORE_CHUNK)); + await onYield?.(); } // Insert file record @@ -2294,6 +2357,81 @@ export class ExtractionOrchestrator { this.queries.upsertFile(fileRecord); } + /** + * Build one file's store bundle for the FRESH-DB path: no existing-file + * check, no cross-file edge snapshot (both are re-index concerns — a fresh + * database has neither). Filters mirror storeExtractionResult exactly. + */ + private buildFreshStoreBundle( + filePath: string, + content: string, + language: Language, + stats: fs.Stats, + result: ExtractionResult + ): StoreBundle { + const validNodes = result.nodes.filter((n) => n.id && n.kind && n.name && n.filePath && n.language); + const insertedIds = new Set(validNodes.map((n) => n.id)); + const validEdges = result.edges.filter( + (e) => insertedIds.has(e.source) && insertedIds.has(e.target) + ); + const validRefs = result.unresolvedReferences + .filter((ref) => insertedIds.has(ref.fromNodeId)) + .map((ref) => ({ + ...ref, + filePath: ref.filePath ?? filePath, + language: ref.language ?? language, + })); + return { + nodes: validNodes, + edges: validEdges, + refs: validRefs, + file: { + path: filePath, + contentHash: hashContent(content), + language, + size: stats.size, + modifiedAt: stats.mtimeMs, + indexedAt: Date.now(), + nodeCount: result.nodes.length, + errors: result.errors.length > 0 ? result.errors : undefined, + }, + }; + } + + /** + * Re-attach cross-file incoming edges snapshotted before a re-index delete + * (#899): re-resolve each edge's target to the re-indexed node's new id by + * (kind, name); targets that vanished are resurrected as their original + * unresolved ref (#1240's removal-side counterpart) when the edge carries + * its refName stamp. + */ + private reattachCrossFileEdges( + crossFileIncomingEdges: Array, + validNodes: Node[] + ): void { + const newNodesByKindName = new Map(); + for (const n of validNodes) { + newNodesByKindName.set(`${n.kind}\0${n.name}`, n.id); + } + const reinserted: Edge[] = []; + const resurrected: UnresolvedReference[] = []; + for (const e of crossFileIncomingEdges) { + const newTargetId = newNodesByKindName.get(`${e.targetKind}\0${e.targetName}`); + if (newTargetId) { + reinserted.push({ source: e.source, target: newTargetId, kind: e.kind, metadata: e.metadata, line: e.line, column: e.column, provenance: e.provenance }); + } else { + const ref = resurrectRefFromDroppedEdge(e); + if (ref) resurrected.push(ref); + } + } + if (reinserted.length > 0) { + this.queries.insertEdges(reinserted); + } + if (resurrected.length > 0) { + this.queries.insertUnresolvedRefsBatch(resurrected); + } + } + /** * Sync the index with the current file state. * diff --git a/src/extraction/parse-pool.ts b/src/extraction/parse-pool.ts index c4a80f2..26f8ca0 100644 --- a/src/extraction/parse-pool.ts +++ b/src/extraction/parse-pool.ts @@ -208,6 +208,21 @@ export class ParseWorkerPool { this.spawnOne(); // one eager warm worker, ready for the first parse } + /** + * Spawn the whole pool up front. The default demand-driven growth avoids + * paying worker boot for small jobs, but a bulk index KNOWS every core will + * be needed — on a fast repo the one-by-one ramp-up otherwise consumes most + * of the parse phase (each worker boot is a fresh Node isolate + grammar + * load, ~hundreds of ms, and growth only triggers as queue pressure builds). + */ + prewarm(): void { + while (this.workers.size < this.maxSize) { + const before = this.workers.size; + this.spawnOne(); + if (this.workers.size === before) break; // spawn failed / breaker tripped + } + } + /** Pool size cap (for logging). */ get size(): number { return this.maxSize; } diff --git a/src/extraction/parse-worker.ts b/src/extraction/parse-worker.ts index 41022cd..d086e32 100644 --- a/src/extraction/parse-worker.ts +++ b/src/extraction/parse-worker.ts @@ -5,6 +5,14 @@ * stays unblocked and the UI animation renders smoothly. */ +// Compile cache FIRST: the worker's boot cost is dominated by re-requiring +// the extraction module graph; the persistent V8 cache (Node ≥22.8) makes +// that a bytecode load instead of a recompile. Safe no-op when unavailable. +try { + // eslint-disable-next-line @typescript-eslint/no-require-imports + (require('node:module') as { enableCompileCache?: () => void }).enableCompileCache?.(); +} catch { /* cache is best-effort */ } + import { parentPort } from 'worker_threads'; import { extractFromSource } from './tree-sitter'; import { detectLanguage, loadGrammarsForLanguages, resetParser } from './grammars'; diff --git a/src/extraction/store-worker.ts b/src/extraction/store-worker.ts new file mode 100644 index 0000000..74b5a1d --- /dev/null +++ b/src/extraction/store-worker.ts @@ -0,0 +1,99 @@ +/** + * Store worker — dedicated writer thread for the bulk-index store phase. + * + * During a fresh full index the main thread's biggest serial cost is executing + * the per-file INSERT batches. This worker owns that work on its own SQLite + * connection: the orchestrator posts one message per file (in file order) and + * the worker applies them in arrival order, which preserves the #1015 + * insertion-order determinism exactly as if the main thread had run the same + * calls. The main thread performs NO database access while the writer is + * active (fresh-DB path only), so there is no cross-connection contention. + * + * Protocol (main → worker): + * {type:'open', dbPath, fastInit} → open connection, reply {type:'ready'} + * {type:'bundle', bundle} → apply one file's store bundle + * {type:'drain', id} → reply {type:'drained', id} (in-order ⇒ all prior bundles applied) + * {type:'close'} → close DB and exit + * Worker → main: {type:'ready'} | {type:'drained', id} | {type:'error', message} + * + * A bundle failure does not kill the worker; the first error is reported and + * the client surfaces it at drain(), matching the main-thread path where a + * store exception propagates out of the ordered-flush chain. + */ + +// Compile cache FIRST — same worker-boot rationale as parse-worker.ts. +try { + // eslint-disable-next-line @typescript-eslint/no-require-imports + (require('node:module') as { enableCompileCache?: () => void }).enableCompileCache?.(); +} catch { /* cache is best-effort */ } + +import { parentPort } from 'worker_threads'; +import { QueryBuilder } from '../db/queries'; +import { createDatabase, SqliteDatabase } from '../db/sqlite-adapter'; +import type { StoreBundle } from './store-writer'; + +if (!parentPort) { + throw new Error('store-worker must be run as a worker thread'); +} +const port = parentPort; + +let db: SqliteDatabase | null = null; +let queries: QueryBuilder | null = null; + +type InMessage = + | { type: 'open'; dbPath: string; fastInit: boolean } + | { type: 'bundle'; bundle: StoreBundle } + | { type: 'drain'; id: number } + | { type: 'close' }; + +port.on('message', (msg: InMessage) => { + try { + switch (msg.type) { + case 'open': { + const created = createDatabase(msg.dbPath); + db = created.db; + // Mirrors db/index.ts configureConnection, with the same fast-init + // durability trade the main connection applies for fresh builds. + db.pragma('busy_timeout = 5000'); + db.pragma('foreign_keys = ON'); + if (msg.fastInit) { + db.pragma('journal_mode = MEMORY'); + db.pragma('synchronous = OFF'); + } else { + db.pragma('synchronous = NORMAL'); + } + db.pragma('cache_size = -64000'); + db.pragma('temp_store = MEMORY'); + queries = new QueryBuilder(db); + port.postMessage({ type: 'ready' }); + break; + } + case 'bundle': { + if (!queries) throw new Error('store-worker: bundle before open'); + queries.storeFileBundle(msg.bundle); + port.postMessage({ type: 'ack' }); + break; + } + case 'drain': { + port.postMessage({ type: 'drained', id: msg.id }); + break; + } + case 'close': { + try { + db?.close(); + } catch { + /* already closed */ + } + process.exit(0); + break; + } + } + } catch (err) { + // The error reply doubles as the bundle's ack so the client's outstanding + // counter still drains after a failure. + port.postMessage({ + type: 'error', + message: err instanceof Error ? err.message : String(err), + }); + } +}); diff --git a/src/extraction/store-writer.ts b/src/extraction/store-writer.ts new file mode 100644 index 0000000..f0372d8 --- /dev/null +++ b/src/extraction/store-writer.ts @@ -0,0 +1,146 @@ +/** + * StoreWriter — main-thread client for the store worker (see store-worker.ts). + * + * Used ONLY on the fresh-DB bulk path: bundles are posted in file order and the + * worker applies them in arrival order, so rowid assignment (and therefore + * resolution's insertion-order disambiguation) is byte-identical to the + * main-thread store. Kill switch: CODEGRAPH_NO_STORE_WORKER=1. + */ + +import { Worker } from 'worker_threads'; +import { Node, Edge, UnresolvedReference, FileRecord } from '../types'; + +/** One file's complete store payload (pre-filtered — see storeFileBundle). */ +export interface StoreBundle { + nodes: Node[]; + edges: Edge[]; + refs: UnresolvedReference[]; + file: FileRecord; +} + +export class StoreWriter { + private worker: Worker; + private readyPromise: Promise; + private firstError: Error | null = null; + private drainWaiters = new Map void; reject: (e: Error) => void }>(); + private nextDrainId = 0; + private exited = false; + /** Bundles posted but not yet acked — the queue-depth backpressure signal. */ + private outstanding = 0; + private belowWaiters: Array<{ limit: number; resolve: () => void }> = []; + + constructor(workerScriptPath: string, dbPath: string, fastInit: boolean) { + this.worker = new Worker(workerScriptPath); + let readyResolve!: () => void; + let readyReject!: (e: Error) => void; + this.readyPromise = new Promise((resolve, reject) => { + readyResolve = resolve; + readyReject = reject; + }); + + this.worker.on('message', (msg: { type: string; id?: number; message?: string }) => { + if (msg.type === 'ready') { + readyResolve(); + } else if (msg.type === 'ack') { + this.settleOne(); + } else if (msg.type === 'drained' && msg.id !== undefined) { + const waiter = this.drainWaiters.get(msg.id); + this.drainWaiters.delete(msg.id); + if (!waiter) return; + if (this.firstError) waiter.reject(this.firstError); + else waiter.resolve(); + } else if (msg.type === 'error') { + if (!this.firstError) this.firstError = new Error(`store worker: ${msg.message}`); + this.settleOne(); // the error reply is also the failed bundle's ack + } + }); + this.worker.on('error', (err) => { + this.failAll(err instanceof Error ? err : new Error(String(err))); + readyReject(this.firstError!); + }); + this.worker.on('exit', (code) => { + this.exited = true; + if (code !== 0) { + this.failAll(new Error(`store worker exited with code ${code}`)); + readyReject(this.firstError!); + } else if (this.drainWaiters.size > 0 || this.belowWaiters.length > 0) { + // A clean exit with waiters pending is a protocol violation (only + // close() should end the worker) — settle the waiters instead of + // hanging the index forever. + this.failAll(new Error('store worker exited before drain completed')); + } + }); + + this.worker.postMessage({ type: 'open', dbPath, fastInit }); + // The worker holds the event loop open only until close(); don't unref — + // bundles must never be dropped because main ran out of work. + } + + private failAll(err: Error): void { + if (!this.firstError) this.firstError = err; + for (const [, waiter] of this.drainWaiters) waiter.reject(this.firstError); + this.drainWaiters.clear(); + this.outstanding = 0; + const waiters = this.belowWaiters; + this.belowWaiters = []; + for (const w of waiters) w.resolve(); // send() will surface firstError + } + + private settleOne(): void { + if (this.outstanding > 0) this.outstanding--; + if (this.belowWaiters.length === 0) return; + const still: typeof this.belowWaiters = []; + for (const w of this.belowWaiters) { + if (this.outstanding < w.limit) w.resolve(); + else still.push(w); + } + this.belowWaiters = still; + } + + ready(): Promise { + return this.readyPromise; + } + + /** Post one file's bundle. Throws immediately if the writer already failed. */ + send(bundle: StoreBundle): void { + if (this.firstError) throw this.firstError; + if (this.exited) throw new Error('store worker already exited'); + this.outstanding++; + this.worker.postMessage({ type: 'bundle', bundle }); + } + + /** Backpressure: resolves once fewer than `limit` bundles are un-acked. */ + waitBelow(limit: number): Promise { + if (this.firstError || this.exited || this.outstanding < limit) return Promise.resolve(); + return new Promise((resolve) => { + this.belowWaiters.push({ limit, resolve }); + }); + } + + /** Resolves when every bundle posted before this call has been applied. */ + drain(): Promise { + if (this.firstError) return Promise.reject(this.firstError); + if (this.exited) return Promise.reject(new Error('store worker already exited')); + const id = this.nextDrainId++; + const p = new Promise((resolve, reject) => { + this.drainWaiters.set(id, { resolve, reject }); + }); + this.worker.postMessage({ type: 'drain', id }); + return p; + } + + /** Close the worker's DB connection and join the thread. */ + async close(): Promise { + if (this.exited) return; + this.worker.postMessage({ type: 'close' }); + await new Promise((resolve) => { + const t = setTimeout(() => { + void this.worker.terminate().then(() => resolve()); + }, 5000); + this.worker.once('exit', () => { + clearTimeout(t); + resolve(); + }); + }); + } +} diff --git a/src/index.ts b/src/index.ts index b60205d..3b02441 100644 --- a/src/index.ts +++ b/src/index.ts @@ -55,6 +55,7 @@ import { deriveProjectNameTokens } from './search/query-utils'; import { CodeGraphPackageVersion } from './mcp/version'; import { segmentLookupVariants, splitIdentifierSegments } from './search/identifier-segments'; import { createYielder } from './resolution/cooperative-yield'; +import { minRefsForPool } from './resolution/resolver-pool'; // Re-export types for consumers export * from './types'; @@ -445,7 +446,21 @@ export class CodeGraph { // the final fold-up before the interval is restored in the finally. // Kill switch: CODEGRAPH_NO_WAL_DEFER=1. Non-WAL journal modes (some // network filesystems) have no WAL to defer — skip. - const deferWal = process.env.CODEGRAPH_NO_WAL_DEFER !== '1' && this.db.getJournalMode() === 'wal'; + // Fast-init: on a COMPLETELY fresh DB, trade crash-durability for speed + // during the bulk build (journal in memory, no fsync). Safe because the + // DB is disposable until the index completes — index_state stays + // 'indexing' and a crashed init is re-run from scratch; existing DBs + // (re-index/sync) never take this path. Kill switch: + // CODEGRAPH_NO_FAST_INIT=1 (same pattern as CODEGRAPH_NO_WAL_DEFER). + const freshDb = this.queries.getNodeAndEdgeCount().nodes === 0; + const fastInit = process.env.CODEGRAPH_NO_FAST_INIT !== '1' && freshDb; + if (fastInit) { + try { + this.db.getDb().pragma('journal_mode = MEMORY'); + this.db.getDb().pragma('synchronous = OFF'); + } catch { /* keep WAL */ } + } + const deferWal = !fastInit && process.env.CODEGRAPH_NO_WAL_DEFER !== '1' && this.db.getJournalMode() === 'wal'; let walValve: WalCheckpointValve | null = null; let priorAutocheckpoint = 1000; if (deferWal) { @@ -470,12 +485,25 @@ export class CodeGraph { // path as every file (re-)indexes below — so a full index is also the // orphan-cleanup pass for names deleted since the last one. try { this.queries.clearNameSegmentVocab(); } catch { /* vocab is advisory — never fail an index over it */ } - const result = await this.orchestrator.indexAll( - options.onProgress, - options.signal, - options.verbose, - walValve ? () => walValve!.backpressure() : undefined - ); + // Bulk FTS mode for the mass-insert phase: drop the per-row FTS sync + // triggers, rebuild nodes_fts once from the nodes table afterwards. + // Crash inside the window is healed on the next DatabaseConnection.open. + this.db.beginBulkNodeLoad(); + let result: IndexResult; + try { + result = await this.orchestrator.indexAll( + options.onProgress, + options.signal, + options.verbose, + walValve ? () => walValve!.backpressure() : undefined, + // Store-writer offload is fresh-DB-only: with any pre-existing + // data the store path must read (existing-file checks, cross-file + // edge snapshots) and delete, which belongs on one thread. + freshDb ? { dbPath: this.db.getPath(), fastInit } : null + ); + } finally { + this.db.endBulkNodeLoad(); + } // Fold the parse phase's WAL BEFORE the first post-parse reads // (resolver re-init and resolution both read on the main thread): @@ -503,6 +531,19 @@ export class CodeGraph { // Get count without loading all refs into memory const unresolvedCount = this.queries.getUnresolvedReferencesCount(); + // Fast-init leaves the DB in memory-journal (rollback) mode, where + // the parallel resolver pool's read connections would contend with + // the main writer's exclusive commits. When the pool will actually + // run (enough pending refs), restore WAL BEFORE resolution so + // readers never block the writer; otherwise stay in the fast mode + // until the finally — sequential resolution has no readers. + if (fastInit && unresolvedCount >= minRefsForPool()) { + try { + this.db.getDb().pragma('synchronous = NORMAL'); + this.db.getDb().pragma('journal_mode = WAL'); + } catch { /* keep current mode; resolution still works sequentially */ } + } + options.onProgress?.({ phase: 'resolving', current: 0, @@ -619,6 +660,14 @@ export class CodeGraph { if (deferWal) { try { this.db.setWalAutocheckpoint(priorAutocheckpoint); } catch { /* connection may be closing */ } } + if (fastInit) { + // Back to the durable defaults; journal_mode=WAL folds the MEMORY + // journal state into a normal WAL-mode database file. + try { + this.db.getDb().pragma('synchronous = NORMAL'); + this.db.getDb().pragma('journal_mode = WAL'); + } catch { /* connection may be closing */ } + } this.fileLock.release(); } }); @@ -1032,7 +1081,9 @@ export class CodeGraph { onProgress?: (current: number, total: number) => void, onSynthesisProgress?: (done: number, total: number) => void ): Promise { - return this.resolver.resolveAndPersistBatched(onProgress, undefined, onSynthesisProgress); + return this.resolver.resolveAndPersistBatched(onProgress, undefined, onSynthesisProgress, { + dbPath: this.db.getPath(), + }); } /** diff --git a/src/resolution/cooperative-yield.ts b/src/resolution/cooperative-yield.ts index 653b2e2..d8c6166 100644 --- a/src/resolution/cooperative-yield.ts +++ b/src/resolution/cooperative-yield.ts @@ -24,8 +24,16 @@ * stop killing work that is demonstrably making progress. */ -/** Yield when more than `budgetMs` of wall-clock has passed since the last yield. */ -export type MaybeYield = () => Promise; +/** + * Yield when more than `budgetMs` of wall-clock has passed since the last + * yield. Returns `undefined` on the (overwhelmingly common) not-due path so a + * hot loop can skip the await entirely — `await`ing an async no-op costs a + * promise allocation + microtask hop, which at hundreds of thousands of calls + * per index is real time. Callers may either `await maybeYield()` (works for + * both return shapes) or use the fast form: + * `const y = maybeYield(); if (y) await y;` + */ +export type MaybeYield = () => Promise | undefined; /** Default budget: well under the watchdog's minimum heartbeat cadence (~1s), so * a heartbeat byte always has a chance to land between yields. */ @@ -33,9 +41,13 @@ export const DEFAULT_YIELD_BUDGET_MS = 250; export function createYielder(budgetMs: number = DEFAULT_YIELD_BUDGET_MS): MaybeYield { let last = Date.now(); - return async function maybeYield(): Promise { - if (Date.now() - last < budgetMs) return; - await new Promise((resolve) => setImmediate(resolve)); - last = Date.now(); + return function maybeYield(): Promise | undefined { + if (Date.now() - last < budgetMs) return undefined; + return new Promise((resolve) => + setImmediate(() => { + last = Date.now(); + resolve(); + }) + ); }; } diff --git a/src/resolution/import-resolver.ts b/src/resolution/import-resolver.ts index bbb1303..d11078c 100644 --- a/src/resolution/import-resolver.ts +++ b/src/resolution/import-resolver.ts @@ -55,11 +55,78 @@ export function isNixPathImportRef(ref: UnresolvedRef): boolean { /** * Resolve an import path to an actual file */ +// Per-context memos for the two hottest pure lookups on the resolution path: +// import-specifier → file resolution and exported-symbol lookup. Both are pure +// given a stable file set + node table, which is exactly the window between +// ReferenceResolver.clearCaches() calls — clearImportResolverMemos() is invoked +// there, so the staleness discipline matches the resolver's own caches. +const importPathMemos = new WeakMap>(); +const exportedSymbolMemos = new WeakMap>(); + +/** + * Per-file index of exported symbols, replacing repeated linear `.find`s over + * `getNodesInFile` arrays (a barrel-heavy repo scans its biggest files once + * per referencing symbol otherwise). First-wins insertion preserves exactly + * the array-order semantics of the `.find` calls it replaces. + */ +interface FileExportIndex { + byName: Map; + defaultComponent: Node | undefined; + defaultFnClass: Node | undefined; +} +const fileExportIndexes = new WeakMap>(); + +function getFileExportIndex(filePath: string, context: ResolutionContext): FileExportIndex { + let perFile = fileExportIndexes.get(context); + if (!perFile) { + perFile = new Map(); + fileExportIndexes.set(context, perFile); + } + let idx = perFile.get(filePath); + if (!idx) { + idx = { byName: new Map(), defaultComponent: undefined, defaultFnClass: undefined }; + for (const n of context.getNodesInFile(filePath)) { + if (!n.isExported) continue; + if (!idx.byName.has(n.name)) idx.byName.set(n.name, n); + if (idx.defaultComponent === undefined && n.kind === 'component') idx.defaultComponent = n; + if (idx.defaultFnClass === undefined && (n.kind === 'function' || n.kind === 'class')) idx.defaultFnClass = n; + } + perFile.set(filePath, idx); + } + return idx; +} + +/** Drop the per-context memo tables (see ReferenceResolver.clearCaches). */ +export function clearImportResolverMemos(context: ResolutionContext): void { + importPathMemos.delete(context); + exportedSymbolMemos.delete(context); + fileExportIndexes.delete(context); +} + export function resolveImportPath( importPath: string, fromFile: string, language: Language, context: ResolutionContext +): string | null { + let memo = importPathMemos.get(context); + if (!memo) { + memo = new Map(); + importPathMemos.set(context, memo); + } + const key = `${language}\0${fromFile}\0${importPath}`; + const hit = memo.get(key); + if (hit !== undefined || memo.has(key)) return hit ?? null; + const resolved = resolveImportPathUncached(importPath, fromFile, language, context); + memo.set(key, resolved); + return resolved; +} + +function resolveImportPathUncached( + importPath: string, + fromFile: string, + language: Language, + context: ResolutionContext ): string | null { // COBOL COPY/EXEC SQL INCLUDE names a copybook member, not a path — the // compiler searches a library, so we match against indexed file basenames. @@ -1972,12 +2039,45 @@ function findExportedSymbol( context: ResolutionContext, visited: Set, depth = 0 +): Node | undefined { + // Memoize fresh (top-level) lookups only: recursive re-export steps carry a + // populated `visited` set, whose contents change the reachable answer. + // Every ref to the same imported symbol repeats this exact walk, so the + // top-level memo removes the re-export chase + per-file linear scans from + // all but the first occurrence. + if (depth === 0 && visited.size === 0) { + let memo = exportedSymbolMemos.get(context); + if (!memo) { + memo = new Map(); + exportedSymbolMemos.set(context, memo); + } + const key = `${filePath}\0${want.isDefault ? 1 : 0}${want.isNamespace ? 1 : 0}\0${want.exportedName}\0${want.memberName ?? ''}\0${language}`; + if (memo.has(key)) return memo.get(key); + const result = findExportedSymbolWalk(filePath, want, language, context, visited, depth); + memo.set(key, result); + return result; + } + return findExportedSymbolWalk(filePath, want, language, context, visited, depth); +} + +function findExportedSymbolWalk( + filePath: string, + want: { + isDefault: boolean; + isNamespace: boolean; + exportedName: string; + memberName: string | null; + }, + language: Language, + context: ResolutionContext, + visited: Set, + depth: number ): Node | undefined { if (depth > REEXPORT_MAX_DEPTH) return undefined; if (visited.has(filePath)) return undefined; visited.add(filePath); - const nodesInFile = context.getNodesInFile(filePath); + const exportIndex = getFileExportIndex(filePath, context); // 1. Direct hit: the symbol is declared in this file. if (want.isDefault) { @@ -1987,21 +2087,13 @@ function findExportedSymbol( // `.ts`/`.tsx` `export default fn`/`class` case. Without the component // branch, an `export { default as X } from './X.svelte'` barrel never // resolves and the component shows a false 0 callers (#629). - const direct = - nodesInFile.find((n) => n.isExported && n.kind === 'component') ?? - nodesInFile.find( - (n) => n.isExported && (n.kind === 'function' || n.kind === 'class') - ); + const direct = exportIndex.defaultComponent ?? exportIndex.defaultFnClass; if (direct) return direct; } else if (want.isNamespace && want.memberName) { - const direct = nodesInFile.find( - (n) => n.name === want.memberName && n.isExported - ); + const direct = exportIndex.byName.get(want.memberName); if (direct) return direct; } else { - const direct = nodesInFile.find( - (n) => n.name === want.exportedName && n.isExported - ); + const direct = exportIndex.byName.get(want.exportedName); if (direct) return direct; } diff --git a/src/resolution/index.ts b/src/resolution/index.ts index e304744..dce4006 100644 --- a/src/resolution/index.ts +++ b/src/resolution/index.ts @@ -17,7 +17,8 @@ import { ImportMapping, } from './types'; import { matchReference, matchFunctionRef, matchDottedCallChain, matchScopedCallChain, matchMethodCall, sameLanguageFamily, crossesKnownFamily } from './name-matcher'; -import { resolveViaImport, resolveJvmImport, extractImportMappings, extractReExports, loadCppIncludeDirs, isPhpIncludePathRef, isCobolCopybookRef, isNixPathImportRef } from './import-resolver'; +import { resolveViaImport, resolveJvmImport, extractImportMappings, extractReExports, loadCppIncludeDirs, isPhpIncludePathRef, isCobolCopybookRef, isNixPathImportRef, clearImportResolverMemos } from './import-resolver'; +import { ResolverPool, minRefsForPool } from './resolver-pool'; import { detectFrameworks } from './frameworks'; import { synthesizeCallbackEdges } from './callback-synthesizer'; import { createYielder, type MaybeYield } from './cooperative-yield'; @@ -372,6 +373,9 @@ export class ReferenceResolver { this.knownNames = null; this.knownFiles = null; this.cachesWarmed = false; + // The import-resolver's per-context memos assume the same stable window + // as the caches above — drop them together. + if (this.context) clearImportResolverMemos(this.context); } /** `readFile` through the LRU content cache (null = read failed, also cached). */ @@ -1236,7 +1240,10 @@ export class ReferenceResolver { } else { unresolved.push(ref); } - await maybeYield(); + // Fast-path the per-ref yield check: awaiting the async no-op costs a + // microtask hop per ref, which dominates at ~10⁵ refs (see MaybeYield). + const y = maybeYield(); + if (y) await y; } return { @@ -1251,6 +1258,64 @@ export class ReferenceResolver { }; } + /** + * Resolve a list of refs and return everything the ADMISSION side needs to + * persist the outcome: resolutions, failures, the deferred post-pass refs + * this run produced (drained, so the caller owns routing them), and stats. + * This is the resolver-worker entry point — it runs the exact per-ref loop + * of resolveBatchYielding, minus the main-thread yields (worker threads have + * no watchdog heartbeat to starve). Results are in input order. + */ + resolveListForAdmission(refs: UnresolvedReference[]): { + resolved: ResolvedRef[]; + unresolved: UnresolvedRef[]; + deferredChain: UnresolvedRef[]; + deferredThisMember: UnresolvedRef[]; + byMethod: Record; + } { + this.warmCaches(); + const resolved: ResolvedRef[] = []; + const unresolved: UnresolvedRef[] = []; + const byMethod: Record = {}; + for (const raw of refs) { + const ref: UnresolvedRef = { + fromNodeId: raw.fromNodeId, + referenceName: raw.referenceName, + referenceKind: raw.referenceKind, + line: raw.line, + column: raw.column, + filePath: raw.filePath || this.getFilePathFromNodeId(raw.fromNodeId), + language: raw.language || this.getLanguageFromNodeId(raw.fromNodeId), + rowId: raw.rowId, + }; + const result = this.resolveOne(ref); + if (result) { + resolved.push(result); + byMethod[result.resolvedBy] = (byMethod[result.resolvedBy] || 0) + 1; + } else { + unresolved.push(ref); + } + } + return { + resolved, + unresolved, + deferredChain: this.deferredChainRefs.splice(0), + deferredThisMember: this.deferredThisMemberRefs.splice(0), + byMethod, + }; + } + + /** + * Re-queue deferred post-pass refs produced by resolver workers, preserving + * their admission order so resolveChainedCallsViaConformance / + * resolveDeferredThisMemberRefs process them exactly as the sequential path + * would have. + */ + appendDeferredFromWorkers(deferredChain: UnresolvedRef[], deferredThisMember: UnresolvedRef[]): void { + this.deferredChainRefs.push(...deferredChain); + this.deferredThisMemberRefs.push(...deferredThisMember); + } + /** * Resolve and persist in batches to keep memory bounded. * Processes unresolved references in chunks, persisting edges and cleaning @@ -1259,7 +1324,12 @@ export class ReferenceResolver { async resolveAndPersistBatched( onProgress?: (current: number, total: number) => void, batchSize: number = 5000, - onSynthesisProgress?: (done: number, total: number) => void + onSynthesisProgress?: (done: number, total: number) => void, + // When provided, big batches fan out across a read-only resolver-worker + // pool with results admitted in canonical order (see resolver-pool.ts). + // Sequential fallback on any pool failure. CODEGRAPH_NO_PARALLEL_RESOLVE=1 + // disables entirely. + parallel?: { dbPath: string } ): 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 @@ -1280,16 +1350,59 @@ export class ReferenceResolver { byMethod: {} as Record, }; + // Parallel pool, started immediately but never awaited up front: early + // batches run sequentially while the workers boot (module load + readonly + // DB open + framework detect + cache warm ≈ hundreds of ms), and the loop + // switches to fan-out the moment the pool reports ready — so pool boot + // costs zero wall-clock. Any failure downgrades to sequential permanently. + let pool: ResolverPool | null = null; + let poolReady = false; + if (parallel && total >= minRefsForPool()) { + pool = ResolverPool.tryCreate(parallel.dbPath, this.projectRoot); + pool?.ready().then( + () => { poolReady = true; }, + () => { void pool?.destroy().catch(() => undefined); pool = null; } + ); + } + // Process in batches. We always read from offset 0 because every ref the // batch processed leaves the pending set (resolved rows are deleted, // unresolvable ones flip to status='failed'), shifting the remaining // pending rows forward. let prevRemaining = Number.POSITIVE_INFINITY; + try { while (true) { const batch = this.queries.getUnresolvedReferencesBatch(0, batchSize); if (batch.length === 0) break; - const result = await this.resolveBatchYielding(batch, maybeYield); + let result: ResolutionResult; + if (pool && poolReady && ResolverPool.worthParallel(batch.length)) { + try { + const out = await pool.resolveBatch(batch); + // Deferred post-pass refs ride back from the workers; re-queue them + // in admission order so the post-passes see the sequential order. + this.appendDeferredFromWorkers(out.deferredChain, out.deferredThisMember); + result = { + resolved: out.resolved, + unresolved: out.unresolved, + stats: { + total: batch.length, + resolved: out.resolved.length, + unresolved: out.unresolved.length, + byMethod: out.byMethod, + }, + }; + } catch (err) { + logDebug('Parallel resolution failed; falling back to sequential', { + error: err instanceof Error ? err.message : String(err), + }); + await pool.destroy().catch(() => undefined); + pool = null; + result = await this.resolveBatchYielding(batch, maybeYield); + } + } else { + result = await this.resolveBatchYielding(batch, maybeYield); + } // Persist in bounded sub-transactions with yields between: a whole // batch's edge insert / keyed deletes are otherwise one solid @@ -1370,6 +1483,9 @@ export class ReferenceResolver { if (remaining >= prevRemaining) break; prevRemaining = remaining; } + } finally { + if (pool) await pool.destroy().catch(() => undefined); + } // Dynamic-edge synthesis: now that all base `calls` edges are persisted, // synthesize observer/callback dispatch edges (dispatcher → registered diff --git a/src/resolution/resolver-pool.ts b/src/resolution/resolver-pool.ts new file mode 100644 index 0000000..564348d --- /dev/null +++ b/src/resolution/resolver-pool.ts @@ -0,0 +1,195 @@ +/** + * ResolverPool — main-thread client for the parallel-resolution workers. + * + * resolveBatch() splits a rowid-ordered batch into ordered chunks, fans the + * chunks across the pool, and reassembles the results IN CHUNK ORDER, so the + * caller's admission (edge inserts, row cleanup, failure parking, deferred + * post-pass queues) is byte-for-byte the sequence the single-threaded loop + * would have produced. Any worker failure fails the batch — the caller falls + * back to the sequential path. Kill switch: CODEGRAPH_NO_PARALLEL_RESOLVE=1. + */ + +import { Worker } from 'worker_threads'; +import * as fs from 'fs'; +import * as path from 'path'; +import * as os from 'os'; +import type { UnresolvedReference } from '../types'; +import type { ResolvedRef, UnresolvedRef } from './types'; + +export interface ChunkResult { + resolved: ResolvedRef[]; + unresolved: UnresolvedRef[]; + deferredChain: UnresolvedRef[]; + deferredThisMember: UnresolvedRef[]; + byMethod: Record; +} + +interface PoolWorker { + worker: Worker; + ready: Promise; + busy: number; +} + +const MIN_PARALLEL_BATCH = 1000; +const CHUNK_SIZE = 500; + +/** + * Minimum TOTAL pending refs before the pool is created at all. Pool boot + * (module load + readonly DB open + framework detect + cache warm, times N + * workers) costs real CPU that CONTENDS with sequential resolution on the + * same cores — measured on a medium repo (~40k refs, ~1.2s of resolution) + * the pool made indexing slower. It pays off when resolution runs for tens + * of seconds to minutes (large JVM/Spring-class repos). Override: + * CODEGRAPH_PARALLEL_RESOLVE_MIN= (0 forces the pool on). + */ +export function minRefsForPool(): number { + const raw = process.env.CODEGRAPH_PARALLEL_RESOLVE_MIN; + if (raw !== undefined) { + const parsed = Number.parseInt(raw, 10); + if (Number.isFinite(parsed) && parsed >= 0) return parsed; + } + return 150_000; +} + +export class ResolverPool { + private workers: PoolWorker[] = []; + private nextId = 0; + private waiters = new Map void; reject: (e: Error) => void }>(); + private failed: Error | null = null; + + /** + * Create a pool when the compiled worker exists (absent when running from + * source in tests → callers use the sequential path), the kill switch is + * off, and the machine has cores to spare. Returns null otherwise. + */ + static tryCreate(dbPath: string, projectRoot: string): ResolverPool | null { + if (process.env.CODEGRAPH_NO_PARALLEL_RESOLVE === '1') return null; + const workerScript = path.join(__dirname, 'resolver-worker.js'); + if (!fs.existsSync(workerScript)) return null; + const size = Math.max(1, Math.min(os.cpus().length - 2, 6)); + if (size < 2) return null; + try { + return new ResolverPool(workerScript, dbPath, projectRoot, size); + } catch { + return null; + } + } + + private constructor(workerScript: string, dbPath: string, projectRoot: string, size: number) { + for (let i = 0; i < size; i++) { + const worker = new Worker(workerScript); + let readyResolve!: () => void; + let readyReject!: (e: Error) => void; + const ready = new Promise((resolve, reject) => { + readyResolve = resolve; + readyReject = reject; + }); + const pw: PoolWorker = { worker, ready, busy: 0 }; + worker.on('message', (msg: { type: string; id?: number; message?: string } & Partial) => { + if (msg.type === 'ready') { + readyResolve(); + } else if (msg.type === 'result' && msg.id !== undefined) { + pw.busy--; + const waiter = this.waiters.get(msg.id); + this.waiters.delete(msg.id); + waiter?.resolve({ + resolved: msg.resolved!, + unresolved: msg.unresolved!, + deferredChain: msg.deferredChain!, + deferredThisMember: msg.deferredThisMember!, + byMethod: msg.byMethod!, + }); + } else if (msg.type === 'error') { + pw.busy--; + const err = new Error(`resolver worker: ${msg.message}`); + if (msg.id !== undefined && this.waiters.has(msg.id)) { + const waiter = this.waiters.get(msg.id)!; + this.waiters.delete(msg.id); + waiter.reject(err); + } else { + this.fail(err); + } + } + }); + worker.on('error', (err) => { + this.fail(err instanceof Error ? err : new Error(String(err))); + readyReject(this.failed!); + }); + worker.on('exit', (code) => { + if (code !== 0) { + this.fail(new Error(`resolver worker exited with code ${code}`)); + readyReject(this.failed!); + } + }); + worker.postMessage({ type: 'open', dbPath, projectRoot }); + this.workers.push(pw); + } + } + + private fail(err: Error): void { + if (!this.failed) this.failed = err; + for (const [, waiter] of this.waiters) waiter.reject(this.failed); + this.waiters.clear(); + } + + /** Whether this batch is worth fanning out. */ + static worthParallel(batchLength: number): boolean { + return batchLength >= MIN_PARALLEL_BATCH; + } + + async ready(): Promise { + await Promise.all(this.workers.map((w) => w.ready)); + } + + /** + * Resolve `refs` across the pool. Chunks preserve input order; the returned + * arrays are the in-order concatenation of the chunk results. + */ + async resolveBatch(refs: UnresolvedReference[]): Promise { + if (this.failed) throw this.failed; + const chunkPromises: Promise[] = []; + for (let i = 0; i < refs.length; i += CHUNK_SIZE) { + const chunk = refs.slice(i, i + CHUNK_SIZE); + const id = this.nextId++; + // Least-busy dispatch keeps workers evenly loaded regardless of chunk + // cost variance; result order is fixed by the promise array, not by + // completion order. + const pw = this.workers.reduce((a, b) => (b.busy < a.busy ? b : a)); + pw.busy++; + chunkPromises.push( + new Promise((resolve, reject) => { + this.waiters.set(id, { resolve, reject }); + pw.worker.postMessage({ type: 'resolve', id, refs: chunk }); + }) + ); + } + const chunks = await Promise.all(chunkPromises); + const out: ChunkResult = { resolved: [], unresolved: [], deferredChain: [], deferredThisMember: [], byMethod: {} }; + for (const c of chunks) { + out.resolved.push(...c.resolved); + out.unresolved.push(...c.unresolved); + out.deferredChain.push(...c.deferredChain); + out.deferredThisMember.push(...c.deferredThisMember); + for (const [k, v] of Object.entries(c.byMethod)) out.byMethod[k] = (out.byMethod[k] || 0) + v; + } + return out; + } + + async destroy(): Promise { + await Promise.all( + this.workers.map( + (pw) => + new Promise((resolve) => { + const t = setTimeout(() => { + void pw.worker.terminate().then(() => resolve()); + }, 5000); + pw.worker.once('exit', () => { + clearTimeout(t); + resolve(); + }); + pw.worker.postMessage({ type: 'close' }); + }) + ) + ); + } +} diff --git a/src/resolution/resolver-worker.ts b/src/resolution/resolver-worker.ts new file mode 100644 index 0000000..36bf9ff --- /dev/null +++ b/src/resolution/resolver-worker.ts @@ -0,0 +1,79 @@ +/** + * Resolver worker — one member of the parallel-resolution pool. + * + * Opens the project database READ-ONLY on its own connection and hosts a full + * ReferenceResolver over it. The main thread partitions each resolution batch + * into ordered chunks, fans them across the pool, and ADMITS the results + * sequentially in chunk order — so edge insertion order (and every cleanup / + * parking side effect) is identical to the single-threaded loop. Workers only + * ever read; all writes stay on the main thread. + * + * Visibility note: the sequential baseline resolves every ref of a batch + * against the DB state committed BEFORE that batch (edges persist after the + * whole batch resolves). Workers read exactly that same committed state, so + * per-ref inputs match the baseline ref-for-ref. + */ + +// Compile cache FIRST — same worker-boot rationale as parse-worker.ts. +try { + // eslint-disable-next-line @typescript-eslint/no-require-imports + (require('node:module') as { enableCompileCache?: () => void }).enableCompileCache?.(); +} catch { /* cache is best-effort */ } + +import { parentPort } from 'worker_threads'; +import { createDatabase, SqliteDatabase } from '../db/sqlite-adapter'; +import { QueryBuilder } from '../db/queries'; +import { ReferenceResolver } from './index'; +import type { UnresolvedReference } from '../types'; + +if (!parentPort) { + throw new Error('resolver-worker must be run as a worker thread'); +} +const port = parentPort; + +let db: SqliteDatabase | null = null; +let resolver: ReferenceResolver | null = null; + +type InMessage = + | { type: 'open'; dbPath: string; projectRoot: string } + | { type: 'resolve'; id: number; refs: UnresolvedReference[] } + | { type: 'close' }; + +port.on('message', (msg: InMessage) => { + try { + switch (msg.type) { + case 'open': { + const created = createDatabase(msg.dbPath, { readOnly: true }); + db = created.db; + db.pragma('busy_timeout = 5000'); + db.pragma('cache_size = -32000'); + const queries = new QueryBuilder(db); + resolver = new ReferenceResolver(msg.projectRoot, queries); + resolver.initialize(); + port.postMessage({ type: 'ready' }); + break; + } + case 'resolve': { + if (!resolver) throw new Error('resolver-worker: resolve before open'); + const out = resolver.resolveListForAdmission(msg.refs); + port.postMessage({ type: 'result', id: msg.id, ...out }); + break; + } + case 'close': { + try { + db?.close(); + } catch { + /* already closed */ + } + process.exit(0); + break; + } + } + } catch (err) { + port.postMessage({ + type: 'error', + id: (msg as { id?: number }).id, + message: err instanceof Error ? err.message : String(err), + }); + } +});