scriptoriumadmin/GitNexus
1
0
Fork 0
mirror of https://github.com/abhigyanpatwari/GitNexus.git synced 2026-09-24 00:51:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(workers): apply code-review fixes (12 findings)

Browse source 
Walks through every finding from ce-code-review run
20260519-094648-3549cf5e. All 12 picked Apply.

Critical:
- F1 — Layer 5 cumulative-timeout exhaustion no longer silently drops
  the rest of the job. `requeueRemainder` is now invoked before
  `handleWorkerDeath` in both Layer 5 and singleton-final-fail give-up
  paths so non-quarantined items get re-tried by another worker.
- F2 — idle-timer recovery overhaul. `!shouldContinue` branch no
  longer calls `replaceWorker` (double-spawn race with the
  `handleWorkerDeath` inside `requeueAfterTimeout`). `shouldContinue`
  branch now enforces `maxRespawnsPerSlot` before respawning, closing
  the budget-bypass for the timeout-retry path. Also fixes premature
  `maybeDone` by simplifying the bookkeeping.
- F3 — `requeueRemainder` no longer pre-charges `cumulativeTimeoutMs`
  by `job.timeoutMs`. The death itself consumed no budget, so the
  next `requeueAfterTimeout` was double-billing the first attempt.
- F4 — `WorkerPool.getQuarantinedPaths` is now optional on the
  interface, matching the defensive `?.()` call site and the existing
  mocks. Removes the contract-vs-callsite contradiction.
- F5 — per-job unattributed-death tracking. When a worker dies with
  no exclusion attribution, `requeueRemainder` tracks death count per
  `startIndex`. First time: re-queue intact. Second time: quarantine
  items[0] as best guess, or drop the job entirely when items lack
  paths. Bounds the death loop the original design admitted to.
- F6 — per-slot consecutive-failure counter. Replaces the pool-wide
  scalar so a chronically-failing slot trips the breaker on its own
  streak instead of being masked by another slot's successes.

Smaller:
- F7 — exhaustiveness `never` check on `WorkerOutgoingMessage` union.
- F8 — recursive `runWorker` on fully-quarantined jobs converted to
  a while-loop.
- F9 — `tripBreaker` calls `reject(err)` BEFORE awaiting
  `worker.terminate()`. A stuck terminate no longer blocks the caller.
- F10 — `parsing-processor.ts` quarantine log de-duplicates per pool
  instance via a `WeakMap`. Only newly-quarantined paths are logged
  in each chunk; the per-chunk count still surfaces via progress.
- F11 — extract `firstPath` local in `requeueAfterTimeout`; eliminates
  double `itemPath` call and the `unknown as string` cast.

Tests (F12, 6 new):
- crash-error event path (errorHandler).
- F5 drop-branch coverage via items without `.path`.
- Common-case unattributable crash falling back to items[0] heuristic.
- `replaceWorker` startup failure (workerFactory emits 'exit' before
  'online').
- All-slots-dropped breaker trip.
- `GITNEXUS_WORKER_MAX_CUMULATIVE_TIMEOUT_MS` env override.

Residual gap (deferred): no unit test exercises the Layer 5
cumulative-budget runtime path — requires fake-timer interleaving
with FakeWorker that's too brittle for this iteration. Tracked.

Unit suite: 257 files / 6056 passed / 30 skipped / 0 failed.
This commit is contained in:
Gergo Magyar 2026-05-19 11:04:21 +01:00
parent 701c5a67a4
commit e6f181493c
3 changed files with 373 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
gitnexus/src/core/ingestion/parsing-processor.ts
View file

