mirror of
https://github.com/morluto/rea.git
synced 2026-10-30 11:25:39 +00:00
fix(process): retain failures through finalization
This commit is contained in:
1 parent
381004b664
commit
95d9d51f72
4 files changed
+149
-9
No files matched your search
@@ -5,7 +5,10 @@ import { join } from "node:path";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js";
|
||||
import { execFileOutput } from "../process/ExecFileOutput.js";
|
||||
import {
|
||||
ExecFileOutputCancelledError,
|
||||
execFileOutput,
|
||||
} from "../process/ExecFileOutput.js";
|
||||
import { inspectLinuxHopperPythonRuntime } from "./LinuxHopperPython.js";
|
||||
|
||||
const response = (value: unknown) => {
|
||||
@@ -128,6 +131,50 @@ describe("distribution Python runtime inspection", () => {
|
||||
});
|
||||
|
||||
describe("distribution Python process cleanup", () => {
|
||||
it("retains output failure when cancellation arrives during command cleanup", async () => {
|
||||
const directory = await createTestTempDirectory("rea-python-failure-stop-");
|
||||
const stopping = join(directory, "stopping");
|
||||
const release = join(directory, "release");
|
||||
const controller = new AbortController();
|
||||
const program = `
|
||||
const fs = require('node:fs');
|
||||
process.on('SIGTERM', () => {
|
||||
fs.writeFileSync(${JSON.stringify(stopping)}, 'stopping');
|
||||
const timer = setInterval(() => {
|
||||
if (fs.existsSync(${JSON.stringify(release)})) {
|
||||
clearInterval(timer);
|
||||
process.exit(0);
|
||||
}
|
||||
}, 10);
|
||||
});
|
||||
process.stdout.write(Buffer.alloc(128 * 1024, 120));
|
||||
setInterval(() => {}, 1000);
|
||||
`;
|
||||
const execution = inspectLinuxHopperPythonRuntime(
|
||||
process.env,
|
||||
controller.signal,
|
||||
(_command, _arguments, options) =>
|
||||
execFileOutput(process.execPath, ["-e", program], options),
|
||||
);
|
||||
try {
|
||||
await vi.waitFor(() => access(stopping), {
|
||||
timeout: 5_000,
|
||||
interval: 10,
|
||||
});
|
||||
controller.abort({ request: "cancel during failed command cleanup" });
|
||||
} finally {
|
||||
await writeFile(release, "release");
|
||||
await execution;
|
||||
}
|
||||
expect(await execution).toMatchObject({
|
||||
ok: false,
|
||||
error: {
|
||||
reason: "query_failed",
|
||||
cause: { code: "ERR_CHILD_PROCESS_STDIO_MAXBUFFER" },
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it.skipIf(process.platform !== "linux" || !existsSync("/usr/bin/python3"))(
|
||||
"waits for a delayed Python exit before completing cancellation",
|
||||
async () => {
|
||||
@@ -194,6 +241,22 @@ while True:
|
||||
});
|
||||
|
||||
describe("distribution Python isolation and cancellation", () => {
|
||||
it("uses the producer's cancellation reason without requiring an aborted caller", async () => {
|
||||
const reason = { producer: "interrupted command" };
|
||||
const result = await inspectLinuxHopperPythonRuntime(
|
||||
{},
|
||||
undefined,
|
||||
async () => {
|
||||
throw new ExecFileOutputCancelledError(reason);
|
||||
},
|
||||
);
|
||||
expect(result).toMatchObject({
|
||||
ok: false,
|
||||
error: { reason: "cancelled", cause: reason },
|
||||
});
|
||||
if (!result.ok) expect(result.error.cause).toBe(reason);
|
||||
});
|
||||
|
||||
it("preserves cancellation and starts no interpreter for an already cancelled query", async () => {
|
||||
const controller = new AbortController();
|
||||
const reason = new Error("caller cancellation");
|
||||
|
||||
@@ -3,7 +3,10 @@ import { posix } from "node:path";
|
||||
import { z } from "zod";
|
||||
|
||||
import { err, ok, type Result } from "../domain/result.js";
|
||||
import { execFileOutput } from "../process/ExecFileOutput.js";
|
||||
import {
|
||||
ExecFileOutputCancelledError,
|
||||
execFileOutput,
|
||||
} from "../process/ExecFileOutput.js";
|
||||
|
||||
const libraryPath = z
|
||||
.string()
|
||||
@@ -76,15 +79,15 @@ export const inspectLinuxHopperPythonRuntime = async (
|
||||
signal?: AbortSignal,
|
||||
execute: typeof execFileOutput = execFileOutput,
|
||||
): Promise<Result<LinuxHopperPythonRuntime, LinuxHopperPythonError>> => {
|
||||
const cancelled = () =>
|
||||
const cancelled = (reason: unknown) =>
|
||||
err(
|
||||
new LinuxHopperPythonError(
|
||||
"cancelled",
|
||||
"Python runtime inspection was cancelled.",
|
||||
signal?.reason,
|
||||
reason,
|
||||
),
|
||||
);
|
||||
if (isAborted(signal)) return cancelled();
|
||||
if (isAborted(signal)) return cancelled(signal?.reason);
|
||||
let output: string;
|
||||
try {
|
||||
const result = await execute(
|
||||
@@ -100,7 +103,8 @@ export const inspectLinuxHopperPythonRuntime = async (
|
||||
);
|
||||
output = result.stdout;
|
||||
} catch (cause: unknown) {
|
||||
if (isAborted(signal)) return cancelled();
|
||||
if (cause instanceof ExecFileOutputCancelledError)
|
||||
return cancelled(cause.cause);
|
||||
return err(
|
||||
new LinuxHopperPythonError(
|
||||
"query_failed",
|
||||
@@ -109,7 +113,7 @@ export const inspectLinuxHopperPythonRuntime = async (
|
||||
),
|
||||
);
|
||||
}
|
||||
if (isAborted(signal)) return cancelled();
|
||||
if (isAborted(signal)) return cancelled(signal?.reason);
|
||||
let value: unknown;
|
||||
try {
|
||||
value = JSON.parse(output);
|
||||
|
||||
@@ -686,3 +686,68 @@ it.each([
|
||||
expect(failure.cause).toBe(reason);
|
||||
},
|
||||
);
|
||||
|
||||
it.skipIf(process.platform === "win32").each([
|
||||
{ code: 7, action: "abort" },
|
||||
{ code: 0, action: "abort" },
|
||||
{ code: 7, action: "overflow" },
|
||||
] as const)(
|
||||
"owns root exit $code before inherited-pipe $action",
|
||||
async ({ code, action }) => {
|
||||
const directory = await createTestTempDirectory("rea-exec-exit-pipes-");
|
||||
const exited = join(directory, "root-exited");
|
||||
const release = join(directory, "release");
|
||||
const descendant = [
|
||||
'const fs = require("node:fs");',
|
||||
"const parent = Number(process.argv[1]);",
|
||||
"const timer = setInterval(() => {",
|
||||
" try { process.kill(parent, 0); } catch {",
|
||||
` fs.writeFileSync(${JSON.stringify(exited)}, "");`,
|
||||
" }",
|
||||
` if (!fs.existsSync(${JSON.stringify(release)})) return;`,
|
||||
` process.stdout.write(${JSON.stringify(action === "overflow" ? "x".repeat(64) : "late")});`,
|
||||
" clearInterval(timer); process.exit(0);",
|
||||
"}, 10);",
|
||||
"setTimeout(() => process.exit(0), 5000).unref();",
|
||||
].join("\n");
|
||||
const program = [
|
||||
'const { spawn } = require("node:child_process");',
|
||||
`spawn(process.execPath, ["-e", ${JSON.stringify(descendant)}, String(process.pid)], { stdio: "inherit" }).unref();`,
|
||||
'process.stderr.write("root diagnostic");',
|
||||
`process.exit(${code});`,
|
||||
].join("\n");
|
||||
const controller = new AbortController();
|
||||
let settled = false;
|
||||
const execution = execFileOutput(process.execPath, ["-e", program], {
|
||||
stopSignal: "SIGTERM",
|
||||
signal: controller.signal,
|
||||
maxBuffer: 32,
|
||||
})
|
||||
.catch((cause: unknown) => cause)
|
||||
.finally(() => {
|
||||
settled = true;
|
||||
});
|
||||
try {
|
||||
expect(await waitForFile(exited)).toBe(true);
|
||||
expect(settled).toBe(false);
|
||||
if (action === "abort") controller.abort("late abort");
|
||||
expect(settled).toBe(false);
|
||||
await writeFile(release, "");
|
||||
const failure = await execution;
|
||||
expect(failure).toMatchObject(
|
||||
code === 0
|
||||
? { _tag: "ExecFileOutputCancelledError", cause: "late abort" }
|
||||
: { code: 7 },
|
||||
);
|
||||
expect(execFileOutputFailure(failure)).toMatchObject({
|
||||
stderr: "root diagnostic",
|
||||
stdout: action === "overflow" ? "x".repeat(32) : "late",
|
||||
outputTruncated: action === "overflow",
|
||||
});
|
||||
} finally {
|
||||
await writeFile(release, "");
|
||||
controller.abort();
|
||||
await execution;
|
||||
}
|
||||
},
|
||||
);
|
||||
@@ -220,6 +220,7 @@ const startStoppableCommand = (
|
||||
}
|
||||
let stopped: "abort" | "timeout" | "output-limit" | undefined;
|
||||
let processError: Error | null = null;
|
||||
let rootFailed = false;
|
||||
let outputError: Error | undefined;
|
||||
let deadline: NodeJS.Timeout | undefined;
|
||||
let grace: NodeJS.Timeout | undefined;
|
||||
@@ -233,7 +234,7 @@ const startStoppableCommand = (
|
||||
});
|
||||
// The first stop request names the failure; a later one is not resent.
|
||||
const stop = (reason: "abort" | "timeout" | "output-limit"): void => {
|
||||
if (stopped !== undefined) return;
|
||||
if (stopped !== undefined || rootFailed || processError !== null) return;
|
||||
stopped = reason;
|
||||
child.kill(stopSignal);
|
||||
grace = setTimeout(() => child.kill("SIGKILL"), STOP_GRACE_MS);
|
||||
@@ -268,6 +269,11 @@ const startStoppableCommand = (
|
||||
child.on("error", (error: Error) => {
|
||||
processError ??= error;
|
||||
});
|
||||
child.on("exit", (code, exitSignal) => {
|
||||
// A failed root has completed its outcome even while descendants retain
|
||||
// its pipes. Finalization cancellation cannot replace that failure.
|
||||
rootFailed = code !== 0 || exitSignal !== null;
|
||||
});
|
||||
child.on("close", (code, exitSignal) => {
|
||||
signal?.removeEventListener("abort", onAbort);
|
||||
clearTimeout(deadline);
|
||||
@@ -290,7 +296,9 @@ const startStoppableCommand = (
|
||||
? new ExecFileOutputCancelledError(signal.reason)
|
||||
: stopped === "timeout" && timeout !== undefined
|
||||
? timeoutError(command, timeout, processError)
|
||||
: (outputError ?? processError);
|
||||
: rootFailed && stopped === undefined
|
||||
? processError
|
||||
: (outputError ?? processError);
|
||||
if (failure !== null && outputError !== undefined)
|
||||
Reflect.set(failure, "outputTruncated", true);
|
||||
complete(failure, out, diagnostic);
|
||||
|
||||
Reference in new issue
Block a user