mirror of
https://github.com/garrytan/gstack.git
synced 2026-11-03 23:37:53 +01:00
refactor(shard-engine): move scripts/test-strict-output.ts to scripts/lib/shard-engine.ts
The shared shard engine grows from the existing strict-output module (runShardChild, killProcessGroup, signal forwarding, strict classifier). scripts/test-strict-output.ts stays as a re-export so existing importers, mock.module paths and the strict-output/run-shard-child tests are unchanged. The engine inherits the global touchfile entry; the free runner's CLI-routing fixture copies the new module.
This commit is contained in:
garrytan committed
1 parent
96764e80a6
commit
b76fab349f
4 files changed
+538
-527
No files matched your search
@@ -0,0 +1,532 @@
|
||||
/**
|
||||
* Shard engine: the one owner of how a test shard runs — process-group spawn,
|
||||
* wall timeout + group kill, signal forwarding, strict Bun output verdicts,
|
||||
* per-shard tmp/Chromium sandbox, log files, duration seeds, CLI flag loop.
|
||||
* Lanes (scripts/test-free-shards.ts, scripts/test-paid-shards.ts) keep only
|
||||
* policy: selection, budgets, manifests, retries, and a LanePolicy, e.g.
|
||||
* const LANE: LanePolicy = { acceptsSeedDuration: ms => ms > 0, zeroExecution: () => 'passed' };
|
||||
* const result = await runShardChild({ command, args, cwd, env, timeoutMs, hookStreams });
|
||||
* Enforced by ratchet (c) (test/module-size-ratchet.test.ts). Moved from
|
||||
* scripts/test-strict-output.ts, which re-exports this module.
|
||||
*/
|
||||
|
||||
import { spawn, type ChildProcess } from 'node:child_process';
|
||||
import { StringDecoder } from 'node:string_decoder';
|
||||
import * as path from 'node:path';
|
||||
|
||||
const ROOT = path.resolve(import.meta.dir, '..', '..');
|
||||
// Strict Bun-test output classification works around a Bun test runner bug
|
||||
// where failures can be printed even though the child exits successfully:
|
||||
// output is forwarded byte-for-byte as it arrives, only complete Bun result
|
||||
// lines and terminal summaries are classified, and `strictTestExitCode`
|
||||
// refuses a zero exit when the output shows failures or fewer files ran.
|
||||
const ANSI_ESCAPE = /\u001B\[[0-?]*[ -/]*[@-~]/g;
|
||||
const BUN_FAIL_RESULT = /^(?:\(fail\)|✗) (.+) \[(?:\d+(?:\.\d+)?)(?:ns|us|µs|ms|s)\]$/;
|
||||
const BUN_BETWEEN_TESTS_ERROR = '# Unhandled error between tests';
|
||||
const BUN_TERMINAL_SUMMARY = /^Ran (\d+) tests? across (\d+) files?\. \[(?:\d+(?:\.\d+)?)(?:ns|us|µs|ms|s)\]$/;
|
||||
// The counts block bun prints just before the terminal summary (" 1 pass",
|
||||
// " 2 skip", " 0 fail"). "Ran N tests" COUNTS skipped tests, so N alone
|
||||
// cannot distinguish a shard that verified work from one whose every test
|
||||
// self-skipped (external-service binary missing, tier mismatch) — the
|
||||
// green-by-skip class. Anchored to whole-line matches; nested bun-test
|
||||
// children can still contribute counts (same known limit as the terminal
|
||||
// summary — see the last-summary-anchoring TODO in the audit).
|
||||
const BUN_SKIP_COUNT = /^\s*(\d+) skip$/;
|
||||
const BUN_PASS_COUNT = /^\s*(\d+) pass$/;
|
||||
const BUN_FAIL_COUNT = /^\s*(\d+) fail$/;
|
||||
|
||||
export type BunTestOutputFinding = 'failed-test' | 'unhandled-between-tests';
|
||||
|
||||
export interface BunTestOutputSummary {
|
||||
failedTests: number;
|
||||
unhandledBetweenTests: number;
|
||||
terminalFileCounts: number[];
|
||||
/** Test counts from the same terminal lines — feeds the hollow-shard guard. */
|
||||
terminalTestCounts: number[];
|
||||
/** Sum of bun's " N skip" count lines. "Ran N tests" includes skips, so
|
||||
* this is what separates verified work from green-by-skip. */
|
||||
skippedTests: number;
|
||||
/** Sum of bun's " N pass" count lines. */
|
||||
passedTests: number;
|
||||
}
|
||||
|
||||
export type ForwardedTerminationSignal = 'SIGINT' | 'SIGTERM';
|
||||
|
||||
export interface TerminationSignalSource {
|
||||
on(event: string, listener: () => void): unknown;
|
||||
off(event: string, listener: () => void): unknown;
|
||||
}
|
||||
|
||||
export interface TerminationTimerApi {
|
||||
schedule(callback: () => void, delayMs: number): unknown;
|
||||
cancel(handle: unknown): void;
|
||||
}
|
||||
|
||||
export interface ChildSignalForwarding {
|
||||
readonly receivedSignal: ForwardedTerminationSignal | null;
|
||||
dispose(): void;
|
||||
}
|
||||
|
||||
const DEFAULT_TERMINATION_TIMER: TerminationTimerApi = {
|
||||
schedule: (callback, delayMs) => setTimeout(callback, delayMs),
|
||||
cancel: (handle) => clearTimeout(handle as ReturnType<typeof setTimeout>),
|
||||
};
|
||||
|
||||
/**
|
||||
* Per-source termination bookkeeping, shared across every forwarder bound to
|
||||
* the same source. Installing ANY signal listener suppresses Node's default
|
||||
* terminate-on-SIGINT/SIGTERM, so without this the parent runner survived
|
||||
* cancellation: it killed the current child, then kept LAUNCHING new shards
|
||||
* (observed: paid runs continuing to burn API spend after Ctrl-C). The first
|
||||
* signal now also schedules the parent's own exit after the children's
|
||||
* SIGKILL grace, and runners consult isTerminationRequested() before
|
||||
* launching more work.
|
||||
*/
|
||||
interface SourceTerminationState {
|
||||
requested: boolean;
|
||||
exitScheduled: boolean;
|
||||
}
|
||||
const SOURCE_TERMINATION_STATE = new WeakMap<TerminationSignalSource, SourceTerminationState>();
|
||||
function terminationStateFor(source: TerminationSignalSource): SourceTerminationState {
|
||||
let state = SOURCE_TERMINATION_STATE.get(source);
|
||||
if (!state) {
|
||||
state = { requested: false, exitScheduled: false };
|
||||
SOURCE_TERMINATION_STATE.set(source, state);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
export function isTerminationRequested(source: TerminationSignalSource = process): boolean {
|
||||
return SOURCE_TERMINATION_STATE.get(source)?.requested ?? false;
|
||||
}
|
||||
const signalExitCode = (signal: ForwardedTerminationSignal): number =>
|
||||
128 + (signal === 'SIGINT' ? 2 : 15);
|
||||
|
||||
/**
|
||||
* Bind one active child to the parent's termination lifecycle. SIGINT and
|
||||
* SIGTERM get a grace period so Bun can clean up; a repeated signal, timeout,
|
||||
* or synchronous parent exit uses SIGKILL so the child cannot be orphaned.
|
||||
* The parent itself exits shortly after the grace window (or immediately on
|
||||
* a repeated signal) — cancellation must terminate the RUN, not just the
|
||||
* currently-running children.
|
||||
*/
|
||||
export function installChildSignalForwarding(
|
||||
child: Pick<ChildProcess, 'kill'>,
|
||||
source: TerminationSignalSource = process,
|
||||
timer: TerminationTimerApi = DEFAULT_TERMINATION_TIMER,
|
||||
graceMs = 5_000,
|
||||
exitImpl: (code: number) => void = (code) => process.exit(code),
|
||||
): ChildSignalForwarding {
|
||||
let receivedSignal: ForwardedTerminationSignal | null = null;
|
||||
let forceTimer: unknown = null;
|
||||
let disposed = false;
|
||||
|
||||
const scheduleParentExit = (signal: ForwardedTerminationSignal, delayMs: number): void => {
|
||||
const state = terminationStateFor(source);
|
||||
state.requested = true;
|
||||
if (state.exitScheduled) return;
|
||||
state.exitScheduled = true;
|
||||
// Never cancelled by dispose(): once cancellation is requested, the run
|
||||
// is going down even if this particular shard finishes cleanly first.
|
||||
timer.schedule(() => exitImpl(signalExitCode(signal)), delayMs);
|
||||
};
|
||||
|
||||
const forward = (signal: ForwardedTerminationSignal): void => {
|
||||
if (disposed) return;
|
||||
if (receivedSignal !== null) {
|
||||
child.kill('SIGKILL');
|
||||
scheduleParentExit(signal, 0);
|
||||
return;
|
||||
}
|
||||
receivedSignal = signal;
|
||||
child.kill(signal);
|
||||
forceTimer = timer.schedule(() => {
|
||||
forceTimer = null;
|
||||
child.kill('SIGKILL');
|
||||
}, graceMs);
|
||||
// Exit AFTER the children's SIGKILL grace so the group kills land first.
|
||||
scheduleParentExit(signal, graceMs + 1_000);
|
||||
};
|
||||
const onSigint = () => forward('SIGINT');
|
||||
const onSigterm = () => forward('SIGTERM');
|
||||
const onExit = () => { child.kill('SIGKILL'); };
|
||||
|
||||
source.on('SIGINT', onSigint);
|
||||
source.on('SIGTERM', onSigterm);
|
||||
source.on('exit', onExit);
|
||||
|
||||
return {
|
||||
get receivedSignal() {
|
||||
return receivedSignal;
|
||||
},
|
||||
dispose() {
|
||||
if (disposed) return;
|
||||
disposed = true;
|
||||
source.off('SIGINT', onSigint);
|
||||
source.off('SIGTERM', onSigterm);
|
||||
source.off('exit', onExit);
|
||||
if (forceTimer !== null) timer.cancel(forceTimer);
|
||||
forceTimer = null;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* SIGKILL the shard's whole process group. Orphaned grandchildren (browsers,
|
||||
* claude sessions) are how a stalled run once burned a core for 15.7 hours.
|
||||
*/
|
||||
export function killProcessGroup(child: ChildProcess, signal: NodeJS.Signals): void {
|
||||
if (process.platform === 'win32' || typeof child.pid !== 'number') {
|
||||
child.kill(signal);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
process.kill(-child.pid, signal);
|
||||
} catch (err) {
|
||||
const code = (err as NodeJS.ErrnoException).code;
|
||||
if (code === 'ESRCH') return; // group already gone
|
||||
if (code !== 'EPERM') throw err;
|
||||
// Observed on macOS after a SIGKILLed group is reaped: signalling the
|
||||
// now-empty group id returns EPERM, not ESRCH. Throwing here loses the
|
||||
// shard's real outcome (a timeout gets recorded as a failure) and, from
|
||||
// the timeout timer, leaves the shard promise unsettled — a hang, which
|
||||
// is the exact failure class this runner exists to kill. Fall back to the
|
||||
// direct pid so a genuinely-live child is still signalled.
|
||||
try {
|
||||
child.kill(signal);
|
||||
} catch {
|
||||
// Best-effort reap: nothing actionable is left if this fails too.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Strip ANSI escapes and a trailing CR from one output line. Every line
|
||||
* matcher (here and in the free runner's console filter / failure
|
||||
* attribution) MUST match against this form — a prior grep for `(fail)`
|
||||
* lines missed real failures because color codes sat inside the line.
|
||||
*/
|
||||
export function stripAnsiLine(rawLine: string): string {
|
||||
return rawLine.replace(ANSI_ESCAPE, '').replace(/\r$/, '');
|
||||
}
|
||||
|
||||
export function classifyBunTestOutputLine(rawLine: string): BunTestOutputFinding | null {
|
||||
const line = stripAnsiLine(rawLine);
|
||||
if (parseBunFailureResult(line) !== null) return 'failed-test';
|
||||
if (line === BUN_BETWEEN_TESTS_ERROR) return 'unhandled-between-tests';
|
||||
return null;
|
||||
}
|
||||
|
||||
export function parseBunFailureResult(rawLine: string): string | null {
|
||||
return BUN_FAIL_RESULT.exec(stripAnsiLine(rawLine))?.[1] ?? null;
|
||||
}
|
||||
|
||||
export function parseBunTerminalSummaryLine(rawLine: string): number | null {
|
||||
return parseBunTerminalSummary(rawLine)?.files ?? null;
|
||||
}
|
||||
|
||||
export function parseBunTerminalSummary(rawLine: string): { tests: number; files: number } | null {
|
||||
const line = stripAnsiLine(rawLine);
|
||||
const match = BUN_TERMINAL_SUMMARY.exec(line);
|
||||
return match
|
||||
? { tests: Number.parseInt(match[1], 10), files: Number.parseInt(match[2], 10) }
|
||||
: null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Incrementally classifies output without assuming process chunks align to
|
||||
* lines. Buffers are PER ORIGIN: stdout and stderr are independent pipes, so
|
||||
* a chunk from one can arrive between two halves of a line from the other.
|
||||
* A single shared buffer would glue those fragments into garbled lines — a
|
||||
* sheared `(fail)` line goes uncounted and a sheared terminal summary reads
|
||||
* as truncation. Counters are shared; only line assembly is per-stream.
|
||||
*/
|
||||
export type ClassifierOrigin = 'stdout' | 'stderr';
|
||||
|
||||
export class BunFailureSummaryParser {
|
||||
private readonly pending: Partial<Record<ClassifierOrigin, { failures: number | null }>> = {};
|
||||
|
||||
consume(rawLine: string, origin: ClassifierOrigin): number | null {
|
||||
const line = stripAnsiLine(rawLine);
|
||||
if (BUN_PASS_COUNT.test(line)) {
|
||||
this.pending[origin] = { failures: null };
|
||||
return null;
|
||||
}
|
||||
const pending = this.pending[origin];
|
||||
if (!pending) return null;
|
||||
const fail = BUN_FAIL_COUNT.exec(line);
|
||||
if (fail) {
|
||||
pending.failures = Math.max(pending.failures ?? 0, Number.parseInt(fail[1], 10));
|
||||
return null;
|
||||
}
|
||||
if (parseBunTerminalSummary(line) !== null) {
|
||||
delete this.pending[origin];
|
||||
return pending.failures;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export class BunTestOutputClassifier {
|
||||
private readonly decoders: Record<ClassifierOrigin, StringDecoder> = {
|
||||
stdout: new StringDecoder('utf8'),
|
||||
stderr: new StringDecoder('utf8'),
|
||||
};
|
||||
private pending: Record<ClassifierOrigin, string> = { stdout: '', stderr: '' };
|
||||
private failedTests = 0;
|
||||
private reportedFailedTests = 0;
|
||||
private readonly failureSummary = new BunFailureSummaryParser();
|
||||
private unhandledBetweenTests = 0;
|
||||
private terminalFileCounts: number[] = [];
|
||||
private terminalTestCounts: number[] = [];
|
||||
private skippedTests = 0;
|
||||
private passedTests = 0;
|
||||
|
||||
write(chunk: Uint8Array | string, origin: ClassifierOrigin = 'stdout'): void {
|
||||
this.pending[origin] += typeof chunk === 'string'
|
||||
? chunk
|
||||
: this.decoders[origin].write(Buffer.from(chunk));
|
||||
this.consumeCompleteLines(origin);
|
||||
}
|
||||
|
||||
end(): BunTestOutputSummary {
|
||||
for (const origin of ['stdout', 'stderr'] as const) {
|
||||
this.pending[origin] += this.decoders[origin].end();
|
||||
if (this.pending[origin].length > 0) this.classify(this.pending[origin], origin);
|
||||
this.pending[origin] = '';
|
||||
}
|
||||
return this.summary();
|
||||
}
|
||||
|
||||
summary(): BunTestOutputSummary {
|
||||
return {
|
||||
failedTests: Math.max(this.failedTests, this.reportedFailedTests),
|
||||
unhandledBetweenTests: this.unhandledBetweenTests,
|
||||
terminalFileCounts: [...this.terminalFileCounts],
|
||||
terminalTestCounts: [...this.terminalTestCounts],
|
||||
skippedTests: this.skippedTests,
|
||||
passedTests: this.passedTests,
|
||||
};
|
||||
}
|
||||
|
||||
private consumeCompleteLines(origin: ClassifierOrigin): void {
|
||||
let newline = this.pending[origin].indexOf('\n');
|
||||
while (newline !== -1) {
|
||||
this.classify(this.pending[origin].slice(0, newline), origin);
|
||||
this.pending[origin] = this.pending[origin].slice(newline + 1);
|
||||
newline = this.pending[origin].indexOf('\n');
|
||||
}
|
||||
}
|
||||
|
||||
private classify(line: string, origin: ClassifierOrigin): void {
|
||||
const finding = classifyBunTestOutputLine(line);
|
||||
if (finding === 'failed-test') this.failedTests += 1;
|
||||
if (finding === 'unhandled-between-tests') this.unhandledBetweenTests += 1;
|
||||
const stripped = stripAnsiLine(line);
|
||||
const skip = BUN_SKIP_COUNT.exec(stripped);
|
||||
if (skip !== null) this.skippedTests += Number.parseInt(skip[1], 10);
|
||||
const pass = BUN_PASS_COUNT.exec(stripped);
|
||||
if (pass !== null) this.passedTests += Number.parseInt(pass[1], 10);
|
||||
const fail = this.failureSummary.consume(stripped, origin);
|
||||
if (fail !== null) this.reportedFailedTests = Math.max(this.reportedFailedTests, fail);
|
||||
const terminal = parseBunTerminalSummary(line);
|
||||
if (terminal !== null) {
|
||||
this.terminalFileCounts.push(terminal.files);
|
||||
this.terminalTestCounts.push(terminal.tests);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function strictTestExitCode(
|
||||
childExitCode: number,
|
||||
summary: BunTestOutputSummary,
|
||||
expectedFiles?: number,
|
||||
): number {
|
||||
if (childExitCode !== 0) return childExitCode;
|
||||
if (summary.failedTests > 0 || summary.unhandledBetweenTests > 0) return 1;
|
||||
if (expectedFiles !== undefined && !summary.terminalFileCounts.includes(expectedFiles)) return 1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
export function normalizeRelativePath(filePath: string): string {
|
||||
return filePath.replace(/\\/g, '/');
|
||||
}
|
||||
|
||||
/**
|
||||
* Bun treats positional test paths as substring filters. Resolve every
|
||||
* canonical relative path before spawning so `test/foo.test.ts` cannot also
|
||||
* select `browse/test/foo.test.ts`.
|
||||
*/
|
||||
export function exactTestFileSelectors(files: string[], rootDir = ROOT): string[] {
|
||||
return files.map((file) => path.isAbsolute(file) ? path.normalize(file) : path.resolve(rootDir, file));
|
||||
}
|
||||
|
||||
export function forwardAndClassify(
|
||||
stream: NodeJS.ReadableStream,
|
||||
destination: NodeJS.WriteStream,
|
||||
classifier: BunTestOutputClassifier,
|
||||
origin: ClassifierOrigin = 'stdout',
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let ended = false;
|
||||
const incomplete = () => reject(new Error(`incomplete ${origin} capture: stream closed before end`));
|
||||
stream.on('data', (chunk: Buffer | string) => {
|
||||
classifier.write(chunk, origin);
|
||||
destination.write(chunk);
|
||||
});
|
||||
stream.once('end', () => { ended = true; resolve(); });
|
||||
stream.on('error', reject);
|
||||
stream.once('close', () => { if (!ended) incomplete(); });
|
||||
// Bun can return an already-destroyed pipe whose close event is past.
|
||||
if ('destroyed' in stream && stream.destroyed && !ended) {
|
||||
if ('errored' in stream && stream.errored) reject(stream.errored);
|
||||
else incomplete();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// --- Shared shard-child lifecycle ---
|
||||
|
||||
export interface RunShardChildOptions {
|
||||
command: string;
|
||||
args: string[];
|
||||
cwd: string;
|
||||
env: NodeJS.ProcessEnv;
|
||||
/** External wall-clock deadline; on expiry the child's process GROUP is SIGKILLed. */
|
||||
timeoutMs: number;
|
||||
deadlineMs?: number;
|
||||
/**
|
||||
* Hook the freshly-spawned child's stdout/stderr. Stream POLICY (classifier
|
||||
* tees, log spooling, console forwarding, reporters) is entirely the
|
||||
* caller's. Runs synchronously right after spawn; child close and every
|
||||
* returned promise must settle within the same deadline. A wall-expired
|
||||
* return reports incomplete capture instead of treating the prefix as final.
|
||||
*/
|
||||
hookStreams: (child: ChildProcess) => Array<Promise<void>>;
|
||||
}
|
||||
|
||||
export interface ShardChildResult {
|
||||
exitCode: number | null;
|
||||
/** True when the shared deadline expired; no further child work is allowed. */
|
||||
timedOut: boolean;
|
||||
/** The child's pid — the process-GROUP id on POSIX (detached spawn). */
|
||||
groupPid: number | null;
|
||||
incompleteCapture?: {
|
||||
childClosed: boolean;
|
||||
pendingStreams: number;
|
||||
failedStreams: number;
|
||||
deadlineMs: number;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* The child lifecycle both sharded runners need, extracted from
|
||||
* scripts/test-paid-shards.ts runPaidShard (scripts/test-free-shards.ts
|
||||
* runFreeShard duplicates the same ~35 lines verbatim today and is designed
|
||||
* to migrate here in a later change):
|
||||
*
|
||||
* - spawn detached on POSIX so the child owns its process group,
|
||||
* - forward parent SIGINT/SIGTERM to the whole group (not just the child),
|
||||
* - arm an EXTERNAL wall-clock timer that SIGKILLs the group — a spinning
|
||||
* child main thread never fires its own in-process timer,
|
||||
* - in EVERY exit path: disarm the timer, detach the signal forwarder, and
|
||||
* signal group survivors with SIGKILL.
|
||||
*
|
||||
* Caller-side cleanup that must run even on a spawn failure (log streams,
|
||||
* reporters, temp dirs) belongs in the caller's own try/finally around this
|
||||
* call: a spawn 'error' event THROWS from here after the finally block runs,
|
||||
* preserving the runners' existing could-not-run handling.
|
||||
*/
|
||||
export async function runShardChild(options: RunShardChildOptions): Promise<ShardChildResult> {
|
||||
const deadlineMs = Math.min(options.deadlineMs ?? Infinity, Date.now() + options.timeoutMs);
|
||||
if (!Number.isFinite(deadlineMs)) throw new Error('Shard deadline must be finite');
|
||||
if (deadlineMs <= Date.now()) {
|
||||
return { exitCode: null, timedOut: true, groupPid: null,
|
||||
incompleteCapture: { childClosed: false, pendingStreams: 0, failedStreams: 0, deadlineMs } };
|
||||
}
|
||||
const child = spawn(options.command, options.args, {
|
||||
cwd: options.cwd,
|
||||
env: options.env,
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
detached: process.platform !== 'win32',
|
||||
windowsHide: true,
|
||||
});
|
||||
const groupPid = child.pid ?? null;
|
||||
// Group-kill on parent SIGINT/SIGTERM too, not just on timeout.
|
||||
const forwarding = installChildSignalForwarding({
|
||||
kill: (signal?: NodeJS.Signals | number) => {
|
||||
killProcessGroup(child, (signal as NodeJS.Signals) ?? 'SIGTERM');
|
||||
return true;
|
||||
},
|
||||
});
|
||||
|
||||
let timedOut = false;
|
||||
let childClosed = false;
|
||||
let pendingStreams = 0;
|
||||
let failedStreams = 0;
|
||||
let exitCode: number | null = null;
|
||||
let failed = false;
|
||||
let firstError: unknown;
|
||||
const rememberError = (error: unknown) => {
|
||||
if (failed) return;
|
||||
failed = true;
|
||||
firstError = error;
|
||||
};
|
||||
const kill = () => {
|
||||
try { killProcessGroup(child, 'SIGKILL'); }
|
||||
catch (error) { rememberError(error); }
|
||||
};
|
||||
let close!: () => void;
|
||||
const closed = new Promise<void>(resolve => { close = resolve; });
|
||||
const onExit = (code: number | null) => { exitCode = code; };
|
||||
const onClose = (code: number | null) => { exitCode = code; childClosed = true; close(); };
|
||||
child.once('exit', onExit);
|
||||
child.once('close', onClose);
|
||||
child.on('error', rememberError);
|
||||
let expire!: () => void;
|
||||
const expired = new Promise<void>(resolve => { expire = resolve; });
|
||||
const killTimer = setTimeout(() => {
|
||||
timedOut = true;
|
||||
kill();
|
||||
expire();
|
||||
}, Math.max(0, deadlineMs - Date.now()));
|
||||
|
||||
try {
|
||||
let streams: Array<Promise<void>> = [];
|
||||
try { streams = options.hookStreams(child); }
|
||||
catch (error) {
|
||||
rememberError(error);
|
||||
kill();
|
||||
child.stdout?.destroy();
|
||||
child.stderr?.destroy();
|
||||
}
|
||||
pendingStreams = streams.length;
|
||||
const drainage = Promise.all(streams.map(stream => Promise.resolve(stream).then(
|
||||
() => { pendingStreams -= 1; },
|
||||
(error: unknown) => { pendingStreams -= 1; failedStreams += 1; rememberError(error); },
|
||||
)));
|
||||
await Promise.race([Promise.all([closed, drainage]), expired]);
|
||||
if (Date.now() >= deadlineMs) timedOut = true;
|
||||
} finally {
|
||||
clearTimeout(killTimer);
|
||||
forwarding.dispose();
|
||||
kill();
|
||||
child.off('exit', onExit);
|
||||
child.off('close', onClose);
|
||||
if (!childClosed || pendingStreams > 0) {
|
||||
child.stdout?.destroy();
|
||||
child.stderr?.destroy();
|
||||
child.unref();
|
||||
}
|
||||
}
|
||||
const result: ShardChildResult = { exitCode, timedOut, groupPid };
|
||||
if (!childClosed || pendingStreams > 0 || failedStreams > 0) {
|
||||
result.incompleteCapture = { childClosed, pendingStreams, failedStreams, deadlineMs };
|
||||
}
|
||||
if (failed) {
|
||||
if (firstError instanceof Error && Object.isExtensible(firstError)) {
|
||||
Reflect.defineProperty(firstError, 'shardResult', { value: result, configurable: true });
|
||||
}
|
||||
throw firstError;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
@@ -1,528 +1,6 @@
|
||||
/**
|
||||
* Strict Bun-test output classification + child lifecycle helpers.
|
||||
*
|
||||
* Works around a Bun test runner bug where failures can be printed even though
|
||||
* the child exits successfully: output is forwarded byte-for-byte as it
|
||||
* arrives, and only complete Bun result lines and terminal summaries are
|
||||
* classified. `strictTestExitCode` then refuses to trust a zero exit when the
|
||||
* output shows failures (or when fewer files ran than expected).
|
||||
*
|
||||
* Shared by the sharded paid-tier runner (scripts/test-paid-shards.ts) and any
|
||||
* future strict wrapper around `bun test`.
|
||||
* Compatibility path for the shard engine, which now lives in
|
||||
* scripts/lib/shard-engine.ts. Existing importers (session runners, the
|
||||
* strict-output and run-shard-child tests, mock.module paths) keep working.
|
||||
*/
|
||||
|
||||
import { spawn, type ChildProcess } from 'node:child_process';
|
||||
import { StringDecoder } from 'node:string_decoder';
|
||||
import * as path from 'node:path';
|
||||
|
||||
const ROOT = path.resolve(import.meta.dir, '..');
|
||||
const ANSI_ESCAPE = /\u001B\[[0-?]*[ -/]*[@-~]/g;
|
||||
const BUN_FAIL_RESULT = /^(?:\(fail\)|✗) (.+) \[(?:\d+(?:\.\d+)?)(?:ns|us|µs|ms|s)\]$/;
|
||||
const BUN_BETWEEN_TESTS_ERROR = '# Unhandled error between tests';
|
||||
const BUN_TERMINAL_SUMMARY = /^Ran (\d+) tests? across (\d+) files?\. \[(?:\d+(?:\.\d+)?)(?:ns|us|µs|ms|s)\]$/;
|
||||
// The counts block bun prints just before the terminal summary (" 1 pass",
|
||||
// " 2 skip", " 0 fail"). "Ran N tests" COUNTS skipped tests, so N alone
|
||||
// cannot distinguish a shard that verified work from one whose every test
|
||||
// self-skipped (external-service binary missing, tier mismatch) — the
|
||||
// green-by-skip class. Anchored to whole-line matches; nested bun-test
|
||||
// children can still contribute counts (same known limit as the terminal
|
||||
// summary — see the last-summary-anchoring TODO in the audit).
|
||||
const BUN_SKIP_COUNT = /^\s*(\d+) skip$/;
|
||||
const BUN_PASS_COUNT = /^\s*(\d+) pass$/;
|
||||
const BUN_FAIL_COUNT = /^\s*(\d+) fail$/;
|
||||
|
||||
export type BunTestOutputFinding = 'failed-test' | 'unhandled-between-tests';
|
||||
|
||||
export interface BunTestOutputSummary {
|
||||
failedTests: number;
|
||||
unhandledBetweenTests: number;
|
||||
terminalFileCounts: number[];
|
||||
/** Test counts from the same terminal lines — feeds the hollow-shard guard. */
|
||||
terminalTestCounts: number[];
|
||||
/** Sum of bun's " N skip" count lines. "Ran N tests" includes skips, so
|
||||
* this is what separates verified work from green-by-skip. */
|
||||
skippedTests: number;
|
||||
/** Sum of bun's " N pass" count lines. */
|
||||
passedTests: number;
|
||||
}
|
||||
|
||||
export type ForwardedTerminationSignal = 'SIGINT' | 'SIGTERM';
|
||||
|
||||
export interface TerminationSignalSource {
|
||||
on(event: string, listener: () => void): unknown;
|
||||
off(event: string, listener: () => void): unknown;
|
||||
}
|
||||
|
||||
export interface TerminationTimerApi {
|
||||
schedule(callback: () => void, delayMs: number): unknown;
|
||||
cancel(handle: unknown): void;
|
||||
}
|
||||
|
||||
export interface ChildSignalForwarding {
|
||||
readonly receivedSignal: ForwardedTerminationSignal | null;
|
||||
dispose(): void;
|
||||
}
|
||||
|
||||
const DEFAULT_TERMINATION_TIMER: TerminationTimerApi = {
|
||||
schedule: (callback, delayMs) => setTimeout(callback, delayMs),
|
||||
cancel: (handle) => clearTimeout(handle as ReturnType<typeof setTimeout>),
|
||||
};
|
||||
|
||||
/**
|
||||
* Per-source termination bookkeeping, shared across every forwarder bound to
|
||||
* the same source. Installing ANY signal listener suppresses Node's default
|
||||
* terminate-on-SIGINT/SIGTERM, so without this the parent runner survived
|
||||
* cancellation: it killed the current child, then kept LAUNCHING new shards
|
||||
* (observed: paid runs continuing to burn API spend after Ctrl-C). The first
|
||||
* signal now also schedules the parent's own exit after the children's
|
||||
* SIGKILL grace, and runners consult isTerminationRequested() before
|
||||
* launching more work.
|
||||
*/
|
||||
interface SourceTerminationState {
|
||||
requested: boolean;
|
||||
exitScheduled: boolean;
|
||||
}
|
||||
const SOURCE_TERMINATION_STATE = new WeakMap<TerminationSignalSource, SourceTerminationState>();
|
||||
function terminationStateFor(source: TerminationSignalSource): SourceTerminationState {
|
||||
let state = SOURCE_TERMINATION_STATE.get(source);
|
||||
if (!state) {
|
||||
state = { requested: false, exitScheduled: false };
|
||||
SOURCE_TERMINATION_STATE.set(source, state);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
export function isTerminationRequested(source: TerminationSignalSource = process): boolean {
|
||||
return SOURCE_TERMINATION_STATE.get(source)?.requested ?? false;
|
||||
}
|
||||
const signalExitCode = (signal: ForwardedTerminationSignal): number =>
|
||||
128 + (signal === 'SIGINT' ? 2 : 15);
|
||||
|
||||
/**
|
||||
* Bind one active child to the parent's termination lifecycle. SIGINT and
|
||||
* SIGTERM get a grace period so Bun can clean up; a repeated signal, timeout,
|
||||
* or synchronous parent exit uses SIGKILL so the child cannot be orphaned.
|
||||
* The parent itself exits shortly after the grace window (or immediately on
|
||||
* a repeated signal) — cancellation must terminate the RUN, not just the
|
||||
* currently-running children.
|
||||
*/
|
||||
export function installChildSignalForwarding(
|
||||
child: Pick<ChildProcess, 'kill'>,
|
||||
source: TerminationSignalSource = process,
|
||||
timer: TerminationTimerApi = DEFAULT_TERMINATION_TIMER,
|
||||
graceMs = 5_000,
|
||||
exitImpl: (code: number) => void = (code) => process.exit(code),
|
||||
): ChildSignalForwarding {
|
||||
let receivedSignal: ForwardedTerminationSignal | null = null;
|
||||
let forceTimer: unknown = null;
|
||||
let disposed = false;
|
||||
|
||||
const scheduleParentExit = (signal: ForwardedTerminationSignal, delayMs: number): void => {
|
||||
const state = terminationStateFor(source);
|
||||
state.requested = true;
|
||||
if (state.exitScheduled) return;
|
||||
state.exitScheduled = true;
|
||||
// Never cancelled by dispose(): once cancellation is requested, the run
|
||||
// is going down even if this particular shard finishes cleanly first.
|
||||
timer.schedule(() => exitImpl(signalExitCode(signal)), delayMs);
|
||||
};
|
||||
|
||||
const forward = (signal: ForwardedTerminationSignal): void => {
|
||||
if (disposed) return;
|
||||
if (receivedSignal !== null) {
|
||||
child.kill('SIGKILL');
|
||||
scheduleParentExit(signal, 0);
|
||||
return;
|
||||
}
|
||||
receivedSignal = signal;
|
||||
child.kill(signal);
|
||||
forceTimer = timer.schedule(() => {
|
||||
forceTimer = null;
|
||||
child.kill('SIGKILL');
|
||||
}, graceMs);
|
||||
// Exit AFTER the children's SIGKILL grace so the group kills land first.
|
||||
scheduleParentExit(signal, graceMs + 1_000);
|
||||
};
|
||||
const onSigint = () => forward('SIGINT');
|
||||
const onSigterm = () => forward('SIGTERM');
|
||||
const onExit = () => { child.kill('SIGKILL'); };
|
||||
|
||||
source.on('SIGINT', onSigint);
|
||||
source.on('SIGTERM', onSigterm);
|
||||
source.on('exit', onExit);
|
||||
|
||||
return {
|
||||
get receivedSignal() {
|
||||
return receivedSignal;
|
||||
},
|
||||
dispose() {
|
||||
if (disposed) return;
|
||||
disposed = true;
|
||||
source.off('SIGINT', onSigint);
|
||||
source.off('SIGTERM', onSigterm);
|
||||
source.off('exit', onExit);
|
||||
if (forceTimer !== null) timer.cancel(forceTimer);
|
||||
forceTimer = null;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* SIGKILL the shard's whole process group. Orphaned grandchildren (browsers,
|
||||
* claude sessions) are how a stalled run once burned a core for 15.7 hours.
|
||||
*/
|
||||
export function killProcessGroup(child: ChildProcess, signal: NodeJS.Signals): void {
|
||||
if (process.platform === 'win32' || typeof child.pid !== 'number') {
|
||||
child.kill(signal);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
process.kill(-child.pid, signal);
|
||||
} catch (err) {
|
||||
const code = (err as NodeJS.ErrnoException).code;
|
||||
if (code === 'ESRCH') return; // group already gone
|
||||
if (code !== 'EPERM') throw err;
|
||||
// Observed on macOS after a SIGKILLed group is reaped: signalling the
|
||||
// now-empty group id returns EPERM, not ESRCH. Throwing here loses the
|
||||
// shard's real outcome (a timeout gets recorded as a failure) and, from
|
||||
// the timeout timer, leaves the shard promise unsettled — a hang, which
|
||||
// is the exact failure class this runner exists to kill. Fall back to the
|
||||
// direct pid so a genuinely-live child is still signalled.
|
||||
try {
|
||||
child.kill(signal);
|
||||
} catch {
|
||||
// Best-effort reap: nothing actionable is left if this fails too.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Strip ANSI escapes and a trailing CR from one output line. Every line
|
||||
* matcher (here and in the free runner's console filter / failure
|
||||
* attribution) MUST match against this form — a prior grep for `(fail)`
|
||||
* lines missed real failures because color codes sat inside the line.
|
||||
*/
|
||||
export function stripAnsiLine(rawLine: string): string {
|
||||
return rawLine.replace(ANSI_ESCAPE, '').replace(/\r$/, '');
|
||||
}
|
||||
|
||||
export function classifyBunTestOutputLine(rawLine: string): BunTestOutputFinding | null {
|
||||
const line = stripAnsiLine(rawLine);
|
||||
if (parseBunFailureResult(line) !== null) return 'failed-test';
|
||||
if (line === BUN_BETWEEN_TESTS_ERROR) return 'unhandled-between-tests';
|
||||
return null;
|
||||
}
|
||||
|
||||
export function parseBunFailureResult(rawLine: string): string | null {
|
||||
return BUN_FAIL_RESULT.exec(stripAnsiLine(rawLine))?.[1] ?? null;
|
||||
}
|
||||
|
||||
export function parseBunTerminalSummaryLine(rawLine: string): number | null {
|
||||
return parseBunTerminalSummary(rawLine)?.files ?? null;
|
||||
}
|
||||
|
||||
export function parseBunTerminalSummary(rawLine: string): { tests: number; files: number } | null {
|
||||
const line = stripAnsiLine(rawLine);
|
||||
const match = BUN_TERMINAL_SUMMARY.exec(line);
|
||||
return match
|
||||
? { tests: Number.parseInt(match[1], 10), files: Number.parseInt(match[2], 10) }
|
||||
: null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Incrementally classifies output without assuming process chunks align to
|
||||
* lines. Buffers are PER ORIGIN: stdout and stderr are independent pipes, so
|
||||
* a chunk from one can arrive between two halves of a line from the other.
|
||||
* A single shared buffer would glue those fragments into garbled lines — a
|
||||
* sheared `(fail)` line goes uncounted and a sheared terminal summary reads
|
||||
* as truncation. Counters are shared; only line assembly is per-stream.
|
||||
*/
|
||||
export type ClassifierOrigin = 'stdout' | 'stderr';
|
||||
|
||||
export class BunFailureSummaryParser {
|
||||
private readonly pending: Partial<Record<ClassifierOrigin, { failures: number | null }>> = {};
|
||||
|
||||
consume(rawLine: string, origin: ClassifierOrigin): number | null {
|
||||
const line = stripAnsiLine(rawLine);
|
||||
if (BUN_PASS_COUNT.test(line)) {
|
||||
this.pending[origin] = { failures: null };
|
||||
return null;
|
||||
}
|
||||
const pending = this.pending[origin];
|
||||
if (!pending) return null;
|
||||
const fail = BUN_FAIL_COUNT.exec(line);
|
||||
if (fail) {
|
||||
pending.failures = Math.max(pending.failures ?? 0, Number.parseInt(fail[1], 10));
|
||||
return null;
|
||||
}
|
||||
if (parseBunTerminalSummary(line) !== null) {
|
||||
delete this.pending[origin];
|
||||
return pending.failures;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export class BunTestOutputClassifier {
|
||||
private readonly decoders: Record<ClassifierOrigin, StringDecoder> = {
|
||||
stdout: new StringDecoder('utf8'),
|
||||
stderr: new StringDecoder('utf8'),
|
||||
};
|
||||
private pending: Record<ClassifierOrigin, string> = { stdout: '', stderr: '' };
|
||||
private failedTests = 0;
|
||||
private reportedFailedTests = 0;
|
||||
private readonly failureSummary = new BunFailureSummaryParser();
|
||||
private unhandledBetweenTests = 0;
|
||||
private terminalFileCounts: number[] = [];
|
||||
private terminalTestCounts: number[] = [];
|
||||
private skippedTests = 0;
|
||||
private passedTests = 0;
|
||||
|
||||
write(chunk: Uint8Array | string, origin: ClassifierOrigin = 'stdout'): void {
|
||||
this.pending[origin] += typeof chunk === 'string'
|
||||
? chunk
|
||||
: this.decoders[origin].write(Buffer.from(chunk));
|
||||
this.consumeCompleteLines(origin);
|
||||
}
|
||||
|
||||
end(): BunTestOutputSummary {
|
||||
for (const origin of ['stdout', 'stderr'] as const) {
|
||||
this.pending[origin] += this.decoders[origin].end();
|
||||
if (this.pending[origin].length > 0) this.classify(this.pending[origin], origin);
|
||||
this.pending[origin] = '';
|
||||
}
|
||||
return this.summary();
|
||||
}
|
||||
|
||||
summary(): BunTestOutputSummary {
|
||||
return {
|
||||
failedTests: Math.max(this.failedTests, this.reportedFailedTests),
|
||||
unhandledBetweenTests: this.unhandledBetweenTests,
|
||||
terminalFileCounts: [...this.terminalFileCounts],
|
||||
terminalTestCounts: [...this.terminalTestCounts],
|
||||
skippedTests: this.skippedTests,
|
||||
passedTests: this.passedTests,
|
||||
};
|
||||
}
|
||||
|
||||
private consumeCompleteLines(origin: ClassifierOrigin): void {
|
||||
let newline = this.pending[origin].indexOf('\n');
|
||||
while (newline !== -1) {
|
||||
this.classify(this.pending[origin].slice(0, newline), origin);
|
||||
this.pending[origin] = this.pending[origin].slice(newline + 1);
|
||||
newline = this.pending[origin].indexOf('\n');
|
||||
}
|
||||
}
|
||||
|
||||
private classify(line: string, origin: ClassifierOrigin): void {
|
||||
const finding = classifyBunTestOutputLine(line);
|
||||
if (finding === 'failed-test') this.failedTests += 1;
|
||||
if (finding === 'unhandled-between-tests') this.unhandledBetweenTests += 1;
|
||||
const stripped = stripAnsiLine(line);
|
||||
const skip = BUN_SKIP_COUNT.exec(stripped);
|
||||
if (skip !== null) this.skippedTests += Number.parseInt(skip[1], 10);
|
||||
const pass = BUN_PASS_COUNT.exec(stripped);
|
||||
if (pass !== null) this.passedTests += Number.parseInt(pass[1], 10);
|
||||
const fail = this.failureSummary.consume(stripped, origin);
|
||||
if (fail !== null) this.reportedFailedTests = Math.max(this.reportedFailedTests, fail);
|
||||
const terminal = parseBunTerminalSummary(line);
|
||||
if (terminal !== null) {
|
||||
this.terminalFileCounts.push(terminal.files);
|
||||
this.terminalTestCounts.push(terminal.tests);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function strictTestExitCode(
|
||||
childExitCode: number,
|
||||
summary: BunTestOutputSummary,
|
||||
expectedFiles?: number,
|
||||
): number {
|
||||
if (childExitCode !== 0) return childExitCode;
|
||||
if (summary.failedTests > 0 || summary.unhandledBetweenTests > 0) return 1;
|
||||
if (expectedFiles !== undefined && !summary.terminalFileCounts.includes(expectedFiles)) return 1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
export function normalizeRelativePath(filePath: string): string {
|
||||
return filePath.replace(/\\/g, '/');
|
||||
}
|
||||
|
||||
/**
|
||||
* Bun treats positional test paths as substring filters. Resolve every
|
||||
* canonical relative path before spawning so `test/foo.test.ts` cannot also
|
||||
* select `browse/test/foo.test.ts`.
|
||||
*/
|
||||
export function exactTestFileSelectors(files: string[], rootDir = ROOT): string[] {
|
||||
return files.map((file) => path.isAbsolute(file) ? path.normalize(file) : path.resolve(rootDir, file));
|
||||
}
|
||||
|
||||
export function forwardAndClassify(
|
||||
stream: NodeJS.ReadableStream,
|
||||
destination: NodeJS.WriteStream,
|
||||
classifier: BunTestOutputClassifier,
|
||||
origin: ClassifierOrigin = 'stdout',
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let ended = false;
|
||||
const incomplete = () => reject(new Error(`incomplete ${origin} capture: stream closed before end`));
|
||||
stream.on('data', (chunk: Buffer | string) => {
|
||||
classifier.write(chunk, origin);
|
||||
destination.write(chunk);
|
||||
});
|
||||
stream.once('end', () => { ended = true; resolve(); });
|
||||
stream.on('error', reject);
|
||||
stream.once('close', () => { if (!ended) incomplete(); });
|
||||
// Bun can return an already-destroyed pipe whose close event is past.
|
||||
if ('destroyed' in stream && stream.destroyed && !ended) {
|
||||
if ('errored' in stream && stream.errored) reject(stream.errored);
|
||||
else incomplete();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// --- Shared shard-child lifecycle ---
|
||||
|
||||
export interface RunShardChildOptions {
|
||||
command: string;
|
||||
args: string[];
|
||||
cwd: string;
|
||||
env: NodeJS.ProcessEnv;
|
||||
/** External wall-clock deadline; on expiry the child's process GROUP is SIGKILLed. */
|
||||
timeoutMs: number;
|
||||
deadlineMs?: number;
|
||||
/**
|
||||
* Hook the freshly-spawned child's stdout/stderr. Stream POLICY (classifier
|
||||
* tees, log spooling, console forwarding, reporters) is entirely the
|
||||
* caller's. Runs synchronously right after spawn; child close and every
|
||||
* returned promise must settle within the same deadline. A wall-expired
|
||||
* return reports incomplete capture instead of treating the prefix as final.
|
||||
*/
|
||||
hookStreams: (child: ChildProcess) => Array<Promise<void>>;
|
||||
}
|
||||
|
||||
export interface ShardChildResult {
|
||||
exitCode: number | null;
|
||||
/** True when the shared deadline expired; no further child work is allowed. */
|
||||
timedOut: boolean;
|
||||
/** The child's pid — the process-GROUP id on POSIX (detached spawn). */
|
||||
groupPid: number | null;
|
||||
incompleteCapture?: {
|
||||
childClosed: boolean;
|
||||
pendingStreams: number;
|
||||
failedStreams: number;
|
||||
deadlineMs: number;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* The child lifecycle both sharded runners need, extracted from
|
||||
* scripts/test-paid-shards.ts runPaidShard (scripts/test-free-shards.ts
|
||||
* runFreeShard duplicates the same ~35 lines verbatim today and is designed
|
||||
* to migrate here in a later change):
|
||||
*
|
||||
* - spawn detached on POSIX so the child owns its process group,
|
||||
* - forward parent SIGINT/SIGTERM to the whole group (not just the child),
|
||||
* - arm an EXTERNAL wall-clock timer that SIGKILLs the group — a spinning
|
||||
* child main thread never fires its own in-process timer,
|
||||
* - in EVERY exit path: disarm the timer, detach the signal forwarder, and
|
||||
* signal group survivors with SIGKILL.
|
||||
*
|
||||
* Caller-side cleanup that must run even on a spawn failure (log streams,
|
||||
* reporters, temp dirs) belongs in the caller's own try/finally around this
|
||||
* call: a spawn 'error' event THROWS from here after the finally block runs,
|
||||
* preserving the runners' existing could-not-run handling.
|
||||
*/
|
||||
export async function runShardChild(options: RunShardChildOptions): Promise<ShardChildResult> {
|
||||
const deadlineMs = Math.min(options.deadlineMs ?? Infinity, Date.now() + options.timeoutMs);
|
||||
if (!Number.isFinite(deadlineMs)) throw new Error('Shard deadline must be finite');
|
||||
if (deadlineMs <= Date.now()) {
|
||||
return { exitCode: null, timedOut: true, groupPid: null,
|
||||
incompleteCapture: { childClosed: false, pendingStreams: 0, failedStreams: 0, deadlineMs } };
|
||||
}
|
||||
const child = spawn(options.command, options.args, {
|
||||
cwd: options.cwd,
|
||||
env: options.env,
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
detached: process.platform !== 'win32',
|
||||
windowsHide: true,
|
||||
});
|
||||
const groupPid = child.pid ?? null;
|
||||
// Group-kill on parent SIGINT/SIGTERM too, not just on timeout.
|
||||
const forwarding = installChildSignalForwarding({
|
||||
kill: (signal?: NodeJS.Signals | number) => {
|
||||
killProcessGroup(child, (signal as NodeJS.Signals) ?? 'SIGTERM');
|
||||
return true;
|
||||
},
|
||||
});
|
||||
|
||||
let timedOut = false;
|
||||
let childClosed = false;
|
||||
let pendingStreams = 0;
|
||||
let failedStreams = 0;
|
||||
let exitCode: number | null = null;
|
||||
let failed = false;
|
||||
let firstError: unknown;
|
||||
const rememberError = (error: unknown) => {
|
||||
if (failed) return;
|
||||
failed = true;
|
||||
firstError = error;
|
||||
};
|
||||
const kill = () => {
|
||||
try { killProcessGroup(child, 'SIGKILL'); }
|
||||
catch (error) { rememberError(error); }
|
||||
};
|
||||
let close!: () => void;
|
||||
const closed = new Promise<void>(resolve => { close = resolve; });
|
||||
const onExit = (code: number | null) => { exitCode = code; };
|
||||
const onClose = (code: number | null) => { exitCode = code; childClosed = true; close(); };
|
||||
child.once('exit', onExit);
|
||||
child.once('close', onClose);
|
||||
child.on('error', rememberError);
|
||||
let expire!: () => void;
|
||||
const expired = new Promise<void>(resolve => { expire = resolve; });
|
||||
const killTimer = setTimeout(() => {
|
||||
timedOut = true;
|
||||
kill();
|
||||
expire();
|
||||
}, Math.max(0, deadlineMs - Date.now()));
|
||||
|
||||
try {
|
||||
let streams: Array<Promise<void>> = [];
|
||||
try { streams = options.hookStreams(child); }
|
||||
catch (error) {
|
||||
rememberError(error);
|
||||
kill();
|
||||
child.stdout?.destroy();
|
||||
child.stderr?.destroy();
|
||||
}
|
||||
pendingStreams = streams.length;
|
||||
const drainage = Promise.all(streams.map(stream => Promise.resolve(stream).then(
|
||||
() => { pendingStreams -= 1; },
|
||||
(error: unknown) => { pendingStreams -= 1; failedStreams += 1; rememberError(error); },
|
||||
)));
|
||||
await Promise.race([Promise.all([closed, drainage]), expired]);
|
||||
if (Date.now() >= deadlineMs) timedOut = true;
|
||||
} finally {
|
||||
clearTimeout(killTimer);
|
||||
forwarding.dispose();
|
||||
kill();
|
||||
child.off('exit', onExit);
|
||||
child.off('close', onClose);
|
||||
if (!childClosed || pendingStreams > 0) {
|
||||
child.stdout?.destroy();
|
||||
child.stderr?.destroy();
|
||||
child.unref();
|
||||
}
|
||||
}
|
||||
const result: ShardChildResult = { exitCode, timedOut, groupPid };
|
||||
if (!childClosed || pendingStreams > 0 || failedStreams > 0) {
|
||||
result.incompleteCapture = { childClosed, pendingStreams, failedStreams, deadlineMs };
|
||||
}
|
||||
if (failed) {
|
||||
if (firstError instanceof Error && Object.isExtensible(firstError)) {
|
||||
Reflect.defineProperty(firstError, 'shardResult', { value: result, configurable: true });
|
||||
}
|
||||
throw firstError;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
export * from './lib/shard-engine';
|
||||
@@ -1560,6 +1560,7 @@ export const LLM_JUDGE_TOUCHFILES: Record<string, string[]> = {
|
||||
*/
|
||||
export const GLOBAL_TOUCHFILES = [
|
||||
'scripts/test-strict-output.ts',
|
||||
'scripts/lib/shard-engine.ts', // The shard engine test-strict-output.ts re-exports (moved there in the W2 refactor)
|
||||
// Canonical paid execution and its shared time allocations affect every paid test.
|
||||
'scripts/test-paid-shards.ts',
|
||||
'scripts/test-pr-profile.ts',
|
||||
|
||||
@@ -467,7 +467,7 @@ describe('test-free-shards: exclusive host-state phase', () => {
|
||||
let child: ReturnType<typeof Bun.spawn> | undefined;
|
||||
let watchdog: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
for (const file of ['scripts/test-free-shards.ts', 'scripts/test-strict-output.ts',
|
||||
for (const file of ['scripts/test-free-shards.ts', 'scripts/test-strict-output.ts', 'scripts/lib/shard-engine.ts',
|
||||
'test/helpers/paid-test-set.ts', 'test/helpers/touchfiles.ts', 'test/helpers/touchfiles-data.ts', 'test/helpers/test-selection.ts']) {
|
||||
const target = path.join(directory, file);
|
||||
fs.mkdirSync(path.dirname(target), { recursive: true });
|
||||
|
||||
Reference in new issue
Block a user