From 681a3520064ff2d3b7382b8a5f9ead0ce24c4e1c Mon Sep 17 00:00:00 2001 From: ChamHerry <51915924+ChamHerry@users.noreply.github.com> Date: Wed, 27 May 2026 00:28:39 +0800 Subject: [PATCH] fix(worker): analyze native worker aborts (#1833) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(analyze): avoid native aborts on generated worker bundles Retire timed-out parse workers instead of force-terminating native parser state, and skip Monaco generated worker bundles by default while preserving explicit .gitnexusignore negation overrides. Constraint: Node native tree-sitter bindings can abort the process when a timed-out worker is terminated while inside parser state. Rejected: Falling back to sequential parsing for native stalls | it can move the same native crash onto the main thread. Confidence: high Scope-risk: moderate Directive: Keep timeout recovery from force-terminating workers until they return to JS or exit naturally. Tested: npm test; npx tsc --noEmit; npm run build; targeted analyze on /Users/wangxc/Code/keep; gitnexus detect_changes --scope staged Not-tested: Node 22 LTS runtime and non-macOS platforms * fix(worker): bound retired parser worker lifetimes Keep timeout recovery from immediately terminating workers that may still be inside native parser state, while making terminal pool shutdown own retired worker cleanup so long-lived processes do not accumulate retired threads. Constraint: Claude review on PR #1833 required retiredWorkers cleanup in pool.terminate() and tripBreaker() without regressing no-immediate-terminate timeout safety. Rejected: clearing the retiredWorkers set without terminating | would remove JS bookkeeping while leaking the underlying worker thread. Confidence: high Scope-risk: moderate Directive: Preserve the distinction between recoverable timeout retirement and terminal pool shutdown; do not reintroduce immediate terminate in removeWorkerFromSlot(..., 'retire'). Tested: npx vitest run test/unit/worker-pool-timeout-retire.test.ts; npx vitest run test/unit/worker-pool-timeout-retire.test.ts test/unit/worker-pool-resilience.test.ts test/unit/worker-pool-cumulative-timeout.test.ts test/unit/worker-pool-slot-generation.test.ts; npx tsc --noEmit; npm run build; npx prettier --check src/core/ingestion/workers/worker-pool.ts test/unit/worker-pool-timeout-retire.test.ts ../docs/todo/pr-1833-retired-worker-cleanup-plan.md; npx eslint src/core/ingestion/workers/worker-pool.ts test/unit/worker-pool-timeout-retire.test.ts; gitnexus detect_changes --scope staged. Not-tested: npm test full suite did not complete green in this environment; two runs each had one unrelated test/unit/hooks.test.ts parseHookOutput null failure, and each failed hook test passed when rerun in isolation. * ci: retrigger checks Co-authored-by: Cursor --------- Co-authored-by: wangxc Co-authored-by: Gergő Magyar Co-authored-by: Test Co-authored-by: Cursor --- gitnexus/src/config/ignore-service.ts | 9 +- .../src/core/ingestion/workers/worker-pool.ts | 141 ++++++++++++-- gitnexus/test/unit/ignore-service.test.ts | 13 ++ .../unit/worker-pool-timeout-retire.test.ts | 180 ++++++++++++++++++ 4 files changed, 319 insertions(+), 24 deletions(-) create mode 100644 gitnexus/test/unit/worker-pool-timeout-retire.test.ts diff --git a/gitnexus/src/config/ignore-service.ts b/gitnexus/src/config/ignore-service.ts index 17b9d14bd..02a26068e 100644 --- a/gitnexus/src/config/ignore-service.ts +++ b/gitnexus/src/config/ignore-service.ts @@ -87,6 +87,7 @@ const DEFAULT_IGNORE_LIST = new Set([ '.generated', 'generated', 'auto-generated', + 'monaco-workers', // Monaco editor web-worker bundles generated for browser runtime '.terraform', '.serverless', @@ -323,14 +324,6 @@ export const shouldIgnorePath = (filePath: string): boolean => { } } - // Ignore hidden files (starting with .) - if (fileName.startsWith('.') && fileName !== '.') { - // But allow some important config files - const allowedDotFiles = ['.env', '.gitignore']; // Already in IGNORED_FILES, so this is redundant - // Actually, let's NOT ignore all dot files - many are important configs - // Just rely on the explicit lists above - } - // Ignore files that look like generated/bundled code if ( fileNameLower.includes('.bundle.') || diff --git a/gitnexus/src/core/ingestion/workers/worker-pool.ts b/gitnexus/src/core/ingestion/workers/worker-pool.ts index d29b352ab..5c326e972 100644 --- a/gitnexus/src/core/ingestion/workers/worker-pool.ts +++ b/gitnexus/src/core/ingestion/workers/worker-pool.ts @@ -598,6 +598,14 @@ export const createWorkerPool = ( const poolOptions = resolveWorkerPoolOptions(options, size); const spawnWorker = options?.workerFactory ?? ((url: URL) => new Worker(url)); const workers: (Worker | undefined)[] = new Array(size); + type RetiredWorkerRecord = { + worker: Worker; + workerIndex: number; + reason: string; + cleanup: () => void; + terminate: () => Promise; + }; + const retiredWorkers = new Set(); const respawnCount: number[] = new Array(size).fill(0); const activeSlots: Set = new Set(); // Layer 3 (quarantine): tracked via the dedicated `quarantine.ts` @@ -625,6 +633,17 @@ export const createWorkerPool = ( let poolBroken = false; let poolFailure: Error | undefined; + const terminateTrackedWorkers = async ( + liveWorkers: readonly (Worker | undefined)[], + ): Promise => { + const retired = Array.from(retiredWorkers); + await Promise.all([ + ...liveWorkers.map((worker) => worker?.terminate().catch(() => undefined)), + ...retired.map((record) => record.terminate()), + ]); + retiredWorkers.clear(); + }; + for (let i = 0; i < size; i++) { workers[i] = spawnWorker(workerUrl); activeSlots.add(i); @@ -755,10 +774,88 @@ export const createWorkerPool = ( onProgress(next); }; - const replaceWorker = async (workerIndex: number): Promise => { + type WorkerRemovalMode = 'terminate' | 'retire'; + + const retireWorkerAfterTimeout = ( + worker: Worker, + workerIndex: number, + reason: string, + ): void => { + let cleaned = false; + let terminateStarted = false; + + function cleanupRetired() { + if (cleaned) return; + cleaned = true; + worker.removeListener('message', onRetiredMessage); + worker.removeListener('error', onRetiredError); + worker.removeListener('exit', onRetiredExit); + worker.removeListener('messageerror', onRetiredMessageError); + retiredWorkers.delete(record); + } + + async function terminateRetired() { + if (terminateStarted) return; + terminateStarted = true; + cleanupRetired(); + await worker.terminate().catch(() => undefined); + } + + function terminateWhenBackInJs() { + void terminateRetired(); + } + + function onRetiredMessage(raw: unknown) { + if (raw === null || typeof raw !== 'object') return; + const type = (raw as { type?: unknown }).type; + if (type === 'sub-batch-done' || type === 'result' || type === 'error') { + terminateWhenBackInJs(); + } + } + + const onRetiredError = () => cleanupRetired(); + const onRetiredExit = () => cleanupRetired(); + const onRetiredMessageError = () => terminateWhenBackInJs(); + const record: RetiredWorkerRecord = { + worker, + workerIndex, + reason, + cleanup: cleanupRetired, + terminate: terminateRetired, + }; + retiredWorkers.add(record); + worker.on('message', onRetiredMessage); + worker.once('error', onRetiredError); + worker.once('exit', onRetiredExit); + worker.once('messageerror', onRetiredMessageError); + (worker as Worker & { unref?: () => void }).unref?.(); + logger.warn( + { workerIndex, reason }, + `Worker ${workerIndex} timed out; retiring without immediate terminate to avoid aborting native parser state.`, + ); + }; + + const removeWorkerFromSlot = async ( + workerIndex: number, + mode: WorkerRemovalMode, + reason: string, + ): Promise => { const existing = workers[workerIndex]; - await existing?.terminate().catch(() => undefined); workers[workerIndex] = undefined; + if (!existing) return; + if (mode === 'retire') { + retireWorkerAfterTimeout(existing, workerIndex, reason); + return; + } + await existing.terminate().catch(() => undefined); + }; + + const replaceWorker = async ( + workerIndex: number, + mode: WorkerRemovalMode = 'terminate', + reason = 'replacing worker', + ): Promise => { + await removeWorkerFromSlot(workerIndex, mode, reason); if (stopped) return false; const replacement = spawnWorker(workerUrl); try { @@ -803,7 +900,7 @@ export const createWorkerPool = ( const liveWorkers = workers.slice(); for (let i = 0; i < workers.length; i++) workers[i] = undefined; activeSlots.clear(); - void Promise.all(liveWorkers.map((worker) => worker?.terminate().catch(() => undefined))); + void terminateTrackedWorkers(liveWorkers); }; const maybeDone = () => { @@ -893,6 +990,7 @@ export const createWorkerPool = ( workerIndex: number, reason: string, excludePaths: readonly string[], + removalMode: WorkerRemovalMode = 'terminate', ) => { if (stopped) return; consecutiveFailuresPerSlot[workerIndex]++; @@ -921,9 +1019,7 @@ export const createWorkerPool = ( }, `Worker ${workerIndex} exceeded respawn budget; dropping slot.`, ); - const dead = workers[workerIndex]; - await dead?.terminate().catch(() => undefined); - workers[workerIndex] = undefined; + await removeWorkerFromSlot(workerIndex, removalMode, reason); activeSlots.delete(workerIndex); if (activeSlots.size === 0) { tripBreaker( @@ -945,7 +1041,7 @@ export const createWorkerPool = ( }, `Worker ${workerIndex} died; respawning slot (attempt ${respawnCount[workerIndex]}/${poolOptions.maxRespawnsPerSlot}).`, ); - const respawned = await replaceWorker(workerIndex); + const respawned = await replaceWorker(workerIndex, removalMode, reason); if (!respawned) { activeSlots.delete(workerIndex); if (activeSlots.size === 0) { @@ -1211,7 +1307,12 @@ export const createWorkerPool = ( activeWorkers--; busySlots.delete(workerIndex); requeueRemainder(job, decision.excludePaths); - await handleWorkerDeath(workerIndex, decision.reason, decision.excludePaths); + await handleWorkerDeath( + workerIndex, + decision.reason, + decision.excludePaths, + 'retire', + ); if (stopped) return; if (activeSlots.has(workerIndex)) runWorker(workerIndex); wakeIdleSlots(); @@ -1255,9 +1356,11 @@ export const createWorkerPool = ( }, `Worker ${workerIndex} hit consecutive-failure threshold on idle-timeout retry; tripping circuit breaker.`, ); - const dead = workers[workerIndex]; - await dead?.terminate().catch(() => undefined); - workers[workerIndex] = undefined; + await removeWorkerFromSlot( + workerIndex, + 'retire', + 'idle-timeout retry consecutive-failure threshold', + ); activeSlots.delete(workerIndex); tripBreaker( new WorkerPoolDispatchError( @@ -1278,12 +1381,18 @@ export const createWorkerPool = ( }, `Worker ${workerIndex} exceeded respawn budget during idle-timeout retry; dropping slot.`, ); - const dead = workers[workerIndex]; - await dead?.terminate().catch(() => undefined); - workers[workerIndex] = undefined; + await removeWorkerFromSlot( + workerIndex, + 'retire', + 'idle-timeout retry respawn budget exhausted', + ); activeSlots.delete(workerIndex); } else { - const respawned = await replaceWorker(workerIndex); + const respawned = await replaceWorker( + workerIndex, + 'retire', + 'idle-timeout retry', + ); if (!respawned) { activeSlots.delete(workerIndex); } @@ -1482,7 +1591,7 @@ export const createWorkerPool = ( // exception when this is called from `runChunkedParseAndResolve`'s // finally block — masking the real failure and leaving `workers[]` // populated with dead references because the lines below never run. - await Promise.all(workers.map((w) => w?.terminate().catch(() => undefined))); + await terminateTrackedWorkers(workers); workers.length = 0; activeSlots.clear(); }; diff --git a/gitnexus/test/unit/ignore-service.test.ts b/gitnexus/test/unit/ignore-service.test.ts index cd5ebdb4e..9f989bbcc 100644 --- a/gitnexus/test/unit/ignore-service.test.ts +++ b/gitnexus/test/unit/ignore-service.test.ts @@ -175,6 +175,11 @@ describe('shouldIgnorePath', () => { expect(shouldIgnorePath('src/api.generated.ts')).toBe(true); }); + it('ignores generated Monaco editor worker bundles', () => { + expect(shouldIgnorePath('public/monaco-workers/json.worker.js')).toBe(true); + expect(shouldIgnorePath('keep-ui/public/monaco-workers/125.js')).toBe(true); + }); + it('ignores TypeScript declaration files', () => { expect(shouldIgnorePath('types/index.d.ts')).toBe(true); }); @@ -217,6 +222,7 @@ describe('isHardcodedIgnoredDirectory', () => { expect(isHardcodedIgnoredDirectory('node_modules')).toBe(true); expect(isHardcodedIgnoredDirectory('.git')).toBe(true); expect(isHardcodedIgnoredDirectory('dist')).toBe(true); + expect(isHardcodedIgnoredDirectory('monaco-workers')).toBe(true); expect(isHardcodedIgnoredDirectory('__pycache__')).toBe(true); }); @@ -318,6 +324,13 @@ describe('.gitnexusignore negation overrides hardcoded DEFAULT_IGNORE_LIST (#771 expect(filter.childrenIgnored(mkPath('.git'))).toBe(true); }); + it('explicit negation can still opt into generated Monaco worker bundles', async () => { + await fs.writeFile(path.join(tmpDir, '.gitnexusignore'), '!public/monaco-workers/\n'); + const filter = await createIgnoreFilter(tmpDir); + expect(filter.childrenIgnored(mkPath('public/monaco-workers'))).toBe(false); + expect(filter.ignored(mkPath('public/monaco-workers/json.worker.js'))).toBe(false); + }); + it('standard `.gitignore` rules (no negation) still layer on top of hardcoded', async () => { // Pre-#771 behaviour: if .gitnexusignore says `my-dir/`, that dir // is ignored in addition to the hardcoded list. Non-negation diff --git a/gitnexus/test/unit/worker-pool-timeout-retire.test.ts b/gitnexus/test/unit/worker-pool-timeout-retire.test.ts new file mode 100644 index 000000000..0579ae6d6 --- /dev/null +++ b/gitnexus/test/unit/worker-pool-timeout-retire.test.ts @@ -0,0 +1,180 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { EventEmitter } from 'node:events'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { pathToFileURL } from 'node:url'; + +import { createWorkerPool } from '../../src/core/ingestion/workers/worker-pool.js'; + +type FirstWorkerBehavior = 'stall' | 'delayed-safe-return'; + +class TimeoutThenHealthyWorker extends EventEmitter { + static instances: TimeoutThenHealthyWorker[] = []; + static firstWorkerBehavior: FirstWorkerBehavior = 'stall'; + static safeReturnDelayMs = 40; + + readonly id: number; + terminateCalls = 0; + unrefCalls = 0; + private currentPaths: string[] = []; + + constructor() { + super(); + this.id = TimeoutThenHealthyWorker.instances.length; + TimeoutThenHealthyWorker.instances.push(this); + queueMicrotask(() => this.emit('message', { type: 'ready' })); + } + + postMessage(msg: unknown): void { + if (msg === null || typeof msg !== 'object') return; + const type = (msg as { type?: unknown }).type; + if (type === 'sub-batch') { + const files = (msg as { files?: Array<{ path: string }> }).files ?? []; + this.currentPaths = files.map((file) => file.path); + if (this.id === 0) { + if (TimeoutThenHealthyWorker.firstWorkerBehavior === 'delayed-safe-return') { + setTimeout(() => { + this.emit('message', { type: 'sub-batch-done' }); + }, TimeoutThenHealthyWorker.safeReturnDelayMs); + } + return; + } + queueMicrotask(() => { + this.emit('message', { type: 'progress', filesProcessed: this.currentPaths.length }); + this.emit('message', { type: 'sub-batch-done' }); + }); + return; + } + if (type === 'flush') { + const paths = this.currentPaths.slice(); + queueMicrotask(() => this.emit('message', { type: 'result', data: { paths } })); + } + } + + async terminate(): Promise { + this.terminateCalls++; + this.emit('exit', 0); + return 0; + } + + unref(): void { + this.unrefCalls++; + } +} + +const waitFor = async ( + predicate: () => boolean, + message: string, + timeoutMs = 250, +): Promise => { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error(message); +}; + +let tempDir: string; +let workerUrl: URL; + +beforeEach(() => { + TimeoutThenHealthyWorker.instances = []; + TimeoutThenHealthyWorker.firstWorkerBehavior = 'stall'; + TimeoutThenHealthyWorker.safeReturnDelayMs = 40; + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'gitnexus-worker-timeout-retire-')); + const workerPath = path.join(tempDir, 'fake-worker.js'); + fs.writeFileSync(workerPath, '// fake worker path for createWorkerPool'); + workerUrl = pathToFileURL(workerPath) as URL; +}); + +afterEach(() => { + fs.rmSync(tempDir, { recursive: true, force: true }); +}); + +describe('worker pool timeout retirement', () => { + it('does not immediately terminate a worker that timed out inside native parsing', async () => { + const pool = createWorkerPool(workerUrl, 1, { + subBatchIdleTimeoutMs: 20, + maxTimeoutRetries: 1, + timeoutBackoffFactor: 2, + workerFactory: () => + new TimeoutThenHealthyWorker() as unknown as import('node:worker_threads').Worker, + }); + + try { + const results = await pool.dispatch<{ path: string; content: string }, { paths: string[] }>([ + { path: 'src/native-stall.ts', content: 'const x = 1;' }, + ]); + + expect(results).toEqual([{ paths: ['src/native-stall.ts'] }]); + expect(TimeoutThenHealthyWorker.instances.length).toBeGreaterThanOrEqual(2); + expect(TimeoutThenHealthyWorker.instances[0].unrefCalls).toBe(1); + expect(TimeoutThenHealthyWorker.instances[0].terminateCalls).toBe(0); + + await pool.terminate(); + + expect(TimeoutThenHealthyWorker.instances[0].terminateCalls).toBe(1); + } finally { + await pool.terminate(); + } + }); + + it('terminates a retired worker once it returns to a JS-visible safe point', async () => { + TimeoutThenHealthyWorker.firstWorkerBehavior = 'delayed-safe-return'; + TimeoutThenHealthyWorker.safeReturnDelayMs = 35; + const pool = createWorkerPool(workerUrl, 1, { + subBatchIdleTimeoutMs: 10, + maxTimeoutRetries: 1, + timeoutBackoffFactor: 2, + workerFactory: () => + new TimeoutThenHealthyWorker() as unknown as import('node:worker_threads').Worker, + }); + + try { + const results = await pool.dispatch<{ path: string; content: string }, { paths: string[] }>([ + { path: 'src/native-stall.ts', content: 'const x = 1;' }, + ]); + + expect(results).toEqual([{ paths: ['src/native-stall.ts'] }]); + expect(TimeoutThenHealthyWorker.instances[0].unrefCalls).toBe(1); + await waitFor( + () => TimeoutThenHealthyWorker.instances[0]?.terminateCalls === 1, + 'Timed out waiting for retired worker to terminate after safe signal', + ); + + await pool.terminate(); + + expect(TimeoutThenHealthyWorker.instances[0].terminateCalls).toBe(1); + } finally { + await pool.terminate(); + } + }); + + it('terminates retired workers when the circuit breaker shuts the pool down', async () => { + const pool = createWorkerPool(workerUrl, 1, { + subBatchIdleTimeoutMs: 10, + maxTimeoutRetries: 1, + timeoutBackoffFactor: 2, + consecutiveFailureThreshold: 1, + workerFactory: () => + new TimeoutThenHealthyWorker() as unknown as import('node:worker_threads').Worker, + }); + + try { + await expect( + pool.dispatch<{ path: string; content: string }, { paths: string[] }>([ + { path: 'src/native-stall.ts', content: 'const x = 1;' }, + ]), + ).rejects.toThrow(/circuit breaker/i); + + await waitFor( + () => TimeoutThenHealthyWorker.instances[0]?.terminateCalls === 1, + 'Timed out waiting for circuit breaker cleanup to terminate retired worker', + ); + } finally { + await pool.terminate(); + } + }); +});