From 95d9d51f7298f855d62c19035f4e3d6491fa9f1e Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:34:46 +0800 Subject: [PATCH] fix(process): retain failures through finalization --- src/application/LinuxHopperPython.test.ts | 65 ++++++++++++++++++++++- src/application/LinuxHopperPython.ts | 16 +++--- src/process/ExecFileOutput.test.ts | 65 +++++++++++++++++++++++ src/process/ExecFileOutput.ts | 12 ++++- 4 files changed, 149 insertions(+), 9 deletions(-) diff --git a/src/application/LinuxHopperPython.test.ts b/src/application/LinuxHopperPython.test.ts index 10b365136d..e6a0011560 100644 --- a/src/application/LinuxHopperPython.test.ts +++ b/src/application/LinuxHopperPython.test.ts @@ -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"); diff --git a/src/application/LinuxHopperPython.ts b/src/application/LinuxHopperPython.ts index e586ca2cbc..f61019ae32 100644 --- a/src/application/LinuxHopperPython.ts +++ b/src/application/LinuxHopperPython.ts @@ -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> => { - 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); diff --git a/src/process/ExecFileOutput.test.ts b/src/process/ExecFileOutput.test.ts index e31ffe62ab..3975990d61 100644 --- a/src/process/ExecFileOutput.test.ts +++ b/src/process/ExecFileOutput.test.ts @@ -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; + } + }, +); diff --git a/src/process/ExecFileOutput.ts b/src/process/ExecFileOutput.ts index 0174e6f7ff..d2c2ca43df 100644 --- a/src/process/ExecFileOutput.ts +++ b/src/process/ExecFileOutput.ts @@ -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);