commit 22c1e668054402cdf126b21b2d1981fa91ac9c4f
parent d2192da12ca9a34ec78458420f3d35fd0d39e875
Author: Florian Dold <dold@taler.net>
Date: Fri, 11 Sep 2026 16:50:32 +0200
taler-harness: clean up worker process groups after failures
Let the parent runner terminate worker process groups after failures and
interruptions, escalating from SIGTERM to SIGKILL when needed.
Issue: https://bugs.taler.net/n/9814
Diffstat:
7 files changed, 918 insertions(+), 434 deletions(-)
diff --git a/packages/taler-harness/README.md b/packages/taler-harness/README.md
@@ -212,6 +212,30 @@ globals still share a process, however. Thus this mode deliberately provides
less isolation than the default; rerun a failure without `--reuse-worker`
before treating it as a product regression.
+### Cleanup after failures
+
+On Linux, each integration-test worker runs in its own process group.
+The parent runner cleans up that group after worker failures, execution
+timeouts, interruptions, and disposal. This also reaches services and shell
+descendants after the worker has exited. Failed workers are discarded even
+with `--reuse-worker`, including after tolerated TODO failures.
+
+Normal teardown has a separate five-second deadline. If it stalls, the runner
+sends `SIGTERM` to the entire worker group, allows five seconds for termination,
+then sends `SIGKILL` to remaining members. It waits up to one more second for
+worker output to close before reporting. The original assertion failure is
+preserved alongside cleanup diagnostics. A passing test that requires forced
+termination is reported as failed.
+
+`--no-timeout` disables the execution deadline, while teardown remains bounded.
+`TALER_TEST_LINGER` still preserves services after ordinary completion; an
+execution timeout or explicit interruption cleans them up. A second interrupt
+escalates termination immediately.
+
+This covers descendants that inherit the worker's process group. Processes
+that create a separate group or session, including some browser automation
+processes, require their own lifecycle handling.
+
### Service warnings and errors
By default, after every integration test the harness parses structured `WARN`,
diff --git a/packages/taler-harness/src/harness/harness.ts b/packages/taler-harness/src/harness/harness.ts
@@ -469,6 +469,7 @@ export class GlobalTestState {
procs: ProcessWrapper[];
servers: http.Server[];
inShutdown: boolean = false;
+ private shutdownPromise?: Promise<void>;
stepSet: Set<string> = new Set();
private logStreamCompletions: Promise<void>[] = [];
private logStreamFailures: Error[] = [];
@@ -740,15 +741,18 @@ export class GlobalTestState {
return procWrap;
}
- async shutdown(): Promise<void> {
- if (this.inShutdown) {
- await this.waitForLogStreams();
- return;
+ shutdown(): Promise<void> {
+ if (this.shutdownPromise) {
+ return this.shutdownPromise;
}
if (shouldLingerInTest()) {
logger.trace("refusing to shut down, lingering was requested");
- return;
+ return Promise.resolve();
}
+ return (this.shutdownPromise = this.performShutdown());
+ }
+
+ private async performShutdown(): Promise<void> {
this.inShutdown = true;
logger.trace("shutting down");
for (const s of this.servers) {
@@ -3181,11 +3185,12 @@ export async function runTestWithState(
testName: string,
linger: boolean = false,
logAudit: LogAuditMode = "warn",
+ onTeardownStart?: (result: TestRunResult) => void | Promise<void>,
): Promise<TestRunResult> {
const startMs = new Date().getTime();
const p = openPromise();
- let status: TestStatus;
+ let status: TestStatus = "fail";
let reason: string | undefined;
let fatalShutdownStarted = false;
@@ -3269,7 +3274,18 @@ export async function runTestWithState(
reason = e instanceof Error ? e.message : String(e);
} finally {
try {
- await gc.shutdown();
+ try {
+ if (!shouldLingerInTest()) {
+ await onTeardownStart?.({
+ name: testName,
+ status,
+ reason,
+ timeSec: (Date.now() - startMs) / 1000,
+ });
+ }
+ } finally {
+ await gc.shutdown();
+ }
} finally {
process.removeListener("SIGINT", handleSignal);
process.removeListener("SIGTERM", handleSignal);
diff --git a/packages/taler-harness/src/harness/lifecycle.test.ts b/packages/taler-harness/src/harness/lifecycle.test.ts
@@ -24,6 +24,37 @@ import { GlobalTestState, WalletService, runTestWithState } from "./harness.js";
import net from "node:net";
import { once } from "node:events";
+test("concurrent shutdown callers wait for children even after their pipes close", async (t) => {
+ const testDir = fs.mkdtempSync(path.join(os.tmpdir(), "harness-lifecycle-"));
+ t.after(() => fs.rmSync(testDir, { recursive: true, force: true }));
+ const state = new GlobalTestState({ testDir });
+ const child = state.spawnService(
+ process.execPath,
+ [
+ "-e",
+ `
+ process.on('SIGTERM', () => setTimeout(() => process.exit(0), 100));
+ setInterval(() => {}, 1000);
+ process.stdout.end();
+ process.stderr.end();
+ `,
+ ],
+ "closed-pipes",
+ );
+ t.after(async () => {
+ child.proc.kill("SIGKILL");
+ await child.wait();
+ });
+ await Promise.all([
+ once(child.proc.stdout!, "end"),
+ once(child.proc.stderr!, "end"),
+ ]);
+ const first = state.shutdown();
+ await state.shutdown();
+ assert.equal(child.proc.exitCode, 0);
+ await first;
+});
+
test("wallet sockets work with long test paths and are removed on shutdown", async (t) => {
const testDir = fs.mkdtempSync(path.join(os.tmpdir(), "harness-lifecycle-"));
t.after(() => fs.rmSync(testDir, { recursive: true, force: true }));
diff --git a/packages/taler-harness/src/integrationtests/testrunner.ts b/packages/taler-harness/src/integrationtests/testrunner.ts
@@ -16,16 +16,20 @@
import { runTransactionFinalAmountsTest } from "./test-transaction-final-amounts.js";
import {
- CancellationToken,
Logger,
minimatch,
setGlobalLogLevelFromString,
} from "@gnu-taler/taler-util";
-import * as child_process from "child_process";
import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import url from "node:url";
+import { finished } from "node:stream/promises";
+import {
+ RunTestChildInstruction,
+ TestWorkerSupervisor,
+ sendTestWorkerMessage,
+} from "./worker-supervisor.js";
import {
GlobalTestState,
LogAuditMode,
@@ -645,152 +649,6 @@ function validateTestMetadata(testCases: TestMainFunction[]): void {
}
}
-interface RunTestChildInstruction {
- testName: string;
- testRootDir: string;
- logAudit: LogAuditMode;
-}
-
-interface ReusableWorkerResultMessage {
- type: "result";
- result: TestRunResult;
-}
-
-interface ReusableWorkerReadyMessage {
- type: "ready";
-}
-
-type ReusableWorkerMessage =
- | ReusableWorkerResultMessage
- | ReusableWorkerReadyMessage;
-
-class ReusableTestWorker {
- readonly child: child_process.ChildProcess;
- private active:
- | {
- resolve: (result: TestRunResult) => void;
- reject: (error: Error) => void;
- }
- | undefined;
- private currentLogStream: fs.WriteStream | undefined;
- private readyResolve!: () => void;
- private readyReject!: (error: Error) => void;
- private readonly ready: Promise<void>;
- private stopped = false;
-
- constructor(myFilename: string, verbosity: number) {
- this.ready = new Promise<void>((resolve, reject) => {
- this.readyResolve = resolve;
- this.readyReject = reject;
- });
- this.child = child_process.fork(
- myFilename,
- ["__TWCLI_REUSABLE_TESTWORKER"],
- {
- env: {
- TWCLI_LOGLEVEL: logger.getGlobalLogLevel(),
- ...process.env,
- },
- stdio: ["pipe", "pipe", "pipe", "ipc"],
- },
- );
-
- this.child.stdout?.on("data", (chunk: Buffer) => {
- this.currentLogStream?.write(chunk);
- if (verbosity > 0) {
- process.stdout.write(chunk);
- }
- });
- this.child.stderr?.on("data", (chunk: Buffer) => {
- this.currentLogStream?.write(chunk);
- if (verbosity > 0) {
- process.stderr.write(chunk);
- }
- });
- this.child.on("message", (message: ReusableWorkerMessage) => {
- if (message.type === "ready") {
- this.readyResolve();
- return;
- }
- if (!this.active) {
- this.stop();
- return;
- }
- const active = this.active;
- this.active = undefined;
- active.resolve(message.result);
- });
- this.child.on("exit", (code, signal) => {
- this.stopped = true;
- const detail = signal ? `signal ${signal}` : `code ${code}`;
- const error = new Error(`reusable test worker exited with ${detail}`);
- this.readyReject(error);
- this.active?.reject(error);
- this.active = undefined;
- });
- this.child.on("error", (error) => {
- this.readyReject(error);
- this.active?.reject(error);
- this.active = undefined;
- });
- }
-
- async run(
- instruction: RunTestChildInstruction,
- logStream: fs.WriteStream,
- ): Promise<TestRunResult> {
- await this.ready;
- if (this.stopped) {
- throw Error("reusable test worker is not running");
- }
- if (this.active) {
- throw Error("reusable test worker already has an active test");
- }
- this.currentLogStream = logStream;
- return new Promise<TestRunResult>((resolve, reject) => {
- this.active = { resolve, reject };
- this.child.send({ type: "run", instruction }, (error) => {
- if (!error) {
- return;
- }
- this.active = undefined;
- reject(error);
- });
- });
- }
-
- finishTest(): void {
- this.currentLogStream = undefined;
- }
-
- stop(): void {
- if (this.stopped) {
- return;
- }
- this.stopped = true;
- this.child.disconnect();
- }
-}
-
-async function waitForWorkerClose(
- worker: child_process.ChildProcess,
-): Promise<void> {
- if (worker.exitCode !== null || worker.signalCode !== null) {
- return;
- }
- await new Promise<void>((resolve) => {
- const timeout = setTimeout(resolve, 5_000);
- worker.once("close", () => {
- clearTimeout(timeout);
- resolve();
- });
- worker.once("error", () => {
- clearTimeout(timeout);
- resolve();
- });
- });
-}
-
export async function runTests(spec: TestRunSpec) {
const logAudit = spec.logAudit ?? "warn";
validateTestMetadata(allTests);
@@ -817,23 +675,15 @@ export async function runTests(spec: TestRunSpec) {
const testResults: TestRunResult[] = [];
- let currentChild: child_process.ChildProcess | undefined;
-
- const handleSignal = (s: NodeJS.Signals) => {
- console.log(`received signal ${s} in test parent`);
- if (currentChild) {
- currentChild.kill("SIGTERM");
- }
- reportAndQuit(testRootDir, testResults, {
- interrupted: true,
- strictTodo: spec.strictTodo,
- });
+ let currentWorker: TestWorkerSupervisor | undefined;
+ let interrupted = false;
+ const handleSignal = (signal: NodeJS.Signals) => {
+ console.log(`received signal ${signal} in test parent`);
+ currentWorker?.interrupt(interrupted);
+ interrupted = true;
};
-
- process.on("SIGINT", (s) => handleSignal(s));
- process.on("SIGTERM", (s) => handleSignal(s));
- //process.on("unhandledRejection", handleSignal);
- //process.on("uncaughtException", handleSignal);
+ process.on("SIGINT", handleSignal);
+ process.on("SIGTERM", handleSignal);
let suites: Set<string> | undefined;
@@ -878,253 +728,215 @@ export async function runTests(spec: TestRunSpec) {
let numFailed = 0;
let numTodoFailed = 0;
- let reusableWorker: ReusableTestWorker | undefined;
-
- for (const [n, testCase] of filteredTests.entries()) {
- const testName = getTestName(testCase);
- const todoBugUrl = testCase.todo;
- const isTodo = todoBugUrl !== undefined;
- if (spec.dryRun) {
- console.log(
- `dry run: would run test ${testName}${isTodo ? ` (todo: ${todoBugUrl})` : ""}`,
- );
- continue;
- }
+ let reusableWorker: TestWorkerSupervisor | undefined;
+ let cleanupFailed = false;
- const testInstr: RunTestChildInstruction = {
- testName,
- testRootDir,
- logAudit,
- };
+ try {
+ for (const [n, testCase] of filteredTests.entries()) {
+ const testName = getTestName(testCase);
+ const todoBugUrl = testCase.todo;
+ const isTodo = todoBugUrl !== undefined;
+ if (spec.dryRun) {
+ console.log(
+ `dry run: would run test ${testName}${isTodo ? ` (todo: ${todoBugUrl})` : ""}`,
+ );
+ continue;
+ }
- const myFilename = url.fileURLToPath(import.meta.url);
+ const testInstr: RunTestChildInstruction = {
+ testName,
+ testRootDir,
+ logAudit,
+ };
- const testDir = path.join(testRootDir, testName);
- fs.mkdirSync(testDir, { recursive: true });
+ const myFilename = url.fileURLToPath(import.meta.url);
- const harnessLogFilename = path.join(testRootDir, testName, "harness.log");
- const harnessLogStream = fs.createWriteStream(harnessLogFilename);
+ const testDir = path.join(testRootDir, testName);
+ fs.mkdirSync(testDir, { recursive: true });
- let reusableResultPromise: Promise<TestRunResult> | undefined;
- if (spec.reuseWorker) {
- reusableWorker ??= new ReusableTestWorker(myFilename, spec.verbosity);
- currentChild = reusableWorker.child;
- reusableResultPromise = reusableWorker.run(testInstr, harnessLogStream);
- } else {
- currentChild = child_process.fork(myFilename, ["__TWCLI_TESTWORKER"], {
- env: {
- TWCLI_RUN_TEST_INSTRUCTION: JSON.stringify(testInstr),
- TWCLI_LOGLEVEL: logger.getGlobalLogLevel(),
- ...process.env,
- },
- stdio: ["pipe", "pipe", "pipe", "ipc"],
- });
+ const harnessLogFilename = path.join(
+ testRootDir,
+ testName,
+ "harness.log",
+ );
+ const harnessLogStream = fs.createWriteStream(harnessLogFilename);
+ const logCompletion = finished(harnessLogStream);
+ // Attach a rejection handler immediately, including during worker cleanup.
+ void logCompletion.catch(() => undefined);
+
+ currentWorker =
+ reusableWorker ??
+ new TestWorkerSupervisor(
+ myFilename,
+ spec.reuseWorker ?? false,
+ spec.verbosity,
+ {
+ env: { TWCLI_LOGLEVEL: logger.getGlobalLogLevel(), ...process.env },
+ },
+ );
+ if (spec.reuseWorker) reusableWorker = currentWorker;
- if (spec.verbosity > 0) {
- currentChild.stderr?.pipe(process.stderr);
- currentChild.stdout?.pipe(process.stdout);
+ // Default timeout when the test doesn't override it.
+ let defaultTimeout = 60000;
+ const overrideDefaultTimeout = process.env.TALER_TEST_TIMEOUT;
+ if (overrideDefaultTimeout) {
+ defaultTimeout = Number.parseInt(overrideDefaultTimeout, 10) * 1000;
}
- currentChild.stdout?.pipe(harnessLogStream);
- currentChild.stderr?.pipe(harnessLogStream);
- }
-
- // Default timeout when the test doesn't override it.
- let defaultTimeout = 60000;
- const overrideDefaultTimeout = process.env.TALER_TEST_TIMEOUT;
- if (overrideDefaultTimeout) {
- defaultTimeout = Number.parseInt(overrideDefaultTimeout, 10) * 1000;
- }
+ // Set the timeout to at least be the default timeout.
+ const testTimeoutMs = testCase.timeoutMs
+ ? Math.max(testCase.timeoutMs, defaultTimeout)
+ : defaultTimeout;
- // Set the timeout to at least be the default timeout.
- const testTimeoutMs = testCase.timeoutMs
- ? Math.max(testCase.timeoutMs, defaultTimeout)
- : defaultTimeout;
+ let progressText = `${n + 1} of ${filteredTests.length}`;
+ if (numFailed > 0) {
+ progressText = progressText + `, failed ${numFailed}`;
+ }
+ if (numTodoFailed > 0) {
+ progressText = progressText + `, todo failed ${numTodoFailed}`;
+ }
- let progressText = `${n + 1} of ${filteredTests.length}`;
- if (numFailed > 0) {
- progressText = progressText + `, failed ${numFailed}`;
- }
- if (numTodoFailed > 0) {
- progressText = progressText + `, todo failed ${numTodoFailed}`;
- }
+ const todoText = isTodo ? ` [todo: ${todoBugUrl}]` : "";
- const todoText = isTodo ? ` [todo: ${todoBugUrl}]` : "";
+ if (spec.noTimeout) {
+ console.log(
+ `running ${testName}${todoText} (${progressText}), no timeout`,
+ );
+ } else {
+ console.log(
+ `running ${testName}${todoText} (${progressText}) with timeout ${testTimeoutMs}ms`,
+ );
+ }
- if (spec.noTimeout) {
- console.log(
- `running ${testName}${todoText} (${progressText}), no timeout`,
+ const outcome = await currentWorker.run(
+ testInstr,
+ harnessLogStream,
+ spec.noTimeout ? undefined : testTimeoutMs,
);
- } else {
- console.log(
- `running ${testName}${todoText} (${progressText}) with timeout ${testTimeoutMs}ms`,
- );
- }
-
- const token = spec.noTimeout
- ? CancellationToken.CONTINUE
- : CancellationToken.timeout(testTimeoutMs).token;
-
- const resultPromise: Promise<TestRunResult> =
- reusableResultPromise ??
- new Promise((resolve, reject) => {
- let msg: TestRunResult | undefined;
- currentChild!.on("message", (m) => {
- if (token.isCancelled) {
- return;
- }
- msg = m as TestRunResult;
- });
- currentChild!.on("exit", (code, signal) => {
- if (token.isCancelled) {
- return;
- }
- logger.info(`process exited code=${code} signal=${signal}`);
- if (signal) {
- reject(new Error(`test worker exited with signal ${signal}`));
- } else if (code != 0) {
- reject(new Error(`test worker exited with code ${code}`));
- } else if (!msg) {
- reject(
- new Error(
- `test worker exited without giving back the test results`,
- ),
- );
- } else {
- resolve(msg);
- }
- });
- currentChild!.on("error", (err) => {
- if (token.isCancelled) {
- return;
- }
- reject(err);
- });
- });
-
- let result: TestRunResult;
- let needsPostMortemLogScan = false;
-
- try {
- result = await token.racePromise(resultPromise);
- } catch (e: any) {
- if (token.isCancelled) {
- result = {
- status: "fail",
- reason: "timeout",
- timeSec: testTimeoutMs / 1000,
- name: testName,
- };
- currentChild.kill("SIGTERM");
- reusableWorker = undefined;
- needsPostMortemLogScan = true;
- } else if (e instanceof Error) {
- result = {
- status: "fail",
- reason: `worker error: ${e.message}`,
- timeSec: testTimeoutMs / 1000,
- name: testName,
- };
- currentChild.kill("SIGTERM");
- reusableWorker = undefined;
- needsPostMortemLogScan = true;
- } else {
- currentChild.kill("SIGTERM");
+ const result = outcome.result;
+ cleanupFailed ||= outcome.cleanupFailed;
+ if (result.status === "fail") reusableWorker = undefined;
+ harnessLogStream.end();
+ try {
+ await logCompletion;
+ } catch (error) {
+ const detail = `harness log write failed: ${error}`;
+ result.status = "fail";
+ result.reason = result.reason ? `${result.reason}; ${detail}` : detail;
+ const cleanup = await currentWorker.stop();
+ cleanupFailed ||= cleanup.error !== undefined;
reusableWorker = undefined;
- // Should never happen
- throw Error("test failed with strange exception");
}
- }
- if (needsPostMortemLogScan) {
- await waitForWorkerClose(currentChild);
- if (logAudit !== "no") {
- try {
- result.postmortemLogEntries = readServiceLogEntries(testDir);
- } catch (error) {
- const detail = error instanceof Error ? error.message : String(error);
- const scanFailure = `service log scan failed: ${detail}`;
- if (logAudit === "warn") {
- logger.warn(scanFailure);
- } else {
- result.reason = `${result.reason}; ${scanFailure}`;
+ if (outcome.needsPostmortemLogScan) {
+ if (logAudit !== "no") {
+ try {
+ result.postmortemLogEntries = readServiceLogEntries(testDir);
+ } catch (error) {
+ const detail =
+ error instanceof Error ? error.message : String(error);
+ const scanFailure = `service log scan failed: ${detail}`;
+ if (logAudit === "warn") {
+ logger.warn(scanFailure);
+ } else {
+ result.reason = `${result.reason}; ${scanFailure}`;
+ }
}
}
}
- }
- if (result.unexpectedLogEntries?.length) {
- console.log("-- unexpected service log entries --");
- for (const entry of result.unexpectedLogEntries) {
- console.log(
- `${entry.file}:${entry.line}: ${entry.component} ${entry.level} ${entry.message}`,
- );
+ if (result.unexpectedLogEntries?.length) {
+ console.log("-- unexpected service log entries --");
+ for (const entry of result.unexpectedLogEntries) {
+ console.log(
+ `${entry.file}:${entry.line}: ${entry.component} ${entry.level} ${entry.message}`,
+ );
+ }
+ console.log("-- end unexpected service log entries --");
}
- console.log("-- end unexpected service log entries --");
- }
- if (result.postmortemLogEntries?.length) {
- console.log("-- service log entries recovered after worker failure --");
- for (const entry of result.postmortemLogEntries) {
- console.log(
- `${entry.file}:${entry.line}: ${entry.component} ${entry.level} ${entry.message}`,
- );
+ if (result.postmortemLogEntries?.length) {
+ console.log("-- service log entries recovered after worker failure --");
+ for (const entry of result.postmortemLogEntries) {
+ console.log(
+ `${entry.file}:${entry.line}: ${entry.component} ${entry.level} ${entry.message}`,
+ );
+ }
+ console.log("-- end recovered service log entries --");
}
- console.log("-- end recovered service log entries --");
- }
- if (isTodo) {
- result.todo = true;
- result.todoBugUrl = todoBugUrl;
- }
+ if (isTodo) {
+ result.todo = true;
+ result.todoBugUrl = todoBugUrl;
+ }
- // A failing todo test is expected, unless we're asked to be strict.
- const toleratedFailure = isTodo && !spec.strictTodo;
+ // A failing todo test is expected, unless we're asked to be strict.
+ const toleratedFailure = isTodo && !spec.strictTodo;
- if (result.status === "fail") {
- if (toleratedFailure) {
- numTodoFailed++;
- } else {
- numFailed++;
+ if (result.status === "fail") {
+ if (toleratedFailure) {
+ numTodoFailed++;
+ } else {
+ numFailed++;
+ }
}
- }
- reusableWorker?.finishTest();
- harnessLogStream.close();
-
- const stepsFile = `${testDir}/steps.txt`;
- if (spec.verbosity > 0 && fs.existsSync(stepsFile)) {
- let stepsLog = fs.readFileSync(stepsFile, {
- encoding: "utf-8",
- });
- console.log("-- test steps --");
- console.log(stepsLog.trim());
- console.log("-- end --");
- }
+ const stepsFile = `${testDir}/steps.txt`;
+ if (spec.verbosity > 0 && fs.existsSync(stepsFile)) {
+ let stepsLog = fs.readFileSync(stepsFile, {
+ encoding: "utf-8",
+ });
+ console.log("-- test steps --");
+ console.log(stepsLog.trim());
+ console.log("-- end --");
+ }
- const { unexpectedLogEntries, postmortemLogEntries, ...resultSummary } =
- result;
- console.log(
- `parent: got result ${JSON.stringify({
- ...resultSummary,
- unexpectedLogEntryCount: unexpectedLogEntries?.length ?? 0,
- postmortemLogEntryCount: postmortemLogEntries?.length ?? 0,
- })}`,
- );
+ const { unexpectedLogEntries, postmortemLogEntries, ...resultSummary } =
+ result;
+ console.log(
+ `parent: got result ${JSON.stringify({
+ ...resultSummary,
+ unexpectedLogEntryCount: unexpectedLogEntries?.length ?? 0,
+ postmortemLogEntryCount: postmortemLogEntries?.length ?? 0,
+ })}`,
+ );
- testResults.push(result);
+ testResults.push(result);
- if (result.status === "fail" && !toleratedFailure && spec.failFast) {
- logger.error("test failed and failing fast, exit!");
- break;
+ if (interrupted || cleanupFailed) break;
+ if (result.status === "fail" && !toleratedFailure && spec.failFast) {
+ logger.error("test failed and failing fast, exit!");
+ break;
+ }
+ }
+ } catch (error) {
+ logger.error(`test runner failed: ${error}`);
+ cleanupFailed = true;
+ } finally {
+ if (currentWorker) {
+ const cleanup = await currentWorker.stop();
+ if (cleanup.error) {
+ console.error(`process cleanup failed: ${cleanup.error}`);
+ cleanupFailed = true;
+ }
+ const lastResult = testResults.at(-1);
+ if (cleanup.forced && lastResult?.status === "pass") {
+ lastResult.status = "fail";
+ lastResult.reason = "process cleanup required SIGKILL";
+ }
}
}
-
- reusableWorker?.stop();
- reportAndQuit(testRootDir, testResults, { strictTodo: spec.strictTodo });
+ process.removeListener("SIGINT", handleSignal);
+ process.removeListener("SIGTERM", handleSignal);
+ reportAndQuit(testRootDir, testResults, {
+ interrupted,
+ strictTodo: spec.strictTodo,
+ cleanupFailed,
+ });
}
export interface ReportOptions {
interrupted?: boolean;
+ cleanupFailed?: boolean;
/**
* Count failures of todo tests as real failures.
*/
@@ -1193,7 +1005,7 @@ export function reportAndQuit(
if (interrupted) {
process.exit(3);
- } else if (numPass < numTotal - numSkip) {
+ } else if (options.cleanupFailed || numPass < numTotal - numSkip) {
process.exit(1);
} else {
process.exit(0);
@@ -1230,6 +1042,7 @@ async function runChildInstruction({
testName,
false,
logAudit,
+ (result) => sendTestWorkerMessage({ type: "teardown", result }),
);
logger.info(`done test ${testName}: ${testResult.status}`);
return testResult;
@@ -1242,41 +1055,8 @@ function flushWorkerOutput(): Promise<unknown[]> {
]);
}
-const runTestInstrStr = process.env["TWCLI_RUN_TEST_INSTRUCTION"];
-if (runTestInstrStr && process.argv.includes("__TWCLI_TESTWORKER")) {
- setGlobalLogLevelFromString(process.env["TWCLI_LOGLEVEL"] ?? "INFO");
- // Test will call taler-wallet-cli, so we must not propagate these variables.
- delete process.env["TWCLI_RUN_TEST_INSTRUCTION"];
- delete process.env["TWCLI_LOGLEVEL"];
- const instruction = JSON.parse(runTestInstrStr) as RunTestChildInstruction;
-
- process.on("disconnect", () => {
- logger.trace("got disconnect from parent");
- process.exit(3);
- });
-
- runChildInstruction(instruction)
- .then((testResult) => {
- if (!process.send) {
- throw Error("can't communicate with parent");
- }
- process.send(testResult);
- })
- .then(() => {
- logger.trace(`test ${instruction.testName} finished in worker`);
- if (shouldLingerInTest()) {
- logger.trace("lingering ...");
- return;
- }
- process.exit(0);
- })
- .catch((e) => {
- logger.error(e);
- process.exit(1);
- });
-}
-
-if (process.argv.includes("__TWCLI_REUSABLE_TESTWORKER")) {
+const isReusableWorker = process.argv.includes("__TWCLI_REUSABLE_TESTWORKER");
+if (isReusableWorker || process.argv.includes("__TWCLI_TESTWORKER")) {
setGlobalLogLevelFromString(process.env["TWCLI_LOGLEVEL"] ?? "INFO");
delete process.env["TWCLI_LOGLEVEL"];
const baselineEnvironment = { ...process.env };
@@ -1284,7 +1064,7 @@ if (process.argv.includes("__TWCLI_REUSABLE_TESTWORKER")) {
let running = false;
process.on("disconnect", () => {
- logger.trace("reusable worker disconnected from parent");
+ logger.trace("test worker disconnected from parent");
process.exit(0);
});
@@ -1292,29 +1072,31 @@ if (process.argv.includes("__TWCLI_REUSABLE_TESTWORKER")) {
"message",
(message: { type?: string; instruction?: RunTestChildInstruction }) => {
if (message.type !== "run" || !message.instruction) {
- logger.error("reusable worker received an invalid instruction");
+ logger.error("test worker received an invalid instruction");
process.exit(2);
}
if (running) {
- logger.error("reusable worker received overlapping tests");
+ logger.error("test worker received overlapping tests");
process.exit(2);
}
running = true;
runChildInstruction(message.instruction)
.then(async (result) => {
- for (const key of Object.keys(process.env)) {
- if (!(key in baselineEnvironment)) {
- delete process.env[key];
+ if (isReusableWorker) {
+ for (const key of Object.keys(process.env)) {
+ if (!(key in baselineEnvironment)) delete process.env[key];
}
+ Object.assign(process.env, baselineEnvironment);
+ process.chdir(baselineWorkingDirectory);
}
- Object.assign(process.env, baselineEnvironment);
- process.chdir(baselineWorkingDirectory);
await flushWorkerOutput();
- if (!process.send) {
- throw Error("can't communicate with parent");
+ // Become idle before the parent can receive the result and send the
+ // next instruction.
+ if (isReusableWorker) running = false;
+ await sendTestWorkerMessage({ type: "result", result });
+ if (!isReusableWorker && !shouldLingerInTest()) {
+ process.exit(0);
}
- running = false;
- process.send({ type: "result", result });
})
.catch((error) => {
logger.error(error);
@@ -1323,8 +1105,8 @@ if (process.argv.includes("__TWCLI_REUSABLE_TESTWORKER")) {
},
);
- if (!process.send) {
- throw Error("can't communicate with parent");
- }
- process.send({ type: "ready" });
+ void sendTestWorkerMessage({ type: "ready" }).catch((error) => {
+ logger.error(error);
+ process.exit(1);
+ });
}
diff --git a/packages/taler-harness/src/integrationtests/worker-supervisor.fixture.ts b/packages/taler-harness/src/integrationtests/worker-supervisor.fixture.ts
@@ -0,0 +1,109 @@
+/*
+ This file is part of GNU Taler
+ (C) 2026 Taler Systems S.A.
+
+ GNU Taler is free software; you can redistribute it and/or modify it under the
+ terms of the GNU General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+
+import fs from "node:fs";
+import path from "node:path";
+import { GlobalTestState, runTestWithState, sh } from "../harness/harness.js";
+import {
+ RunTestChildInstruction,
+ sendTestWorkerMessage,
+} from "./worker-supervisor.js";
+
+const reusable = process.argv.includes("__TWCLI_REUSABLE_TESTWORKER");
+process.on("disconnect", () => process.exit(0));
+process.on(
+ "message",
+ async ({ instruction }: { instruction: RunTestChildInstruction }) => {
+ const { testName, testRootDir } = instruction;
+ const testDir = path.join(testRootDir, testName);
+ fs.mkdirSync(testDir, { recursive: true });
+ const state = new GlobalTestState({ testDir });
+ const result = await runTestWithState(
+ state,
+ async (state) => {
+ const graceful =
+ testName.startsWith("pass") && testName !== "pass-stuck";
+ const serviceScript = `
+ const fs = require('node:fs');
+ const net = require('node:net');
+ const server = net.createServer();
+ process.on('SIGTERM', () => {
+ fs.appendFileSync(${JSON.stringify(path.join(testDir, "signals.txt"))}, 'SIGTERM\\n');
+ if (${graceful}) {
+ console.error('graceful shutdown log');
+ server.close(() => process.exit(0));
+ }
+ });
+ // Stay below the usual ephemeral client-port range so concurrent
+ // browser tests cannot claim this port between cleanup and rebinding.
+ const listen = () => server.listen(20000 + Math.floor(Math.random() * 10000), '127.0.0.1');
+ server.on('error', (error) => {
+ if (error.code === 'EADDRINUSE') listen();
+ else throw error;
+ });
+ server.on('listening', () => {
+ fs.writeFileSync(${JSON.stringify(path.join(testDir, "service.json"))}, JSON.stringify({ pid: process.pid, port: server.address().port }));
+ console.log('ready');
+ });
+ listen();
+ `;
+ if (testName === "shell") {
+ // The shell and its immediate child exit, leaving their grandchild and
+ // its inherited stderr pipe alive. Neither is registered in state.procs.
+ const launcher = `require('node:child_process').spawn(process.execPath, ['-e', ${JSON.stringify(serviceScript)}], {stdio: 'inherit'}).unref();`;
+ const quote = (s: string) => "'" + s.replaceAll("'", "'\\''") + "'";
+ await sh(
+ state,
+ "shell",
+ `${quote(process.execPath)} -e ${quote(launcher)}`,
+ );
+ } else {
+ state.spawnService(
+ process.execPath,
+ ["-e", serviceScript],
+ "service",
+ );
+ }
+ while (!fs.existsSync(path.join(testDir, "service.json"))) {
+ await new Promise((resolve) => setTimeout(resolve, 10));
+ }
+ if (testName.startsWith("pass")) return;
+ if (testName === "crash") process.exit(17);
+ if (testName === "sigkill") process.kill(process.pid, "SIGKILL");
+ if (testName === "rejection") {
+ void Promise.reject(Error("deliberate unhandled rejection"));
+ await new Promise(() => {});
+ }
+ if (testName === "blocked") {
+ // Deliberately prevent this worker's signal handler from running.
+ while (true) {
+ /* wait for runner SIGKILL */
+ }
+ }
+ if (testName === "timeout" || testName === "interrupt")
+ await new Promise(() => {});
+ throw Error("original assertion failure");
+ },
+ testName,
+ false,
+ "no",
+ (result) => sendTestWorkerMessage({ type: "teardown", result }),
+ );
+ await sendTestWorkerMessage({ type: "result", result });
+ if (!reusable && !process.env.TALER_TEST_LINGER) process.exit(0);
+ },
+);
+await sendTestWorkerMessage({ type: "ready" });
diff --git a/packages/taler-harness/src/integrationtests/worker-supervisor.test.ts b/packages/taler-harness/src/integrationtests/worker-supervisor.test.ts
@@ -0,0 +1,238 @@
+/*
+ This file is part of GNU Taler
+ (C) 2026 Taler Systems S.A.
+
+ GNU Taler is free software; you can redistribute it and/or modify it under the
+ terms of the GNU General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { once } from "node:events";
+import fs from "node:fs";
+import net from "node:net";
+import os from "node:os";
+import path from "node:path";
+import { finished } from "node:stream/promises";
+import { test, type TestContext } from "node:test";
+import { setTimeout as delay } from "node:timers/promises";
+import { fileURLToPath } from "node:url";
+import { TestWorkerSupervisor } from "./worker-supervisor.js";
+
+const fixture = fileURLToPath(
+ new URL("./worker-supervisor.fixture.js", import.meta.url),
+);
+const testOptions = { timeout: 15_000 };
+
+function setup(t: TestContext, reusable = false, linger = false) {
+ const root = fs.mkdtempSync(path.join(os.tmpdir(), "harness-supervisor-"));
+ t.after(() => fs.rmSync(root, { recursive: true, force: true }));
+ const env = { ...process.env };
+ delete env.TALER_HARNESS_VALGRIND;
+ delete env.TALER_TEST_LINGER;
+ if (linger) env.TALER_TEST_LINGER = "1";
+ const worker = new TestWorkerSupervisor(fixture, reusable, 0, {
+ env,
+ teardownMs: 500,
+ terminateMs: 150,
+ closeMs: 1_000,
+ });
+ t.after(async () => {
+ await worker.stop();
+ });
+ async function run(testName: string, timeoutMs?: number) {
+ const log = fs.createWriteStream(path.join(root, `${testName}.log`));
+ const completion = finished(log);
+ void completion.catch(() => undefined);
+ try {
+ return await worker.run(
+ { testName, testRootDir: root, logAudit: "no" },
+ log,
+ timeoutMs,
+ );
+ } finally {
+ log.end();
+ await completion;
+ }
+ }
+ return { root, worker, run };
+}
+
+async function serviceInfo(
+ root: string,
+ name: string,
+): Promise<{ pid: number; port: number }> {
+ const file = path.join(root, name, "service.json");
+ const deadline = Date.now() + 5_000;
+ while (!fs.existsSync(file) && Date.now() < deadline) await delay(10);
+ return JSON.parse(fs.readFileSync(file, "utf8"));
+}
+
+function isRunning(pid: number): boolean {
+ try {
+ process.kill(pid, 0);
+ // Orphans may be zombies until init reaps them, but are already dead.
+ const stat = fs.readFileSync(`/proc/${pid}/stat`, "utf8");
+ return stat.slice(stat.lastIndexOf(")") + 2).split(" ")[0] !== "Z";
+ } catch (error) {
+ if (
+ ["ESRCH", "ENOENT"].includes((error as NodeJS.ErrnoException).code ?? "")
+ )
+ return false;
+ throw error;
+ }
+}
+
+async function assertCleaned(root: string, name: string, workerPid: number) {
+ const { pid, port } = await serviceInfo(root, name);
+ const deadline = Date.now() + 2_000;
+ while ((isRunning(pid) || isRunning(workerPid)) && Date.now() < deadline)
+ await delay(10);
+ assert.equal(isRunning(pid), false, "service survived cleanup");
+ assert.equal(isRunning(workerPid), false, "worker survived cleanup");
+ const server = net.createServer();
+ server.listen(port, "127.0.0.1");
+ await once(server, "listening");
+ await new Promise<void>((resolve, reject) =>
+ server.close((error) => (error ? reject(error) : resolve())),
+ );
+}
+
+for (const name of ["fail", "rejection", "crash", "sigkill", "shell"]) {
+ test(
+ `cleans up ${name} without an execution timeout`,
+ testOptions,
+ async (t) => {
+ const { root, worker, run } = setup(t);
+ const outcome = await run(name);
+ assert.equal(outcome.result.status, "fail");
+ assert.equal(outcome.cleanupFailed, false);
+ assert.equal(outcome.needsPostmortemLogScan, true);
+ if (name === "fail" || name === "shell") {
+ assert.match(
+ outcome.result.reason!,
+ /original assertion failure; test teardown timed out/,
+ );
+ }
+ await assertCleaned(root, name, worker.child.pid!);
+ },
+ );
+}
+
+for (const name of ["timeout", "blocked"]) {
+ test(`kills a ${name} worker and its services`, testOptions, async (t) => {
+ const { root, worker, run } = setup(t);
+ const outcome = await run(name, 2_000);
+ assert.match(outcome.result.reason!, /^timeout/);
+ assert.equal(outcome.cleanupFailed, false);
+ await assertCleaned(root, name, worker.child.pid!);
+ });
+}
+
+test(
+ "graceful shutdown flushes output and does not fail a passing test",
+ testOptions,
+ async (t) => {
+ const { root, worker, run } = setup(t);
+ const outcome = await run("pass");
+ assert.equal(outcome.result.status, "pass", outcome.result.reason);
+ assert.equal(outcome.needsPostmortemLogScan, false);
+ assert.match(
+ fs.readFileSync(path.join(root, "pass", "service-stderr.log"), "utf8"),
+ /graceful shutdown log/,
+ );
+ await assertCleaned(root, "pass", worker.child.pid!);
+ },
+);
+
+test(
+ "a passing body fails when its teardown requires forced termination",
+ testOptions,
+ async (t) => {
+ const { root, worker, run } = setup(t);
+ const outcome = await run("pass-stuck");
+ assert.equal(outcome.result.status, "fail");
+ assert.match(
+ outcome.result.reason!,
+ /test teardown timed out; process cleanup required SIGKILL/,
+ );
+ await assertCleaned(root, "pass-stuck", worker.child.pid!);
+ },
+);
+
+test(
+ "reuses healthy workers and discards failed workers",
+ testOptions,
+ async (t) => {
+ const { root, worker, run } = setup(t, true);
+ const pid = worker.child.pid!;
+ for (const name of ["pass-first", "pass-second"]) {
+ const outcome = await run(name);
+ assert.equal(outcome.result.status, "pass", outcome.result.reason);
+ assert.equal(isRunning(pid), true);
+ }
+ const failure = await run("fail");
+ assert.equal(failure.result.status, "fail");
+ await assertCleaned(root, "fail", pid);
+ const replacement = setup(t, true);
+ assert.notEqual(replacement.worker.child.pid, pid);
+ assert.equal((await replacement.run("pass")).result.status, "pass");
+ assert.equal((await replacement.worker.stop()).error, undefined);
+ },
+);
+
+test(
+ "interruption is idempotent and leaves unrelated processes alive",
+ testOptions,
+ async (t) => {
+ const unrelated = spawn(
+ process.execPath,
+ ["-e", "setInterval(() => {}, 1000)"],
+ { stdio: "ignore" },
+ );
+ t.after(async () => {
+ const closed = once(unrelated, "close");
+ unrelated.kill("SIGKILL");
+ await closed;
+ });
+ const { root, worker, run } = setup(t);
+ const pending = run("interrupt");
+ await serviceInfo(root, "interrupt");
+ worker.interrupt();
+ worker.interrupt(true);
+ const outcome = await pending;
+ assert.match(outcome.result.reason!, /interrupted/);
+ assert.equal(outcome.cleanupFailed, false);
+ assert.strictEqual(worker.stop(), worker.stop());
+ await assertCleaned(root, "interrupt", worker.child.pid!);
+ assert.equal(isRunning(unrelated.pid!), true);
+ },
+);
+
+test(
+ "linger preserves services until explicit interruption",
+ testOptions,
+ async (t) => {
+ const { root, worker, run } = setup(t, false, true);
+ let completed = false;
+ const pending = run("pass").then((outcome) => {
+ completed = true;
+ return outcome;
+ });
+ const service = await serviceInfo(root, "pass");
+ await delay(300);
+ assert.equal(completed, false);
+ assert.equal(isRunning(service.pid), true);
+ worker.interrupt();
+ await pending;
+ await assertCleaned(root, "pass", worker.child.pid!);
+ },
+);
diff --git a/packages/taler-harness/src/integrationtests/worker-supervisor.ts b/packages/taler-harness/src/integrationtests/worker-supervisor.ts
@@ -0,0 +1,284 @@
+/*
+ This file is part of GNU Taler
+ (C) 2026 Taler Systems S.A.
+
+ GNU Taler is free software; you can redistribute it and/or modify it under the
+ terms of the GNU General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ GNU Taler; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+
+import { fork, type ChildProcess } from "node:child_process";
+import type { WriteStream } from "node:fs";
+import { setTimeout as delay } from "node:timers/promises";
+import type { LogAuditMode, TestRunResult } from "../harness/harness.js";
+
+export interface RunTestChildInstruction {
+ testName: string;
+ testRootDir: string;
+ logAudit: LogAuditMode;
+}
+
+export type TestWorkerMessage =
+ | { type: "ready" }
+ | { type: "teardown"; result: TestRunResult }
+ | { type: "result"; result: TestRunResult };
+
+export function sendTestWorkerMessage(
+ message: TestWorkerMessage,
+): Promise<void> {
+ return new Promise((resolve, reject) => {
+ if (!process.send || !process.connected) {
+ reject(Error("can't communicate with test parent"));
+ return;
+ }
+ process.send(message, (error) => (error ? reject(error) : resolve()));
+ });
+}
+
+interface CleanupResult {
+ forced: boolean;
+ error?: string;
+}
+
+interface ActiveTest {
+ resolve: (result: TestRunResult) => void;
+ reject: (error: Error) => void;
+ preliminary?: TestRunResult;
+ result?: TestRunResult;
+ teardownTimer?: NodeJS.Timeout;
+}
+
+export interface SupervisedTestResult {
+ result: TestRunResult;
+ needsPostmortemLogScan: boolean;
+ cleanupFailed: boolean;
+}
+
+/** Owns the worker's process group, even after the worker itself has exited. */
+export class TestWorkerSupervisor {
+ readonly child: ChildProcess;
+ private readonly teardownMs: number;
+ private readonly terminateMs: number;
+ private readonly closeMs: number;
+ private active?: ActiveTest;
+ private logStream?: WriteStream;
+ private readyResolve!: () => void;
+ private readonly ready = new Promise<void>((resolve) => {
+ this.readyResolve = resolve;
+ });
+ private exitError?: Error;
+ private closed = false;
+ private groupGone = false;
+ private stopPromise?: Promise<CleanupResult>;
+ private force = false;
+
+ constructor(
+ filename: string,
+ private readonly reusable: boolean,
+ verbosity: number,
+ options: {
+ env?: NodeJS.ProcessEnv;
+ teardownMs?: number;
+ terminateMs?: number;
+ closeMs?: number;
+ } = {},
+ ) {
+ this.teardownMs = options.teardownMs ?? 5_000;
+ this.terminateMs = options.terminateMs ?? 5_000;
+ this.closeMs = options.closeMs ?? 1_000;
+ this.child = fork(
+ filename,
+ [reusable ? "__TWCLI_REUSABLE_TESTWORKER" : "__TWCLI_TESTWORKER"],
+ {
+ env: options.env ?? process.env,
+ detached: true,
+ stdio: ["pipe", "pipe", "pipe", "ipc"],
+ },
+ );
+ this.child.stdout!.on("data", (chunk: Buffer) => {
+ this.logStream?.write(chunk);
+ if (verbosity > 0) process.stdout.write(chunk);
+ });
+ this.child.stderr!.on("data", (chunk: Buffer) => {
+ this.logStream?.write(chunk);
+ if (verbosity > 0) process.stderr.write(chunk);
+ });
+ this.child.on("message", this.handleMessage);
+ this.child.on("exit", (code, signal) => {
+ this.exitError = Error(
+ `test worker exited with ${signal ? `signal ${signal}` : `code ${code}`}`,
+ );
+ this.readyResolve();
+ if (code === 0 && this.active?.result) {
+ this.active.resolve(this.active.result);
+ } else {
+ this.active?.reject(this.exitError);
+ }
+ });
+ this.child.on("error", (error) => {
+ this.exitError = error;
+ this.readyResolve();
+ this.active?.reject(error);
+ });
+ this.child.once("close", () => {
+ this.closed = true;
+ });
+ }
+
+ private readonly handleMessage = (message: TestWorkerMessage): void => {
+ if (message.type === "ready") {
+ this.readyResolve();
+ return;
+ }
+ const active = this.active;
+ if (!active) return;
+ if (message.type === "teardown") {
+ if (active.teardownTimer) return;
+ active.preliminary = message.result;
+ active.teardownTimer = setTimeout(
+ () => active.reject(Error("test teardown timed out")),
+ this.teardownMs,
+ );
+ } else if (message.type === "result") {
+ clearTimeout(active.teardownTimer);
+ active.result = message.result;
+ if (this.reusable) active.resolve(message.result);
+ }
+ };
+
+ async run(
+ instruction: RunTestChildInstruction,
+ logStream: WriteStream,
+ timeoutMs?: number,
+ ): Promise<SupervisedTestResult> {
+ if (this.active || this.stopPromise) {
+ throw Error("test worker is busy or stopped");
+ }
+ this.logStream = logStream;
+ const started = Date.now();
+ let executionTimer: NodeJS.Timeout | undefined;
+ let active!: ActiveTest;
+ const resultPromise = new Promise<TestRunResult>((resolve, reject) => {
+ active = { resolve, reject };
+ this.active = active;
+ if (timeoutMs !== undefined) {
+ executionTimer = setTimeout(() => reject(Error("timeout")), timeoutMs);
+ }
+ void this.ready
+ .then(() => {
+ if (this.active !== active || this.stopPromise) return;
+ if (this.exitError) {
+ reject(this.exitError);
+ return;
+ }
+ this.child.send({ type: "run", instruction }, (error) => {
+ if (error) reject(error);
+ });
+ })
+ .catch(reject);
+ });
+
+ let result: TestRunResult;
+ let needsPostmortemLogScan = false;
+ try {
+ result = await resultPromise;
+ } catch (error) {
+ const detail = error instanceof Error ? error.message : String(error);
+ const originalReason = active.preliminary?.reason;
+ result = {
+ name: instruction.testName,
+ status: "fail",
+ timeSec: (Date.now() - started) / 1000,
+ reason: originalReason ? `${originalReason}; ${detail}` : detail,
+ };
+ needsPostmortemLogScan = true;
+ } finally {
+ clearTimeout(executionTimer);
+ clearTimeout(active.teardownTimer);
+ this.active = undefined;
+ }
+
+ let cleanupFailed = false;
+ if (!this.reusable || result.status === "fail" || this.stopPromise) {
+ const cleanup = await this.stop();
+ if (cleanup.forced || cleanup.error) {
+ const detail = cleanup.error
+ ? `process cleanup failed: ${cleanup.error}`
+ : "process cleanup required SIGKILL";
+ result.status = "fail";
+ result.reason = result.reason ? `${result.reason}; ${detail}` : detail;
+ needsPostmortemLogScan = true;
+ }
+ cleanupFailed = cleanup.error !== undefined;
+ }
+ this.logStream = undefined;
+ return { result, needsPostmortemLogScan, cleanupFailed };
+ }
+
+ interrupt(force = false): void {
+ this.force ||= force;
+ this.active?.reject(Error("test suite interrupted"));
+ void this.stop();
+ }
+
+ stop(): Promise<CleanupResult> {
+ return (this.stopPromise ??= this.stopGroup());
+ }
+
+ private signal(signal: NodeJS.Signals | 0): boolean {
+ const pid = this.child.pid;
+ if (this.groupGone || pid === undefined) return false;
+ if (pid <= 0 || pid === process.pid) {
+ throw Error(`invalid test worker pid ${pid}`);
+ }
+ try {
+ process.kill(-pid, signal);
+ return true;
+ } catch (error) {
+ if ((error as NodeJS.ErrnoException).code === "ESRCH") {
+ this.groupGone = true;
+ return false;
+ }
+ throw error;
+ }
+ }
+
+ private async stopGroup(): Promise<CleanupResult> {
+ let forced = false;
+ try {
+ if (this.signal("SIGTERM")) {
+ const deadline = Date.now() + this.terminateMs;
+ while (!this.force && Date.now() < deadline && this.signal(0)) {
+ await delay(25);
+ }
+ forced = this.signal("SIGKILL");
+ }
+ // An orphan can remain a zombie until its new parent reaps it. Do not
+ // require the group to disappear before accepting closed worker pipes.
+ const deadline = Date.now() + this.closeMs;
+ while (!this.closed && Date.now() < deadline) await delay(10);
+ if (!this.closed)
+ throw Error("worker output did not close after termination");
+ return { forced };
+ } catch (error) {
+ return {
+ forced,
+ error: error instanceof Error ? error.message : String(error),
+ };
+ } finally {
+ this.child.removeListener("message", this.handleMessage);
+ // A failed cleanup must not keep the reporting parent alive on pipes.
+ this.child.stdin?.destroy();
+ this.child.stdout?.destroy();
+ this.child.stderr?.destroy();
+ if (this.child.connected) this.child.disconnect();
+ }
+ }
+}
