mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-18 08:31:49 -06:00
fix(qa): settle Unix commands after PGID cleanup (#120010)
Punchcard-Session: cobalt-valley-meadow-mg
This commit is contained in:
Vincent Koc
@@ -1,13 +1,7 @@
|
||||
// Qa Lab plugin module implements gateway child behavior.
|
||||
import { spawn, spawnSync, type ChildProcess } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import {
|
||||
createWriteStream,
|
||||
existsSync,
|
||||
readFileSync,
|
||||
readdirSync,
|
||||
type WriteStream,
|
||||
} from "node:fs";
|
||||
import { createWriteStream, existsSync, type WriteStream } from "node:fs";
|
||||
import fs from "node:fs/promises";
|
||||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
@@ -54,6 +48,11 @@ import {
|
||||
import { startQaGatewayRpcClient } from "./gateway-rpc-client.js";
|
||||
import { splitQaModelRef, type QaProviderMode } from "./model-selection.js";
|
||||
import { resolveQaNodeExecPath } from "./node-exec.js";
|
||||
import {
|
||||
inspectLinuxProcessGroup,
|
||||
inspectLinuxProcessGroupStats,
|
||||
type QaLinuxProcessGroupInspector,
|
||||
} from "./posix-process-group.js";
|
||||
import { readProcessTreeCpuMs, readProcessTreeRssBytes } from "./process-tree-cpu.js";
|
||||
import {
|
||||
normalizeQaProviderModeEnv,
|
||||
@@ -89,7 +88,6 @@ const QA_GATEWAY_CHILD_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 30_000;
|
||||
// Loaded Docker runners can take several seconds to reap a force-killed process group.
|
||||
const QA_GATEWAY_CHILD_FORCE_SHUTDOWN_TIMEOUT_MS = 10_000;
|
||||
const QA_GATEWAY_LOG_CLOSE_TIMEOUT_MS = 5_000;
|
||||
const QA_GATEWAY_PROCESS_TREE_DIAGNOSTIC_MAX_CHARS = 2_048;
|
||||
const QA_MOCK_OPENAI_API_KEY = ["qa", "mock", "openai", "key"].join("-");
|
||||
const QA_GATEWAY_CHILD_BLOCKED_SECRET_ENV_VARS = Object.freeze([
|
||||
"OPENCLAW_QA_CONVEX_SECRET_CI",
|
||||
@@ -794,98 +792,11 @@ function isProcessAlreadyExitedError(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException | undefined)?.code === "ESRCH";
|
||||
}
|
||||
|
||||
function parseLinuxProcessStat(raw: string) {
|
||||
const commandStart = raw.indexOf("(");
|
||||
const commandEnd = raw.lastIndexOf(")");
|
||||
if (commandStart <= 0 || commandEnd <= commandStart) {
|
||||
return null;
|
||||
}
|
||||
const pid = Number.parseInt(raw.slice(0, commandStart).trim(), 10);
|
||||
const fields = raw
|
||||
.slice(commandEnd + 1)
|
||||
.trim()
|
||||
.split(/\s+/u);
|
||||
const state = fields[0];
|
||||
const processGroupId = Number.parseInt(fields[2] ?? "", 10);
|
||||
if (
|
||||
!Number.isSafeInteger(pid) ||
|
||||
pid <= 0 ||
|
||||
!state ||
|
||||
!Number.isSafeInteger(processGroupId) ||
|
||||
processGroupId <= 0
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
command: raw.slice(commandStart + 1, commandEnd),
|
||||
pid,
|
||||
processGroupId,
|
||||
state,
|
||||
};
|
||||
}
|
||||
|
||||
function boundQaGatewayProcessTreeDiagnostics(details: string) {
|
||||
if (details.length <= QA_GATEWAY_PROCESS_TREE_DIAGNOSTIC_MAX_CHARS) {
|
||||
if (details.length <= 2_048) {
|
||||
return details;
|
||||
}
|
||||
return `${sliceUtf16Safe(details, 0, QA_GATEWAY_PROCESS_TREE_DIAGNOSTIC_MAX_CHARS - 3)}...`;
|
||||
}
|
||||
|
||||
function inspectLinuxProcessGroupStats(processGroupId: number, stats: readonly string[]) {
|
||||
const members = stats
|
||||
.map((raw) => parseLinuxProcessStat(raw))
|
||||
.filter(
|
||||
(entry): entry is NonNullable<ReturnType<typeof parseLinuxProcessStat>> =>
|
||||
entry?.processGroupId === processGroupId,
|
||||
)
|
||||
.toSorted((left, right) => left.pid - right.pid);
|
||||
const diagnostics = members
|
||||
.map(
|
||||
(member) =>
|
||||
`pid=${member.pid} state=${member.state} command=${JSON.stringify(member.command)}`,
|
||||
)
|
||||
.join(", ");
|
||||
return {
|
||||
alive:
|
||||
members.length === 0
|
||||
? null
|
||||
: members.some((entry) => entry.state !== "Z" && entry.state !== "X"),
|
||||
diagnostics: boundQaGatewayProcessTreeDiagnostics(
|
||||
`pgid=${processGroupId} members=[${diagnostics}]`,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
type QaLinuxProcessGroupInspection = ReturnType<typeof inspectLinuxProcessGroupStats>;
|
||||
type QaLinuxProcessGroupInspector = (
|
||||
processGroupId: number,
|
||||
) => QaLinuxProcessGroupInspection | null;
|
||||
|
||||
function inspectLinuxProcessGroup(processGroupId: number): QaLinuxProcessGroupInspection | null {
|
||||
if (process.platform !== "linux") {
|
||||
return null;
|
||||
}
|
||||
let entries;
|
||||
try {
|
||||
entries = readdirSync("/proc", { withFileTypes: true });
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
const stats: string[] = [];
|
||||
for (const entry of entries) {
|
||||
if (!entry.isDirectory() || !/^\d+$/u.test(entry.name)) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
stats.push(readFileSync(path.join("/proc", entry.name, "stat"), "utf8"));
|
||||
} catch (error) {
|
||||
// Processes can exit while /proc is being scanned.
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
return inspectLinuxProcessGroupStats(processGroupId, stats);
|
||||
return `${sliceUtf16Safe(details, 0, 2_045)}...`;
|
||||
}
|
||||
|
||||
function isQaGatewayChildProcessTreeAlive(
|
||||
|
||||
+14
-4
@@ -115,9 +115,13 @@ describe("Matrix QA CLI runtime stream errors", () => {
|
||||
childProcessMocks.spawn.mockClear();
|
||||
});
|
||||
|
||||
it.each(["stdout", "stderr"] as const)(
|
||||
"rejects after cleaning up when %s emits a stream error",
|
||||
async (streamName) => {
|
||||
it.each([
|
||||
["stdout", false],
|
||||
["stderr", false],
|
||||
["stdout", true],
|
||||
] as const)(
|
||||
"rejects after cleaning up when %s emits a stream error (after exit: %s)",
|
||||
async (streamName, afterExit) => {
|
||||
const { grandchildPidPath, grandchildReadyPath, root } = await createCliRoot();
|
||||
let child: ChildProcess | undefined;
|
||||
let grandchildPid: number | undefined;
|
||||
@@ -142,7 +146,13 @@ describe("Matrix QA CLI runtime stream errors", () => {
|
||||
grandchildPid = await waitForPidFile(grandchildPidPath, 2_000);
|
||||
await waitForFile(grandchildReadyPath, 2_000);
|
||||
|
||||
child[streamName]?.emit("error", new Error(`${streamName} pipe failed`));
|
||||
const streamError = new Error(`${streamName} pipe failed`);
|
||||
if (afterExit) {
|
||||
child.once("exit", () => child?.[streamName]?.emit("error", streamError));
|
||||
child.kill("SIGTERM");
|
||||
} else {
|
||||
child[streamName]?.emit("error", streamError);
|
||||
}
|
||||
|
||||
await expect(session.wait()).rejects.toThrow(
|
||||
`${streamName} stream error: ${streamName} pipe failed`,
|
||||
|
||||
@@ -7,6 +7,7 @@ import { setTimeout as sleep } from "node:timers/promises";
|
||||
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
|
||||
import { redactSensitiveText } from "openclaw/plugin-sdk/logging-core";
|
||||
import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path";
|
||||
import { createQaPosixCommandSettlement } from "../../../posix-command-settlement.js";
|
||||
import {
|
||||
killMatrixQaCliChild,
|
||||
resolveMatrixQaOpenClawCliEntryPath,
|
||||
@@ -100,20 +101,6 @@ function formatMatrixQaCliTimeoutError(result: MatrixQaCliRunResult, timeoutMs:
|
||||
.join("\n");
|
||||
}
|
||||
|
||||
function isMatrixQaCliChildProcessGroupRunning(
|
||||
child: ReturnType<typeof startOpenClawCliProcess>,
|
||||
): boolean {
|
||||
if (process.platform === "win32" || !child.pid) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
process.kill(-child.pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export function startMatrixQaOpenClawCli(params: {
|
||||
allowNonZero?: boolean;
|
||||
args: string[];
|
||||
@@ -129,11 +116,6 @@ export function startMatrixQaOpenClawCli(params: {
|
||||
let closed = false;
|
||||
let closeError: Error | undefined;
|
||||
let closeResult: MatrixQaCliRunResult | undefined;
|
||||
let killRequested = false;
|
||||
let timedOut = false;
|
||||
let forceKillTimeout: NodeJS.Timeout | undefined;
|
||||
let forceSettleTimeout: NodeJS.Timeout | undefined;
|
||||
let streamFailure: Error | undefined;
|
||||
let settleWait:
|
||||
| {
|
||||
reject: (error: Error) => void;
|
||||
@@ -167,120 +149,67 @@ export function startMatrixQaOpenClawCli(params: {
|
||||
settleWait.resolve(result);
|
||||
}
|
||||
};
|
||||
const finishTimeout = (result: MatrixQaCliRunResult) => {
|
||||
finish(result, new Error(formatMatrixQaCliTimeoutError(result, params.timeoutMs)));
|
||||
};
|
||||
const finishResult = (result: MatrixQaCliRunResult) => {
|
||||
if (result.exitCode !== 0 && params.allowNonZero !== true) {
|
||||
finish(result, new Error(formatMatrixQaCliExitError(result)));
|
||||
return;
|
||||
}
|
||||
finish(result);
|
||||
};
|
||||
const clearForcedTimeouts = () => {
|
||||
if (forceKillTimeout) {
|
||||
clearTimeout(forceKillTimeout);
|
||||
forceKillTimeout = undefined;
|
||||
}
|
||||
if (forceSettleTimeout) {
|
||||
clearTimeout(forceSettleTimeout);
|
||||
forceSettleTimeout = undefined;
|
||||
}
|
||||
};
|
||||
const finishForcedCleanup = (result: MatrixQaCliRunResult) => {
|
||||
if (timedOut) {
|
||||
finishTimeout(result);
|
||||
return;
|
||||
}
|
||||
if (streamFailure) {
|
||||
finish(result, streamFailure);
|
||||
return;
|
||||
}
|
||||
finishResult(result);
|
||||
};
|
||||
const scheduleForcedCleanup = () => {
|
||||
if (forceKillTimeout || forceSettleTimeout) {
|
||||
return;
|
||||
}
|
||||
forceKillTimeout = setTimeout(() => {
|
||||
forceKillTimeout = undefined;
|
||||
killMatrixQaCliChild(child, "SIGKILL");
|
||||
forceSettleTimeout = setTimeout(() => {
|
||||
forceSettleTimeout = undefined;
|
||||
finishForcedCleanup(
|
||||
buildMatrixQaCliResult({
|
||||
args: params.args,
|
||||
exitCode: 1,
|
||||
output: readOutput(),
|
||||
}),
|
||||
);
|
||||
}, MATRIX_QA_CLI_TIMEOUT_FORCE_SETTLE_MS);
|
||||
}, MATRIX_QA_CLI_TIMEOUT_KILL_GRACE_MS);
|
||||
};
|
||||
|
||||
const timeout = setTimeout(() => {
|
||||
timedOut = true;
|
||||
killMatrixQaCliChild(child, "SIGTERM");
|
||||
scheduleForcedCleanup();
|
||||
}, params.timeoutMs);
|
||||
const handleStreamError = (stream: "stderr" | "stdout", error: Error) => {
|
||||
if (closed || timedOut || killRequested) {
|
||||
return;
|
||||
}
|
||||
clearTimeout(timeout);
|
||||
killRequested = true;
|
||||
streamFailure = new Error(`${stream} stream error: ${formatErrorMessage(error)}`, {
|
||||
cause: error,
|
||||
});
|
||||
// Keep stream failures on the normal kill path so detached descendants are gone
|
||||
// before the session reports the parent-side pipe failure.
|
||||
killMatrixQaCliChild(child, "SIGTERM");
|
||||
scheduleForcedCleanup();
|
||||
};
|
||||
|
||||
child.stdout.on("data", (chunk) => stdout.push(Buffer.from(chunk)));
|
||||
child.stdout.on("error", (error) => handleStreamError("stdout", error));
|
||||
child.stderr.on("data", (chunk) => stderr.push(Buffer.from(chunk)));
|
||||
child.stderr.on("error", (error) => handleStreamError("stderr", error));
|
||||
const isWindows = process.platform === "win32";
|
||||
const settlement = createQaPosixCommandSettlement({
|
||||
child,
|
||||
settlementFailureMessage: `${formatMatrixQaCliCommand(params.args)} settlement failed`,
|
||||
forceKillAfterMs: MATRIX_QA_CLI_TIMEOUT_KILL_GRACE_MS,
|
||||
initialSignal: "SIGTERM",
|
||||
...(isWindows
|
||||
? {
|
||||
windowsCleanup: {
|
||||
closeCompletesCleanup: true,
|
||||
signal: (signal: NodeJS.Signals) => {
|
||||
try {
|
||||
killMatrixQaCliChild(child, signal);
|
||||
return undefined;
|
||||
} catch (error) {
|
||||
return error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
executionTimeoutMs: params.timeoutMs,
|
||||
onSettled: (outcome) => {
|
||||
const primary = outcome.primary;
|
||||
const result = buildMatrixQaCliResult({
|
||||
args: params.args,
|
||||
exitCode: primary.type === "exit" ? (primary.exitCode ?? 1) : 1,
|
||||
output: readOutput(),
|
||||
});
|
||||
const primaryError =
|
||||
primary.type === "spawn-error"
|
||||
? primary.error
|
||||
: primary.type === "stream-error"
|
||||
? new Error(`${primary.stream} stream error: ${formatErrorMessage(primary.error)}`, {
|
||||
cause: primary.error,
|
||||
})
|
||||
: primary.type === "timeout"
|
||||
? new Error(formatMatrixQaCliTimeoutError(result, params.timeoutMs))
|
||||
: result.exitCode !== 0 && params.allowNonZero !== true
|
||||
? new Error(formatMatrixQaCliExitError(result))
|
||||
: undefined;
|
||||
finish(
|
||||
result,
|
||||
outcome.settlementFailure
|
||||
? primaryError
|
||||
? new AggregateError(
|
||||
[primaryError, outcome.settlementFailure],
|
||||
"Matrix QA CLI command and settlement failed",
|
||||
)
|
||||
: outcome.settlementFailure
|
||||
: primaryError,
|
||||
);
|
||||
},
|
||||
onStderrData: (chunk) => stderr.push(Buffer.from(chunk)),
|
||||
onStdoutData: (chunk) => stdout.push(Buffer.from(chunk)),
|
||||
processGroupId: isWindows ? undefined : child.pid,
|
||||
verifyAfterMs: MATRIX_QA_CLI_TIMEOUT_FORCE_SETTLE_MS,
|
||||
});
|
||||
if (params.stdin !== undefined) {
|
||||
child.stdin.end(params.stdin);
|
||||
}
|
||||
child.on("error", (error) => {
|
||||
if (streamFailure) {
|
||||
// Forced cleanup owns settlement after a stream failure. Finishing here could
|
||||
// reject before detached descendants are proven gone.
|
||||
return;
|
||||
}
|
||||
clearTimeout(timeout);
|
||||
clearForcedTimeouts();
|
||||
finish(
|
||||
buildMatrixQaCliResult({
|
||||
args: params.args,
|
||||
exitCode: 1,
|
||||
output: readOutput(),
|
||||
}),
|
||||
error,
|
||||
);
|
||||
});
|
||||
child.on("close", (exitCode) => {
|
||||
clearTimeout(timeout);
|
||||
const result = buildMatrixQaCliResult({
|
||||
args: params.args,
|
||||
exitCode: exitCode ?? 1,
|
||||
output: readOutput(),
|
||||
});
|
||||
if (timedOut || killRequested) {
|
||||
// A closed parent is not proof that detached, ignored-stdio descendants are gone.
|
||||
if (isMatrixQaCliChildProcessGroupRunning(child)) {
|
||||
return;
|
||||
}
|
||||
clearForcedTimeouts();
|
||||
finishForcedCleanup(result);
|
||||
return;
|
||||
}
|
||||
clearForcedTimeouts();
|
||||
finishResult(result);
|
||||
});
|
||||
|
||||
return {
|
||||
args: params.args,
|
||||
@@ -335,10 +264,7 @@ export function startMatrixQaOpenClawCli(params: {
|
||||
},
|
||||
kill: () => {
|
||||
if (!closed) {
|
||||
clearTimeout(timeout);
|
||||
killRequested = true;
|
||||
killMatrixQaCliChild(child, "SIGTERM");
|
||||
scheduleForcedCleanup();
|
||||
settlement.requestCleanup();
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { createQaPosixCommandSettlement } from "./posix-command-settlement.js";
|
||||
|
||||
type TestOutcome = Parameters<Parameters<typeof createQaPosixCommandSettlement>[0]["onSettled"]>[0];
|
||||
|
||||
function createChild() {
|
||||
const child = new EventEmitter() as ChildProcess;
|
||||
const childKill = vi.fn(() => true);
|
||||
const stderrDestroy = vi.fn();
|
||||
const stdoutDestroy = vi.fn();
|
||||
Object.defineProperty(child, "pid", { value: 42 });
|
||||
child.stdout = Object.assign(new EventEmitter(), { destroy: stdoutDestroy }) as never;
|
||||
child.stderr = Object.assign(new EventEmitter(), { destroy: stderrDestroy }) as never;
|
||||
child.kill = childKill as ChildProcess["kill"];
|
||||
return { child, childKill, stderrDestroy, stdoutDestroy };
|
||||
}
|
||||
|
||||
describe("POSIX command settlement", () => {
|
||||
let processGroupAlive: boolean;
|
||||
let processKill: ReturnType<typeof vi.spyOn>;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
processGroupAlive = false;
|
||||
processKill = vi.spyOn(process, "kill").mockImplementation((pid, signal) => {
|
||||
if (pid !== -42) {
|
||||
return true;
|
||||
}
|
||||
if (signal === 0 && !processGroupAlive) {
|
||||
throw Object.assign(new Error("gone"), { code: "ESRCH" });
|
||||
}
|
||||
if (signal === "SIGKILL") {
|
||||
processGroupAlive = false;
|
||||
}
|
||||
return true;
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
function start(
|
||||
params: {
|
||||
executionTimeoutMs?: number;
|
||||
onSettled?: (outcome: TestOutcome) => void;
|
||||
} = {},
|
||||
) {
|
||||
const childFixture = createChild();
|
||||
const { child } = childFixture;
|
||||
const settled = vi.fn(params.onSettled);
|
||||
const controller = createQaPosixCommandSettlement({
|
||||
child,
|
||||
settlementFailureMessage: "settlement failed",
|
||||
executionTimeoutMs: params.executionTimeoutMs,
|
||||
forceKillAfterMs: 20,
|
||||
initialSignal: "SIGTERM",
|
||||
onSettled: settled,
|
||||
processGroupId: 42,
|
||||
verifyAfterMs: 10,
|
||||
});
|
||||
return { ...childFixture, controller, settled };
|
||||
}
|
||||
|
||||
it("keeps the exit tuple, resets only idle, and disposes listeners once", async () => {
|
||||
const { child, settled } = start();
|
||||
|
||||
child.emit("exit", 7, null);
|
||||
await vi.advanceTimersByTimeAsync(90);
|
||||
child.stdout?.emit("data", Buffer.from("tail"));
|
||||
await vi.advanceTimersByTimeAsync(99);
|
||||
expect(settled).not.toHaveBeenCalled();
|
||||
child.emit("close", 7, null);
|
||||
|
||||
expect(settled).toHaveBeenCalledOnce();
|
||||
expect(settled).toHaveBeenCalledWith({
|
||||
primary: { type: "exit", exitCode: 7, signal: null },
|
||||
});
|
||||
expect(child.listenerCount("exit")).toBe(0);
|
||||
expect(child.listenerCount("close")).toBe(0);
|
||||
child.emit("close", 7, null);
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
expect(settled).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("caps active output at one second and destroys readers only after cleanup", async () => {
|
||||
processGroupAlive = true;
|
||||
const { child, settled, stderrDestroy, stdoutDestroy } = start();
|
||||
child.emit("exit", 0, null);
|
||||
|
||||
for (let index = 0; index < 10; index += 1) {
|
||||
await vi.advanceTimersByTimeAsync(90);
|
||||
child.stdout?.emit("data", Buffer.from(String(index)));
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(99);
|
||||
expect(stdoutDestroy).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(processKill).toHaveBeenCalledWith(-42, "SIGTERM");
|
||||
expect(stdoutDestroy).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
expect(processKill).toHaveBeenCalledWith(-42, "SIGKILL");
|
||||
await vi.advanceTimersByTimeAsync(10);
|
||||
|
||||
expect(stdoutDestroy).toHaveBeenCalledOnce();
|
||||
expect(stderrDestroy).toHaveBeenCalledOnce();
|
||||
expect(settled).toHaveBeenCalledWith({
|
||||
settlementFailure: expect.objectContaining({ message: "stdio-drain-timeout" }),
|
||||
primary: { type: "exit", exitCode: 0, signal: null },
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves the timeout primary when process-group signaling fails", async () => {
|
||||
processGroupAlive = true;
|
||||
processKill.mockImplementation((pid: number, signal?: NodeJS.Signals | 0) => {
|
||||
if (pid === -42 && signal !== 0) {
|
||||
throw Object.assign(new Error(`cannot send ${String(signal)}`), { code: "EPERM" });
|
||||
}
|
||||
return true;
|
||||
});
|
||||
const { child, childKill, settled } = start({ executionTimeoutMs: 100 });
|
||||
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
child.emit("exit", 0, null);
|
||||
child.emit("close", 0, null);
|
||||
await vi.advanceTimersByTimeAsync(30);
|
||||
|
||||
expect(settled).toHaveBeenCalledOnce();
|
||||
expect(settled.mock.calls[0]?.[0]).toMatchObject({
|
||||
primary: { type: "timeout" },
|
||||
settlementFailure: expect.any(Error),
|
||||
});
|
||||
expect(childKill).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("bounds stdio draining when a timeout cleans the group without close", async () => {
|
||||
const { settled, stderrDestroy, stdoutDestroy } = start({ executionTimeoutMs: 100 });
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1_099);
|
||||
expect(settled).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
|
||||
expect(stdoutDestroy).toHaveBeenCalledOnce();
|
||||
expect(stderrDestroy).toHaveBeenCalledOnce();
|
||||
expect(settled).toHaveBeenCalledWith({
|
||||
settlementFailure: expect.objectContaining({ message: "stdio-drain-timeout" }),
|
||||
primary: { type: "timeout" },
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves a stream failure that follows a successful exit", async () => {
|
||||
const { child, settled } = start();
|
||||
const streamFailure = new Error("stdout pipe failed");
|
||||
|
||||
child.emit("exit", 0, null);
|
||||
child.stdout?.emit("error", streamFailure);
|
||||
child.emit("close", 0, null);
|
||||
|
||||
expect(settled).toHaveBeenCalledWith({
|
||||
primary: { type: "exit", exitCode: 0, signal: null },
|
||||
settlementFailure: expect.objectContaining({
|
||||
cause: streamFailure,
|
||||
message: "stdout stream error: stdout pipe failed",
|
||||
}),
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,267 @@
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import { isQaPosixProcessGroupAlive, signalQaPosixProcessGroup } from "./posix-process-group.js";
|
||||
|
||||
type QaPosixCommandPrimary =
|
||||
| { type: "exit"; exitCode: number | null; signal: NodeJS.Signals | null }
|
||||
| { type: "manual" | "timeout" }
|
||||
| { type: "parent-signal"; signal: "SIGINT" | "SIGTERM" }
|
||||
| { type: "spawn-error"; error: Error }
|
||||
| { type: "stream-error"; error: Error; stream: "stderr" | "stdout" };
|
||||
|
||||
type QaPosixCommandSettlementParams = {
|
||||
child: ChildProcess;
|
||||
settlementFailureMessage: string;
|
||||
executionTimeoutMs?: number;
|
||||
forceKillAfterMs: number;
|
||||
forwardParentSignals?: boolean;
|
||||
initialSignal: NodeJS.Signals;
|
||||
onStderrData?: (chunk: Buffer) => void;
|
||||
onSettled: (outcome: { primary: QaPosixCommandPrimary; settlementFailure?: Error }) => void;
|
||||
onStdoutData?: (chunk: Buffer) => void;
|
||||
processGroupId: number | undefined;
|
||||
verifyAfterMs: number;
|
||||
windowsCleanup?: {
|
||||
alive?: () => boolean;
|
||||
closeCompletesCleanup?: boolean;
|
||||
signal: (signal: NodeJS.Signals) => Error | undefined;
|
||||
};
|
||||
};
|
||||
|
||||
export function createQaPosixCommandSettlement(params: QaPosixCommandSettlementParams) {
|
||||
const timers: NodeJS.Timeout[] = [];
|
||||
const errors: Error[] = [];
|
||||
let cleanupDone = false;
|
||||
let cleanupStarted = false;
|
||||
let disposed = false;
|
||||
let drainDeadline: NodeJS.Timeout | undefined;
|
||||
let drainIdle: NodeJS.Timeout | undefined;
|
||||
let drainTimedOut = false;
|
||||
let executionTimer: NodeJS.Timeout | undefined;
|
||||
let primary: QaPosixCommandPrimary | undefined;
|
||||
let stdioDrained = false;
|
||||
const windows = params.windowsCleanup;
|
||||
|
||||
const schedule = (fn: () => void, delay: number) => {
|
||||
const timer = setTimeout(() => {
|
||||
fn();
|
||||
}, delay);
|
||||
timers.push(timer);
|
||||
return timer;
|
||||
};
|
||||
const cancel = (timer: NodeJS.Timeout | undefined) => {
|
||||
if (timer) {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
};
|
||||
// Cleanup owns only the original PGID; a descendant that calls setsid can escape it.
|
||||
const alive = () =>
|
||||
windows?.alive?.() ??
|
||||
(windows
|
||||
? true
|
||||
: params.processGroupId !== undefined && isQaPosixProcessGroupAlive(params.processGroupId));
|
||||
const signal = (nextSignal: NodeJS.Signals) => {
|
||||
const error = windows
|
||||
? windows.signal(nextSignal)
|
||||
: params.processGroupId === undefined
|
||||
? undefined
|
||||
: signalQaPosixProcessGroup(params.processGroupId, nextSignal);
|
||||
if (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
};
|
||||
const dispose = () => {
|
||||
if (disposed) {
|
||||
return;
|
||||
}
|
||||
disposed = true;
|
||||
for (const timer of timers) {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
params.child.stdout?.removeListener("data", onStdoutData);
|
||||
params.child.stdout?.removeListener("error", onStdoutError);
|
||||
params.child.stderr?.removeListener("data", onStderrData);
|
||||
params.child.stderr?.removeListener("error", onStderrError);
|
||||
params.child.removeListener("error", onChildError);
|
||||
params.child.removeListener("exit", onExit);
|
||||
params.child.removeListener("close", onClose);
|
||||
process.removeListener("exit", onParentExit);
|
||||
process.removeListener("SIGINT", onParentSigint);
|
||||
process.removeListener("SIGTERM", onParentSigterm);
|
||||
};
|
||||
const settle = () => {
|
||||
if (disposed || !primary || !cleanupDone || !stdioDrained) {
|
||||
return;
|
||||
}
|
||||
const parentSignal = primary.type === "parent-signal" ? primary.signal : undefined;
|
||||
const settlementFailure =
|
||||
errors.length > 1 ? new AggregateError(errors, params.settlementFailureMessage) : errors[0];
|
||||
dispose();
|
||||
params.onSettled({ primary, ...(settlementFailure ? { settlementFailure } : {}) });
|
||||
if (parentSignal) {
|
||||
process.kill(process.pid, parentSignal);
|
||||
}
|
||||
};
|
||||
const finishCleanup = () => {
|
||||
cleanupDone = true;
|
||||
if (drainTimedOut && !stdioDrained) {
|
||||
params.child.stdout?.destroy();
|
||||
params.child.stderr?.destroy();
|
||||
stdioDrained = true;
|
||||
}
|
||||
settle();
|
||||
};
|
||||
const verify = () => {
|
||||
if (alive() && !windows) {
|
||||
errors.push(
|
||||
new Error(`${params.settlementFailureMessage}: pgid=${params.processGroupId} alive`),
|
||||
);
|
||||
}
|
||||
if (!stdioDrained) {
|
||||
drainTimedOut = true;
|
||||
if (!windows && !errors.some((error) => error.message === "stdio-drain-timeout")) {
|
||||
errors.push(new Error("stdio-drain-timeout"));
|
||||
}
|
||||
}
|
||||
finishCleanup();
|
||||
};
|
||||
const forceKill = () => {
|
||||
if (!alive()) {
|
||||
finishCleanup();
|
||||
return;
|
||||
}
|
||||
signal("SIGKILL");
|
||||
schedule(verify, params.verifyAfterMs);
|
||||
};
|
||||
const startCleanup = (initialSignal = params.initialSignal) => {
|
||||
if (cleanupStarted) {
|
||||
return;
|
||||
}
|
||||
cleanupStarted = true;
|
||||
if (!alive()) {
|
||||
finishCleanup();
|
||||
return;
|
||||
}
|
||||
signal(initialSignal);
|
||||
schedule(
|
||||
initialSignal === "SIGKILL" ? verify : forceKill,
|
||||
initialSignal === "SIGKILL" ? params.verifyAfterMs : params.forceKillAfterMs,
|
||||
);
|
||||
};
|
||||
const onDrainDeadline = () => {
|
||||
drainTimedOut = true;
|
||||
errors.push(new Error("stdio-drain-timeout"));
|
||||
startCleanup();
|
||||
if (cleanupDone) {
|
||||
finishCleanup();
|
||||
}
|
||||
};
|
||||
const armDrainDeadline = () => {
|
||||
if (!drainDeadline && !stdioDrained) {
|
||||
drainDeadline = schedule(onDrainDeadline, 1_000);
|
||||
}
|
||||
};
|
||||
const freeze = (nextPrimary: QaPosixCommandPrimary, initialSignal = params.initialSignal) => {
|
||||
primary ??= nextPrimary;
|
||||
cancel(executionTimer);
|
||||
// Every terminal path needs a drain bound: an escaped descendant can retain
|
||||
// inherited stdio even after the original process group is gone.
|
||||
armDrainDeadline();
|
||||
startCleanup(initialSignal);
|
||||
settle();
|
||||
};
|
||||
const armIdle = () => {
|
||||
cancel(drainIdle);
|
||||
drainIdle = schedule(startCleanup, 100);
|
||||
};
|
||||
const onOutput = () => {
|
||||
if (primary?.type === "exit" && !drainTimedOut && !stdioDrained) {
|
||||
armIdle();
|
||||
}
|
||||
};
|
||||
function onStdoutData(chunk: Buffer) {
|
||||
params.onStdoutData?.(chunk);
|
||||
onOutput();
|
||||
}
|
||||
function onStderrData(chunk: Buffer) {
|
||||
params.onStderrData?.(chunk);
|
||||
onOutput();
|
||||
}
|
||||
const freezeStreamError = (stream: "stderr" | "stdout", error: Error) => {
|
||||
if (primary) {
|
||||
errors.push(new Error(`${stream} stream error: ${error.message}`, { cause: error }));
|
||||
}
|
||||
freeze({ type: "stream-error", stream, error });
|
||||
};
|
||||
function onStdoutError(error: Error) {
|
||||
freezeStreamError("stdout", error);
|
||||
}
|
||||
function onStderrError(error: Error) {
|
||||
freezeStreamError("stderr", error);
|
||||
}
|
||||
function onChildError(error: Error) {
|
||||
freeze({ type: "spawn-error", error });
|
||||
}
|
||||
// `exit` freezes the leader tuple; only `close` can prove stdio drained.
|
||||
function onExit(exitCode: number | null, nextSignal: NodeJS.Signals | null) {
|
||||
primary ??= { type: "exit", exitCode, signal: nextSignal };
|
||||
cancel(executionTimer);
|
||||
armDrainDeadline();
|
||||
armIdle();
|
||||
if (cleanupStarted && !cleanupDone && !alive()) {
|
||||
finishCleanup();
|
||||
}
|
||||
}
|
||||
function onClose() {
|
||||
stdioDrained = true;
|
||||
cancel(drainIdle);
|
||||
cancel(drainDeadline);
|
||||
if (!cleanupStarted && windows) {
|
||||
cleanupDone = true;
|
||||
} else if (cleanupStarted && windows?.closeCompletesCleanup) {
|
||||
finishCleanup();
|
||||
} else {
|
||||
startCleanup();
|
||||
if (!cleanupDone && !alive()) {
|
||||
finishCleanup();
|
||||
}
|
||||
}
|
||||
settle();
|
||||
}
|
||||
function onParentExit() {
|
||||
signal("SIGKILL");
|
||||
}
|
||||
const onParentSignal = (nextSignal: "SIGINT" | "SIGTERM") => {
|
||||
if (windows) {
|
||||
primary ??= { type: "parent-signal", signal: nextSignal };
|
||||
signal(nextSignal);
|
||||
dispose();
|
||||
process.kill(process.pid, nextSignal);
|
||||
return;
|
||||
}
|
||||
freeze({ type: "parent-signal", signal: nextSignal }, nextSignal);
|
||||
};
|
||||
function onParentSigint() {
|
||||
onParentSignal("SIGINT");
|
||||
}
|
||||
function onParentSigterm() {
|
||||
onParentSignal("SIGTERM");
|
||||
}
|
||||
|
||||
params.child.stdout?.on("data", onStdoutData);
|
||||
params.child.stdout?.on("error", onStdoutError);
|
||||
params.child.stderr?.on("data", onStderrData);
|
||||
params.child.stderr?.on("error", onStderrError);
|
||||
params.child.on("error", onChildError);
|
||||
params.child.on("exit", onExit);
|
||||
params.child.on("close", onClose);
|
||||
if (params.forwardParentSignals) {
|
||||
process.on("exit", onParentExit);
|
||||
process.on("SIGINT", onParentSigint);
|
||||
process.on("SIGTERM", onParentSigterm);
|
||||
}
|
||||
if (params.executionTimeoutMs !== undefined) {
|
||||
executionTimer = schedule(() => freeze({ type: "timeout" }), params.executionTimeoutMs);
|
||||
}
|
||||
|
||||
return { requestCleanup: () => freeze({ type: "manual" }) };
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
inspectLinuxProcessGroupStats,
|
||||
isQaPosixProcessGroupAlive,
|
||||
signalQaPosixProcessGroup,
|
||||
} from "./posix-process-group.js";
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("POSIX process group inspection", () => {
|
||||
it("treats Linux zombie and dead members as stopped", () => {
|
||||
expect(
|
||||
inspectLinuxProcessGroupStats(123, [
|
||||
"123 (leader) Z 1 123 123 0 -1 0",
|
||||
"124 (helper (worker)) X 1 123 123 0 -1 0",
|
||||
"125 (unrelated) S 1 999 999 0 -1 0",
|
||||
]),
|
||||
).toEqual({
|
||||
alive: false,
|
||||
diagnostics:
|
||||
'pgid=123 members=[pid=123 state=Z command="leader", pid=124 state=X command="helper (worker)"]',
|
||||
});
|
||||
});
|
||||
|
||||
it("treats runnable members as alive and empty snapshots as unknown", () => {
|
||||
expect(
|
||||
inspectLinuxProcessGroupStats(123, [
|
||||
"123 (leader) Z 1 123 123 0 -1 0",
|
||||
"124 (worker) D 1 123 123 0 -1 0",
|
||||
]).alive,
|
||||
).toBe(true);
|
||||
expect(inspectLinuxProcessGroupStats(123, ["125 (other) S 1 999 999 0 -1 0"])).toEqual({
|
||||
alive: null,
|
||||
diagnostics: "pgid=123 members=[]",
|
||||
});
|
||||
});
|
||||
|
||||
it("fails closed when the Linux member snapshot is unavailable", () => {
|
||||
const platform = vi.spyOn(process, "platform", "get").mockReturnValue("linux");
|
||||
const processKill = vi.spyOn(process, "kill").mockImplementation(() => true);
|
||||
try {
|
||||
expect(isQaPosixProcessGroupAlive(123, () => null)).toBe(true);
|
||||
expect(processKill).toHaveBeenCalledWith(-123, 0);
|
||||
} finally {
|
||||
platform.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("treats a kill-visible Linux group with only zombie members as stopped", () => {
|
||||
const platform = vi.spyOn(process, "platform", "get").mockReturnValue("linux");
|
||||
const processKill = vi.spyOn(process, "kill").mockImplementation(() => true);
|
||||
try {
|
||||
expect(
|
||||
isQaPosixProcessGroupAlive(123, () => ({
|
||||
alive: false,
|
||||
diagnostics: 'pgid=123 members=[pid=123 state=Z command="leader"]',
|
||||
})),
|
||||
).toBe(false);
|
||||
expect(processKill).toHaveBeenCalledWith(-123, 0);
|
||||
} finally {
|
||||
platform.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("stops on ESRCH and never falls back to a positive pid", () => {
|
||||
const processKill = vi.spyOn(process, "kill").mockImplementation((pid, signal) => {
|
||||
expect(pid).toBe(-123);
|
||||
if (signal === 0) {
|
||||
throw Object.assign(new Error("gone"), { code: "ESRCH" });
|
||||
}
|
||||
return true;
|
||||
});
|
||||
|
||||
expect(isQaPosixProcessGroupAlive(123)).toBe(false);
|
||||
expect(signalQaPosixProcessGroup(123, "SIGTERM")).toBeUndefined();
|
||||
expect(processKill).not.toHaveBeenCalledWith(123, expect.anything());
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,126 @@
|
||||
import { readFileSync, readdirSync } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { sliceUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime";
|
||||
|
||||
function parseLinuxProcessStat(raw: string) {
|
||||
const commandStart = raw.indexOf("(");
|
||||
const commandEnd = raw.lastIndexOf(")");
|
||||
if (commandStart <= 0 || commandEnd <= commandStart) {
|
||||
return null;
|
||||
}
|
||||
const pid = Number.parseInt(raw.slice(0, commandStart).trim(), 10);
|
||||
const fields = raw
|
||||
.slice(commandEnd + 1)
|
||||
.trim()
|
||||
.split(/\s+/u);
|
||||
const state = fields[0];
|
||||
const processGroupId = Number.parseInt(fields[2] ?? "", 10);
|
||||
if (
|
||||
!Number.isSafeInteger(pid) ||
|
||||
pid <= 0 ||
|
||||
!state ||
|
||||
!Number.isSafeInteger(processGroupId) ||
|
||||
processGroupId <= 0
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
command: raw.slice(commandStart + 1, commandEnd),
|
||||
pid,
|
||||
processGroupId,
|
||||
state,
|
||||
};
|
||||
}
|
||||
|
||||
function boundProcessGroupDiagnostics(details: string) {
|
||||
if (details.length <= 2_048) {
|
||||
return details;
|
||||
}
|
||||
return `${sliceUtf16Safe(details, 0, 2_045)}...`;
|
||||
}
|
||||
|
||||
export function inspectLinuxProcessGroupStats(processGroupId: number, stats: readonly string[]) {
|
||||
const members = stats
|
||||
.map((raw) => parseLinuxProcessStat(raw))
|
||||
.filter(
|
||||
(entry): entry is NonNullable<ReturnType<typeof parseLinuxProcessStat>> =>
|
||||
entry?.processGroupId === processGroupId,
|
||||
)
|
||||
.toSorted((left, right) => left.pid - right.pid);
|
||||
const diagnostics = members
|
||||
.map(
|
||||
(member) =>
|
||||
`pid=${member.pid} state=${member.state} command=${JSON.stringify(member.command)}`,
|
||||
)
|
||||
.join(", ");
|
||||
return {
|
||||
alive:
|
||||
members.length === 0
|
||||
? null
|
||||
: members.some((entry) => entry.state !== "Z" && entry.state !== "X"),
|
||||
diagnostics: boundProcessGroupDiagnostics(`pgid=${processGroupId} members=[${diagnostics}]`),
|
||||
};
|
||||
}
|
||||
|
||||
type QaLinuxProcessGroupInspection = ReturnType<typeof inspectLinuxProcessGroupStats>;
|
||||
export type QaLinuxProcessGroupInspector = (
|
||||
processGroupId: number,
|
||||
) => QaLinuxProcessGroupInspection | null;
|
||||
|
||||
export function inspectLinuxProcessGroup(
|
||||
processGroupId: number,
|
||||
): QaLinuxProcessGroupInspection | null {
|
||||
if (process.platform !== "linux") {
|
||||
return null;
|
||||
}
|
||||
let entries;
|
||||
try {
|
||||
entries = readdirSync("/proc", { withFileTypes: true });
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
const stats: string[] = [];
|
||||
for (const entry of entries) {
|
||||
if (!entry.isDirectory() || !/^\d+$/u.test(entry.name)) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
stats.push(readFileSync(path.join("/proc", entry.name, "stat"), "utf8"));
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
return inspectLinuxProcessGroupStats(processGroupId, stats);
|
||||
}
|
||||
|
||||
export function isQaPosixProcessGroupAlive(
|
||||
processGroupId: number,
|
||||
inspectLinuxProcessGroupFn: QaLinuxProcessGroupInspector = inspectLinuxProcessGroup,
|
||||
) {
|
||||
try {
|
||||
process.kill(-processGroupId, 0);
|
||||
} catch (error) {
|
||||
return (error as NodeJS.ErrnoException).code !== "ESRCH";
|
||||
}
|
||||
if (process.platform !== "linux") {
|
||||
return true;
|
||||
}
|
||||
return inspectLinuxProcessGroupFn(processGroupId)?.alive ?? true;
|
||||
}
|
||||
|
||||
export function signalQaPosixProcessGroup(
|
||||
processGroupId: number,
|
||||
signal: NodeJS.Signals,
|
||||
): Error | undefined {
|
||||
try {
|
||||
process.kill(-processGroupId, signal);
|
||||
return undefined;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ESRCH") {
|
||||
return undefined;
|
||||
}
|
||||
return error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
}
|
||||
@@ -60,6 +60,17 @@ function createMockEmitter() {
|
||||
|
||||
function createSpawnedProcess(params: { pid?: number } = {}) {
|
||||
const child = createMockEmitter() as MockChildProcess;
|
||||
const emit = child.emit.bind(child);
|
||||
let exited = false;
|
||||
child.emit = (eventName, ...args) => {
|
||||
if (eventName === "exit") {
|
||||
exited = true;
|
||||
} else if (eventName === "close" && !exited) {
|
||||
exited = true;
|
||||
emit("exit", ...args);
|
||||
}
|
||||
return emit(eventName, ...args);
|
||||
};
|
||||
child.pid = params.pid;
|
||||
child.stdout = createMockEmitter();
|
||||
child.stderr = createMockEmitter();
|
||||
@@ -173,7 +184,16 @@ describe("qa suite runtime agent process helpers", () => {
|
||||
});
|
||||
|
||||
it.runIf(process.platform !== "win32")("kills timed-out qa cli process groups", async () => {
|
||||
const killSpy = vi.spyOn(process, "kill").mockImplementation(() => true);
|
||||
let processGroupAlive = true;
|
||||
const killSpy = vi.spyOn(process, "kill").mockImplementation((pid, signal) => {
|
||||
if (pid === -12345 && signal === "SIGKILL") {
|
||||
processGroupAlive = false;
|
||||
}
|
||||
if (pid === -12345 && signal === 0 && !processGroupAlive) {
|
||||
throw Object.assign(new Error("gone"), { code: "ESRCH" });
|
||||
}
|
||||
return true;
|
||||
});
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const child = createSpawnedProcess({ pid: 12345 });
|
||||
@@ -198,6 +218,8 @@ describe("qa suite runtime agent process helpers", () => {
|
||||
),
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
child.emit("exit", null, "SIGKILL");
|
||||
child.emit("close", null, "SIGKILL");
|
||||
|
||||
const error = await errorPromise;
|
||||
expect(error).toMatchObject({ code: "qa_cli_timeout" });
|
||||
@@ -217,6 +239,41 @@ describe("qa suite runtime agent process helpers", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.runIf(process.platform !== "win32")(
|
||||
"preserves a nonzero qa cli failure when process-group cleanup also fails",
|
||||
async () => {
|
||||
const killSpy = vi.spyOn(process, "kill").mockImplementation((pid, signal) => {
|
||||
if (pid === -12345 && signal === "SIGKILL") {
|
||||
throw Object.assign(new Error("cleanup denied"), { code: "EPERM" });
|
||||
}
|
||||
return true;
|
||||
});
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const child = createSpawnedProcess({ pid: 12345 });
|
||||
const { pending } = startMockQaCli({ args: ["qa", "suite"], child });
|
||||
const errorPromise = pending.catch((value: unknown) => value);
|
||||
await Promise.resolve();
|
||||
child.stderr.emit("data", Buffer.from("suite failed\n"));
|
||||
child.emit("exit", 7, null);
|
||||
child.emit("close", 7, null);
|
||||
await vi.advanceTimersByTimeAsync(500);
|
||||
|
||||
const error = await errorPromise;
|
||||
expect(error).toBeInstanceOf(AggregateError);
|
||||
expect(error).toMatchObject({ message: "qa cli command and settlement failed" });
|
||||
const failures = error instanceof AggregateError ? error.errors : [];
|
||||
expect(failures).toEqual([
|
||||
expect.objectContaining({ message: "qa cli failed (7): suite failed" }),
|
||||
expect.any(Error),
|
||||
]);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
killSpy.mockRestore();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("force-kills timed-out Windows qa cli process trees with taskkill", async () => {
|
||||
const platformDescriptor = Object.getOwnPropertyDescriptor(process, "platform");
|
||||
const originalSystemRoot = process.env.SystemRoot;
|
||||
|
||||
@@ -19,6 +19,7 @@ import {
|
||||
import { QaSuiteInfraError } from "./errors.js";
|
||||
import { extractGatewayMessageText } from "./gateway-log-sentinel.js";
|
||||
import { resolveQaNodeExecPath } from "./node-exec.js";
|
||||
import { createQaPosixCommandSettlement } from "./posix-command-settlement.js";
|
||||
import { liveTurnTimeoutMs } from "./suite-runtime-agent-common.js";
|
||||
import { readSessionTranscriptSummary } from "./suite-runtime-agent-session.js";
|
||||
import { waitForGatewayHealthy, waitForTransportReady } from "./suite-runtime-gateway.js";
|
||||
@@ -234,37 +235,22 @@ function parseQaCliJsonOutput(text: string, args: readonly string[]) {
|
||||
}
|
||||
}
|
||||
|
||||
function signalQaCliProcessTree(
|
||||
child: Pick<ChildProcessWithoutNullStreams, "kill" | "pid">,
|
||||
signal: NodeJS.Signals,
|
||||
) {
|
||||
if (process.platform === "win32") {
|
||||
if (typeof child.pid === "number") {
|
||||
const result = spawnSync(
|
||||
resolveQaWindowsSystem32ExePath("taskkill.exe"),
|
||||
["/PID", String(child.pid), "/T", "/F"],
|
||||
{
|
||||
stdio: "ignore",
|
||||
windowsHide: true,
|
||||
timeout: 5_000,
|
||||
},
|
||||
);
|
||||
if (!result.error && result.status === 0) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
child.kill(signal);
|
||||
return;
|
||||
}
|
||||
if (typeof child.pid === "number") {
|
||||
try {
|
||||
process.kill(-child.pid, signal);
|
||||
function killQaCliWindowsProcessTree(child: Pick<ChildProcessWithoutNullStreams, "kill" | "pid">) {
|
||||
if (child.pid) {
|
||||
const result = spawnSync(
|
||||
resolveQaWindowsSystem32ExePath("taskkill.exe"),
|
||||
["/PID", String(child.pid), "/T", "/F"],
|
||||
{
|
||||
stdio: "ignore",
|
||||
windowsHide: true,
|
||||
timeout: 5_000,
|
||||
},
|
||||
);
|
||||
if (!result.error && result.status === 0) {
|
||||
return;
|
||||
} catch {
|
||||
// The detached process group may already be gone; fall back to the child handle.
|
||||
}
|
||||
}
|
||||
child.kill(signal);
|
||||
child.kill("SIGKILL");
|
||||
}
|
||||
|
||||
async function runQaCli(
|
||||
@@ -291,8 +277,7 @@ async function runQaCli(
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
const timeoutMs = resolveTimerTimeoutMs(opts?.timeoutMs, 60_000);
|
||||
const timeout = setTimeout(() => {
|
||||
signalQaCliProcessTree(child, "SIGKILL");
|
||||
const rejectTimeout = () => {
|
||||
const stdoutText = formatQaChildOutputTail(stdoutTail, "qa cli stdout");
|
||||
const stderrText = formatQaChildOutputTail(stderr, "qa cli stderr");
|
||||
const diagnostics = [
|
||||
@@ -301,12 +286,68 @@ async function runQaCli(
|
||||
]
|
||||
.filter(Boolean)
|
||||
.join("\n");
|
||||
reject(
|
||||
new QaSuiteInfraError(
|
||||
"qa_cli_timeout",
|
||||
`qa cli timed out: openclaw ${args.join(" ")}${diagnostics ? `\n${diagnostics}` : ""}`,
|
||||
),
|
||||
return new QaSuiteInfraError(
|
||||
"qa_cli_timeout",
|
||||
`qa cli timed out: openclaw ${args.join(" ")}${diagnostics ? `\n${diagnostics}` : ""}`,
|
||||
);
|
||||
};
|
||||
const getExitError = (code: number | null) => {
|
||||
if (code === 0) {
|
||||
if (stdout.exceeded) {
|
||||
return new Error(
|
||||
`qa cli stdout exceeded ${QA_CHILD_STDOUT_MAX_BYTES} bytes; refusing to parse truncated output`,
|
||||
);
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
const stderrText = formatQaChildOutputTail(stderr, "qa cli stderr");
|
||||
return new Error(`qa cli failed (${code ?? "unknown"}): ${stderrText}`);
|
||||
};
|
||||
if (process.platform !== "win32") {
|
||||
createQaPosixCommandSettlement({
|
||||
child,
|
||||
settlementFailureMessage: "qa cli settlement failed",
|
||||
executionTimeoutMs: timeoutMs,
|
||||
forceKillAfterMs: 0,
|
||||
initialSignal: "SIGKILL",
|
||||
onSettled: (outcome) => {
|
||||
const primary = outcome.primary;
|
||||
const primaryError =
|
||||
primary.type === "spawn-error" || primary.type === "stream-error"
|
||||
? primary.error
|
||||
: primary.type === "timeout"
|
||||
? rejectTimeout()
|
||||
: getExitError(primary.type === "exit" ? primary.exitCode : 1);
|
||||
if (outcome.settlementFailure) {
|
||||
reject(
|
||||
primaryError
|
||||
? new AggregateError(
|
||||
[primaryError, outcome.settlementFailure],
|
||||
"qa cli command and settlement failed",
|
||||
)
|
||||
: outcome.settlementFailure,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (primaryError) {
|
||||
reject(primaryError);
|
||||
return;
|
||||
}
|
||||
resolve();
|
||||
},
|
||||
onStderrData: (chunk) => appendQaChildOutputTail(stderr, chunk),
|
||||
onStdoutData: (chunk) => {
|
||||
appendQaChildOutput(stdout, chunk);
|
||||
appendQaChildOutputTail(stdoutTail, chunk);
|
||||
},
|
||||
processGroupId: child.pid,
|
||||
verifyAfterMs: 500,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const timeout = setTimeout(() => {
|
||||
killQaCliWindowsProcessTree(child);
|
||||
reject(rejectTimeout());
|
||||
}, timeoutMs);
|
||||
child.stdout.on("data", (chunk) => {
|
||||
appendQaChildOutput(stdout, chunk);
|
||||
@@ -319,20 +360,8 @@ async function runQaCli(
|
||||
});
|
||||
child.once("close", (code) => {
|
||||
clearTimeout(timeout);
|
||||
if (code === 0) {
|
||||
if (stdout.exceeded) {
|
||||
reject(
|
||||
new Error(
|
||||
`qa cli stdout exceeded ${QA_CHILD_STDOUT_MAX_BYTES} bytes; refusing to parse truncated output`,
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
const stderrText = formatQaChildOutputTail(stderr, "qa cli stderr");
|
||||
reject(new Error(`qa cli failed (${code ?? "unknown"}): ${stderrText}`));
|
||||
const error = getExitError(code);
|
||||
return error ? reject(error) : resolve();
|
||||
});
|
||||
});
|
||||
const text = readQaChildOutput(stdout).trim();
|
||||
|
||||
@@ -1,17 +1,33 @@
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { mkdtemp, readFile, rm } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { setTimeout as sleep } from "node:timers/promises";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const spawnMock = vi.hoisted(() => vi.fn());
|
||||
const spawnSyncMock = vi.hoisted(() => vi.fn());
|
||||
const actualSpawn = vi.hoisted(
|
||||
() =>
|
||||
({
|
||||
value: undefined,
|
||||
}) as {
|
||||
value: typeof import("node:child_process").spawn | undefined;
|
||||
},
|
||||
);
|
||||
|
||||
vi.mock("node:child_process", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("node:child_process")>();
|
||||
actualSpawn.value = actual.spawn;
|
||||
return {
|
||||
...actual,
|
||||
spawn: spawnMock,
|
||||
spawnSync: spawnSyncMock,
|
||||
};
|
||||
});
|
||||
|
||||
import { isQaPosixProcessGroupAlive } from "./posix-process-group.js";
|
||||
import {
|
||||
resetQaScenarioCommandCleanupTimings,
|
||||
runQaScenarioCommandLifecycle,
|
||||
@@ -28,8 +44,8 @@ function spyOnProcessKill() {
|
||||
function createChild(pid = 42) {
|
||||
const child = new EventEmitter() as ChildProcess;
|
||||
Object.defineProperty(child, "pid", { value: pid });
|
||||
child.stdout = new EventEmitter() as NonNullable<ChildProcess["stdout"]>;
|
||||
child.stderr = new EventEmitter() as NonNullable<ChildProcess["stderr"]>;
|
||||
child.stdout = Object.assign(new EventEmitter(), { destroy: vi.fn() }) as never;
|
||||
child.stderr = Object.assign(new EventEmitter(), { destroy: vi.fn() }) as never;
|
||||
child.kill = vi.fn(() => true) as ChildProcess["kill"];
|
||||
spawnMock.mockReturnValue(child);
|
||||
return child;
|
||||
@@ -45,6 +61,207 @@ function runCommand(timeoutMs?: number) {
|
||||
});
|
||||
}
|
||||
|
||||
function isProcessRunning(pid: number) {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForPidFile(filePath: string, timeoutMs = 2_000) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
const value = await readFile(filePath, "utf8").catch(() => "");
|
||||
if (/^[1-9]\d*$/u.test(value.trim())) {
|
||||
return Number(value.trim());
|
||||
}
|
||||
await sleep(10);
|
||||
}
|
||||
throw new Error(`timed out waiting for pid file ${filePath}`);
|
||||
}
|
||||
|
||||
async function waitForProcessExit(pid: number, timeoutMs = 2_000) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (!isProcessRunning(pid)) {
|
||||
return;
|
||||
}
|
||||
await sleep(10);
|
||||
}
|
||||
throw new Error(`timed out waiting for process ${pid} to exit`);
|
||||
}
|
||||
|
||||
describe.skipIf(process.platform === "win32")("qa scenario command real POSIX lifecycle", () => {
|
||||
afterEach(() => {
|
||||
resetQaScenarioCommandCleanupTimings();
|
||||
spawnMock.mockReset();
|
||||
});
|
||||
|
||||
it("settles within a bound after the leader writes its final result with inherited stdio open", async () => {
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "qa-command-settlement-"));
|
||||
const descendantPidPath = path.join(root, "descendant.pid");
|
||||
let descendantPid: number | undefined;
|
||||
spawnMock.mockImplementation((...args: Parameters<NonNullable<typeof actualSpawn.value>>) => {
|
||||
if (!actualSpawn.value) {
|
||||
throw new Error("real spawn unavailable");
|
||||
}
|
||||
return actualSpawn.value(...args);
|
||||
});
|
||||
setQaScenarioCommandCleanupTimings({ killGraceMs: 100, forceSettleMs: 100 });
|
||||
try {
|
||||
const descendantScript = [
|
||||
"const { writeFileSync } = require('node:fs');",
|
||||
"process.on('SIGTERM', () => {});",
|
||||
`writeFileSync(${JSON.stringify(descendantPidPath)}, String(process.pid));`,
|
||||
"setTimeout(() => process.stdout.write('delayed descendant output\\n'), 40);",
|
||||
"setInterval(() => {}, 1000);",
|
||||
].join(" ");
|
||||
const leaderScript = [
|
||||
"const { spawn } = require('node:child_process');",
|
||||
"const { existsSync } = require('node:fs');",
|
||||
`spawn(process.execPath, ['-e', ${JSON.stringify(descendantScript)}], { stdio: ['ignore', 'inherit', 'inherit'] }).unref();`,
|
||||
`const ready = setInterval(() => { if (!existsSync(${JSON.stringify(descendantPidPath)})) return; clearInterval(ready); process.stdout.write('Docker scheduling finished\\n', () => process.exit(7)); }, 5);`,
|
||||
].join("\n");
|
||||
|
||||
const pending = runQaScenarioCommandLifecycle({
|
||||
command: process.execPath,
|
||||
args: ["-e", leaderScript],
|
||||
cwd: root,
|
||||
env: process.env,
|
||||
timeoutMs: 5_000,
|
||||
});
|
||||
descendantPid = await waitForPidFile(descendantPidPath);
|
||||
const processGroupId = (spawnMock.mock.results[0]?.value as ChildProcess | undefined)?.pid;
|
||||
if (!processGroupId) {
|
||||
throw new Error("scenario command did not expose its process group id");
|
||||
}
|
||||
const startedAt = Date.now();
|
||||
const deadline = new AbortController();
|
||||
const result = await Promise.race([
|
||||
pending,
|
||||
sleep(1_500, undefined, { signal: deadline.signal }).then(() => {
|
||||
throw new Error("command did not settle after process-group cleanup");
|
||||
}),
|
||||
]).finally(() => deadline.abort());
|
||||
|
||||
expect(Date.now() - startedAt).toBeLessThan(1_500);
|
||||
expect(result).toEqual({
|
||||
exitCode: 7,
|
||||
signal: null,
|
||||
stdout: "Docker scheduling finished\ndelayed descendant output\n",
|
||||
stderr: "",
|
||||
});
|
||||
expect(isQaPosixProcessGroupAlive(processGroupId)).toBe(false);
|
||||
} finally {
|
||||
if (descendantPid && isProcessRunning(descendantPid)) {
|
||||
process.kill(descendantPid, "SIGKILL");
|
||||
}
|
||||
await rm(root, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("reports that a self-detached descendant escaped and cleans it explicitly", async () => {
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "qa-command-setsid-escape-"));
|
||||
const descendantPidPath = path.join(root, "descendant.pid");
|
||||
let descendantPid: number | undefined;
|
||||
spawnMock.mockImplementation((...args: Parameters<NonNullable<typeof actualSpawn.value>>) => {
|
||||
if (!actualSpawn.value) {
|
||||
throw new Error("real spawn unavailable");
|
||||
}
|
||||
return actualSpawn.value(...args);
|
||||
});
|
||||
try {
|
||||
const descendantScript = [
|
||||
"const { writeFileSync } = require('node:fs');",
|
||||
`writeFileSync(${JSON.stringify(descendantPidPath)}, String(process.pid));`,
|
||||
"process.stdout.write('escaped descendant output\\n');",
|
||||
"setInterval(() => {}, 1000);",
|
||||
].join(" ");
|
||||
const leaderScript = [
|
||||
"const { spawn } = require('node:child_process');",
|
||||
`spawn(process.execPath, ['-e', ${JSON.stringify(descendantScript)}], { detached: true, stdio: ['ignore', 'inherit', 'inherit'] }).unref();`,
|
||||
"process.exit(0);",
|
||||
].join("\n");
|
||||
|
||||
const result = await runQaScenarioCommandLifecycle({
|
||||
command: process.execPath,
|
||||
args: ["-e", leaderScript],
|
||||
cwd: root,
|
||||
env: process.env,
|
||||
timeoutMs: 5_000,
|
||||
});
|
||||
|
||||
descendantPid = await waitForPidFile(descendantPidPath);
|
||||
expect(result.exitCode).toBe(1);
|
||||
expect(result.failureMessage).toBe("stdio-drain-timeout");
|
||||
expect(result.stdout).toContain("escaped descendant output");
|
||||
expect(isProcessRunning(descendantPid)).toBe(true);
|
||||
} finally {
|
||||
// A true setsid descendant is outside the original PGID by design.
|
||||
if (descendantPid && isProcessRunning(descendantPid)) {
|
||||
process.kill(descendantPid, "SIGKILL");
|
||||
await waitForProcessExit(descendantPid).catch(() => undefined);
|
||||
}
|
||||
await rm(root, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("cleans the command group before re-raising a parent SIGTERM", async () => {
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "qa-command-parent-signal-"));
|
||||
const descendantPidPath = path.join(root, "descendant.pid");
|
||||
const moduleUrl = new URL("./test-file-scenario-command-lifecycle.ts", import.meta.url).href;
|
||||
let descendantPid: number | undefined;
|
||||
if (!actualSpawn.value) {
|
||||
throw new Error("real spawn unavailable");
|
||||
}
|
||||
const controllerScript = [
|
||||
`import { runQaScenarioCommandLifecycle, setQaScenarioCommandCleanupTimings } from ${JSON.stringify(moduleUrl)};`,
|
||||
"setQaScenarioCommandCleanupTimings({ killGraceMs: 50, forceSettleMs: 50 });",
|
||||
"const nested = [",
|
||||
" \"const { writeFileSync } = require('node:fs');\",",
|
||||
` ${JSON.stringify(`writeFileSync(${JSON.stringify(descendantPidPath)}, String(process.pid));`)},`,
|
||||
" \"process.on('SIGTERM', () => {});\",",
|
||||
' "setInterval(() => {}, 1000);",',
|
||||
"].join(' ');",
|
||||
"await runQaScenarioCommandLifecycle({",
|
||||
" command: process.execPath,",
|
||||
" args: ['-e', nested],",
|
||||
` cwd: ${JSON.stringify(root)},`,
|
||||
" env: process.env,",
|
||||
" timeoutMs: 5000,",
|
||||
"});",
|
||||
].join("\n");
|
||||
const controller = actualSpawn.value(
|
||||
process.execPath,
|
||||
["--import", "tsx", "--input-type=module", "-e", controllerScript],
|
||||
{ cwd: process.cwd(), env: process.env, stdio: "ignore" },
|
||||
);
|
||||
try {
|
||||
descendantPid = await waitForPidFile(descendantPidPath, 10_000);
|
||||
controller.kill("SIGTERM");
|
||||
const [exitCode, signal] = await new Promise<[number | null, NodeJS.Signals | null]>(
|
||||
(resolve) => {
|
||||
controller.once("close", (code, nextSignal) => resolve([code, nextSignal]));
|
||||
},
|
||||
);
|
||||
|
||||
expect(exitCode).toBeNull();
|
||||
expect(signal).toBe("SIGTERM");
|
||||
await waitForProcessExit(descendantPid);
|
||||
} finally {
|
||||
if (controller.exitCode === null && controller.signalCode === null) {
|
||||
controller.kill("SIGKILL");
|
||||
}
|
||||
if (descendantPid && isProcessRunning(descendantPid)) {
|
||||
process.kill(descendantPid, "SIGKILL");
|
||||
}
|
||||
await rm(root, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", () => {
|
||||
const parentHandlers = new Map<ParentSignal | "exit", ParentHandler>();
|
||||
let processKill: ReturnType<typeof spyOnProcessKill>;
|
||||
@@ -52,10 +269,15 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", (
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
spawnMock.mockReset();
|
||||
spawnSyncMock.mockReset();
|
||||
vi.spyOn(process, "once").mockImplementation((event, listener) => {
|
||||
parentHandlers.set(event as ParentSignal | "exit", listener as ParentHandler);
|
||||
return process;
|
||||
});
|
||||
vi.spyOn(process, "on").mockImplementation((event, listener) => {
|
||||
parentHandlers.set(event as ParentSignal | "exit", listener as ParentHandler);
|
||||
return process;
|
||||
});
|
||||
vi.spyOn(process, "removeListener").mockImplementation((event, listener) => {
|
||||
if (parentHandlers.get(event as ParentSignal | "exit") === listener) {
|
||||
parentHandlers.delete(event as ParentSignal | "exit");
|
||||
@@ -83,6 +305,7 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", (
|
||||
|
||||
child.stdout?.emit("data", Buffer.from("out\n"));
|
||||
child.stderr?.emit("data", Buffer.from("err\n"));
|
||||
child.emit("exit", 3, null);
|
||||
child.emit("close", 3, null);
|
||||
|
||||
await expect(resultPromise).resolves.toEqual({
|
||||
@@ -114,6 +337,38 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", (
|
||||
expect(parentHandlers.size).toBe(0);
|
||||
});
|
||||
|
||||
it("preserves the Windows taskkill timeout lifecycle", async () => {
|
||||
const platformDescriptor = Object.getOwnPropertyDescriptor(process, "platform");
|
||||
const originalSystemRoot = process.env.SystemRoot;
|
||||
Object.defineProperty(process, "platform", { value: "win32", configurable: true });
|
||||
process.env.SystemRoot = "C:\\Windows";
|
||||
spawnSyncMock.mockReturnValue({ status: 0 });
|
||||
createChild(12345);
|
||||
setQaScenarioCommandCleanupTimings({ killGraceMs: 20, forceSettleMs: 10 });
|
||||
try {
|
||||
const resultPromise = runCommand(100);
|
||||
await vi.advanceTimersByTimeAsync(130);
|
||||
|
||||
await expect(resultPromise).resolves.toMatchObject({
|
||||
exitCode: 1,
|
||||
failureMessage: "scenario-command timed out after 100ms",
|
||||
signal: null,
|
||||
});
|
||||
expect(spawnSyncMock).toHaveBeenCalledTimes(2);
|
||||
expect(spawnSyncMock.mock.calls[0]?.[1]).toEqual(["/pid", "12345", "/T"]);
|
||||
expect(spawnSyncMock.mock.calls[1]?.[1]).toEqual(["/pid", "12345", "/T", "/F"]);
|
||||
} finally {
|
||||
if (platformDescriptor) {
|
||||
Object.defineProperty(process, "platform", platformDescriptor);
|
||||
}
|
||||
if (originalSystemRoot === undefined) {
|
||||
delete process.env.SystemRoot;
|
||||
} else {
|
||||
process.env.SystemRoot = originalSystemRoot;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("escalates timed-out commands and preserves the timeout result", async () => {
|
||||
createChild();
|
||||
setQaScenarioCommandCleanupTimings({ killGraceMs: 20, forceSettleMs: 10 });
|
||||
@@ -129,7 +384,11 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", (
|
||||
});
|
||||
|
||||
const resultPromise = runCommand(100);
|
||||
await vi.advanceTimersByTimeAsync(130);
|
||||
await vi.advanceTimersByTimeAsync(120);
|
||||
const child = spawnMock.mock.results[0]?.value as ChildProcess;
|
||||
child.emit("exit", null, "SIGKILL");
|
||||
child.emit("close", null, "SIGKILL");
|
||||
await vi.advanceTimersByTimeAsync(10);
|
||||
|
||||
await expect(resultPromise).resolves.toEqual({
|
||||
exitCode: 1,
|
||||
@@ -163,7 +422,11 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", (
|
||||
| undefined;
|
||||
expect(signalHandler).toBeDefined();
|
||||
signalHandler?.("SIGTERM");
|
||||
await vi.advanceTimersByTimeAsync(30);
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
const child = spawnMock.mock.results[0]?.value as ChildProcess;
|
||||
child.emit("exit", null, "SIGKILL");
|
||||
child.emit("close", null, "SIGKILL");
|
||||
await vi.advanceTimersByTimeAsync(10);
|
||||
|
||||
await expect(resultPromise).resolves.toEqual({
|
||||
exitCode: 1,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { spawn, spawnSync, type ChildProcess } from "node:child_process";
|
||||
import { spawn, spawnSync } from "node:child_process";
|
||||
import path from "node:path";
|
||||
import { createQaPosixCommandSettlement } from "./posix-command-settlement.js";
|
||||
import { resolveQaWindowsSystem32ExePath } from "./windows-system-tools.js";
|
||||
|
||||
export type QaScenarioCommandExecution = {
|
||||
@@ -24,14 +25,9 @@ type QaScenarioCommandTerminalResult = Pick<
|
||||
>;
|
||||
|
||||
type QaScenarioTaskkillRunner = typeof spawnSync;
|
||||
type QaScenarioCommandTimers = Partial<
|
||||
Record<"timeout" | "forceKill" | "forceSettle", NodeJS.Timeout>
|
||||
>;
|
||||
|
||||
const QA_SCENARIO_COMMAND_TIMEOUT_KILL_GRACE_MS = 2_000;
|
||||
const QA_SCENARIO_COMMAND_TIMEOUT_FORCE_SETTLE_MS = 500;
|
||||
const QA_SCENARIO_COMMAND_PARENT_SIGNALS = ["SIGINT", "SIGTERM"] as const;
|
||||
type QaScenarioParentSignal = (typeof QA_SCENARIO_COMMAND_PARENT_SIGNALS)[number];
|
||||
let timeoutKillGraceMs = QA_SCENARIO_COMMAND_TIMEOUT_KILL_GRACE_MS;
|
||||
let timeoutForceSettleMs = QA_SCENARIO_COMMAND_TIMEOUT_FORCE_SETTLE_MS;
|
||||
|
||||
@@ -55,214 +51,92 @@ export function killQaScenarioWindowsProcessTree(
|
||||
return signal === "SIGKILL" ? run(true) : run(false) || run(true);
|
||||
}
|
||||
|
||||
// One owner keeps timers, parent handlers, child signals, and final result
|
||||
// settlement symmetric across every command exit path.
|
||||
class QaScenarioCommandLifecycle {
|
||||
private readonly stderr: Buffer[] = [];
|
||||
private readonly stdout: Buffer[] = [];
|
||||
private resolve: ((result: QaScenarioCommandResult) => void) | undefined;
|
||||
private settled = false;
|
||||
private timers: QaScenarioCommandTimers = {};
|
||||
private timedOut = false;
|
||||
|
||||
constructor(
|
||||
private readonly execution: QaScenarioCommandExecution,
|
||||
private readonly child: ChildProcess,
|
||||
private readonly useProcessGroup: boolean,
|
||||
) {}
|
||||
|
||||
start(resolve: (result: QaScenarioCommandResult) => void, reject: (reason?: unknown) => void) {
|
||||
this.resolve = resolve;
|
||||
this.armTimeout();
|
||||
this.child.stdout?.on("data", (chunk: Buffer) => this.stdout.push(chunk));
|
||||
this.child.stderr?.on("data", (chunk: Buffer) => this.stderr.push(chunk));
|
||||
process.once("exit", this.handleParentExit);
|
||||
for (const signal of QA_SCENARIO_COMMAND_PARENT_SIGNALS) {
|
||||
process.once(signal, this.handleParentSignal);
|
||||
}
|
||||
this.child.on("error", (error) => {
|
||||
if (this.settled) {
|
||||
return;
|
||||
}
|
||||
this.clearTimers();
|
||||
this.cleanupParentHandlers();
|
||||
reject(error);
|
||||
});
|
||||
this.child.on("close", this.handleClose);
|
||||
}
|
||||
|
||||
private readonly handleParentExit = () => {
|
||||
this.signalChild("SIGKILL");
|
||||
};
|
||||
|
||||
private readonly handleParentSignal = (signal: QaScenarioParentSignal) => {
|
||||
this.removeParentSignalHandlers();
|
||||
this.signalChild(signal);
|
||||
this.scheduleForcedCleanup({
|
||||
exitCode: 1,
|
||||
failureMessage: `${this.commandLabel()} interrupted by ${signal}`,
|
||||
signal,
|
||||
});
|
||||
process.kill(process.pid, signal);
|
||||
};
|
||||
|
||||
private readonly handleClose = (exitCode: number | null, signal: NodeJS.Signals | null) => {
|
||||
if (this.settled) {
|
||||
return;
|
||||
}
|
||||
if (!this.timedOut) {
|
||||
this.clearTimeoutTimer();
|
||||
}
|
||||
const result = {
|
||||
exitCode: this.timedOut ? 1 : (exitCode ?? (signal ? 1 : 0)),
|
||||
signal,
|
||||
...(this.timedOut
|
||||
? {
|
||||
failureMessage: `${this.commandLabel()} timed out after ${this.execution.timeoutMs}ms`,
|
||||
}
|
||||
: {}),
|
||||
};
|
||||
if (
|
||||
this.timedOut &&
|
||||
!this.useProcessGroup &&
|
||||
(this.timers.forceKill || this.timers.forceSettle)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (this.isProcessGroupRunning()) {
|
||||
if (!this.timedOut) {
|
||||
this.signalChild("SIGTERM");
|
||||
}
|
||||
this.scheduleForcedCleanup(result);
|
||||
return;
|
||||
}
|
||||
this.finish(result);
|
||||
};
|
||||
|
||||
private armTimeout() {
|
||||
const timeoutMs = this.execution.timeoutMs;
|
||||
if (timeoutMs === undefined) {
|
||||
return;
|
||||
}
|
||||
this.timers.timeout = setTimeout(() => {
|
||||
delete this.timers.timeout;
|
||||
this.timedOut = true;
|
||||
this.signalChild("SIGTERM");
|
||||
this.scheduleForcedCleanup({
|
||||
exitCode: 1,
|
||||
failureMessage: `${this.commandLabel()} timed out after ${timeoutMs}ms`,
|
||||
signal: null,
|
||||
});
|
||||
}, timeoutMs);
|
||||
}
|
||||
|
||||
private signalChild(signal: NodeJS.Signals) {
|
||||
if (this.useProcessGroup && this.child.pid) {
|
||||
try {
|
||||
process.kill(-this.child.pid, signal);
|
||||
return;
|
||||
} catch {
|
||||
// The process group may already be gone; fall back to the direct child.
|
||||
}
|
||||
}
|
||||
if (!this.useProcessGroup && process.platform === "win32") {
|
||||
if (killQaScenarioWindowsProcessTree(this.child.pid, signal)) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
this.child.kill(signal);
|
||||
}
|
||||
|
||||
private isProcessGroupRunning() {
|
||||
if (!this.useProcessGroup || !this.child.pid) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
process.kill(-this.child.pid, 0);
|
||||
return true;
|
||||
} catch (error) {
|
||||
return (error as NodeJS.ErrnoException).code === "EPERM";
|
||||
}
|
||||
}
|
||||
|
||||
private scheduleForcedCleanup(result: QaScenarioCommandTerminalResult) {
|
||||
if (this.timers.forceKill || this.timers.forceSettle) {
|
||||
return;
|
||||
}
|
||||
this.timers.forceKill = setTimeout(() => {
|
||||
delete this.timers.forceKill;
|
||||
this.signalChild("SIGKILL");
|
||||
this.timers.forceSettle = setTimeout(() => {
|
||||
delete this.timers.forceSettle;
|
||||
const stillRunning = this.isProcessGroupRunning();
|
||||
const failureMessage =
|
||||
result.failureMessage ??
|
||||
(stillRunning ? `${this.commandLabel()} left background processes running` : undefined);
|
||||
this.finish({
|
||||
exitCode: stillRunning ? 1 : result.exitCode,
|
||||
signal: result.signal,
|
||||
...(failureMessage ? { failureMessage } : {}),
|
||||
});
|
||||
}, timeoutForceSettleMs);
|
||||
}, timeoutKillGraceMs);
|
||||
}
|
||||
|
||||
private finish(result: QaScenarioCommandTerminalResult) {
|
||||
if (this.settled) {
|
||||
return;
|
||||
}
|
||||
this.settled = true;
|
||||
this.clearTimers();
|
||||
this.cleanupParentHandlers();
|
||||
this.resolve?.({
|
||||
...result,
|
||||
stdout: Buffer.concat(this.stdout).toString("utf8"),
|
||||
stderr: Buffer.concat(this.stderr).toString("utf8"),
|
||||
});
|
||||
}
|
||||
|
||||
private commandLabel() {
|
||||
return path.basename(this.execution.command);
|
||||
}
|
||||
|
||||
private clearTimeoutTimer() {
|
||||
if (this.timers.timeout) {
|
||||
clearTimeout(this.timers.timeout);
|
||||
delete this.timers.timeout;
|
||||
}
|
||||
}
|
||||
|
||||
private clearTimers() {
|
||||
for (const timer of Object.values(this.timers)) {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
this.timers = {};
|
||||
}
|
||||
|
||||
private removeParentSignalHandlers() {
|
||||
for (const signal of QA_SCENARIO_COMMAND_PARENT_SIGNALS) {
|
||||
process.removeListener(signal, this.handleParentSignal);
|
||||
}
|
||||
}
|
||||
|
||||
private cleanupParentHandlers() {
|
||||
this.removeParentSignalHandlers();
|
||||
process.removeListener("exit", this.handleParentExit);
|
||||
}
|
||||
}
|
||||
|
||||
export function runQaScenarioCommandLifecycle(
|
||||
execution: QaScenarioCommandExecution,
|
||||
): Promise<QaScenarioCommandResult> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const useProcessGroup = process.platform !== "win32";
|
||||
const isWindows = process.platform === "win32";
|
||||
const child = spawn(execution.command, execution.args, {
|
||||
cwd: execution.cwd,
|
||||
detached: useProcessGroup,
|
||||
detached: !isWindows,
|
||||
env: execution.env,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
new QaScenarioCommandLifecycle(execution, child, useProcessGroup).start(resolve, reject);
|
||||
const stdout: Buffer[] = [];
|
||||
const stderr: Buffer[] = [];
|
||||
const commandLabel = path.basename(execution.command);
|
||||
createQaPosixCommandSettlement({
|
||||
child,
|
||||
settlementFailureMessage: `${commandLabel} settlement failed`,
|
||||
forceKillAfterMs: timeoutKillGraceMs,
|
||||
...(isWindows
|
||||
? {
|
||||
windowsCleanup: {
|
||||
alive: () => child.pid !== undefined,
|
||||
signal: (signal: NodeJS.Signals) => {
|
||||
try {
|
||||
if (!killQaScenarioWindowsProcessTree(child.pid, signal)) {
|
||||
child.kill(signal);
|
||||
}
|
||||
return undefined;
|
||||
} catch (error) {
|
||||
return error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
executionTimeoutMs: execution.timeoutMs,
|
||||
forwardParentSignals: true,
|
||||
initialSignal: "SIGTERM",
|
||||
onSettled: (outcome) => {
|
||||
const primary = outcome.primary;
|
||||
if (primary.type === "spawn-error" || primary.type === "stream-error") {
|
||||
reject(
|
||||
outcome.settlementFailure
|
||||
? new AggregateError(
|
||||
[primary.error, outcome.settlementFailure],
|
||||
`${commandLabel} command and settlement failed`,
|
||||
)
|
||||
: primary.error,
|
||||
);
|
||||
return;
|
||||
}
|
||||
const result: QaScenarioCommandTerminalResult =
|
||||
primary.type === "exit"
|
||||
? {
|
||||
exitCode: primary.exitCode ?? (primary.signal ? 1 : 0),
|
||||
signal: primary.signal,
|
||||
}
|
||||
: primary.type === "parent-signal"
|
||||
? {
|
||||
exitCode: 1,
|
||||
failureMessage: `${commandLabel} interrupted by ${primary.signal}`,
|
||||
signal: primary.signal,
|
||||
}
|
||||
: {
|
||||
exitCode: 1,
|
||||
failureMessage: `${commandLabel} timed out after ${execution.timeoutMs}ms`,
|
||||
signal: null,
|
||||
};
|
||||
const settlementFailure = outcome.settlementFailure?.message;
|
||||
resolve({
|
||||
...result,
|
||||
...(settlementFailure && result.exitCode === 0 ? { exitCode: 1 } : {}),
|
||||
stdout: Buffer.concat(stdout).toString("utf8"),
|
||||
stderr: Buffer.concat(stderr).toString("utf8"),
|
||||
...(settlementFailure
|
||||
? result.failureMessage
|
||||
? { failureMessage: `${result.failureMessage}; settlement: ${settlementFailure}` }
|
||||
: { failureMessage: settlementFailure }
|
||||
: {}),
|
||||
});
|
||||
},
|
||||
onStderrData: (chunk) => stderr.push(Buffer.from(chunk)),
|
||||
onStdoutData: (chunk) => stdout.push(Buffer.from(chunk)),
|
||||
processGroupId: isWindows ? undefined : child.pid,
|
||||
verifyAfterMs: timeoutForceSettleMs,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user