From b9292516f4eedc2dedba713abf57f665ef1fb596 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Wed, 7 Oct 2026 18:39:10 +0800 Subject: [PATCH] fix(process): retain provider output until streams close (#928) --- src/hopper/LinuxPrivateDisplayProbe.ts | 8 +- src/javascript/recovery/WakaruCommand.test.ts | 115 +++++++++++++++++- src/javascript/recovery/WakaruCommand.ts | 2 +- .../hopper/linuxPrivateDisplay.test.ts | 21 ++++ tests/fixtures/x11ProbeOutput.py | 9 ++ 5 files changed, 149 insertions(+), 6 deletions(-) create mode 100644 tests/fixtures/x11ProbeOutput.py diff --git a/src/hopper/LinuxPrivateDisplayProbe.ts b/src/hopper/LinuxPrivateDisplayProbe.ts index 26d081189..1ec6190d0 100644 --- a/src/hopper/LinuxPrivateDisplayProbe.ts +++ b/src/hopper/LinuxPrivateDisplayProbe.ts @@ -123,7 +123,7 @@ export const runLinuxPrivateDisplayProbe: LinuxPrivateDisplayProbeRunner = cleanup: () => cleanupOwnedProcessGroup(started.ownership), }; const supervisor = new ProviderProcessSupervisor(launch); - const outcome = await waitForProbeExit( + const outcome = await waitForProbeOutputClose( supervisor, options.timeoutMs, options.signal, @@ -300,7 +300,7 @@ const emptyProcessResult = ( cleanupIncomplete: false, }); -const waitForProbeExit = async ( +const waitForProbeOutputClose = async ( supervisor: ProviderProcessSupervisor, timeoutMs: number, signal: AbortSignal | undefined, @@ -316,8 +316,8 @@ const waitForProbeExit = async ( try { return await Promise.race([ supervisor - .waitForExit(Math.max(1, timeoutMs)) - .then((exited): "exit" | "timeout" => (exited ? "exit" : "timeout")), + .waitForOutputClose(Math.max(1, timeoutMs)) + .then((closed): "exit" | "timeout" => (closed ? "exit" : "timeout")), aborted, ]); } finally { diff --git a/src/javascript/recovery/WakaruCommand.test.ts b/src/javascript/recovery/WakaruCommand.test.ts index 194733248..bdfe54dd9 100644 --- a/src/javascript/recovery/WakaruCommand.test.ts +++ b/src/javascript/recovery/WakaruCommand.test.ts @@ -1,5 +1,8 @@ +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; import { expect, it } from "vitest"; -import { runWakaruCommand } from "./WakaruCommand.js"; +import { runWakaruCommand, type WakaruLauncher } from "./WakaruCommand.js"; +import { RECOVERY_LIMITS } from "./WakaruRelease.js"; it("rejects an exhausted shared execution budget before launching the next command", async () => { await expect( @@ -36,3 +39,113 @@ it("preserves caller cancellation before acquiring an expired execution budget", }), ).rejects.toMatchObject({ _tag: "AnalysisCancelledError" }); }); + +const commandFixture = ( + emit: (producer: ReturnType) => void, +) => { + const producer = createProducer(); + const launcher: WakaruLauncher = async (options) => { + setImmediate(() => emit(producer)); + return { + process: producer, + ownership: { + runId: options.runId, + leaderPid: 12345, + processGroupId: 12345, + expectedParentPid: process.pid, + }, + cleanup: async () => ({ cleaned: true, signaled: false }), + }; + }; + return { + producer, + context: { + command: "/selected/wakaru", + limiter: "/usr/bin/prlimit", + arguments: ["--version"], + cwd: "/tmp/unused", + environment: {}, + deadline: Date.now() + 5_000, + launcher, + }, + }; +}; + +const createProducer = () => + Object.assign(new EventEmitter(), { + stdout: new PassThrough(), + stderr: new PassThrough(), + exitCode: null, + signalCode: null, + kill: () => false, + }); + +it("retains stdout and stderr delivered after launcher exit", async () => { + const { context } = commandFixture((producer) => { + producer.stdout.write("first"); + producer.emit("exit", 0, null); + setImmediate(() => { + producer.stdout.end("last"); + producer.stderr.end("diagnostic"); + producer.emit("close", 0, null); + }); + }); + await expect(runWakaruCommand(context)).resolves.toMatchObject({ + exit_code: 0, + signal: null, + stdout: "firstlast", + stderr: "diagnostic", + }); +}); + +it("checks the diagnostic byte limit against output delivered after exit", async () => { + const { context } = commandFixture((producer) => { + producer.emit("exit", 0, null); + setImmediate(() => { + producer.stderr.end(Buffer.alloc(RECOVERY_LIMITS.reportBytes + 1, 97)); + producer.stdout.end(); + producer.emit("close", 0, null); + }); + }); + await expect(runWakaruCommand(context)).rejects.toMatchObject({ + _tag: "AnalysisOutputError", + operation: "recover_javascript_sources", + }); +}); + +it("enforces the shared deadline while exited output streams remain open", async () => { + const { context, producer } = commandFixture((process) => { + process.emit("exit", 0, null); + }); + try { + await expect( + runWakaruCommand({ ...context, deadline: Date.now() + 50 }), + ).rejects.toMatchObject({ + _tag: "AnalysisTimeoutError", + operation: "recover_javascript_sources", + }); + expect(producer.stderr.listenerCount("data")).toBe(0); + } finally { + producer.stdout.destroy(); + producer.stderr.destroy(); + } +}); + +it("honors cancellation while waiting for output after exit", async () => { + const controller = new AbortController(); + const { context, producer } = commandFixture((process) => { + process.emit("exit", 0, null); + setImmediate(() => controller.abort()); + }); + try { + await expect( + runWakaruCommand({ ...context, signal: controller.signal }), + ).rejects.toMatchObject({ + _tag: "AnalysisCancelledError", + }); + expect(producer.stderr.listenerCount("data")).toBe(0); + } finally { + producer.stdout.destroy(); + producer.stderr.destroy(); + } +}); diff --git a/src/javascript/recovery/WakaruCommand.ts b/src/javascript/recovery/WakaruCommand.ts index bed1b47fd..ad90338ef 100644 --- a/src/javascript/recovery/WakaruCommand.ts +++ b/src/javascript/recovery/WakaruCommand.ts @@ -132,7 +132,7 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => { ); let failure: unknown; try { - while (!(await supervisor.waitForExit(100))) { + while (!(await supervisor.waitForOutputClose(100))) { checkRecoveryDeadline(context.deadline, context.signal); if (oversized) throw new AnalysisOutputError( diff --git a/tests/boundary/providers/hopper/linuxPrivateDisplay.test.ts b/tests/boundary/providers/hopper/linuxPrivateDisplay.test.ts index bb309d63e..b4e378f82 100644 --- a/tests/boundary/providers/hopper/linuxPrivateDisplay.test.ts +++ b/tests/boundary/providers/hopper/linuxPrivateDisplay.test.ts @@ -220,3 +220,24 @@ describe("Linux private display selection", () => { expect(processes).not.toContain(hangingHelperPath); }); }); + +describe("private display probe output settlement", () => { + it.skipIf(process.platform === "win32")( + "retains inherited diagnostics written after the launcher exits", + async () => { + const result = await runLinuxPrivateDisplayProbe("direct", { + helperPath: fileURLToPath( + new URL("../../../fixtures/x11ProbeOutput.py", import.meta.url), + ), + timeoutMs: 5_000, + }); + expect(result).toEqual({ + outcome: "exited", + exitCode: 0, + stderr: "firstlast", + stderrBytes: Buffer.byteLength("firstlast"), + cleanupIncomplete: false, + }); + }, + ); +}); diff --git a/tests/fixtures/x11ProbeOutput.py b/tests/fixtures/x11ProbeOutput.py new file mode 100644 index 000000000..f762c9506 --- /dev/null +++ b/tests/fixtures/x11ProbeOutput.py @@ -0,0 +1,9 @@ +import subprocess +import sys + +sys.stderr.write("first") +sys.stderr.flush() +subprocess.Popen( + [sys.executable, "-c", "import sys, time; time.sleep(0.05); sys.stderr.write('last'); sys.stderr.flush()"], + stdout=subprocess.DEVNULL, +)