mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-09-08 22:22:52 +00:00
* 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>
706 lines
27 KiB
TypeScript
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'),
|
|
);
|
|
});
|
|
});
|