fix(process): preserve command completion and input ownership

This commit is contained in:
morluto
2026-10-27 07:54:56 +08:00
parent 8a47cbbcce
commit 7fa54e2d62
4 changed files with 229 additions and 37 deletions
+162
View File
@@ -437,3 +437,165 @@ test.each([
}
},
);
// The injected launcher runs real owned Node processes on every POSIX host;
// Linux firmware binaries and prlimit are deliberately outside this lifecycle seam.
test.skipIf(process.platform === "win32").each([
{
overflow: false,
exitCode: 7,
acceptFailure: false,
cancelBeforeExit: false,
expected: "ProviderAdapterError",
},
{
overflow: false,
exitCode: 0,
acceptFailure: false,
cancelBeforeExit: false,
expected: "AnalysisCancelledError",
},
{
overflow: false,
exitCode: 1,
acceptFailure: true,
cancelBeforeExit: false,
expected: "AnalysisCancelledError",
},
{
overflow: false,
exitCode: 7,
acceptFailure: false,
cancelBeforeExit: true,
expected: "AnalysisCancelledError",
},
{
overflow: true,
exitCode: 7,
acceptFailure: false,
cancelBeforeExit: false,
expected: "ProviderAdapterError",
},
])(
"retains root exit $exitCode while a handshaken descendant holds output (accepted: $acceptFailure, earlier cancellation: $cancelBeforeExit, overflow: $overflow)",
async ({ exitCode, acceptFailure, cancelBeforeExit, expected, overflow }) => {
const root = await createTestTempDirectory(
"rea-firmware-inherited-output-",
);
const caller = new AbortController();
const reason = { caller: "cancel with inherited output still open" };
let cleanupCalls = 0;
let closed = false;
let exited = false;
const failure: unknown = await runFirmwareCommand({
engine: "binwalk",
operation: "inspect_firmware_regions",
command: process.execPath,
limiter: process.execPath,
fileSizeCeiling: Number.MAX_SAFE_INTEGER,
args: [],
cwd: root,
environment: process.env,
signal: caller.signal,
deadline: {
phase: "analysis",
signal: new AbortController().signal,
timeoutMs: 30_000,
expiresAtMs: performance.now() + 30_000,
},
acceptReportedExtractionFailure: acceptFailure,
launcher: async (options) => {
const exitRelease = join(root, "exit-release");
const tailRelease = join(root, "tail-release");
const descendant = `
const { existsSync } = require("node:fs");
setInterval(() => {}, 1000);
const tail = setInterval(() => {
if (!existsSync(${JSON.stringify(tailRelease)})) return;
clearInterval(tail);
process.stderr.write(Buffer.alloc(${overflow ? FIRMWARE_LIMITS.diagnosticBytes : 0}, 120), () => {
process.stderr.write("descendant-tail\\n");
});
}, 10);
process.send("ready");
`;
const owned = await spawnOwnedProviderProcess({
...options,
command: process.execPath,
arguments: [
"-e",
`
const { spawn } = require("node:child_process");
const child = spawn(process.execPath, ["-e", ${JSON.stringify(descendant)}], {
stdio: ["ignore", "inherit", "inherit", "ipc"],
});
child.once("message", () => {
process.stderr.write("descendant-ready\\n");
child.disconnect();
child.unref();
if (${cancelBeforeExit}) {
const release = setInterval(() => {
if (!require("node:fs").existsSync(${JSON.stringify(exitRelease)})) return;
clearInterval(release);
process.exitCode = ${exitCode};
}, 10);
} else process.exitCode = ${exitCode};
});
`,
],
});
owned.process.once("close", () => {
closed = true;
});
let diagnostics = "";
owned.process.stderr?.on("data", (chunk: Buffer) => {
diagnostics += chunk.toString();
if (cancelBeforeExit && !caller.signal.aborted) {
expect(exited).toBe(false);
caller.abort(reason);
void writeFile(exitRelease, "release");
} else if (
!cancelBeforeExit &&
diagnostics.includes("descendant-tail")
) {
expect(exited).toBe(true);
expect(closed).toBe(false);
caller.abort(reason);
}
});
owned.process.once("exit", () => {
exited = true;
// Release descendant diagnostics only after the supervisor has also
// observed the root's exit, leaving inherited pipes open until cleanup.
queueMicrotask(() => {
void writeFile(tailRelease, "release");
});
});
return {
...owned,
cleanup: async () => {
cleanupCalls += 1;
return cleanupOwnedProcessGroup(owned.ownership);
},
};
},
retainCleanup: () => {
throw new Error("Real descendant cleanup must complete");
},
}).catch((cause: unknown) => cause);
expect(exited).toBe(true);
expect(cleanupCalls).toBe(1);
expect(failure).toMatchObject({ _tag: expected });
if (expected === "ProviderAdapterError")
expect(failure).toMatchObject({
diagnostics: {
exit_code: 7,
stderr: overflow
? expect.stringMatching(/^descendant-ready\nxxx/)
: "descendant-ready\ndescendant-tail\n",
truncated: overflow,
},
});
else expect(failure).toMatchObject({ cause: reason });
},
);
+47 -34
View File
@@ -236,6 +236,39 @@ export const runFirmwareCommand = async (context: {
throw firmwareCancellation(operation, signal.reason, context.deadline);
throw cause;
});
let failure: AnalysisError | undefined;
const isFailedExit = (code: number | null | undefined) =>
code !== undefined &&
code !== 0 &&
!(context.acceptReportedExtractionFailure === true && code === 1);
const completedFailure = (snapshot: ProviderProcessSnapshot) => {
if (isFailedExit(snapshot.exitCode)) {
const signalLimit = firmwareSignalLimitError(
operation,
snapshot.signal,
fileSizeBytes,
fileSizeAdjustable,
capturedFirmwareOutput(snapshot),
);
if (signalLimit !== undefined) return signalLimit;
return new ProviderAdapterError(engine, operation, {
diagnostics: {
reason:
"Tool failed; inspect exit status and stderr for format, dependency, sandbox or resource-limit failures",
exit_code: snapshot.exitCode ?? null,
signal: snapshot.signal ?? null,
stdout: snapshot.stdout.text,
stderr: snapshot.stderr.text,
stdout_bytes: snapshot.stdout.bytes,
stderr_bytes: snapshot.stderr.bytes,
truncated: snapshot.diagnosticTruncated === true,
},
remediationAction: `Inspect exit status and stderr for format, dependency, sandbox, or resource-limit failures. ${firmwareProcessLimitRecovery(fileSizeBytes, fileSizeAdjustable)} A format or dependency failure is not repaired by changing analysis_timeout_ms.`,
});
}
return undefined;
};
let completedFailureObserved = false;
let diagnosticTruncated = false;
const supervisor = new ProviderProcessSupervisor(
{
@@ -249,6 +282,17 @@ export const runFirmwareCommand = async (context: {
onDiagnostic: (event) => {
if (event.type === "output" && event.truncated === true)
diagnosticTruncated = true;
// Root exit and inherited-pipe closure are different observations.
// Retain a completed failure before drain or cleanup can be interrupted,
// while cancellation already present at exit keeps its precedence.
if (
event.type === "exit" &&
failure === undefined &&
!diagnosticTruncated &&
firmwarePhaseInterruption(operation, signal, context.deadline) ===
undefined
)
completedFailureObserved = isFailedExit(event.code);
},
},
);
@@ -272,7 +316,6 @@ export const runFirmwareCommand = async (context: {
);
}
};
let failure: AnalysisError | undefined;
try {
while (!(await supervisor.waitForOutputClose(100))) {
checkDiagnostics();
@@ -288,46 +331,16 @@ export const runFirmwareCommand = async (context: {
}
checkDiagnostics();
checkInterruption();
// Physical completion belongs to this observation, before awaited cleanup.
// A later cancellation must not replace an already completed tool failure.
const snapshot = supervisor.snapshot();
if (
snapshot.exitCode !== 0 &&
!(
context.acceptReportedExtractionFailure === true &&
snapshot.exitCode === 1
)
) {
const signalLimit = firmwareSignalLimitError(
operation,
snapshot.signal,
fileSizeBytes,
fileSizeAdjustable,
capturedFirmwareOutput(snapshot),
);
if (signalLimit !== undefined) throw signalLimit;
throw new ProviderAdapterError(engine, operation, {
diagnostics: {
reason:
"Tool failed; inspect exit status and stderr for format, dependency, sandbox or resource-limit failures",
exit_code: snapshot.exitCode ?? null,
signal: snapshot.signal ?? null,
stdout: snapshot.stdout.text,
stderr: snapshot.stderr.text,
stdout_bytes: snapshot.stdout.bytes,
stderr_bytes: snapshot.stderr.bytes,
},
remediationAction: `Inspect exit status and stderr for format, dependency, sandbox, or resource-limit failures. ${firmwareProcessLimitRecovery(fileSizeBytes, fileSizeAdjustable)} A format or dependency failure is not repaired by changing analysis_timeout_ms.`,
});
}
} catch (cause: unknown) {
failure = firmwareFailure(cause, context);
failure ??= firmwareFailure(cause, context);
}
const stopped = await supervisor.stop();
// Cleanup is an awaited part of this phase. Preserve a prior observed failure,
// otherwise retain an interruption that arrived while stopping owned resources.
failure ??= firmwarePhaseInterruption(operation, signal, context.deadline);
const snapshot = supervisor.snapshot();
// Capture diagnostics only after owned cleanup has drained the inherited pipes.
if (completedFailureObserved) failure = completedFailure(snapshot);
if (stopped.status === "incomplete") {
context.retainCleanup(supervisor);
const previous =
+15
View File
@@ -53,6 +53,21 @@ it.each([undefined, "SIGTERM"] as const)(
},
);
it.each([undefined, "SIGTERM"] as const)(
"closes unused command input with stop signal %s",
async (stopSignal) => {
const result = await execFileOutput(
process.execPath,
[
"-e",
'process.stdin.resume(); process.stdin.on("end", () => process.stdout.write("input closed"));',
],
{ timeout: 2_000, ...(stopSignal === undefined ? {} : { stopSignal }) },
);
expect(result.stdout).toBe("input closed");
},
);
describe("execFileOutput", () => {
it("ignores non-Error and incomplete failure metadata", () => {
expect(execFileOutputFailure("failure")).toBeUndefined();
+5 -3
View File
@@ -142,7 +142,7 @@ export const execFileFailureCapturedOutput = (
truncated: failure.outputTruncated,
});
/** Run a shell-free command while capturing its complete output bytes. */
/** Run a shell-free command with no input while capturing its complete output bytes. */
export const execFileBytes = (
command: string,
arguments_: readonly string[],
@@ -175,7 +175,7 @@ export const execFileBytes = (
resolve({ stdout, stderr });
};
if (stopSignal === undefined) {
execFile(
const child = execFile(
command,
[...arguments_],
{
@@ -194,6 +194,8 @@ export const execFileBytes = (
stderr,
),
);
// This promise API exposes no input writer; EOF belongs to the owner.
child.stdin?.end();
return;
}
startStoppableCommand(
@@ -229,7 +231,7 @@ const startStoppableCommand = (
// pipe lifetime of descendants still using caller-owned temporary files.
const child = spawn(command, [...arguments_], {
...execOptions,
stdio: "pipe",
stdio: ["ignore", "pipe", "pipe"],
killSignal: stopSignal,
});
// The first stop request names the failure; a later one is not resent.