@ -832,6 +832,14 @@ const processParsingSequential = async (
// Public API
// ============================================================================
/**
* Per-`WorkerPool` log-dedup state for quarantine reporting. Keyed on the
* pool instance so multiple concurrent pools (test fixtures, future
* multi-pool callers) each get their own seen-set. WeakMap entries vanish
* when the pool is garbage-collected.
*/
const loggedQuarantineByPool = new WeakMap<WorkerPool, Set<string>>();
export const processParsing = async (
graph: KnowledgeGraph,
files: { path: string; content: string }[],
@ -890,14 +898,34 @@ export const processParsing = async (
// out of dispatch; we only need to log + progress-report. Quarantine
// is session-scoped per pool instance — a fresh `createWorkerPool`
// call clears it.
const quarantineSet = new Set(workerPool.getQuarantinedPaths?.() ?? []);
//
// Dedup: log full path list only for entries newly quarantined since
// the previous dispatch on the same pool. The per-chunk progress
// message still surfaces the count for UX continuity, but the
// structured `quarantinedFiles` payload is only emitted when there
// is new signal — prevents O(quarantine × chunks) log spam.
const quarantineSnapshot = workerPool.getQuarantinedPaths?.() ?? [];
const quarantineSet = new Set(quarantineSnapshot);
if (quarantineSet.size > 0) {
const quarantinedInChunk = files.filter((file) => quarantineSet.has(file.path));
if (quarantinedInChunk.length > 0) {
logger.warn(
{ quarantinedFiles: quarantinedInChunk.map((file) => file.path) },
`Worker quarantine: ${quarantinedInChunk.length} file(s) skipped in this chunk (cumulative pool quarantine: ${quarantineSet.size}).`,
);
const seenForPool = loggedQuarantineByPool.get(workerPool) ?? new Set<string>();
const newlyQuarantined = quarantinedInChunk
.map((file) => file.path)
.filter((p) => !seenForPool.has(p));
for (const p of newlyQuarantined) seenForPool.add(p);
loggedQuarantineByPool.set(workerPool, seenForPool);
if (newlyQuarantined.length > 0) {
logger.warn(
{
newlyQuarantined,
cumulativeQuarantine: quarantineSet.size,
chunkSkipped: quarantinedInChunk.length,
},
`Worker quarantine: ${newlyQuarantined.length} new file(s) skipped this chunk ` +
`(${quarantinedInChunk.length} skipped total, ${quarantineSet.size} cumulative).`,
);
}
reportProgress?.(
lastProgress,
files.length,

221
gitnexus/src/core/ingestion/workers/worker-pool.ts
View file

@ -31,8 +31,13 @@ export interface WorkerPool {
* worker dies with an authoritative in-flight file (Layer 4 starting-file
* message) or a singleton-timeout exclusion. Cleared only by pool teardown
* — quarantine is session-scoped per `createWorkerPool` invocation.
*
* Optional so external `WorkerPool` shapes (test doubles, alternate
* implementations) can omit the method without compile errors. Callers
* (`processParsing`) use optional chaining at the call site to handle
* absence gracefully.
*/
getQuarantinedPaths(): readonly string[];
getQuarantinedPaths?(): readonly string[];
}
export interface WorkerPoolOptions {
@ -328,7 +333,11 @@ export const createWorkerPool = (
const respawnCount: number[] = new Array(size).fill(0);
const activeSlots: Set<number> = new Set();
const quarantined: Set<string> = new Set();
let consecutiveFailures = 0;
// Per-slot consecutive-failure counter (F6): replaces the prior pool-wide
// scalar so a chronically-failing slot trips the breaker on its own
// failure streak instead of being masked by another slot's successes.
// Reset to 0 on that slot's next successful job.
const consecutiveFailuresPerSlot: number[] = new Array(size).fill(0);
let poolBroken = false;
let poolFailure: Error | undefined;
@ -380,6 +389,11 @@ export const createWorkerPool = (
// Tracks which slots are currently mid-job so the "wake idle slots"
// pass after a requeue doesn't double-dispatch to a busy slot.
const busySlots: Set<number> = new Set();
// Per-conceptual-job (identified by startIndex) death count for the
// unattributable-crash path (F5). On the 2nd time a job dies with
// no exclusion attribution, requeueRemainder quarantines items[0]
// as a best-guess culprit to break the death loop.
const unattributedJobDeaths: Map<number, number> = new Map();
let completedFiles = 0;
let activeWorkers = 0;
let stopped = false;
@ -434,15 +448,21 @@ export const createWorkerPool = (
// outer dispatch promise with the cumulative exclude paths. This is the
// ONLY place that sets `poolBroken = true` — recoverable single-worker
// failures stay local to `handleWorkerDeath`.
const tripBreaker = async (err: WorkerPoolDispatchError) => {
//
// Reject the caller's promise BEFORE awaiting `worker.terminate()` so a
// stuck terminate (OOM-killed thread, hung native addon) can't block
// the caller indefinitely. Worker cleanup runs in the background; the
// next `dispatch` call sees `poolBroken=true` and rejects up front.
const tripBreaker = (err: WorkerPoolDispatchError) => {
poolBroken = true;
poolFailure = err;
if (stopped) return;
stopped = true;
await Promise.all(workers.map((worker) => worker?.terminate().catch(() => undefined)));
reject(err);
const liveWorkers = workers.slice();
for (let i = 0; i < workers.length; i++) workers[i] = undefined;
activeSlots.clear();
reject(err);
void Promise.all(liveWorkers.map((worker) => worker?.terminate().catch(() => undefined)));
};
const maybeDone = () => {
@ -460,27 +480,66 @@ export const createWorkerPool = (
// healthy worker can finish the work. Earlier items in the dead job
// were never flushed back to the main thread, so they must be
// re-processed. The new job carries the existing job's startIndex so
// result ordering is preserved.
// result ordering is preserved. `cumulativeTimeoutMs` is carried
// forward unchanged — the death itself consumed no timeout budget,
// so charging another timeoutMs here would double-bill the next
// `requeueAfterTimeout` call's accumulation.
//
// Unattributed-death tracking (F5): when called with `excluded=[]`
// the worker died without identifying a culprit (no `starting-file`
// observed, `lastProgress=0`, `items[lastProgress]` heuristic empty).
// The first time, re-queue the job intact and hope another worker
// succeeds. On the second such death of the SAME conceptual job
// (same `startIndex`), quarantine `items[0]` as a best-guess
// culprit so the next attempt isn't condemned to the same death.
// This bounds the unattributable-crash death loop and ensures the
// pool's final `fallbackExcludePaths` carries SOME signal for
// sequential fallback instead of silently re-hitting the bad file.
const requeueRemainder = (job: WorkerJob<TInput>, excluded: readonly string[]) => {
let effectiveExcluded = excluded;
if (excluded.length === 0) {
jobs.unshift(job);
return;
const deaths = (unattributedJobDeaths.get(job.startIndex) ?? 0) + 1;
unattributedJobDeaths.set(job.startIndex, deaths);
if (deaths < 2) {
jobs.unshift(job);
return;
}
const firstPath = itemPath(job.items[0]);
if (firstPath !== undefined) {
quarantined.add(firstPath);
logger.warn(
{ startIndex: job.startIndex, firstPath, deaths },
`Conceptual job ${job.startIndex} died ${deaths} times unattributably; ` +
`quarantining items[0] (${firstPath}) as best-guess culprit.`,
);
effectiveExcluded = [firstPath];
} else {
// No identifiable file on items[0] either — drop the job to
// break the loop. The breaker counter still increments via
// handleWorkerDeath, so consecutive unattributable deaths
// eventually trip it even without quarantine signal.
logger.warn(
{ startIndex: job.startIndex, deaths },
`Conceptual job ${job.startIndex} died ${deaths} times unattributably with ` +
`no identifiable file; dropping job to break the death loop.`,
);
return;
}
}
const excludeSet = new Set(excluded);
const excludeSet = new Set(effectiveExcluded);
const filtered = job.items.filter((item) => {
const p = itemPath(item);
return p === undefined || !excludeSet.has(p);
});
if (filtered.length === 0) return;
const requeueTimeoutMs = job.timeoutMs;
jobs.unshift({
startIndex: job.startIndex,
items: filtered,
estimatedBytes: filtered.reduce((sum, item) => sum + estimateItemBytes(item), 0),
attempt: job.attempt,
splitDepth: job.splitDepth,
timeoutMs: requeueTimeoutMs,
cumulativeTimeoutMs: job.cumulativeTimeoutMs + requeueTimeoutMs,
timeoutMs: job.timeoutMs,
cumulativeTimeoutMs: job.cumulativeTimeoutMs,
});
};
@ -494,15 +553,16 @@ export const createWorkerPool = (
excludePaths: readonly string[],
) => {
if (stopped) return;
consecutiveFailures++;
consecutiveFailuresPerSlot[workerIndex]++;
for (const p of excludePaths) {
if (p) quarantined.add(p);
}
if (consecutiveFailures >= poolOptions.consecutiveFailureThreshold) {
void tripBreaker(
if (consecutiveFailuresPerSlot[workerIndex] >= poolOptions.consecutiveFailureThreshold) {
tripBreaker(
new WorkerPoolDispatchError(
`${reason}. Pool circuit breaker tripped after ${consecutiveFailures} ` +
`consecutive failures (threshold: ${poolOptions.consecutiveFailureThreshold}).`,
`${reason}. Pool circuit breaker tripped: slot ${workerIndex} hit ` +
`${consecutiveFailuresPerSlot[workerIndex]} consecutive failures ` +
`(threshold: ${poolOptions.consecutiveFailureThreshold}).`,
Array.from(quarantined),
),
);
@ -524,7 +584,7 @@ export const createWorkerPool = (
workers[workerIndex] = undefined;
activeSlots.delete(workerIndex);
if (activeSlots.size === 0) {
void tripBreaker(
tripBreaker(
new WorkerPoolDispatchError(
`${reason}. All ${size} worker slot(s) exhausted their respawn budget.`,
Array.from(quarantined),
@ -547,7 +607,7 @@ export const createWorkerPool = (
if (!respawned) {
activeSlots.delete(workerIndex);
if (activeSlots.size === 0) {
void tripBreaker(
tripBreaker(
new WorkerPoolDispatchError(
`${reason}. Replacement worker startup failed and no slots remain.`,
Array.from(quarantined),
@ -571,11 +631,12 @@ export const createWorkerPool = (
// exhausted, surface the in-flight file via WorkerPoolDispatchError
// instead of letting exponential backoff stall further.
if (nextCumulative > poolOptions.maxCumulativeTimeoutMs) {
const exhausted =
const firstPath = itemPath(job.items[0]);
const exhausted: string[] =
inFlightPath !== undefined
? [inFlightPath]
: itemPath(job.items[0])
? [itemPath(job.items[0]) as string]
: firstPath !== undefined
? [firstPath]
: [];
logger.warn(
{
@ -587,6 +648,10 @@ export const createWorkerPool = (
},
`Worker ${workerIndex} parse job exhausted cumulative timeout budget. Surfacing in-flight file(s).`,
);
// Re-queue the rest of the job so other workers can finish the
// non-exhausted items. Without this, the job was already shifted
// off `jobs` in `runWorker` and would be silently lost.
requeueRemainder(job, exhausted);
void handleWorkerDeath(
workerIndex,
`Worker ${workerIndex} parse job exhausted cumulative timeout budget ` +
@ -668,6 +733,12 @@ export const createWorkerPool = (
},
`Worker ${workerIndex} parse job idle timeout exhausted retries; quarantining file and respawning slot.`,
);
// Defensive re-queue for symmetry with the Layer 5 path. Singleton
// jobs have at most one item; filtering by `excludes` typically
// drops it (filtered.length === 0 → no-op). When `excludes` is
// empty (unidentifiable stall), F5's unattributed-death tracking
// in requeueRemainder bounds the loop.
requeueRemainder(job, excludes);
void handleWorkerDeath(
workerIndex,
`Worker ${workerIndex} parse job idle timeout after ${job.timeoutMs / 1000}s ` +
@ -681,25 +752,19 @@ export const createWorkerPool = (
const runWorker = (workerIndex: number) => {
if (stopped) return;
if (!activeSlots.has(workerIndex)) return;
const job = jobs.shift();
if (!job) {
maybeDone();
return;
}
// Drop quarantined items that may have been re-queued before a death
// added them to quarantine — keeps the worker from ever seeing a
// known-bad file.
if (quarantined.size > 0) {
// known-bad file. Loops until we find a job with dispatchable items
// or exhaust the queue (avoids recursion depth growth when many
// queued jobs are fully quarantined back-to-back).
let job: WorkerJob<TInput> | undefined;
while ((job = jobs.shift()) !== undefined) {
if (quarantined.size === 0) break;
const dispatchable = job.items.filter((item) => {
const p = itemPath(item);
return p === undefined || !quarantined.has(p);
});
if (dispatchable.length === 0) {
// Whole job was quarantined; drop and try next.
runWorker(workerIndex);
return;
}
if (dispatchable.length === 0) continue;
if (dispatchable.length !== job.items.length) {
job.items = dispatchable;
job.estimatedBytes = dispatchable.reduce(
@ -707,6 +772,11 @@ export const createWorkerPool = (
0,
);
}
break;
}
if (!job) {
maybeDone();
return;
}
activeWorkers++;
@ -790,38 +860,61 @@ export const createWorkerPool = (
stalledPath,
);
if (!shouldContinue) {
// handleWorkerDeath path was taken by requeueAfterTimeout;
// recover the slot (respawn if budget allows) and continue.
void (async () => {
activeWorkers--;
busySlots.delete(workerIndex);
if (stopped) return;
if (activeSlots.has(workerIndex)) {
const respawned = await replaceWorker(workerIndex);
if (!respawned) activeSlots.delete(workerIndex);
}
if (stopped) return;
if (activeSlots.has(workerIndex)) {
runWorker(workerIndex);
}
wakeIdleSlots();
maybeDone();
})();
// Give-up path: `requeueAfterTimeout` already kicked off
// `void handleWorkerDeath(...)` which owns slot management
// (quarantine, respawn, budget enforcement, breaker). We
// only need to update local bookkeeping and let
// `wakeIdleSlots` pick up the requeued remainder on any
// live slot. handleWorkerDeath's own `runWorker` post-
// respawn fires from inside that async chain — we do NOT
// re-call replaceWorker here, which previously double-
// spawned the slot.
activeWorkers--;
busySlots.delete(workerIndex);
wakeIdleSlots();
maybeDone();
return;
}
// Timeout-retry path: spawn a fresh worker on this slot to
// pick up the next attempt.
// Timeout-retry path: enforce the per-slot respawn budget
// BEFORE spawning a fresh worker. The previous version
// called `replaceWorker` unconditionally, letting a
// chronically-timing-out slot respawn forever.
void (async () => {
try {
const respawned = await replaceWorker(workerIndex);
if (!respawned) {
respawnCount[workerIndex]++;
if (respawnCount[workerIndex] > poolOptions.maxRespawnsPerSlot) {
logger.warn(
{
workerIndex,
respawnCount: respawnCount[workerIndex],
maxRespawns: poolOptions.maxRespawnsPerSlot,
},
`Worker ${workerIndex} exceeded respawn budget during idle-timeout retry; dropping slot.`,
);
const dead = workers[workerIndex];
await dead?.terminate().catch(() => undefined);
workers[workerIndex] = undefined;
activeSlots.delete(workerIndex);
} else {
const respawned = await replaceWorker(workerIndex);
if (!respawned) {
activeSlots.delete(workerIndex);
}
}
} finally {
activeWorkers--;
busySlots.delete(workerIndex);
}
if (stopped) return;
if (activeSlots.size === 0) {
tripBreaker(
new WorkerPoolDispatchError(
`Worker pool exhausted all slots during idle-timeout retry.`,
Array.from(quarantined),
),
);
return;
}
reportProgress();
if (activeSlots.has(workerIndex)) runWorker(workerIndex);
wakeIdleSlots();
@ -861,7 +954,7 @@ export const createWorkerPool = (
if (!waitingForFlush) {
settled = true;
cleanup();
void tripBreaker(
tripBreaker(
new WorkerPoolDispatchError(
`Worker ${workerIndex} protocol error: result before flush`,
Array.from(quarantined),
@ -873,12 +966,18 @@ export const createWorkerPool = (
cleanup();
results.push({ startIndex: job.startIndex, data: msg.data as TResult });
completedFiles += job.items.length;
// Layer 2: a successful job resets the consecutive-failure
// counter so transient bursts of bad files don't trip the
// breaker prematurely.
consecutiveFailures = 0;
// Layer 2 (F6): a successful job resets THIS slot's
// consecutive-failure counter so the breaker only trips
// when a specific slot is chronically failing — another
// slot's successes can't mask a single bad slot.
consecutiveFailuresPerSlot[workerIndex] = 0;
reportProgress();
finishJob();
} else {
// F7: exhaustiveness check — drift-catcher when a future
// WorkerOutgoingMessage variant is added without a handler.
const _exhaustive: never = msg;
void _exhaustive;
}
};

180
gitnexus/test/unit/worker-pool-resilience.test.ts
View file

@ -306,6 +306,186 @@ describe('worker pool resilience', () => {
expect(workerInstances.length).toBe(baselineWorkers);
await pool.terminate();
});
it('quarantines on worker `error` event (errorHandler path)', async () => {
const pool = createWorkerPool(workerUrl, 1, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
consecutiveFailureThreshold: 5,
maxRespawnsPerSlot: 3,
});
nextActions.push({ kind: 'crash-error', message: 'segfault', afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/ok.ts' }],
result: { fileCount: 1 },
});
const results = await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/bad.ts', content: '' },
{ path: 'src/ok.ts', content: '' },
]);
expect(results).toEqual([{ fileCount: 1 }]);
expect(pool.getQuarantinedPaths?.() ?? []).toEqual(['src/bad.ts']);
await pool.terminate();
});
it('drops the job on second unattributable death when items have no paths (F5 drop branch)', async () => {
const pool = createWorkerPool(workerUrl, 1, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
consecutiveFailureThreshold: 5,
maxRespawnsPerSlot: 5,
});
// Items without a `path` field — itemPath returns undefined, so
// inFlightExcludePath returns [] and F5's unattributed-death branch
// is the only path that fires. First death re-queues intact; second
// death drops the job entirely to break the loop (no identifiable
// file to quarantine).
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 0 });
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 0 });
const results = await pool.dispatch<{ content: string }, unknown>([
{ content: 'no-path-1' },
{ content: 'no-path-2' },
]);
// F5 dropped the job; no results, no quarantine (no path to quarantine).
expect(results).toEqual([]);
expect(pool.getQuarantinedPaths?.() ?? []).toEqual([]);
await pool.terminate();
});
it('common-case unattributable crash falls back to the items[0] heuristic for attribution', async () => {
const pool = createWorkerPool(workerUrl, 1, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
consecutiveFailureThreshold: 5,
maxRespawnsPerSlot: 5,
});
// Worker dies BEFORE emitting starting-file or progress. The pool's
// heuristic attributes to items[0] (lastProgress=0, items.length>0,
// path-bearing item). Validates the heuristic fallback before F5
// would take over — confirms today's behavior for the most common
// unattributable-crash mode.
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 0 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/clean.ts' }],
result: { fileCount: 1 },
});
const results = await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/heuristic-target.ts', content: '' },
{ path: 'src/clean.ts', content: '' },
]);
expect(results).toEqual([{ fileCount: 1 }]);
expect(pool.getQuarantinedPaths?.() ?? []).toEqual(['src/heuristic-target.ts']);
await pool.terminate();
});
it('drops slot when waitForWorkerOnline rejects (replaceWorker failure path)', 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));
}
return worker as unknown as import('node:worker_threads').Worker;
},
consecutiveFailureThreshold: 10,
maxRespawnsPerSlot: 5,
});
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/b.ts' }, { path: 'src/c.ts' }],
result: { fileCount: 2 },
});
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: '' },
]);
expect(results).toEqual([{ fileCount: 2 }]);
expect(pool.getQuarantinedPaths?.() ?? []).toEqual(['src/a.ts']);
// Initial 2 workers + 1 failed replacement = 3 factory calls.
expect(factoryCallCount).toBe(3);
await pool.terminate();
});
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,
consecutiveFailureThreshold: 100,
maxRespawnsPerSlot: 0,
});
// Both slots die on first job: budget=0 means slot is dropped on first death.
// After both slots dropped, activeSlots.size === 0 trips the breaker.
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
await expect(
pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/x.ts', content: '' },
{ path: 'src/y.ts', content: '' },
]),
).rejects.toBeInstanceOf(WorkerPoolDispatchError);
// After breaker, no respawns happen so workerInstances === initial 2.
expect(workerInstances.length).toBe(2);
await pool.terminate();
});
});
describe('worker pool option resolution', () => {
it('resolves maxRespawnsPerSlot from explicit options', () => {
const opts = resolveWorkerPoolOptions({ maxRespawnsPerSlot: 7 }, 4);
expect(opts.maxRespawnsPerSlot).toBe(7);
});
it('defaults consecutiveFailureThreshold to max(3, poolSize)', () => {
expect(resolveWorkerPoolOptions({}, 1).consecutiveFailureThreshold).toBe(3);
expect(resolveWorkerPoolOptions({}, 8).consecutiveFailureThreshold).toBe(8);
});
it('defaults maxCumulativeTimeoutMs to 5x subBatchIdleTimeoutMs', () => {
const opts = resolveWorkerPoolOptions({ subBatchIdleTimeoutMs: 1000 }, 1);
expect(opts.maxCumulativeTimeoutMs).toBe(5000);
});
it('reads GITNEXUS_WORKER_MAX_RESPAWNS_PER_SLOT env override', () => {
vi.stubEnv('GITNEXUS_WORKER_MAX_RESPAWNS_PER_SLOT', '2');
try {
expect(resolveWorkerPoolOptions({}, 1).maxRespawnsPerSlot).toBe(2);
} finally {
vi.unstubAllEnvs();
}
});
it('reads GITNEXUS_WORKER_CONSECUTIVE_FAILURE_THRESHOLD env override', () => {
vi.stubEnv('GITNEXUS_WORKER_CONSECUTIVE_FAILURE_THRESHOLD', '12');
try {
expect(resolveWorkerPoolOptions({}, 1).consecutiveFailureThreshold).toBe(12);
} finally {
vi.unstubAllEnvs();
}
});
it('reads GITNEXUS_WORKER_MAX_CUMULATIVE_TIMEOUT_MS env override', () => {
vi.stubEnv('GITNEXUS_WORKER_MAX_CUMULATIVE_TIMEOUT_MS', '60000');
try {
expect(resolveWorkerPoolOptions({}, 1).maxCumulativeTimeoutMs).toBe(60000);
} finally {
vi.unstubAllEnvs();
}
});
});
describe('worker pool option resolution', () => {