perf(index): faster fresh indexing + parallel reference resolution, byte-identical graphs (#1305)
* perf(index): ~34% faster fresh indexing, byte-identical graphs Profiling a fresh init on a medium TS repo (excalidraw, 657 files) showed the main thread as the critical path: per-row SQLite statement calls, repeated import-resolution walks, and per-row FTS trigger firings, with the parse workers ~75% idle behind it. This lands the semantics-preserving tranche of fixes: - Multi-row batched INSERTs (nodes/edges/unresolved refs/name segments) behind cached per-batch-size prepared statements; row order preserved, so rowid-based resolution determinism (#1015) is unchanged. - storeFileBundle: one transaction per file instead of four; nested transaction() calls now flatten (BEGIN-in-BEGIN previously threw, so no caller depended on nested rollback). - Dedicated store-writer thread for the fresh-DB bulk path (bundles applied in file order on a single writer connection; main thread does no DB work during the parse loop). Kill switch: CODEGRAPH_NO_STORE_WORKER=1. - Bulk FTS mode: drop the nodes_fts sync triggers during the bulk load, rebuild once at the end; crash inside the window self-heals on the next open. - Per-context memos for resolveImportPath/findExportedSymbol + a per-file exported-symbol index, invalidated exactly where clearCaches() already resets the resolver's own caches. - Fast-init on completely fresh DBs (journal in memory, no fsync until the index completes; interrupted init re-runs from scratch). Kill switch: CODEGRAPH_NO_FAST_INIT=1. - MaybeYield returns undefined on the not-due path so per-ref yield checks stop paying a promise + microtask hop each. - Parse pool prewarm for bulk indexing; compile-cache enabled at CLI and worker entry points. Excalidraw fresh init: 5.11s -> 3.36s median (n=5, warm cache, M-series). Graph dumps byte-identical across init, re-index, and sync paths; full suite green (2403 passed). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * perf(resolution): parallel reference resolution with canonical admission Fan resolution batches across a pool of read-only worker threads, each hosting a full ReferenceResolver over its own SQLite connection; results are admitted on the main thread in chunk order, so edge insertion order, row cleanup, failure parking, and deferred post-pass queues are exactly the sequence the single-threaded loop produces. Per-ref inputs match the baseline because the sequential path already resolves each batch against the state committed BEFORE that batch. Validated byte-identical on excalidraw (pool forced on) and apache/dubbo (4,048 Java files): dubbo full index 39s -> 19s (2.05x) with identical graph dumps (91,495 nodes / 223,953 edges). The pool only engages when total pending refs clear a threshold (default 150k, CODEGRAPH_PARALLEL_RESOLVE_MIN to tune, CODEGRAPH_NO_PARALLEL_RESOLVE=1 to disable): measured on a ~58k-ref repo the workers' boot CPU contends with resolution on the same cores and makes indexing slower, so small repos keep the sequential path. When fast-init left the DB in memory-journal mode, WAL is restored before resolution only when the pool will run (readers + rollback-journal writers don't mix). Also: sqlite adapter readOnly open support. TreeCursor spine rewrite of the body walker was built, measured neutral on real repos and equal in a 20k-child microbench (web-tree-sitter's namedChild(i) is not quadratic in this binding), and rejected — per-node JS<->WASM marshaling is the floor, which a traversal swap cannot remove. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Colby Mchenry
co-authored by
Claude Fable 5
Claude Fable 5
parent
246aee8373
commit
5736e24bb6
+181
-43
@@ -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<void> | null
|
||||
walBackpressure?: () => Promise<void> | 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<IndexResult> {
|
||||
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<string, string>();
|
||||
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<Edge & { targetKind: string; targetName: string; sourceFilePath: string; sourceLanguage: Language }>,
|
||||
validNodes: Node[]
|
||||
): void {
|
||||
const newNodesByKindName = new Map<string, string>();
|
||||
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.
|
||||
*
|
||||
|
||||
@@ -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; }
|
||||
|
||||
|
||||
@@ -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';
|
||||
|
||||
@@ -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),
|
||||
});
|
||||
}
|
||||
});
|
||||
@@ -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<void>;
|
||||
private firstError: Error | null = null;
|
||||
private drainWaiters = new Map<number, { resolve: () => 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<void>((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<void> {
|
||||
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<void> {
|
||||
if (this.firstError || this.exited || this.outstanding < limit) return Promise.resolve();
|
||||
return new Promise<void>((resolve) => {
|
||||
this.belowWaiters.push({ limit, resolve });
|
||||
});
|
||||
}
|
||||
|
||||
/** Resolves when every bundle posted before this call has been applied. */
|
||||
drain(): Promise<void> {
|
||||
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<void>((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<void> {
|
||||
if (this.exited) return;
|
||||
this.worker.postMessage({ type: 'close' });
|
||||
await new Promise<void>((resolve) => {
|
||||
const t = setTimeout(() => {
|
||||
void this.worker.terminate().then(() => resolve());
|
||||
}, 5000);
|
||||
this.worker.once('exit', () => {
|
||||
clearTimeout(t);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user