diff --git a/gitnexus/src/core/ingestion/workers/parse-worker.ts b/gitnexus/src/core/ingestion/workers/parse-worker.ts index 0e8c8c65d..e6b776228 100644 --- a/gitnexus/src/core/ingestion/workers/parse-worker.ts +++ b/gitnexus/src/core/ingestion/workers/parse-worker.ts @@ -1543,6 +1543,10 @@ function reportWarning(message: string): void { } } +// Keep compiled queries across jobs in this worker. A language can select +// multiple native grammars, so both grammar identity and query text matter. +const compiledQueries = new WeakMap>(); + const processFileGroup = ( files: ParseWorkerInput[], language: SupportedLanguages, @@ -1553,7 +1557,14 @@ const processFileGroup = ( let query: Parser.Query; try { const lang = parser.getLanguage(); - query = new Parser.Query(lang, queryString); + let queries = compiledQueries.get(lang); + if (!queries) { + queries = new Map(); + compiledQueries.set(lang, queries); + } + const cached = queries.get(queryString); + query = cached ?? new Parser.Query(lang, queryString); + if (!cached) queries.set(queryString, query); } catch (err) { reportWarning( `Query compilation failed for ${language}: ${err instanceof Error ? err.message : String(err)}`, diff --git a/gitnexus/src/core/ingestion/workers/worker-pool.ts b/gitnexus/src/core/ingestion/workers/worker-pool.ts index de102ebef..497100c64 100644 --- a/gitnexus/src/core/ingestion/workers/worker-pool.ts +++ b/gitnexus/src/core/ingestion/workers/worker-pool.ts @@ -1316,9 +1316,16 @@ export const createWorkerPool = ( } if (dispatchableItems.length === 0) return []; + // Stable cache packs can be much smaller than either job ceiling. Split + // those packs across the live slots too, otherwise each serial dispatch + // feeds only one worker. Keep both configured ceilings as upper bounds. + const maxItemsPerJob = Math.min( + poolOptions.subBatchSize, + Math.max(1, Math.floor(dispatchableItems.length / activeSlots.size)), + ); const jobs = createJobs( dispatchableItems, - poolOptions.subBatchSize, + maxItemsPerJob, poolOptions.subBatchMaxBytes, poolOptions.subBatchIdleTimeoutMs, chunkHash, @@ -1456,7 +1463,19 @@ export const createWorkerPool = ( retireWorkerAfterTimeout(existing, workerIndex, reason); return; } - await existing.terminate().catch(() => undefined); + // Recovery must settle before dispatch returns, but a failed thread + // may never acknowledge termination. Bound that wait as in shutdown. + const termination = existing.terminate().then( + () => undefined, + () => undefined, + ); + if (!(await settledWithin(termination, poolOptions.shutdownDrainMs))) { + existing.unref?.(); + logger.warn( + { workerIndex, drainMs: poolOptions.shutdownDrainMs, reason }, + `Worker ${workerIndex} did not finish terminating within the shutdown drain; continuing recovery.`, + ); + } }; const replaceWorker = async ( @@ -1898,11 +1917,13 @@ export const createWorkerPool = ( // (`error`, `exit`, msg-channel error). Bridges the per-job teardown // into the pool-level handleWorkerDeath recovery + breaker logic. const recoverAndResume = async (reason: string, excludePaths: readonly string[]) => { - activeWorkers--; busySlots.delete(workerIndex); inFlightProgress[workerIndex] = 0; requeueRemainder(job, excludePaths); + // Keep recovery in flight so another slot finishing cannot settle + // this dispatch before the replacement is ready for the next one. await handleWorkerDeath(workerIndex, reason, excludePaths); + activeWorkers--; if (stopped) return; // Slot may have been dropped or respawned. Kick the current slot // if still active, then wake any other idle live slots so the @@ -1949,7 +1970,6 @@ export const createWorkerPool = ( // is respawned (or dropped) and can dispatch the next // job deterministically. void (async () => { - activeWorkers--; busySlots.delete(workerIndex); requeueRemainder(job, decision.excludePaths); await handleWorkerDeath( @@ -1958,6 +1978,7 @@ export const createWorkerPool = ( decision.excludePaths, 'retire', ); + activeWorkers--; if (stopped) return; if (activeSlots.has(workerIndex)) runWorker(workerIndex); wakeIdleSlots(); diff --git a/gitnexus/test/integration/worker-pool.test.ts b/gitnexus/test/integration/worker-pool.test.ts index d9f6577a9..da910a736 100644 --- a/gitnexus/test/integration/worker-pool.test.ts +++ b/gitnexus/test/integration/worker-pool.test.ts @@ -14,6 +14,7 @@ import { } from '../../src/core/ingestion/workers/worker-pool.js'; import { pathToFileURL } from 'node:url'; import { spawn } from 'node:child_process'; +import { createRequire } from 'node:module'; import path from 'node:path'; import fs from 'node:fs'; import os from 'node:os'; @@ -245,10 +246,8 @@ describe('worker pool integration', () => { const results = await pool.dispatch(files); - // All 7 files fit one default sub-batch (size 200 / budget 8MB), - // so the dispatch returns exactly one chunk result regardless of - // pool size. - expect(results).toHaveLength(1); + // Small inputs must still split across the available workers. + expect(results.length).toBeGreaterThan(1); // Total files parsed should match input const totalParsed = results.reduce((sum: number, r: any) => sum + r.fileCount, 0); @@ -793,6 +792,54 @@ describe('worker pool integration', () => { } }); + it.each([2, 4, 7, 9])( + 'uses available workers for a small %i-file cache pack', + async (fileCount) => { + const { tempDir, workerPath } = writeTempWorker( + 'gitnexus-worker-small-pack-', + ` + const { parentPort, threadId } = require('node:worker_threads'); + let paths = []; + parentPort.on('message', (msg) => { + if (msg && msg.type === 'sub-batch') { + paths = msg.files.map((file) => file.path); + parentPort.postMessage({ type: 'progress', filesProcessed: paths.length }); + parentPort.postMessage({ type: 'sub-batch-done' }); + } else if (msg && msg.type === 'flush') { + parentPort.postMessage({ type: 'result', data: { paths, threadId } }); + } + }); + `, + ); + pool = createWorkerPool(pathToFileURL(workerPath), 4, { + subBatchMaxBytes: 256 * 1024, + }); + + try { + // Stable cache packs are often smaller than the byte budget. They must + // still use the pool, including when a later dispatch has fewer files. + for (const count of [fileCount, 1]) { + const files = Array.from({ length: count }, (_, i) => ({ + path: `file-${i}.ts`, + content: 'export const value = 1;', + })); + const progress: number[] = []; + const results = await pool.dispatch< + (typeof files)[number], + { paths: string[]; threadId: number } + >(files, (completed) => progress.push(completed), `pack-${count}`); + expect(new Set(results.map((result) => result.threadId)).size).toBe(Math.min(4, count)); + expect(results.flatMap((result) => result.paths)).toEqual(files.map((file) => file.path)); + expect(progress).toEqual([...progress].sort((a, b) => a - b)); + expect(progress.at(-1)).toBe(count); + } + } finally { + await pool.terminate(); + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }, + ); + it('bounds worker jobs by byte budget as well as file count', async () => { const { tempDir, workerPath } = writeTempWorker( 'gitnexus-worker-byte-budget-', @@ -831,6 +878,69 @@ describe('worker pool integration', () => { } }); + it.skipIf(!hasDistWorker)( + 'reuses compiled queries across jobs while keeping TS and TSX grammars separate', + async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'gitnexus-worker-query-cache-')); + const workerPath = path.join(tempDir, 'worker.cjs'); + const parserPath = createRequire(import.meta.url).resolve('tree-sitter'); + fs.writeFileSync( + workerPath, + ` + const { parentPort } = require('node:worker_threads'); + const Parser = require(${JSON.stringify(parserPath)}); + let queryCompilations = 0; + Parser.Query = new Proxy(Parser.Query, { + construct(target, args, newTarget) { + queryCompilations++; + return Reflect.construct(target, args, newTarget); + }, + }); + const send = parentPort.postMessage.bind(parentPort); + parentPort.postMessage = (message, ...args) => { + if (message.type === 'result') message.data.queryCompilations = queryCompilations; + return send(message, ...args); + }; + import(${JSON.stringify(pathToFileURL(DIST_WORKER).href)}); + `, + ); + pool = createWorkerPool(pathToFileURL(workerPath), 1, { workerReadyTimeoutMs: 30_000 }); + type QueryResult = { + queryCompilations: number; + fileCount: number; + nodes: Array<{ properties: { name: string } }>; + }; + try { + const counts: number[] = []; + for (const [extension, name] of [ + ['ts', 'first'], + ['ts', 'second'], + ['tsx', 'view'], + ['tsx', 'otherView'], + ['ts', 'last'], + ]) { + const file = { + path: `${name}.${extension}`, + content: `export function ${name}() { return ${extension === 'tsx' ? '
' : '1'}; }`, + }; + const [result] = await pool.dispatch([file]); + expect(result.fileCount).toBe(1); + expect(result.nodes.map((node) => node.properties.name)).toContain(name); + counts.push(result.queryCompilations); + } + expect(counts[0]).toBeGreaterThan(0); + expect(counts[1]).toBe(counts[0]); + expect(counts[2]).toBeGreaterThan(counts[1]); + expect(counts[3]).toBe(counts[2]); + expect(counts[4]).toBe(counts[2]); + } finally { + await pool.terminate(); + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }, + 60_000, + ); + it.skipIf(!hasDistWorker)('createWorkerPool with size 0 creates pool with zero workers', () => { const workerUrl = pathToFileURL(DIST_WORKER) as URL; const zeroPool = createWorkerPool(workerUrl, 0); diff --git a/gitnexus/test/unit/worker-pool-resilience.test.ts b/gitnexus/test/unit/worker-pool-resilience.test.ts index 002d47de8..1427f45f1 100644 --- a/gitnexus/test/unit/worker-pool-resilience.test.ts +++ b/gitnexus/test/unit/worker-pool-resilience.test.ts @@ -58,7 +58,7 @@ let workerInstances: FakeWorker[] = []; class FakeWorker extends EventEmitter { readonly seenMessages: unknown[] = []; - constructor() { + constructor(startupExitCode?: number) { super(); workerInstances.push(this); // Real Worker fires 'online' asynchronously after the runtime is ready; @@ -67,6 +67,10 @@ class FakeWorker extends EventEmitter { // message instead — emit that too so replacement-worker tests don't // hit the WORKER_READY_TIMEOUT_MS budget (5s). queueMicrotask(() => { + if (startupExitCode !== undefined) { + this.emit('exit', startupExitCode); + return; + } this.emit('online'); this.emit('message', { type: 'ready' }); }); @@ -262,8 +266,8 @@ describe('worker pool resilience', () => { consecutiveFailureThreshold: 10, maxRespawnsPerSlot: 1, }); - // Slot 0 dies twice, exceeding budget=1; slot 1 succeeds with the - // requeued remainder. + // Separate dispatches target the first live slot twice, independently + // of how multi-file packs are split across workers. nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 }); nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 }); nextActions.push({ @@ -272,9 +276,10 @@ describe('worker pool resilience', () => { result: { fileCount: 2 }, }); + await pool.dispatch([{ path: 'src/a.ts', content: '' }]); + await pool.dispatch([{ path: 'src/b.ts', content: '' }]); + expect(pool.getStats().activeSlots).toBe(1); const results = await pool.dispatch<{ path: string; content: string }, unknown>([ - { path: 'src/a.ts', content: '' }, - { path: 'src/b.ts', content: '' }, { path: 'src/c.ts', content: '' }, { path: 'src/d.ts', content: '' }, ]); @@ -495,18 +500,13 @@ describe('worker pool resilience', () => { await pool.terminate(); }); - it('drops slot when waitForWorkerOnline rejects (replaceWorker failure path)', async () => { + it('drops slot when replacement readiness rejects', async () => { let factoryCallCount = 0; const pool = createWorkerPool(workerUrl, 2, { workerFactory: () => { factoryCallCount++; - const worker = new FakeWorker(); - // Slot 0's initial worker is healthy; the replacement (3rd factory - // call after slot 0 dies once) exits before emitting 'online'. - if (factoryCallCount === 3) { - // Override the queued 'online' microtask with an immediate 'exit'. - queueMicrotask(() => worker.emit('exit', 1)); - } + // The replacement exits INSTEAD OF reporting ready. + const worker = new FakeWorker(factoryCallCount === 3 ? 1 : undefined); return worker as unknown as import('node:worker_threads').Worker; }, consecutiveFailureThreshold: 10, @@ -519,8 +519,9 @@ describe('worker pool resilience', () => { result: { fileCount: 2 }, }); + await pool.dispatch([{ path: 'src/a.ts', content: '' }]); + expect(pool.getStats().activeSlots).toBe(1); const results = await pool.dispatch<{ path: string; content: string }, unknown>([ - { path: 'src/a.ts', content: '' }, { path: 'src/b.ts', content: '' }, { path: 'src/c.ts', content: '' }, ]); @@ -532,6 +533,33 @@ describe('worker pool resilience', () => { await pool.terminate(); }); + it('finishes recovery when a failed worker never acknowledges termination', async () => { + const pool = createWorkerPool(workerUrl, 2, { + workerFactory: () => { + const worker = new FakeWorker(); + if (workerInstances.length === 1) { + worker.terminate = () => new Promise(() => {}); + } + return worker as unknown as import('node:worker_threads').Worker; + }, + shutdownDrainMs: 10, + }); + nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 }); + nextActions.push({ kind: 'parse-ok', files: [{ path: 'src/good.ts' }] }); + try { + const results = await pool.dispatch([ + { path: 'src/bad.ts', content: '' }, + { path: 'src/good.ts', content: '' }, + ]); + expect(results).toEqual([{ fileCount: 1 }]); + expect(pool.getQuarantinedPaths()).toEqual(['src/bad.ts']); + expect(pool.getStats().slotGenerations).toEqual([1, 0]); + expect(pool.getStats().activeSlots).toBe(2); + } finally { + await pool.terminate(); + } + }, 1000); + it('trips the breaker when all slots exhaust their respawn budget', async () => { const pool = createWorkerPool(workerUrl, 2, { workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,