fix(runner-e2e): bound completed cell teardown (#12890)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The runner E2E suite verifies complete agent tasks against real providers. > - A Codex Plan test finished in 76 seconds, but its Playwright process stayed alive for 25 more minutes. > - The launcher accepted the saved passing result after its watchdog killed the process. > - The existing Plan limits also allowed much more time than recent successful runs need. > - This pull request adds a bounded result-to-exit check and safe process evidence. > - It also reduces the Plan limits while it keeps large headroom over measured success times. > - The benefit is faster diagnosis and no false green result after a teardown stall. ## Linked Issues or Issue Description **Pre-submission checklist** I searched open pull requests for runner E2E timeout and Playwright cleanup changes. I found no duplicate. The problem reproduces on `master`. **What happened?** The local Codex Plan cell completed its test in 76 seconds. Playwright then stayed alive for about 25 minutes. The launcher watchdog killed it after 26.5 minutes, but the launcher still accepted the saved passing result. **Expected behavior** The launcher must stop a process that stays alive after all results exist. It must report a cleanup failure instead of a pass. Plan tests must also use limits that match measured successful runs. **Steps to reproduce** 1. Run `core-compatibility.runner-codex.local.plan-revise-accept`. 2. Observe a valid result and the Playwright pass output. 3. Observe that the process can stay alive until the old launcher watchdog stops it. **Paperclip version or commit** The evidence came from `bcc6fe7a442dae74ab0321ad472f7536ffa58f04` in [Actions run 33963318820](https://github.com/paperclipai/paperclip/actions/runs/33963318820). ## What Changed - Reduce the Plan attempt limit from 20 to 8 minutes for local execution. - Reduce the Plan attempt limit from 35 to 12 minutes for Daytona execution. - Stop Playwright after it stays alive for 120 seconds after every result exists. - Record only allowlisted process kinds in the stall diagnostic. - Validate process identities before cleanup and retain continuously live process groups through member replacement. - Treat watchdog, post-result, cleanup, and nonzero-exit conflicts as cleanup failures. - Keep interactive `--ui` and `--debug` sessions exempt from the result-to-exit check. ## Verification - Prettier completed for all changed files. - `git diff --check` passed. - Static review confirmed the timeout derivation and cleanup boundaries. - An independent review found no blocking issue in the final patch. - I did not run local tests, builds, or type checks because this workstation must use the lightweight workflow. - GitHub CI and the exact paid Codex Plan cell will verify this commit. ## Risks The main risk is a false cleanup failure when Playwright needs more than 120 seconds after it writes all results. The allowance is separate from the task limit. Interactive modes are exempt. The diagnostic does not print command arguments or environment values. ## Model Used OpenAI Codex with GPT-5.6, reasoning, tool use, and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta
GitHub
parent
4a7172f5ac
commit
64d8929ce9
|
|
@ -52,6 +52,10 @@ describe("runner E2E catalog", () => {
|
|||
0,
|
||||
),
|
||||
).toBe(114);
|
||||
expect(
|
||||
runnerTasks.find((task) => task.id === "plan-revise-accept")
|
||||
?.attemptTimeoutMs,
|
||||
).toEqual({ local: 8 * 60_000, daytona: 12 * 60_000 });
|
||||
});
|
||||
|
||||
it("derives the qualified local native OpenCode profiles from the ranked snapshot", () => {
|
||||
|
|
@ -458,7 +462,7 @@ describe("runner E2E selectors", () => {
|
|||
job.executionId ===
|
||||
"core-compatibility.runner-acpx-claude.local.plan-revise-accept",
|
||||
)?.timeoutMinutes,
|
||||
).toBe(48);
|
||||
).toBe(25);
|
||||
expect(
|
||||
jobs.find(
|
||||
(job) =>
|
||||
|
|
|
|||
|
|
@ -432,8 +432,8 @@ export const runnerTasks: readonly RunnerTaskFixture[] = [
|
|||
flow: "plan_revision_acceptance",
|
||||
expectedRunCount: 3,
|
||||
attemptTimeoutMs: {
|
||||
local: 20 * 60_000,
|
||||
daytona: 35 * 60_000,
|
||||
local: 8 * 60_000,
|
||||
daytona: 12 * 60_000,
|
||||
},
|
||||
expectedTerminalState: { issue: "done", run: "succeeded" },
|
||||
buildTitle: (nonce) => `Runner E2E plan lifecycle ${nonce}`,
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import { createRequire } from "node:module";
|
|||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import {
|
||||
access,
|
||||
chmod,
|
||||
cp,
|
||||
lstat,
|
||||
|
|
@ -51,12 +52,39 @@ import {
|
|||
snapshotDarwinSharedMemory,
|
||||
} from "./shared-memory.js";
|
||||
import { reserveRunnerE2EServerPort } from "./ports.js";
|
||||
import {
|
||||
createResultExitGuard,
|
||||
enforceResultProcessIntegrity,
|
||||
} from "./result-exit-guard.js";
|
||||
import {
|
||||
observeDescendantProcessTree,
|
||||
refreshContinuouslyLiveProcessGroups,
|
||||
revalidateObservedProcessGroups,
|
||||
safeProcessGroupTerminationOrder,
|
||||
type ObservedProcessGroup,
|
||||
type ProcessObservation,
|
||||
} from "./process-tree.js";
|
||||
|
||||
const repositoryRoot = path.resolve(import.meta.dirname, "../..");
|
||||
const localEnvPath = path.join(repositoryRoot, ".env.runner-e2e.local");
|
||||
const resultsRoot = path.join(repositoryRoot, "tests/runner-e2e/results");
|
||||
const activeProcessGroups = new Set<number>();
|
||||
const activeProcessCleanup = new Map<number, Promise<string | null>>();
|
||||
const activeProcessTerminators = new Map<number, () => void>();
|
||||
const completedResultExitGraceMs = 120_000;
|
||||
const diagnosticProcessKinds = new Set([
|
||||
"bash",
|
||||
"chrome",
|
||||
"codex",
|
||||
"google-chrome",
|
||||
"node",
|
||||
"paperclip-runnerd",
|
||||
"playwright",
|
||||
"pnpm",
|
||||
"postgres",
|
||||
"sh",
|
||||
"tsx",
|
||||
]);
|
||||
let cancelled = false;
|
||||
|
||||
function cleanId(value: string) {
|
||||
|
|
@ -121,12 +149,117 @@ async function terminateProcessGroup(pid: number) {
|
|||
: null;
|
||||
}
|
||||
|
||||
function wait(milliseconds: number) {
|
||||
return new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
|
||||
}
|
||||
|
||||
interface ProcessTreeDiagnostic {
|
||||
summary: string;
|
||||
groups: ObservedProcessGroup[];
|
||||
}
|
||||
|
||||
function observedGroupsSelectedForTermination(
|
||||
groups: readonly ObservedProcessGroup[],
|
||||
processGroupIds: readonly number[],
|
||||
) {
|
||||
const selected = new Set(processGroupIds);
|
||||
return groups.filter((group) => selected.has(group.processGroupId));
|
||||
}
|
||||
|
||||
async function readProcessTable(): Promise<ProcessObservation[] | null> {
|
||||
if (process.platform === "win32") {
|
||||
return null;
|
||||
}
|
||||
return await new Promise<ProcessObservation[] | null>((resolve) => {
|
||||
const inspector = spawn(
|
||||
"ps",
|
||||
["-e", "-o", "pid=,ppid=,pgid=,lstart=,comm="],
|
||||
{
|
||||
env: { PATH: "/usr/bin:/bin", LANG: "C", LC_ALL: "C" },
|
||||
stdio: ["ignore", "pipe", "ignore"],
|
||||
},
|
||||
);
|
||||
let output = "";
|
||||
let settled = false;
|
||||
const finish = (value: ProcessObservation[] | null) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(inspectionTimeout);
|
||||
resolve(value);
|
||||
};
|
||||
const inspectionTimeout = setTimeout(() => {
|
||||
inspector.kill("SIGKILL");
|
||||
finish(null);
|
||||
}, 5_000);
|
||||
inspectionTimeout.unref();
|
||||
inspector.stdout?.setEncoding("utf8").on("data", (chunk: string) => {
|
||||
output = `${output}${chunk}`.slice(-1024 * 1024);
|
||||
});
|
||||
inspector.once("error", () => finish(null));
|
||||
inspector.once("close", () => {
|
||||
const observations = output
|
||||
.split(/\r?\n/)
|
||||
.map((line) =>
|
||||
/^\s*(\d+)\s+(\d+)\s+(\d+)\s+(\S+\s+\S+\s+\d+\s+\d{2}:\d{2}:\d{2}\s+\d{4})\s+(.+?)\s*$/.exec(
|
||||
line,
|
||||
),
|
||||
)
|
||||
.filter((match): match is RegExpExecArray => match !== null)
|
||||
.map((match): ProcessObservation => {
|
||||
// A target process can choose its own argv and process name. Emit a
|
||||
// fixed category instead of target-controlled text so diagnostics
|
||||
// can never turn that metadata into a secret-exfiltration channel.
|
||||
const command = path.basename(match[5]!);
|
||||
const kind = diagnosticProcessKinds.has(command) ? command : "other";
|
||||
return {
|
||||
pid: Number(match[1]),
|
||||
parentPid: Number(match[2]),
|
||||
processGroupId: Number(match[3]),
|
||||
started: match[4]!,
|
||||
kind,
|
||||
};
|
||||
});
|
||||
finish(observations);
|
||||
});
|
||||
}).catch(() => null);
|
||||
}
|
||||
|
||||
async function processTreeDiagnostic(
|
||||
rootPid: number,
|
||||
): Promise<ProcessTreeDiagnostic> {
|
||||
const table = await readProcessTable();
|
||||
if (!table) {
|
||||
return {
|
||||
summary: `process tree ${rootPid} (member inspection unavailable)`,
|
||||
groups: [],
|
||||
};
|
||||
}
|
||||
const observed = observeDescendantProcessTree(table, rootPid);
|
||||
const members = observed.members
|
||||
.slice(0, 64)
|
||||
.map(
|
||||
({ process: candidate, depth }) =>
|
||||
`pid=${candidate.pid} ppid=${candidate.parentPid} pgid=${candidate.processGroupId} depth=${depth} kind=${candidate.kind}`,
|
||||
);
|
||||
const descendantGroupIds = observed.groups
|
||||
.filter((group) => group.processGroupId !== rootPid)
|
||||
.map((group) => group.processGroupId);
|
||||
return {
|
||||
summary:
|
||||
members.length > 0
|
||||
? `process tree ${rootPid}: ${members.join("; ")}; descendant pgids=${descendantGroupIds.join(",") || "none"}`
|
||||
: `process tree ${rootPid}: no members reported`,
|
||||
groups: observed.groups,
|
||||
};
|
||||
}
|
||||
|
||||
for (const signal of ["SIGINT", "SIGTERM", "SIGHUP"] as const) {
|
||||
process.on(signal, () => {
|
||||
cancelled = true;
|
||||
for (const pid of activeProcessGroups) {
|
||||
if (activeProcessCleanup.has(pid)) {
|
||||
stopProcessGroup(pid, "SIGKILL");
|
||||
const terminate = activeProcessTerminators.get(pid);
|
||||
if (terminate) {
|
||||
terminate();
|
||||
continue;
|
||||
}
|
||||
activeProcessCleanup.set(pid, terminateProcessGroup(pid));
|
||||
|
|
@ -205,6 +338,8 @@ async function runProcess(
|
|||
env: NodeJS.ProcessEnv,
|
||||
timeoutMs: number | null,
|
||||
logPath: string,
|
||||
completionPaths: readonly string[],
|
||||
interactive: boolean,
|
||||
) {
|
||||
const log = createWriteStream(logPath, { flags: "a", mode: 0o600 });
|
||||
const child = spawn("pnpm", args, {
|
||||
|
|
@ -229,40 +364,171 @@ async function runProcess(
|
|||
child.stderr?.on("data", (chunk: Buffer) =>
|
||||
recordOutput(chunk, process.stderr),
|
||||
);
|
||||
let childSettled = false;
|
||||
let postResultStallError: string | null = null;
|
||||
let boundedCleanup: Promise<string | null> | undefined;
|
||||
const forceStopDirectChild = () => {
|
||||
if (!childSettled && child.exitCode === null && child.signalCode === null) {
|
||||
child.kill("SIGKILL");
|
||||
}
|
||||
};
|
||||
const stopChildTree = (diagnostic?: ProcessTreeDiagnostic) => {
|
||||
if (boundedCleanup) return;
|
||||
const rootProcessGroupId = child.pid!;
|
||||
boundedCleanup = (async () => {
|
||||
const snapshot = diagnostic ?? (await processTreeDiagnostic(child.pid!));
|
||||
const validationTable = await readProcessTable();
|
||||
const currentProcessGroupId = validationTable
|
||||
? (validationTable.find((candidate) => candidate.pid === process.pid)
|
||||
?.processGroupId ?? null)
|
||||
: null;
|
||||
const observedGroups = validationTable
|
||||
? revalidateObservedProcessGroups(snapshot.groups, validationTable)
|
||||
: [];
|
||||
const terminationOrder = safeProcessGroupTerminationOrder({
|
||||
rootProcessGroupId,
|
||||
currentProcessGroupId,
|
||||
groups: observedGroups,
|
||||
});
|
||||
const verifiedGroups = observedGroupsSelectedForTermination(
|
||||
observedGroups,
|
||||
terminationOrder,
|
||||
);
|
||||
if (verifiedGroups.length === 0) {
|
||||
forceStopDirectChild();
|
||||
return "Could not revalidate an owned process group before cleanup";
|
||||
}
|
||||
for (const processGroupId of terminationOrder) {
|
||||
stopProcessGroup(processGroupId, "SIGTERM");
|
||||
}
|
||||
|
||||
let remainingGroups = verifiedGroups;
|
||||
const gracefulDeadline = Date.now() + 5_000;
|
||||
while (remainingGroups.length > 0 && Date.now() < gracefulDeadline) {
|
||||
const table = await readProcessTable();
|
||||
if (table) {
|
||||
remainingGroups = refreshContinuouslyLiveProcessGroups(
|
||||
remainingGroups,
|
||||
table,
|
||||
);
|
||||
}
|
||||
if (remainingGroups.length > 0) await wait(50);
|
||||
}
|
||||
const remainingGroupIds = new Set(
|
||||
remainingGroups.map((group) => group.processGroupId),
|
||||
);
|
||||
const forcedOrder = safeProcessGroupTerminationOrder({
|
||||
rootProcessGroupId,
|
||||
currentProcessGroupId,
|
||||
groups: remainingGroups,
|
||||
}).filter((processGroupId) => remainingGroupIds.has(processGroupId));
|
||||
for (const processGroupId of forcedOrder) {
|
||||
stopProcessGroup(processGroupId, "SIGKILL");
|
||||
}
|
||||
|
||||
const forcedDeadline = Date.now() + 5_000;
|
||||
let forcedVerificationUnavailable = false;
|
||||
while (Date.now() < forcedDeadline) {
|
||||
const table = await readProcessTable();
|
||||
if (!table) {
|
||||
forcedVerificationUnavailable = true;
|
||||
await wait(50);
|
||||
continue;
|
||||
}
|
||||
forcedVerificationUnavailable = false;
|
||||
remainingGroups = refreshContinuouslyLiveProcessGroups(
|
||||
remainingGroups,
|
||||
table,
|
||||
);
|
||||
if (remainingGroups.length === 0) return null;
|
||||
await wait(50);
|
||||
}
|
||||
if (forcedVerificationUnavailable) {
|
||||
return "Could not verify descendant process exit after SIGKILL";
|
||||
}
|
||||
return `Verified descendant process groups ${remainingGroups
|
||||
.map((group) => group.processGroupId)
|
||||
.join(",")} survived SIGKILL`;
|
||||
})();
|
||||
activeProcessCleanup.set(rootProcessGroupId, boundedCleanup);
|
||||
};
|
||||
activeProcessTerminators.set(child.pid, stopChildTree);
|
||||
const resultExitGuard = createResultExitGuard({
|
||||
resultPaths: completionPaths,
|
||||
interactive,
|
||||
graceMs: completedResultExitGraceMs,
|
||||
now: Date.now,
|
||||
pathExists: (candidate) =>
|
||||
access(candidate).then(
|
||||
() => true,
|
||||
() => false,
|
||||
),
|
||||
onExpired: async () => {
|
||||
const diagnostic = await processTreeDiagnostic(child.pid!);
|
||||
if (childSettled) return;
|
||||
postResultStallError = `Playwright remained alive for ${completedResultExitGraceMs}ms after every result was written`;
|
||||
recordOutput(
|
||||
Buffer.from(
|
||||
`\n${postResultStallError}; forcing bounded cleanup. ${diagnostic.summary}\n`,
|
||||
),
|
||||
process.stderr,
|
||||
);
|
||||
stopChildTree(diagnostic);
|
||||
},
|
||||
});
|
||||
const completionPoll = resultExitGuard.enabled
|
||||
? setInterval(() => {
|
||||
void resultExitGuard.poll().catch(() => {
|
||||
if (childSettled || postResultStallError) return;
|
||||
postResultStallError =
|
||||
"Completed-result exit guard failed during bounded inspection";
|
||||
recordOutput(
|
||||
Buffer.from(`\n${postResultStallError}; forcing cleanup.\n`),
|
||||
process.stderr,
|
||||
);
|
||||
stopChildTree();
|
||||
});
|
||||
}, 500)
|
||||
: undefined;
|
||||
completionPoll?.unref();
|
||||
let timedOut = false;
|
||||
const timer =
|
||||
timeoutMs === null
|
||||
? undefined
|
||||
: setTimeout(() => {
|
||||
timedOut = true;
|
||||
stopProcessGroup(child.pid!, "SIGTERM");
|
||||
setTimeout(
|
||||
() => stopProcessGroup(child.pid!, "SIGKILL"),
|
||||
10_000,
|
||||
).unref();
|
||||
stopChildTree();
|
||||
}, timeoutMs);
|
||||
timer?.unref();
|
||||
let spawnError: string | null = null;
|
||||
const exitCode = await new Promise<number>((resolve, reject) => {
|
||||
child.once("error", reject);
|
||||
child.once("exit", (code) => resolve(code ?? 1));
|
||||
child.once("exit", (code) => {
|
||||
childSettled = true;
|
||||
resolve(code ?? 1);
|
||||
});
|
||||
}).catch((error) => {
|
||||
childSettled = true;
|
||||
spawnError = error instanceof Error ? error.message : String(error);
|
||||
return 1;
|
||||
});
|
||||
if (timer) clearTimeout(timer);
|
||||
if (completionPoll) clearInterval(completionPoll);
|
||||
const processCleanupError = child.pid
|
||||
? await (activeProcessCleanup.get(child.pid) ??
|
||||
? await (boundedCleanup ??
|
||||
activeProcessCleanup.get(child.pid) ??
|
||||
terminateProcessGroup(child.pid))
|
||||
: null;
|
||||
if (child.pid) {
|
||||
activeProcessGroups.delete(child.pid);
|
||||
activeProcessCleanup.delete(child.pid);
|
||||
activeProcessTerminators.delete(child.pid);
|
||||
}
|
||||
await new Promise<void>((resolve) => log.end(resolve));
|
||||
return {
|
||||
exitCode,
|
||||
timedOut,
|
||||
postResultStallError,
|
||||
processCleanupError,
|
||||
spawnError,
|
||||
outputTail,
|
||||
|
|
@ -461,6 +727,10 @@ async function runAttempt(input: {
|
|||
childEnv,
|
||||
watchdog,
|
||||
path.join(privateDir, "playwright.log"),
|
||||
executions.map((candidate) =>
|
||||
path.join(privateDir, "cases", candidate.task.id, "result.json"),
|
||||
),
|
||||
options.ui || options.debug,
|
||||
);
|
||||
const processFailure = processResult.spawnError
|
||||
? `Playwright failed to start: ${processResult.spawnError}`
|
||||
|
|
@ -493,15 +763,7 @@ async function runAttempt(input: {
|
|||
processFailureClass,
|
||||
);
|
||||
const result = await readResult(resultPath, fallback);
|
||||
return processResult.processCleanupError
|
||||
? {
|
||||
...result,
|
||||
status: "failed" as const,
|
||||
failureClass: "cleanup_failure" as const,
|
||||
error: processResult.processCleanupError,
|
||||
cleanup: "failed" as const,
|
||||
}
|
||||
: result;
|
||||
return enforceResultProcessIntegrity(result, processResult);
|
||||
}),
|
||||
);
|
||||
let isolationError: unknown;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,106 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
observeDescendantProcessTree,
|
||||
refreshContinuouslyLiveProcessGroups,
|
||||
revalidateObservedProcessGroups,
|
||||
safeProcessGroupTerminationOrder,
|
||||
type ProcessObservation,
|
||||
} from "./process-tree.js";
|
||||
|
||||
function process(
|
||||
pid: number,
|
||||
parentPid: number,
|
||||
processGroupId: number,
|
||||
started = `start-${pid}`,
|
||||
): ProcessObservation {
|
||||
return { pid, parentPid, processGroupId, started, kind: "node" };
|
||||
}
|
||||
|
||||
describe("runner E2E process-tree cleanup", () => {
|
||||
it("orders verified nested groups before the outer group and excludes unsafe groups", () => {
|
||||
const table = [
|
||||
process(100, 10, 100),
|
||||
process(101, 100, 100),
|
||||
process(200, 101, 200),
|
||||
process(300, 200, 300),
|
||||
process(301, 300, 300),
|
||||
process(400, 101, 10),
|
||||
process(500, 999, 500),
|
||||
];
|
||||
const observed = observeDescendantProcessTree(table, 100);
|
||||
expect(
|
||||
safeProcessGroupTerminationOrder({
|
||||
rootProcessGroupId: 100,
|
||||
currentProcessGroupId: 10,
|
||||
groups: observed.groups,
|
||||
}),
|
||||
).toEqual([300, 200, 100]);
|
||||
});
|
||||
|
||||
it("rejects a reused pid whose start identity no longer matches", () => {
|
||||
const observed = observeDescendantProcessTree(
|
||||
[process(100, 10, 100), process(200, 100, 200)],
|
||||
100,
|
||||
);
|
||||
expect(
|
||||
revalidateObservedProcessGroups(observed.groups, [
|
||||
process(100, 10, 100),
|
||||
process(200, 1, 200, "reused-process"),
|
||||
]).map((group) => group.processGroupId),
|
||||
).toEqual([100]);
|
||||
});
|
||||
|
||||
it("refuses every group when launcher identity is unavailable", () => {
|
||||
const observed = observeDescendantProcessTree(
|
||||
[process(100, 10, 100), process(200, 100, 200)],
|
||||
100,
|
||||
);
|
||||
expect(
|
||||
safeProcessGroupTerminationOrder({
|
||||
rootProcessGroupId: 100,
|
||||
currentProcessGroupId: null,
|
||||
groups: observed.groups,
|
||||
}),
|
||||
).toEqual([]);
|
||||
});
|
||||
|
||||
it("does not add an unobserved root process group after revalidation", () => {
|
||||
expect(
|
||||
safeProcessGroupTerminationOrder({
|
||||
rootProcessGroupId: 100,
|
||||
currentProcessGroupId: 10,
|
||||
groups: [
|
||||
{
|
||||
processGroupId: 200,
|
||||
depth: 1,
|
||||
members: [{ pid: 200, started: "start-200" }],
|
||||
},
|
||||
],
|
||||
}),
|
||||
).toEqual([200]);
|
||||
});
|
||||
|
||||
it("retains a continuously live group when a cleanup helper replaces its original member", () => {
|
||||
const observed = observeDescendantProcessTree(
|
||||
[process(100, 10, 100), process(200, 100, 200)],
|
||||
100,
|
||||
);
|
||||
const verified = revalidateObservedProcessGroups(observed.groups, [
|
||||
process(100, 10, 100),
|
||||
process(200, 100, 200),
|
||||
]);
|
||||
const refreshed = refreshContinuouslyLiveProcessGroups(verified, [
|
||||
process(100, 10, 100),
|
||||
process(201, 1, 200),
|
||||
]);
|
||||
|
||||
expect(
|
||||
refreshed.find((group) => group.processGroupId === 200)?.members,
|
||||
).toEqual([{ pid: 201, started: "start-201" }]);
|
||||
expect(
|
||||
refreshContinuouslyLiveProcessGroups(refreshed, [
|
||||
process(100, 10, 100),
|
||||
]).map((group) => group.processGroupId),
|
||||
).toEqual([100]);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,142 @@
|
|||
export interface ProcessObservation {
|
||||
pid: number;
|
||||
parentPid: number;
|
||||
processGroupId: number;
|
||||
started: string;
|
||||
kind: string;
|
||||
}
|
||||
|
||||
export interface ObservedProcessGroup {
|
||||
processGroupId: number;
|
||||
depth: number;
|
||||
members: Array<Pick<ProcessObservation, "pid" | "started">>;
|
||||
}
|
||||
|
||||
export interface ObservedProcessTreeMember {
|
||||
process: ProcessObservation;
|
||||
depth: number;
|
||||
}
|
||||
|
||||
export function observeDescendantProcessTree(
|
||||
table: readonly ProcessObservation[],
|
||||
rootPid: number,
|
||||
) {
|
||||
const byPid = new Map(table.map((candidate) => [candidate.pid, candidate]));
|
||||
const byParent = new Map<number, ProcessObservation[]>();
|
||||
for (const candidate of table) {
|
||||
const children = byParent.get(candidate.parentPid) ?? [];
|
||||
children.push(candidate);
|
||||
byParent.set(candidate.parentPid, children);
|
||||
}
|
||||
const observed = new Map<number, ObservedProcessTreeMember>();
|
||||
const pending = [{ pid: rootPid, depth: 0 }];
|
||||
while (pending.length > 0) {
|
||||
const next = pending.shift()!;
|
||||
if (observed.has(next.pid)) continue;
|
||||
const candidate = byPid.get(next.pid);
|
||||
if (!candidate) continue;
|
||||
observed.set(next.pid, { process: candidate, depth: next.depth });
|
||||
for (const child of byParent.get(next.pid) ?? []) {
|
||||
pending.push({ pid: child.pid, depth: next.depth + 1 });
|
||||
}
|
||||
}
|
||||
const groupsById = new Map<number, ObservedProcessGroup>();
|
||||
for (const { process: candidate, depth } of observed.values()) {
|
||||
const group = groupsById.get(candidate.processGroupId) ?? {
|
||||
processGroupId: candidate.processGroupId,
|
||||
depth,
|
||||
members: [],
|
||||
};
|
||||
group.depth = Math.max(group.depth, depth);
|
||||
group.members.push({ pid: candidate.pid, started: candidate.started });
|
||||
groupsById.set(candidate.processGroupId, group);
|
||||
}
|
||||
return {
|
||||
members: [...observed.values()].sort(
|
||||
(left, right) => left.depth - right.depth,
|
||||
),
|
||||
groups: [...groupsById.values()].sort(
|
||||
(left, right) => right.depth - left.depth,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
export function safeProcessGroupTerminationOrder(input: {
|
||||
rootProcessGroupId: number;
|
||||
currentProcessGroupId: number | null;
|
||||
groups: readonly ObservedProcessGroup[];
|
||||
}) {
|
||||
if (input.currentProcessGroupId === null) return [];
|
||||
const safe = new Map<number, number>();
|
||||
for (const group of input.groups) {
|
||||
if (
|
||||
group.processGroupId <= 1 ||
|
||||
group.processGroupId === input.currentProcessGroupId
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
safe.set(
|
||||
group.processGroupId,
|
||||
Math.max(safe.get(group.processGroupId) ?? -1, group.depth),
|
||||
);
|
||||
}
|
||||
if (
|
||||
input.rootProcessGroupId > 1 &&
|
||||
input.rootProcessGroupId !== input.currentProcessGroupId &&
|
||||
input.groups.some(
|
||||
(group) => group.processGroupId === input.rootProcessGroupId,
|
||||
)
|
||||
) {
|
||||
safe.set(input.rootProcessGroupId, -1);
|
||||
}
|
||||
return [...safe.entries()]
|
||||
.sort((left, right) => right[1] - left[1])
|
||||
.map(([processGroupId]) => processGroupId);
|
||||
}
|
||||
|
||||
export function revalidateObservedProcessGroups(
|
||||
groups: readonly ObservedProcessGroup[],
|
||||
table: readonly ProcessObservation[],
|
||||
) {
|
||||
const byPid = new Map(table.map((candidate) => [candidate.pid, candidate]));
|
||||
return groups.filter((group) =>
|
||||
group.members.some((member) => {
|
||||
const candidate = byPid.get(member.pid);
|
||||
return (
|
||||
candidate?.processGroupId === group.processGroupId &&
|
||||
candidate.started === member.started
|
||||
);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Refresh groups only after their original members have been revalidated.
|
||||
* A process may replace itself or fork a final cleanup helper after SIGTERM.
|
||||
* Retain that continuously live group until one poll observes it as empty.
|
||||
*/
|
||||
export function refreshContinuouslyLiveProcessGroups(
|
||||
groups: readonly ObservedProcessGroup[],
|
||||
table: readonly ProcessObservation[],
|
||||
) {
|
||||
const membersByGroup = new Map<number, ProcessObservation[]>();
|
||||
for (const candidate of table) {
|
||||
const members = membersByGroup.get(candidate.processGroupId) ?? [];
|
||||
members.push(candidate);
|
||||
membersByGroup.set(candidate.processGroupId, members);
|
||||
}
|
||||
return groups.flatMap((group) => {
|
||||
const members = membersByGroup.get(group.processGroupId);
|
||||
return members
|
||||
? [
|
||||
{
|
||||
...group,
|
||||
members: members.map((candidate) => ({
|
||||
pid: candidate.pid,
|
||||
started: candidate.started,
|
||||
})),
|
||||
},
|
||||
]
|
||||
: [];
|
||||
});
|
||||
}
|
||||
|
|
@ -0,0 +1,157 @@
|
|||
import { spawn } from "node:child_process";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
createResultExitGuard,
|
||||
enforceResultProcessIntegrity,
|
||||
} from "./result-exit-guard.js";
|
||||
import type { RunnerE2EResult } from "./types.js";
|
||||
|
||||
const passingResult = {
|
||||
status: "passed",
|
||||
cleanup: "passed",
|
||||
} as RunnerE2EResult;
|
||||
|
||||
describe("runner E2E result-exit guard", () => {
|
||||
it("expires once after every result exists and stops a controlled child", async () => {
|
||||
const child = spawn(
|
||||
process.execPath,
|
||||
["--eval", "setInterval(() => {}, 1_000)"],
|
||||
{ stdio: "ignore" },
|
||||
);
|
||||
const exited = new Promise<void>((resolve, reject) => {
|
||||
child.once("exit", () => resolve());
|
||||
child.once("error", reject);
|
||||
});
|
||||
let now = 1_000;
|
||||
const onExpired = vi.fn(() => {
|
||||
child.kill("SIGTERM");
|
||||
});
|
||||
const guard = createResultExitGuard({
|
||||
resultPaths: ["first.json", "second.json"],
|
||||
interactive: false,
|
||||
graceMs: 120_000,
|
||||
now: () => now,
|
||||
pathExists: async () => true,
|
||||
onExpired,
|
||||
});
|
||||
let exitTimeout: NodeJS.Timeout | undefined;
|
||||
|
||||
try {
|
||||
await guard.poll();
|
||||
now += 119_999;
|
||||
await guard.poll();
|
||||
expect(onExpired).not.toHaveBeenCalled();
|
||||
now += 1;
|
||||
await guard.poll();
|
||||
const stopped = await Promise.race([
|
||||
exited.then(() => true),
|
||||
new Promise<false>((resolve) => {
|
||||
exitTimeout = setTimeout(() => resolve(false), 2_000);
|
||||
}),
|
||||
]);
|
||||
expect(stopped).toBe(true);
|
||||
await guard.poll();
|
||||
expect(onExpired).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
if (exitTimeout) clearTimeout(exitTimeout);
|
||||
if (child.exitCode === null && child.signalCode === null) {
|
||||
child.kill("SIGKILL");
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("does not monitor interactive sessions", async () => {
|
||||
const pathExists = vi.fn(async () => true);
|
||||
const onExpired = vi.fn();
|
||||
const guard = createResultExitGuard({
|
||||
resultPaths: ["result.json"],
|
||||
interactive: true,
|
||||
graceMs: 1,
|
||||
now: () => 10,
|
||||
pathExists,
|
||||
onExpired,
|
||||
});
|
||||
|
||||
expect(guard.enabled).toBe(false);
|
||||
await guard.poll();
|
||||
expect(pathExists).not.toHaveBeenCalled();
|
||||
expect(onExpired).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("requires every result before it starts the grace period", async () => {
|
||||
let now = 1_000;
|
||||
let complete = false;
|
||||
const onExpired = vi.fn();
|
||||
const guard = createResultExitGuard({
|
||||
resultPaths: ["first.json", "second.json"],
|
||||
interactive: false,
|
||||
graceMs: 10,
|
||||
now: () => now,
|
||||
pathExists: async (path) => path === "first.json" || complete,
|
||||
onExpired,
|
||||
});
|
||||
|
||||
await guard.poll();
|
||||
now += 100;
|
||||
complete = true;
|
||||
await guard.poll();
|
||||
now += 9;
|
||||
await guard.poll();
|
||||
expect(onExpired).not.toHaveBeenCalled();
|
||||
now += 1;
|
||||
await guard.poll();
|
||||
expect(onExpired).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("changes a saved pass into a cleanup failure after process failure", () => {
|
||||
expect(
|
||||
enforceResultProcessIntegrity(passingResult, {
|
||||
exitCode: 1,
|
||||
timedOut: false,
|
||||
postResultStallError: null,
|
||||
processCleanupError: null,
|
||||
}),
|
||||
).toMatchObject({
|
||||
status: "failed",
|
||||
cleanup: "failed",
|
||||
failureClass: "cleanup_failure",
|
||||
error: "Playwright exited 1 after writing a passing result",
|
||||
});
|
||||
expect(
|
||||
enforceResultProcessIntegrity(passingResult, {
|
||||
exitCode: 0,
|
||||
timedOut: false,
|
||||
postResultStallError: "Playwright stayed alive after every result",
|
||||
processCleanupError: null,
|
||||
}),
|
||||
).toMatchObject({
|
||||
status: "failed",
|
||||
cleanup: "failed",
|
||||
failureClass: "cleanup_failure",
|
||||
error: "Playwright stayed alive after every result",
|
||||
});
|
||||
expect(
|
||||
enforceResultProcessIntegrity(passingResult, {
|
||||
exitCode: 1,
|
||||
timedOut: true,
|
||||
postResultStallError: null,
|
||||
processCleanupError: null,
|
||||
}),
|
||||
).toMatchObject({
|
||||
status: "failed",
|
||||
cleanup: "failed",
|
||||
failureClass: "cleanup_failure",
|
||||
error:
|
||||
"Playwright exceeded its process watchdog after writing every result",
|
||||
});
|
||||
|
||||
expect(
|
||||
enforceResultProcessIntegrity(passingResult, {
|
||||
exitCode: 0,
|
||||
timedOut: false,
|
||||
postResultStallError: null,
|
||||
processCleanupError: null,
|
||||
}),
|
||||
).toBe(passingResult);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
import type { RunnerE2EResult } from "./types.js";
|
||||
|
||||
export interface ResultExitGuard {
|
||||
enabled: boolean;
|
||||
poll: () => Promise<void>;
|
||||
}
|
||||
|
||||
export function createResultExitGuard(input: {
|
||||
resultPaths: readonly string[];
|
||||
interactive: boolean;
|
||||
graceMs: number;
|
||||
now: () => number;
|
||||
pathExists: (path: string) => Promise<boolean>;
|
||||
onExpired: () => Promise<void> | void;
|
||||
}): ResultExitGuard {
|
||||
const enabled = !input.interactive && input.resultPaths.length > 0;
|
||||
let completionObservedAt: number | null = null;
|
||||
let pollActive = false;
|
||||
let expired = false;
|
||||
|
||||
return {
|
||||
enabled,
|
||||
async poll() {
|
||||
if (!enabled || pollActive || expired) return;
|
||||
pollActive = true;
|
||||
try {
|
||||
const complete = (
|
||||
await Promise.all(input.resultPaths.map(input.pathExists))
|
||||
).every(Boolean);
|
||||
if (!complete) return;
|
||||
const now = input.now();
|
||||
completionObservedAt ??= now;
|
||||
if (now - completionObservedAt < input.graceMs) return;
|
||||
expired = true;
|
||||
await input.onExpired();
|
||||
} finally {
|
||||
pollActive = false;
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function enforceResultProcessIntegrity(
|
||||
result: RunnerE2EResult,
|
||||
processResult: {
|
||||
exitCode: number;
|
||||
timedOut: boolean;
|
||||
postResultStallError: string | null;
|
||||
processCleanupError: string | null;
|
||||
},
|
||||
): RunnerE2EResult {
|
||||
const integrityError =
|
||||
processResult.processCleanupError ??
|
||||
processResult.postResultStallError ??
|
||||
(processResult.timedOut && result.status === "passed"
|
||||
? "Playwright exceeded its process watchdog after writing every result"
|
||||
: processResult.exitCode !== 0 && result.status === "passed"
|
||||
? `Playwright exited ${processResult.exitCode} after writing a passing result`
|
||||
: null);
|
||||
return integrityError
|
||||
? {
|
||||
...result,
|
||||
status: "failed",
|
||||
failureClass: "cleanup_failure",
|
||||
error: integrityError,
|
||||
cleanup: "failed",
|
||||
}
|
||||
: result;
|
||||
}
|
||||
Loading…
Reference in New Issue