mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-09-24 00:51:53 +00:00
test(workers): integration tests for resilience layers + fix requeue-after-timeout flow
Adds 6 new real-worker integration tests covering the PR #1693 resilience layers + fixes 3 follow-on bugs surfaced while writing them. New integration coverage (real worker threads + temp fixture scripts): - `respawns the slot after worker process.exit and finishes the work on the replacement` — exercises Layer 1 auto-respawn + Layer 3 quarantine through real IPC. - `attributes exactly via authoritative starting-file message on worker crash` — Layer 4 end-to-end: starting-file message → exact quarantine attribution (not the items[0] heuristic). - `quarantine filters subsequent dispatches without sending to a worker` — second dispatch's sub-batch payload audited via filesystem; the quarantined path is never sent across the message channel. - `drops a slot after maxRespawnsPerSlot and continues on the survivor` — 2-slot pool, slot dies twice past budget, survivor finishes re-queued remainder. - `trips the circuit breaker on cascading per-slot consecutive failures` — single-slot pool, dies on every job, breaker trips after consecutiveFailureThreshold with WorkerPoolDispatchError carrying the cumulative quarantine. - `survives a worker error event (uncaught throw) the same as a process.exit` — validates recoverAndResume on the errorHandler path via a real worker `throw` (not just process.exit). Bug fixes uncovered while writing these tests: 1. **Stack-overflow recursion in runWorker's no-worker branch** — `if (!worker) { ...; wakeIdleSlots(); maybeDone(); }` recursed indefinitely when multiple slots were mid-respawn simultaneously (wakeIdleSlots → runWorker → no worker → wakeIdleSlots → …). Removed the wakeIdleSlots call: the slot's own respawn IIFE owns runWorker post-respawn, and other slots will pick up work via finishJob's runWorker. 2. **requeueAfterTimeout dispatched work before respawn completed** — the F2 fix had `requeueAfterTimeout` `void`-discarding `handleWorkerDeath`, so the `!shouldContinue` IIFE had no way to know when the respawn finished. New design: `requeueAfterTimeout` returns a `TimeoutDecision` discriminated union; the IIFE owns the death-and-respawn-and-dispatch orchestration in an async closure so it can `await handleWorkerDeath` and then call `runWorker` deterministically. 3. **Stalled-singleton + protocol-error + replacement-startup-crash tests** had stale contracts predating the resilience refactor. The stalled-singleton no longer rejects (it quarantines + resolves `[]`); the protocol-error rejection message now mentions "circuit breaker tripped"; the replacement-startup-crash test documents the known `waitForWorkerOnline` race (online fires before the worker's main script runs, so a top-level throw looks like a successful spawn) — the test asserts the file is quarantined via the second-idle-timeout give-up path. Full suite: 334 files / 8982 passed / 43 skipped / 0 failed (second run; first run had a Vitest-reported flake from an uncaught worker exception bleeding into the test report — repeated runs are clean).
This commit is contained in:
Gergo Magyar
parent
e6f181493c
commit
4190aa0f50
2 changed files with 437 additions and 57 deletions
|
|
@ -618,12 +618,20 @@ export const createWorkerPool = (
|
|||
}
|
||||
};
|
||||
|
||||
// Decision returned by `requeueAfterTimeout`. The caller owns the
|
||||
// post-decision orchestration so the death + respawn + dispatch
|
||||
// sequence can `await` cleanly (which is required to know when the
|
||||
// slot is ready to pick up new work after a give-up).
|
||||
type TimeoutDecision =
|
||||
| { kind: 'retry' }
|
||||
| { kind: 'give-up'; reason: string; excludePaths: readonly string[] };
|
||||
|
||||
const requeueAfterTimeout = (
|
||||
workerIndex: number,
|
||||
job: WorkerJob<TInput>,
|
||||
lastProgress: number,
|
||||
inFlightPath: string | undefined,
|
||||
): boolean => {
|
||||
): TimeoutDecision => {
|
||||
const nextTimeout = Math.ceil(job.timeoutMs * poolOptions.timeoutBackoffFactor);
|
||||
const nextCumulative = job.cumulativeTimeoutMs + nextTimeout;
|
||||
|
||||
|
|
@ -648,17 +656,13 @@ 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 ` +
|
||||
return {
|
||||
kind: 'give-up',
|
||||
reason:
|
||||
`Worker ${workerIndex} parse job exhausted cumulative timeout budget ` +
|
||||
`(${(nextCumulative / 1000).toFixed(0)}s > ${(poolOptions.maxCumulativeTimeoutMs / 1000).toFixed(0)}s cap)`,
|
||||
exhausted,
|
||||
);
|
||||
return false;
|
||||
excludePaths: exhausted,
|
||||
};
|
||||
}
|
||||
|
||||
if (job.items.length > 1) {
|
||||
|
|
@ -698,7 +702,7 @@ export const createWorkerPool = (
|
|||
);
|
||||
// Preserve intuitive retry order; final result order is still enforced by startIndex sort.
|
||||
jobs.unshift(first, second);
|
||||
return true;
|
||||
return { kind: 'retry' };
|
||||
}
|
||||
|
||||
const nextAttempt = job.attempt + 1;
|
||||
|
|
@ -719,7 +723,7 @@ export const createWorkerPool = (
|
|||
timeoutMs: nextTimeout,
|
||||
cumulativeTimeoutMs: nextCumulative,
|
||||
});
|
||||
return true;
|
||||
return { kind: 'retry' };
|
||||
}
|
||||
|
||||
const stalledPath = inFlightPath ?? itemPath(job.items[0]);
|
||||
|
|
@ -733,20 +737,14 @@ 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 ` +
|
||||
return {
|
||||
kind: 'give-up',
|
||||
reason:
|
||||
`Worker ${workerIndex} parse job idle timeout after ${job.timeoutMs / 1000}s ` +
|
||||
`(single item${stalledPath ? `: ${stalledPath}` : ''}, ` +
|
||||
`${job.estimatedBytes} bytes, last progress: ${lastProgress})`,
|
||||
excludes,
|
||||
);
|
||||
return false;
|
||||
excludePaths: excludes,
|
||||
};
|
||||
};
|
||||
|
||||
const runWorker = (workerIndex: number) => {
|
||||
|
|
@ -784,12 +782,19 @@ export const createWorkerPool = (
|
|||
inFlightProgress[workerIndex] = 0;
|
||||
const worker = workers[workerIndex];
|
||||
if (!worker) {
|
||||
// Slot was dropped between scheduling and execution; requeue the
|
||||
// job for another slot and bail.
|
||||
// Slot's worker is undefined — typically mid-respawn (replaceWorker
|
||||
// clears `workers[i]` before awaiting `waitForWorkerOnline`). The
|
||||
// respawn IIFE / handleWorkerDeath that started the respawn owns
|
||||
// calling runWorker when the new worker is online; we just
|
||||
// unshift the job and bail.
|
||||
//
|
||||
// Do NOT call wakeIdleSlots from here: it would iterate
|
||||
// `activeSlots` and re-enter `runWorker` for this same slot
|
||||
// (now non-busy), find `workers[i]` still undefined, and
|
||||
// recurse until the call stack overflows.
|
||||
activeWorkers--;
|
||||
busySlots.delete(workerIndex);
|
||||
jobs.unshift(job);
|
||||
wakeIdleSlots();
|
||||
maybeDone();
|
||||
return;
|
||||
}
|
||||
|
|
@ -853,26 +858,22 @@ export const createWorkerPool = (
|
|||
cleanup();
|
||||
inFlightProgress[workerIndex] = 0;
|
||||
const stalledPath = inFlightPath;
|
||||
const shouldContinue = requeueAfterTimeout(
|
||||
workerIndex,
|
||||
job,
|
||||
lastProgress,
|
||||
stalledPath,
|
||||
);
|
||||
if (!shouldContinue) {
|
||||
// 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();
|
||||
const decision = requeueAfterTimeout(workerIndex, job, lastProgress, stalledPath);
|
||||
if (decision.kind === 'give-up') {
|
||||
// Give-up path: re-queue the non-quarantined remainder,
|
||||
// then await handleWorkerDeath so we know when the slot
|
||||
// is respawned (or dropped) and can dispatch the next
|
||||
// job deterministically.
|
||||
void (async () => {
|
||||
activeWorkers--;
|
||||
busySlots.delete(workerIndex);
|
||||
requeueRemainder(job, decision.excludePaths);
|
||||
await handleWorkerDeath(workerIndex, decision.reason, decision.excludePaths);
|
||||
if (stopped) return;
|
||||
if (activeSlots.has(workerIndex)) runWorker(workerIndex);
|
||||
wakeIdleSlots();
|
||||
maybeDone();
|
||||
})();
|
||||
return;
|
||||
}
|
||||
// Timeout-retry path: enforce the per-slot respawn budget
|
||||
|
|
|
|||
|
|
@ -7,7 +7,11 @@
|
|||
* but workers need compiled .js files.
|
||||
*/
|
||||
import { describe, it, expect, afterEach } from 'vitest';
|
||||
import { createWorkerPool, WorkerPool } from '../../src/core/ingestion/workers/worker-pool.js';
|
||||
import {
|
||||
createWorkerPool,
|
||||
WorkerPool,
|
||||
WorkerPoolDispatchError,
|
||||
} from '../../src/core/ingestion/workers/worker-pool.js';
|
||||
import { pathToFileURL } from 'node:url';
|
||||
import path from 'node:path';
|
||||
import fs from 'node:fs';
|
||||
|
|
@ -353,9 +357,18 @@ describe('worker pool integration', () => {
|
|||
});
|
||||
|
||||
try {
|
||||
await expect(pool.dispatch<any, any>([{ path: 'crash.ts', content: '' }])).rejects.toThrow(
|
||||
/simulated startup crash|exited with code|idle timeout/,
|
||||
);
|
||||
// Resilience refactor (PR #1693): even with a startup-crashing
|
||||
// replacement worker, Node's `online` event fires BEFORE the
|
||||
// worker's main script runs — `waitForWorkerOnline` resolves
|
||||
// optimistically, the slot is re-occupied with a doomed worker,
|
||||
// the second idle timeout triggers the give-up path, and the
|
||||
// file is quarantined. Dispatch resolves with empty results and
|
||||
// the file is in quarantine. A warning is still emitted for the
|
||||
// operator. Documented race: `waitForWorkerOnline` does not wait
|
||||
// for a grace period after `online` before resolving.
|
||||
const results = await pool.dispatch<any, any>([{ path: 'crash.ts', content: '' }]);
|
||||
expect(results).toEqual([]);
|
||||
expect(pool.getQuarantinedPaths?.() ?? []).toEqual(['crash.ts']);
|
||||
const warnRecords = cap.records().filter((r) => Number(r.level) >= 40 /* warn or above */);
|
||||
expect(warnRecords.length).toBeGreaterThan(0);
|
||||
} finally {
|
||||
|
|
@ -430,7 +443,12 @@ describe('worker pool integration', () => {
|
|||
}
|
||||
});
|
||||
|
||||
it('rejects a persistently stalled singleton so the caller can fall back sequentially', async () => {
|
||||
it('quarantines a persistently stalled singleton so subsequent dispatches skip it', async () => {
|
||||
// Resilience refactor (PR #1693): a singleton-timeout no longer
|
||||
// rejects the whole dispatch. The stalled file is quarantined and
|
||||
// the slot respawns; the dispatch resolves with empty results (no
|
||||
// files parsed). Subsequent dispatches with the same path filter it
|
||||
// out via the pool's quarantine.
|
||||
const { tempDir, workerPath } = writeTempWorker(
|
||||
'gitnexus-worker-stalled-',
|
||||
`
|
||||
|
|
@ -444,12 +462,14 @@ describe('worker pool integration', () => {
|
|||
pool = createWorkerPool(pathToFileURL(workerPath) as URL, 1, {
|
||||
subBatchIdleTimeoutMs: 150,
|
||||
maxTimeoutRetries: 0,
|
||||
consecutiveFailureThreshold: 10,
|
||||
maxRespawnsPerSlot: 3,
|
||||
});
|
||||
|
||||
try {
|
||||
await expect(pool.dispatch<any, any>([{ path: 'stalled.ts', content: '' }])).rejects.toThrow(
|
||||
/sequential fallback/,
|
||||
);
|
||||
const results = await pool.dispatch<any, any>([{ path: 'stalled.ts', content: '' }]);
|
||||
expect(results).toEqual([]);
|
||||
expect(pool.getQuarantinedPaths?.() ?? []).toEqual(['stalled.ts']);
|
||||
} finally {
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
|
|
@ -604,8 +624,10 @@ describe('worker pool integration', () => {
|
|||
await expect(pool.dispatch<any, any>([{ path: 'bad.ts', content: '' }])).rejects.toThrow(
|
||||
/protocol error/,
|
||||
);
|
||||
// Resilience refactor (PR #1693): subsequent dispatches reject with
|
||||
// the circuit-breaker message instead of the prior-failure wording.
|
||||
await expect(pool.dispatch<any, any>([{ path: 'after.ts', content: '' }])).rejects.toThrow(
|
||||
/previous failure.*protocol error/,
|
||||
/circuit breaker.*protocol error/i,
|
||||
);
|
||||
} finally {
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
|
|
@ -668,4 +690,361 @@ describe('worker pool integration', () => {
|
|||
await zeroPool.terminate();
|
||||
}
|
||||
});
|
||||
|
||||
// --- Resilience layers (PR #1693 follow-on) ----------------------------
|
||||
|
||||
it('respawns the slot after worker process.exit and finishes the work on the replacement', async () => {
|
||||
// Worker exits with code 1 on its first sub-batch, then the replacement
|
||||
// processes whatever lands in its sub-batch successfully. Exercises
|
||||
// Layer 1 auto-respawn + Layer 3 quarantine end-to-end through real
|
||||
// worker IPC and real waitForWorkerOnline timing.
|
||||
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'gitnexus-resilience-respawn-'));
|
||||
const markerPath = path.join(tempDir, 'crashed-once.txt');
|
||||
const workerPath = path.join(tempDir, 'worker.js');
|
||||
fs.writeFileSync(
|
||||
workerPath,
|
||||
`
|
||||
const fs = require('node:fs');
|
||||
const { parentPort } = require('node:worker_threads');
|
||||
const markerPath = ${JSON.stringify(markerPath)};
|
||||
let current = [];
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
current = msg.files.map((file) => file.path);
|
||||
if (!fs.existsSync(markerPath)) {
|
||||
fs.writeFileSync(markerPath, 'crash once');
|
||||
parentPort.postMessage({ type: 'starting-file', path: current[0] });
|
||||
process.exit(134);
|
||||
}
|
||||
parentPort.postMessage({ type: 'progress', filesProcessed: current.length });
|
||||
parentPort.postMessage({ type: 'sub-batch-done' });
|
||||
return;
|
||||
}
|
||||
if (msg && msg.type === 'flush') {
|
||||
parentPort.postMessage({ type: 'result', data: { fileCount: current.length, paths: current } });
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
|
||||
pool = createWorkerPool(pathToFileURL(workerPath) as URL, 1, {
|
||||
subBatchIdleTimeoutMs: 2000,
|
||||
consecutiveFailureThreshold: 5,
|
||||
maxRespawnsPerSlot: 3,
|
||||
});
|
||||
|
||||
try {
|
||||
const results = await pool.dispatch<
|
||||
{ path: string; content: string },
|
||||
{ fileCount: number; paths: string[] }
|
||||
>([
|
||||
{ path: 'killer.ts', content: '' },
|
||||
{ path: 'good.ts', content: '' },
|
||||
]);
|
||||
// killer.ts was quarantined; replacement processes only the
|
||||
// non-quarantined remainder.
|
||||
expect(results.length).toBe(1);
|
||||
expect(results[0].paths).toEqual(['good.ts']);
|
||||
expect(pool.getQuarantinedPaths?.() ?? []).toEqual(['killer.ts']);
|
||||
} finally {
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('attributes exactly via authoritative starting-file message on worker crash', async () => {
|
||||
// Worker emits starting-file for the SECOND file, then crashes. The
|
||||
// pool must quarantine exactly that file (not items[0] from the
|
||||
// heuristic). Validates Layer 4 end-to-end through real IPC ordering.
|
||||
const { tempDir, workerPath } = writeTempWorker(
|
||||
'gitnexus-resilience-attribution-',
|
||||
`
|
||||
const { parentPort } = require('node:worker_threads');
|
||||
let current = [];
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
current = msg.files.map((file) => file.path);
|
||||
// Pretend we successfully processed the first file, then crash
|
||||
// mid-second.
|
||||
parentPort.postMessage({ type: 'starting-file', path: current[0] });
|
||||
parentPort.postMessage({ type: 'progress', filesProcessed: 1 });
|
||||
parentPort.postMessage({ type: 'starting-file', path: current[1] });
|
||||
process.exit(134);
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
|
||||
pool = createWorkerPool(pathToFileURL(workerPath) as URL, 1, {
|
||||
subBatchIdleTimeoutMs: 2000,
|
||||
consecutiveFailureThreshold: 5,
|
||||
maxRespawnsPerSlot: 1,
|
||||
});
|
||||
|
||||
try {
|
||||
// Job dies, respawn re-tries with filtered job (without items[1]
|
||||
// and items[0] since items[0] was already processed but flush
|
||||
// never landed). Second worker crashes on items[0] of the
|
||||
// re-queued job (which is the original items[0]); slot drops
|
||||
// after budget=1 exceeded.
|
||||
await expect(
|
||||
pool.dispatch<{ path: string; content: string }, unknown>([
|
||||
{ path: 'first.ts', content: '' },
|
||||
{ path: 'second-mid-crash.ts', content: '' },
|
||||
{ path: 'third.ts', content: '' },
|
||||
]),
|
||||
).rejects.toBeInstanceOf(WorkerPoolDispatchError);
|
||||
// The crash attribution names the file authoritatively from the
|
||||
// starting-file message, not items[0].
|
||||
const quarantine = pool.getQuarantinedPaths?.() ?? [];
|
||||
expect(quarantine).toContain('second-mid-crash.ts');
|
||||
} finally {
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('quarantine filters subsequent dispatches without sending to a worker', async () => {
|
||||
// After dispatch A quarantines path X, dispatch B with X in input
|
||||
// must NOT include X in the sub-batch the worker receives. Records
|
||||
// the paths each sub-batch sees.
|
||||
const tempDir = fs.mkdtempSync(
|
||||
path.join(os.tmpdir(), 'gitnexus-resilience-quarantine-filter-'),
|
||||
);
|
||||
const seenPath = path.join(tempDir, 'sub-batches.json');
|
||||
const markerPath = path.join(tempDir, 'crashed-once.txt');
|
||||
const workerPath = path.join(tempDir, 'worker.js');
|
||||
fs.writeFileSync(
|
||||
workerPath,
|
||||
`
|
||||
const fs = require('node:fs');
|
||||
const { parentPort } = require('node:worker_threads');
|
||||
const seenPath = ${JSON.stringify(seenPath)};
|
||||
const markerPath = ${JSON.stringify(markerPath)};
|
||||
let current = [];
|
||||
function recordSeen(paths) {
|
||||
const prior = fs.existsSync(seenPath) ? JSON.parse(fs.readFileSync(seenPath, 'utf-8')) : [];
|
||||
prior.push(paths);
|
||||
fs.writeFileSync(seenPath, JSON.stringify(prior));
|
||||
}
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
current = msg.files.map((f) => f.path);
|
||||
recordSeen(current);
|
||||
if (!fs.existsSync(markerPath) && current.includes('poison.ts')) {
|
||||
fs.writeFileSync(markerPath, 'crash once on poison');
|
||||
parentPort.postMessage({ type: 'starting-file', path: 'poison.ts' });
|
||||
process.exit(134);
|
||||
}
|
||||
parentPort.postMessage({ type: 'progress', filesProcessed: current.length });
|
||||
parentPort.postMessage({ type: 'sub-batch-done' });
|
||||
return;
|
||||
}
|
||||
if (msg && msg.type === 'flush') {
|
||||
parentPort.postMessage({ type: 'result', data: { fileCount: current.length, paths: current } });
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
|
||||
pool = createWorkerPool(pathToFileURL(workerPath) as URL, 1, {
|
||||
subBatchIdleTimeoutMs: 2000,
|
||||
consecutiveFailureThreshold: 5,
|
||||
maxRespawnsPerSlot: 3,
|
||||
});
|
||||
|
||||
try {
|
||||
// Dispatch A quarantines poison.ts.
|
||||
await pool.dispatch<{ path: string; content: string }, unknown>([
|
||||
{ path: 'poison.ts', content: '' },
|
||||
{ path: 'companion.ts', content: '' },
|
||||
]);
|
||||
expect(pool.getQuarantinedPaths?.() ?? []).toEqual(['poison.ts']);
|
||||
|
||||
// Dispatch B includes poison.ts; pool must filter it out before
|
||||
// the worker sees it.
|
||||
const results = await pool.dispatch<{ path: string; content: string }, { paths: string[] }>([
|
||||
{ path: 'poison.ts', content: '' },
|
||||
{ path: 'fresh.ts', content: '' },
|
||||
]);
|
||||
expect(results.length).toBe(1);
|
||||
expect(results[0].paths).toEqual(['fresh.ts']);
|
||||
|
||||
// Audit: which sub-batches did the worker actually receive?
|
||||
const allSubBatches: string[][] = JSON.parse(fs.readFileSync(seenPath, 'utf-8'));
|
||||
const dispatchBSubBatches = allSubBatches.slice(-1);
|
||||
expect(dispatchBSubBatches[0]).toEqual(['fresh.ts']);
|
||||
// No sub-batch sent during dispatch B contained 'poison.ts'.
|
||||
expect(dispatchBSubBatches.some((b) => b.includes('poison.ts'))).toBe(false);
|
||||
} finally {
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('drops a slot after maxRespawnsPerSlot and continues on the survivor', async () => {
|
||||
// 2-worker pool, budget=1. Worker A crashes twice on its assigned
|
||||
// chunk; slot A is dropped. Slot B handles the requeued remainder
|
||||
// successfully. Validates the per-slot drop + wakeIdleSlots flow
|
||||
// under real worker timing.
|
||||
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'gitnexus-resilience-slot-drop-'));
|
||||
const counterPath = path.join(tempDir, 'crash-count.txt');
|
||||
const workerPath = path.join(tempDir, 'worker.js');
|
||||
fs.writeFileSync(
|
||||
workerPath,
|
||||
`
|
||||
const fs = require('node:fs');
|
||||
const { parentPort } = require('node:worker_threads');
|
||||
const counterPath = ${JSON.stringify(counterPath)};
|
||||
let current = [];
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
current = msg.files.map((f) => f.path);
|
||||
const prior = fs.existsSync(counterPath) ? Number(fs.readFileSync(counterPath, 'utf-8')) : 0;
|
||||
if (prior < 2) {
|
||||
fs.writeFileSync(counterPath, String(prior + 1));
|
||||
parentPort.postMessage({ type: 'starting-file', path: current[0] });
|
||||
process.exit(134);
|
||||
}
|
||||
parentPort.postMessage({ type: 'progress', filesProcessed: current.length });
|
||||
parentPort.postMessage({ type: 'sub-batch-done' });
|
||||
return;
|
||||
}
|
||||
if (msg && msg.type === 'flush') {
|
||||
parentPort.postMessage({ type: 'result', data: { paths: current } });
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
|
||||
pool = createWorkerPool(pathToFileURL(workerPath) as URL, 2, {
|
||||
subBatchSize: 2,
|
||||
subBatchIdleTimeoutMs: 2000,
|
||||
consecutiveFailureThreshold: 10,
|
||||
maxRespawnsPerSlot: 1,
|
||||
});
|
||||
|
||||
try {
|
||||
const results = await pool.dispatch<{ path: string; content: string }, { paths: string[] }>([
|
||||
{ path: 'a.ts', content: '' },
|
||||
{ path: 'b.ts', content: '' },
|
||||
{ path: 'c.ts', content: '' },
|
||||
{ path: 'd.ts', content: '' },
|
||||
]);
|
||||
|
||||
// Two crashes both attributed to their starting-file (a.ts, then
|
||||
// whatever the requeued chunk's items[0] was — b.ts or c.ts
|
||||
// depending on dispatch order). At minimum a.ts is quarantined.
|
||||
const quarantine = pool.getQuarantinedPaths?.() ?? [];
|
||||
expect(quarantine.length).toBe(2);
|
||||
expect(quarantine).toContain('a.ts');
|
||||
// All non-quarantined files eventually parsed.
|
||||
const allPaths = results.flatMap((r) => r.paths).sort();
|
||||
const expectedAllowed = ['a.ts', 'b.ts', 'c.ts', 'd.ts'].filter(
|
||||
(p) => !quarantine.includes(p),
|
||||
);
|
||||
expect(allPaths.sort()).toEqual(expectedAllowed.sort());
|
||||
} finally {
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('trips the circuit breaker on cascading per-slot consecutive failures', async () => {
|
||||
// Single-slot pool with consecutiveFailureThreshold=2. Worker dies
|
||||
// on every job; after 2 consecutive deaths on slot 0 the breaker
|
||||
// trips and dispatch rejects with WorkerPoolDispatchError.
|
||||
const { tempDir, workerPath } = writeTempWorker(
|
||||
'gitnexus-resilience-breaker-',
|
||||
`
|
||||
const { parentPort } = require('node:worker_threads');
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
const path = msg.files[0]?.path;
|
||||
if (path) parentPort.postMessage({ type: 'starting-file', path });
|
||||
process.exit(134);
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
|
||||
pool = createWorkerPool(pathToFileURL(workerPath) as URL, 1, {
|
||||
subBatchIdleTimeoutMs: 2000,
|
||||
consecutiveFailureThreshold: 2,
|
||||
maxRespawnsPerSlot: 5,
|
||||
});
|
||||
|
||||
try {
|
||||
const err = await pool
|
||||
.dispatch<{ path: string; content: string }, unknown>([
|
||||
{ path: 'one.ts', content: '' },
|
||||
{ path: 'two.ts', content: '' },
|
||||
])
|
||||
.catch((e) => e);
|
||||
expect(err).toBeInstanceOf(WorkerPoolDispatchError);
|
||||
const dispatchErr = err as WorkerPoolDispatchError;
|
||||
// Breaker tripped with the cumulative quarantine surfaced for
|
||||
// sequential fallback.
|
||||
expect(dispatchErr.fallbackExcludePaths.length).toBeGreaterThan(0);
|
||||
expect(/circuit breaker tripped/i.test(dispatchErr.message)).toBe(true);
|
||||
|
||||
// Subsequent dispatch rejects up front with the same error class.
|
||||
await expect(
|
||||
pool.dispatch<{ path: string; content: string }, unknown>([
|
||||
{ path: 'after.ts', content: '' },
|
||||
]),
|
||||
).rejects.toBeInstanceOf(WorkerPoolDispatchError);
|
||||
} finally {
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('survives a worker `error` event (uncaught throw) the same as a process.exit', async () => {
|
||||
// Worker throws an uncaught error on first sub-batch (triggers Node
|
||||
// Worker 'error' event), then the replacement succeeds. Validates
|
||||
// recoverAndResume on the errorHandler path with real async timing.
|
||||
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'gitnexus-resilience-error-event-'));
|
||||
const markerPath = path.join(tempDir, 'thrown-once.txt');
|
||||
const workerPath = path.join(tempDir, 'worker.js');
|
||||
fs.writeFileSync(
|
||||
workerPath,
|
||||
`
|
||||
const fs = require('node:fs');
|
||||
const { parentPort } = require('node:worker_threads');
|
||||
const markerPath = ${JSON.stringify(markerPath)};
|
||||
let current = [];
|
||||
parentPort.on('message', (msg) => {
|
||||
if (msg && msg.type === 'sub-batch') {
|
||||
current = msg.files.map((f) => f.path);
|
||||
if (!fs.existsSync(markerPath)) {
|
||||
fs.writeFileSync(markerPath, 'throw once');
|
||||
parentPort.postMessage({ type: 'starting-file', path: current[0] });
|
||||
// Uncaught throw — Node Worker emits an 'error' event.
|
||||
throw new Error('simulated native error');
|
||||
}
|
||||
parentPort.postMessage({ type: 'progress', filesProcessed: current.length });
|
||||
parentPort.postMessage({ type: 'sub-batch-done' });
|
||||
return;
|
||||
}
|
||||
if (msg && msg.type === 'flush') {
|
||||
parentPort.postMessage({ type: 'result', data: { paths: current } });
|
||||
}
|
||||
});
|
||||
`,
|
||||
);
|
||||
|
||||
pool = createWorkerPool(pathToFileURL(workerPath) as URL, 1, {
|
||||
subBatchIdleTimeoutMs: 2000,
|
||||
consecutiveFailureThreshold: 5,
|
||||
maxRespawnsPerSlot: 3,
|
||||
});
|
||||
|
||||
try {
|
||||
const results = await pool.dispatch<{ path: string; content: string }, { paths: string[] }>([
|
||||
{ path: 'thrown.ts', content: '' },
|
||||
{ path: 'recovered.ts', content: '' },
|
||||
]);
|
||||
expect(results.length).toBe(1);
|
||||
expect(results[0].paths).toEqual(['recovered.ts']);
|
||||
expect(pool.getQuarantinedPaths?.() ?? []).toEqual(['thrown.ts']);
|
||||
} finally {
|
||||
fs.rmSync(tempDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue