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(); + } + }); +});