GitNexus/gitnexus/test/unit/worker-pool-resilience.test.ts
Gergő Magyar c4b69402e1
feat(workers): self-healing worker pool + deferred-resolution observability (#1741) (#1947)
* fix(workers): fail fast instead of silently degrading on worker-pool startup failure (#1741)

When an explicitly-sized worker pool (--workers <N>) fails to start because
every worker crashes during top-of-script init, the parse phase used to log a
swallowed `logger.warn` and silently fall back to the ~10x slower sequential
parser. In #1741 (rc99) that turned a worker-startup regression into a
123-minute "stuck" parse with no explanation.

This change:
- Surfaces the real crash: the pool now spawns workers with `{ stderr: true }`,
  tees + captures each worker's stderr, and attaches the tail to its
  readiness-failure messages (propagated via
  WorkerPoolInitializationError.readinessFailures). "did not report ready"
  now carries the underlying native-binding/import error.
- Gates the fallback: when --workers was explicit and fallback was not opted
  into, a total startup failure throws an actionable error instead of
  degrading. Auto-sized pools still fall back, but loudly (logger.error +
  progress warning). New --allow-sequential-fallback flag (+ i18n) opts back in.
- Adds env-gated worker bootstrap-stage logging (GITNEXUS_WORKER_BOOTSTRAP /
  --verbose): imports+grammars loaded -> ready sent -> first task received, so
  a slow/crashing startup is diagnosable.

Tests: all-workers-failed gating (fatal vs loud degrade), stderr surfacing,
and the updated lazy-cache fallback contract (opt-in flag + fail-fast).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(ingestion): always-on slow-file watchdog for deferred call resolution (#1741)

The original #1741 symptom is a run that appears stuck at "Resolving calls
(all chunks)... (9000/18066 files)" — the progress bar freezes inside a single
file's call resolution and nothing reaches the log. Rich per-file deferred
diagnostics already exist, but only behind --verbose / GITNEXUS_PROFILE_DEFERRED,
so a plain `analyze` run gives the user a frozen bar and silence.

Add an always-on (not verbose-gated) per-file watchdog in
processCallsFromExtracted: when a single file's call resolution exceeds
alwaysOnSlowFileWarnMs() (default 15s, override GITNEXUS_SLOW_FILE_WARN_MS,
0 disables) it emits a throttled logger.warn naming the culprit file and the
files-resolved-so-far — turning the silent stall into one actionable line.
Throttled (>=30s between warnings) so a genuinely slow repo can't storm the log.
The watchdog is observation-only; resolution behavior is unchanged.

Note: deliberately did NOT add a heritage child x parent product cap — the name
lookups are O(1) (type-registry Map.get) and the product is bounded, so the
heritage build is not the bottleneck; a cap would risk dropping real edges for
no measured gain.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(ingestion): worker-vs-sequential parity guard for binding/edge collapse (#1741)

rc99 produced almost no bindings/edges (13 bindings vs rc91's 106,305) because
a worker-path failure left extracted results unmerged while the run still
reported success. Rather than an arbitrary "implausibly low" runtime threshold
(which false-positives on legitimately low-binding repos/languages), pin the
invariant directly: for the same repo, worker mode and sequential mode must
produce the same graph.

The test runs the ts-simple cross-file fixture through worker mode
(workerPoolSize + lowered threshold) and sequential mode (skipWorkers), and
asserts: usedWorkerPool is true/false respectively (guards the test itself
against a silent fallback masking divergence), identical CALLS/IMPORTS/DEFINES/
HAS_METHOD edge sets and Class/Function/Method defs, and non-zero CALLS/IMPORTS
(the rc99 collapse signature).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(workers): arm fail-fast for env-sized pools + fix watchdog /0 denominator (#1741)

Addresses two review findings on the #1741 worker-startup PR:

- Fail-fast gate missed the env channel. `explicitWorkers` keyed only off
  the `--workers` flag, so a pool sized via `GITNEXUS_WORKER_POOL_SIZE`
  (with no `--workers`) silently degraded to sequential on a total
  worker-startup crash — reproducing the original #1741 symptom for
  env-channel operators. The gate now arms on a non-zero size from either
  channel, via a single-source `envWorkerPoolSize()` helper exported from
  worker-pool.ts (also rewired through resolveAutoPoolSize). The fatal
  message now names the channel actually used instead of "--workers undefined".

- Always-on slow-file watchdog printed "Resolved N/0 files". `resolvedTotal`
  was pre-counted only on the profile path, but the watchdog reads it on
  every run, so a plain `analyze` showed a bogus /0 denominator on exactly
  the unprofiled hang the watchdog exists to explain. Pre-count now runs
  whenever its result is read (profile path OR watchdog active).

Tests: strengthened the watchdog test to assert "1/1" (not "/0"); added
env-channel fail-fast/degrade cases and made the gating suite hermetic
against an ambient GITNEXUS_WORKER_POOL_SIZE.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(workers): self-healing worker pool replaces the fail-fast flag (#1741)

Replaces the interim --allow-sequential-fallback flag with automatic,
bounded self-healing in the worker pool — industry-standard supervision
(OTP restart-intensity, systemd StartLimit, circuit-breaker, AWS jittered
backoff) translated to the Node worker_threads pool.

worker-pool.ts — bounded startup self-heal (the missing layer):
- A worker that crashes during top-of-script init is now RETRIED with
  capped, full-jitter backoff (BASE 250ms, CAP 2s) up to a small per-slot
  budget, so a transient blip heals itself with no operator action. The
  prior code dropped an unready initial slot on its first crash.
- A DETERMINISTIC crash-loop (>=2 fresh workers crash with the same
  normalized signature before any reaches ready — the #1741 missing
  native-binding case) is detected and short-circuited, so the pool gives
  up in ~1s instead of burning every slot's budget. Correctness rests on
  the STRUCTURAL signal (zero workers ever ready + budget exhausted), so a
  missed signature only costs a few seconds, never a misfire; even a
  stderr-less crash groups via its normalized "exited with code N" message.
- Backoff sleeps are cancellable (unref'd timer + abort on terminate), so
  terminate() can't be wedged for the backoff duration.
- WorkerPoolInitializationError now carries a crashClass for an accurate,
  flag-free message. The runtime respawn/breaker path is unchanged.

parse-impl.ts — collapse to automatic fail-fast:
- handleWorkerStartupFailure always logs the real cause then THROWS with
  the captured crash + `--workers 0` as the explicit sequential escape.
  No more degrade branch; no dependence on how the pool was sized. This is
  reached only after the bounded self-heal is exhausted, so it can't
  resurrect the #1741 silent 123-minute sequential grind. Construction
  failure (broken install) also fails fast instead of degrading silently.

Removed --allow-sequential-fallback end to end (CLI, run-analyze, pipeline,
i18n). --workers 0 remains the explicit "parse sequentially" path; one flag
removed, none added. Grounded in a research+critique pass; the critique's
hazards (N-parallel race, empty-stderr timing, non-cancellable sleep,
runtime-breaker regression) are addressed or scoped out by design.

Tests: startup self-heal (transient recovers; deterministic fails fast
without burning the budget); gating test rewritten to the fail-fast-always
contract; obsolete degrade test removed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(workers): ref + cancel startup backoff so transient retries aren't dropped (#1741 U1)

abortableSleep unref'd its backoff timer, so a transient startup retry could be
silently dropped if that timer was the last ref'd handle on the event loop —
the process could exit mid-recovery. Keep the timer ref'd (a pending retry is
necessary work) and register a cancel fn in a pool-scoped set; terminate() now
clears pending backoffs so it can't be wedged for the backoff cap. A normally
fired timer self-deregisters (clear-on-settle), so no timer lingers after a
slot's retry loop exits. Exposes pendingStartupTimers in getStats.

Tests: terminate-during-backoff cancels + spawns nothing after (R2); the
recovery test now asserts no startup timer lingers after settle (R1).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(workers): route GITNEXUS_WORKER_POOL_SIZE=0 to sequential, not a phantom fail-fast (#1741 U2)

env=0 (no --workers) built a size-0 pool that threw a fabricated "retry budget
exhausted / native binding" crash. The shouldUseWorkers gate now routes env=0
to the sequential path before pool construction — but only when no explicit
--workers <N> was given, so an explicit positive size wins over an ambient
env=0. The route emits one log line so the undocumented (possibly accidental)
env=0 case is observable instead of a silent degrade.

envWorkerPoolSize is un-exported (module-internal sizing reader); a new
workerPoolDisabledByEnv() predicate serves the gate. Empty/whitespace env is
now treated as unset (auto formula), not 0 — an empty assignment is an accident,
not a request for zero workers. Reattached the detached resolveAutoPoolSize
JSDoc and corrected the stale docstring.

Tests: env=0 → sequential (no spawn); explicit --workers wins over env=0;
workerPoolDisabledByEnv unit (0=true, positive/empty/invalid=false); getStats
shape updated for pendingStartupTimers.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(workers): make deterministic crash-loop detection conservative (#1741 U3)

The old tally counted crash EVENTS in a shared signature->count map, so a
simultaneous transient crash storm (e.g. spawn EAGAIN under fork pressure) or
a single slot crashing identically twice falsely tripped "deterministic" and
hard-aborted work that would have self-healed. Replace it: a crash counts
toward deterministic only after its signature REPRODUCES across a respawn on
the same slot, and the short-circuit fires once >=2 distinct slots reproduced
(or 1 for a size-1 pool). Every slot now gets >=1 self-heal attempt before any
short-circuit; the structural budget floor still bounds the worst case.

crashSignature now also collapses Windows backslash paths and bare (no-0x) hex
runs so the fast-path fires on those platforms; exported for unit testing.

Tests: simultaneous storm self-heals (the discriminator vs an attempt-0 rule);
distinct-per-attempt crashes classify transient-exhausted; single-slot
reproduction classifies deterministic; crashSignature normalization unit.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(workers): class-aware startup failure hint + reattach detached JSDoc (#1741 U4)

The "often a missing/broken native binding" hint was appended to every
failure class, including a pool *construction* failure where no worker ever
ran (a missing build / bad worker path). Make the hint class-aware: keep it
for the readiness/init classes, use a construction-specific hint otherwise,
and surface the construction error (e.g. "Worker script not found: …")
verbatim. Reattach the waitForWorkerReady JSDoc that the stderr-capture block
had detached from its function. (The abortableSleep docstring was already
corrected in U1.)

Tests: construction message surfaces the real error + drops the native-binding
guess; deterministic/transient messages keep the hint (regression guard).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-31 13:52:04 +01:00

706 lines
27 KiB
TypeScript

import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest';
import { EventEmitter } from 'node:events';
import path from 'node:path';
import { pathToFileURL } from 'node:url';
import fs from 'node:fs';
import os from 'node:os';
import {
createWorkerPool,
WorkerPoolDispatchError,
resolveWorkerPoolOptions,
resolveAutoPoolSize,
workerPoolDisabledByEnv,
crashSignature,
} from '../../src/core/ingestion/workers/worker-pool.js';
/**
* The pool now sends sub-batch dispatches via native `worker.postMessage`
* with the shape `{type:'sub-batch', files:[{path, content: Uint8Array}]}`.
* Test action logic inspects only `msg.files[*].path`, but the Uint8Array
* content is decoded back to a string here so any future test that
* reads it sees the legacy POJO shape.
*/
const __sharedDecoder = new TextDecoder('utf-8');
function decodeDispatchedMessage(rawMsg: unknown): unknown {
if (
rawMsg !== null &&
typeof rawMsg === 'object' &&
(rawMsg as { type?: unknown }).type === 'sub-batch' &&
Array.isArray((rawMsg as { files?: unknown }).files)
) {
const files = (rawMsg as { files: Array<{ path: string; content: Uint8Array | string }> })
.files;
return {
type: 'sub-batch',
files: files.map((f) => ({
path: f.path,
content: typeof f.content === 'string' ? f.content : __sharedDecoder.decode(f.content),
})),
};
}
return rawMsg;
}
/**
* Minimal `node:worker_threads` Worker double for unit-testing the pool's
* resilience layers (auto-respawn, circuit breaker, quarantine, retry
* budget). Tests script behaviour via `nextActions`: each action runs on
* the next dispatched sub-batch postMessage. `'crash'` and `'exit'` mimic
* real worker failures; `'parse-ok'` mimics a healthy completion.
*/
type FakeWorkerAction =
| { kind: 'parse-ok'; files: { path: string }[]; result?: unknown }
| { kind: 'crash-exit'; code: number; afterStartingFiles?: number }
| { kind: 'crash-error'; message: string; afterStartingFiles?: number };
const nextActions: FakeWorkerAction[] = [];
let workerInstances: FakeWorker[] = [];
class FakeWorker extends EventEmitter {
readonly seenMessages: unknown[] = [];
constructor() {
super();
workerInstances.push(this);
// Real Worker fires 'online' asynchronously after the runtime is ready;
// replicate so any code still listening on `online` is satisfied. The
// pool's `waitForWorkerReady` (post-M4) waits for a `{type:'ready'}`
// message instead — emit that too so replacement-worker tests don't
// hit the WORKER_READY_TIMEOUT_MS budget (5s).
queueMicrotask(() => {
this.emit('online');
this.emit('message', { type: 'ready' });
});
}
postMessage(rawMsg: unknown): void {
// U17: production pool now sends Buffer-encoded dispatch frames.
// Decode them here so this in-process mock can keep its existing
// POJO-shaped action-scripting API — the action queue still sees
// `{type, files}` shapes regardless of whether the pool encoded
// the message on the way in. Store the DECODED payload in
// `seenMessages` so test-side introspection assertions (which
// expect `msg.type` / `msg.files`) keep working after the wire
// format flipped to Buffer.
const msg = decodeDispatchedMessage(rawMsg);
this.seenMessages.push(msg);
if (typeof msg !== 'object' || msg === null) return;
const m = msg as { type?: string; files?: { path: string }[] };
if (m.type !== 'sub-batch') return;
const action = nextActions.shift();
if (!action) {
// No script set; behave as a hung worker (no reply) — the idle timer
// will eventually fire. Tests should always script enough actions.
return;
}
queueMicrotask(() => this.runAction(action, m.files ?? []));
}
private async runAction(action: FakeWorkerAction, files: { path: string }[]): Promise<void> {
if (action.kind === 'parse-ok') {
for (const file of action.files) {
this.emit('message', { type: 'starting-file', path: file.path });
}
this.emit('message', { type: 'progress', filesProcessed: action.files.length });
this.emit('message', { type: 'sub-batch-done' });
// sub-batch-done triggers the pool to post {type:'flush'} which we
// ignore in postMessage above (only 'sub-batch' triggers actions).
// For the result, wait one microtask so the flush is observed.
await Promise.resolve();
this.emit('message', {
type: 'result',
data: action.result ?? { fileCount: action.files.length },
});
return;
}
if (action.kind === 'crash-exit') {
const upTo = Math.min(action.afterStartingFiles ?? 0, files.length);
for (let i = 0; i < upTo; i++) {
this.emit('message', { type: 'starting-file', path: files[i].path });
}
this.emit('exit', action.code);
return;
}
if (action.kind === 'crash-error') {
const upTo = Math.min(action.afterStartingFiles ?? 0, files.length);
for (let i = 0; i < upTo; i++) {
this.emit('message', { type: 'starting-file', path: files[i].path });
}
this.emit('error', new Error(action.message));
return;
}
}
async terminate(): Promise<number> {
this.emit('exit', 0);
return 0;
}
removeListener(event: string | symbol, listener: (...args: unknown[]) => void): this {
return super.removeListener(event, listener);
}
}
// Create a real on-disk worker script so createWorkerPool's existsSync gate
// passes. The script is never actually executed because we inject
// FakeWorker via workerFactory; it just has to exist as a file path.
let tempDir: string;
let workerUrl: URL;
beforeEach(() => {
nextActions.length = 0;
workerInstances = [];
tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'gitnexus-worker-pool-resilience-'));
const workerPath = path.join(tempDir, 'fake-worker.js');
fs.writeFileSync(workerPath, '// fake');
workerUrl = pathToFileURL(workerPath) as URL;
});
afterEach(() => {
try {
fs.rmSync(tempDir, { recursive: true, force: true });
} catch {
// best-effort cleanup — directory may already be gone if a test removed it
}
});
describe('worker pool resilience', () => {
it('seeds an empty quarantine on a fresh pool', () => {
const pool = createWorkerPool(workerUrl, 1, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
});
expect(pool.getQuarantinedPaths()).toEqual([]);
void pool.terminate();
});
it('exposes a healthy stats snapshot on a fresh pool', () => {
const pool = createWorkerPool(workerUrl, 3, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
});
expect(pool.getStats?.()).toEqual({
size: 3,
activeSlots: 3,
droppedSlots: 0,
quarantined: 0,
poolBroken: false,
// Code-review F16: `terminated` distinguishes graceful shutdown
// from a circuit-breaker trip. Fresh pool has not been
// terminated.
terminated: false,
// #1741: no startup backoff is pending once every slot reached ready.
pendingStartupTimers: 0,
// U12: every slot starts at generation 0; no respawns yet on a
// fresh pool. Per-slot zeros (not a single scalar) because each
// slot tracks its own respawn history independently.
slotGenerations: [0, 0, 0],
});
void pool.terminate();
});
it('reports droppedSlots + quarantined after a recoverable death', async () => {
const pool = createWorkerPool(workerUrl, 2, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
consecutiveFailureThreshold: 10,
maxRespawnsPerSlot: 0,
});
// Slot 0 dies on its only job; budget=0 means it gets dropped.
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/ok.ts' }],
result: { fileCount: 1 },
});
await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/bad.ts', content: '' },
{ path: 'src/ok.ts', content: '' },
]);
expect(pool.getStats?.()).toEqual({
size: 2,
activeSlots: 1,
droppedSlots: 1,
quarantined: 1,
poolBroken: false,
// F16: pool is still alive (just lost a slot); terminated=false.
terminated: false,
// #1741: a runtime death is unrelated to startup backoff timers.
pendingStartupTimers: 0,
// U12: slot 0 was dropped before any successful respawn (budget=0),
// so its generation stays at 0. Slot 1 never died, also 0.
slotGenerations: [0, 0],
});
await pool.terminate();
});
it('quarantines the in-flight file on worker exit and respawns the slot', 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-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/good.ts' }],
result: { fileCount: 1 },
});
const results = await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/bad.ts', content: '' },
{ path: 'src/good.ts', content: '' },
]);
expect(results).toEqual([{ fileCount: 1 }]);
expect(pool.getQuarantinedPaths()).toEqual(['src/bad.ts']);
// First FakeWorker died; second is the respawn. Total = 2.
expect(workerInstances.length).toBe(2);
await pool.terminate();
});
it('drops a slot after maxRespawnsPerSlot exceeded and continues on other slots', async () => {
const pool = createWorkerPool(workerUrl, 2, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
consecutiveFailureThreshold: 10,
maxRespawnsPerSlot: 1,
});
// Slot 0 dies twice, exceeding budget=1; slot 1 succeeds with the
// requeued remainder.
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/c.ts' }, { path: 'src/d.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: '' },
{ path: 'src/d.ts', content: '' },
]);
expect(results).toEqual([{ fileCount: 2 }]);
// Two bad files quarantined; both pre-crash 'starting-file' targets.
expect(pool.getQuarantinedPaths().sort()).toEqual(['src/a.ts', 'src/b.ts']);
await pool.terminate();
});
it('trips the circuit breaker after consecutiveFailureThreshold deaths', async () => {
const pool = createWorkerPool(workerUrl, 1, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
consecutiveFailureThreshold: 2,
maxRespawnsPerSlot: 5,
});
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);
expect(pool.getQuarantinedPaths().sort()).toEqual(['src/x.ts', 'src/y.ts']);
// Subsequent dispatch on a tripped pool rejects without running anything.
await expect(
pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/z.ts', content: '' },
]),
).rejects.toBeInstanceOf(WorkerPoolDispatchError);
await pool.terminate();
});
it('resets consecutive-failure counter on a successful job', async () => {
const pool = createWorkerPool(workerUrl, 1, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
consecutiveFailureThreshold: 2,
maxRespawnsPerSlot: 5,
});
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/recovered.ts' }],
result: { fileCount: 1 },
});
const r1 = await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/bad.ts', content: '' },
{ path: 'src/recovered.ts', content: '' },
]);
expect(r1).toEqual([{ fileCount: 1 }]);
// Second dispatch: another death. Counter was reset by the prior success,
// so this single failure should not trip the breaker (threshold=2).
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/ok.ts' }],
result: { fileCount: 1 },
});
const r2 = await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/bad2.ts', content: '' },
{ path: 'src/ok.ts', content: '' },
]);
expect(r2).toEqual([{ fileCount: 1 }]);
expect(pool.getQuarantinedPaths().sort()).toEqual(['src/bad.ts', 'src/bad2.ts']);
await pool.terminate();
});
it('filters already-quarantined paths from new dispatches', async () => {
const pool = createWorkerPool(workerUrl, 1, {
workerFactory: () => new FakeWorker() as unknown as import('node:worker_threads').Worker,
consecutiveFailureThreshold: 5,
maxRespawnsPerSlot: 3,
});
// First dispatch: quarantine 'src/poison.ts'
nextActions.push({ kind: 'crash-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/a.ts' }],
result: { fileCount: 1 },
});
await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/poison.ts', content: '' },
{ path: 'src/a.ts', content: '' },
]);
expect(pool.getQuarantinedPaths()).toEqual(['src/poison.ts']);
// Second dispatch including the quarantined file: pool filters before
// workers see it. The action should never be popped because the only
// dispatchable item is src/b.ts.
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/b.ts' }],
result: { fileCount: 1 },
});
const results = await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/poison.ts', content: '' },
{ path: 'src/b.ts', content: '' },
]);
expect(results).toEqual([{ fileCount: 1 }]);
// The most recent sub-batch the pool dispatched is the dispatch-2
// payload. With poison already in the quarantine when dispatch 2 ran,
// the pool must have filtered it out before reaching a worker.
const allSubBatches = workerInstances
.flatMap((w) => w.seenMessages)
.filter(
(m): m is { type: string; files: { path: string }[] } =>
typeof m === 'object' && m !== null && (m as { type?: string }).type === 'sub-batch',
);
const lastSubBatch = allSubBatches[allSubBatches.length - 1];
expect(lastSubBatch.files.map((f) => f.path)).toEqual(['src/b.ts']);
await pool.terminate();
});
it('returns an empty result without dispatching when every item is quarantined', 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-exit', code: 134, afterStartingFiles: 1 });
nextActions.push({
kind: 'parse-ok',
files: [{ path: 'src/a.ts' }],
result: { fileCount: 1 },
});
await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/poison.ts', content: '' },
{ path: 'src/a.ts', content: '' },
]);
const baselineWorkers = workerInstances.length;
const results = await pool.dispatch<{ path: string; content: string }, unknown>([
{ path: 'src/poison.ts', content: '' },
]);
expect(results).toEqual([]);
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('resolveAutoPoolSize', () => {
it('honors GITNEXUS_WORKER_POOL_SIZE env override (positive integer)', () => {
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '12');
try {
expect(resolveAutoPoolSize()).toBe(12);
} finally {
vi.unstubAllEnvs();
}
});
it('honors GITNEXUS_WORKER_POOL_SIZE=0 (sequential-fallback signal)', () => {
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '0');
try {
expect(resolveAutoPoolSize()).toBe(0);
} finally {
vi.unstubAllEnvs();
}
});
it('honors GITNEXUS_WORKER_POOL_SIZE override above the auto cap', () => {
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '32');
try {
expect(resolveAutoPoolSize()).toBe(32);
} finally {
vi.unstubAllEnvs();
}
});
it('ignores invalid env values and falls back to the auto formula', () => {
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', 'abc');
try {
const expected = Math.min(16, Math.max(1, os.cpus().length - 1));
expect(resolveAutoPoolSize()).toBe(expected);
} finally {
vi.unstubAllEnvs();
}
});
it('matches the auto formula min(16, max(1, cores - 1)) with no env override', () => {
// Exact-count per DoD §2.7: compute the expected value the same
// way the resolver does so the assertion stays deterministic on
// any machine.
const expected = Math.min(16, Math.max(1, os.cpus().length - 1));
expect(resolveAutoPoolSize()).toBe(expected);
});
it('returns an integer (never a float)', () => {
expect(Number.isInteger(resolveAutoPoolSize())).toBe(true);
});
});
describe('workerPoolDisabledByEnv (#1741 — env=0 → sequential signal)', () => {
afterEach(() => {
vi.unstubAllEnvs();
});
it('is true only for a literal GITNEXUS_WORKER_POOL_SIZE=0', () => {
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '0');
expect(workerPoolDisabledByEnv()).toBe(true);
});
it('is false for a positive env size (the pool is used)', () => {
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '4');
expect(workerPoolDisabledByEnv()).toBe(false);
});
it('treats empty/whitespace as unset (not a disable signal — auto formula applies)', () => {
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', '');
expect(workerPoolDisabledByEnv()).toBe(false);
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', ' ');
expect(workerPoolDisabledByEnv()).toBe(false);
});
it('is false for an invalid value', () => {
vi.stubEnv('GITNEXUS_WORKER_POOL_SIZE', 'abc');
expect(workerPoolDisabledByEnv()).toBe(false);
});
});
describe('crashSignature (#1741 — deterministic-loop fingerprint normalization)', () => {
it('collapses Windows backslash temp paths that differ only in a random token', () => {
const a = crashSignature("Cannot find module 'C:\\Users\\ci\\Temp\\worker-7f3a.js'");
const b = crashSignature("Cannot find module 'C:\\Users\\ci\\Temp\\worker-2b9c.js'");
expect(a).toBe(b);
});
it('collapses bare (no-0x) hex backtrace tokens', () => {
expect(crashSignature('SIGSEGV at 00007f8a2b1c4d')).toBe(
crashSignature('SIGSEGV at 00007fcc3d2e5a'),
);
});
it('collapses POSIX paths and exit codes (stderr-less crashes still group)', () => {
expect(crashSignature('Worker exited with code 1 (/tmp/pool-9/w.js)')).toBe(
crashSignature('Worker exited with code 139 (/tmp/pool-4/w.js)'),
);
});
it('keeps genuinely different crashes distinct', () => {
expect(crashSignature('Error: Cannot find module tree-sitter-c-sharp')).not.toBe(
crashSignature('Error: out of memory'),
);
});
});