diff --git a/.changeset/exec-output-after-exit.md b/.changeset/exec-output-after-exit.md new file mode 100644 index 000000000..b8a37068e --- /dev/null +++ b/.changeset/exec-output-after-exit.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computerd": patch +--- + +`computerd` no longer exits when a command's output arrives after the command has exited. diff --git a/packages/computerd/src/exec/runner.test.ts b/packages/computerd/src/exec/runner.test.ts index 0d94c7e38..6d317c394 100644 --- a/packages/computerd/src/exec/runner.test.ts +++ b/packages/computerd/src/exec/runner.test.ts @@ -52,6 +52,23 @@ function decode(value: Uint8Array): string { return new TextDecoder().decode(value); } +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +// Records errors that escape to the process. A throw inside a child +// stream's `data` listener lands here rather than in the test, and in +// computerd the same throw reaches the logger's handler and exits the +// daemon. +function captureUncaught(): { errors: unknown[]; stop: () => void } { + const errors: unknown[] = []; + const onError = (error: unknown): void => { + errors.push(error); + }; + process.on("uncaughtException", onError); + return { errors, stop: () => process.off("uncaughtException", onError) }; +} + test("exec captures stdout and propagates exit code", async () => { const { runner, dispose } = fixture(); try { @@ -572,3 +589,117 @@ test("rejects a shell that is not an absolute path", () => { storage.close?.(); } }); + +// A shell that starts a background job hands that job its stdout and +// stderr. The shell's exit then arrives while the pipes are still open, +// and the job can write to them at any later time. + +test("output from a background job after the shell exits does not escape the runner", async () => { + const uncaught = captureUncaught(); + const { runner, dispose } = fixture(); + try { + const handle = runner.exec("echo start; (sleep 0.6; echo late) & exit 0", { id: "bg" }); + const events = await drain(handle.events); + expect(events.at(-1)?.name).toBe("exit"); + await sleep(900); + expect(uncaught.errors).toEqual([]); + } finally { + uncaught.stop(); + dispose(); + } +}); + +test("an exec does not wait for a background job that keeps its output open", async () => { + const { runner, dispose } = fixture(); + try { + const start = Date.now(); + const handle = runner.exec("(sleep 2; echo late) & exit 0", { id: "bg-wait" }); + const events = await drain(handle.events); + expect(events.find((event) => event.name === "exit")?.code).toBe(0); + expect(Date.now() - start).toBeLessThan(1_500); + } finally { + dispose(); + } +}); + +test("output that arrives after exit but before the pipes close is delivered before the exit event", async () => { + const uncaught = captureUncaught(); + const { runner, dispose } = fixture({ exitDrainMs: 2_000 }); + try { + const handle = runner.exec("(sleep 0.1; echo late) & exit 0", { id: "bg-drain" }); + const events = await drain(handle.events); + const stdout = events + .filter((event) => event.name === "stdout") + .map((event) => decode(event.value as Uint8Array)) + .join(""); + expect(stdout).toBe("late\n"); + expect(events.at(-1)?.name).toBe("exit"); + expect(uncaught.errors).toEqual([]); + } finally { + uncaught.stop(); + dispose(); + } +}); + +test("output held back by a slow reader is delivered in full after the child exits", async () => { + // The live stream pauses the child's pipes as soon as its queue is + // full, so a reader that starts late leaves output buffered in the + // runner when the child exits. + const uncaught = captureUncaught(); + const { runner, dispose } = fixture(); + try { + const size = 256 * 1024; + const handle = runner.exec(`head -c ${size} /dev/zero`, { id: "slow-reader" }); + await sleep(300); + const events = await drain(handle.events); + const bytes = events + .filter((event) => event.name === "stdout") + .reduce((total, event) => total + (event.value as Uint8Array).byteLength, 0); + expect(bytes).toBe(size); + expect(events.at(-1)?.name).toBe("exit"); + expect(uncaught.errors).toEqual([]); + } finally { + uncaught.stop(); + dispose(); + } +}); + +test("output from a background job after dispose() does not escape the runner", async () => { + const uncaught = captureUncaught(); + const { runner, dispose } = fixture(); + try { + const handle = runner.exec("(sleep 0.6; echo late) & exit 0", { id: "bg-dispose" }); + await drain(handle.events); + runner.dispose("bg-dispose"); + await sleep(900); + expect(uncaught.errors).toEqual([]); + const next = await drain(runner.exec("echo still-here").events); + expect(next.find((event) => event.name === "exit")?.code).toBe(0); + } finally { + uncaught.stop(); + dispose(); + } +}); + +test("an exec finishes at exitDrainMaxMs when its reader stops reading", async () => { + // The reader takes one event and then neither reads nor cancels. The + // background job's output fills the live stream's queue, which pauses + // the pipes, and the job keeps them open, so "close" never arrives. + const uncaught = captureUncaught(); + const { runner, dispose } = fixture({ exitDrainMaxMs: 500 }); + try { + const handle = runner.exec("echo first; (sleep 0.1; head -c 4000000 /dev/zero) & exit 0", { + id: "stalled", + }); + const reader = handle.events.getReader(); + await reader.read(); + await sleep(1_500); + const replay = await drain(runner.get("stalled", { after: 0 }).events); + expect(replay.at(-1)).toMatchObject({ name: "exit", code: 0 }); + expect(uncaught.errors).toEqual([]); + reader.releaseLock(); + } finally { + uncaught.stop(); + dispose(); + } +}); diff --git a/packages/computerd/src/exec/runner.ts b/packages/computerd/src/exec/runner.ts index 6904e8d7c..e80017dfe 100644 --- a/packages/computerd/src/exec/runner.ts +++ b/packages/computerd/src/exec/runner.ts @@ -37,6 +37,8 @@ interface ExecRecord { timeoutTimer?: NodeJS.Timeout; killTimer?: NodeJS.Timeout; heartbeatTimer?: NodeJS.Timeout; + drainTimer?: NodeJS.Timeout; + drainDeadlineTimer?: NodeJS.Timeout; } interface LiveSubscriber { @@ -61,6 +63,12 @@ const DEFAULTS = { // After SIGTERM, give the child this long to exit on its own // before sending SIGKILL. killGraceMs: 5_000, + // After the child exits, wait this long for its stdout and stderr to + // close before finishing the exec. See `RunnerOptions.exitDrainMs`. + exitDrainMs: 250, + // Finish the exec this long after the child exits even if the reader + // has stopped reading. See `RunnerOptions.exitDrainMaxMs`. + exitDrainMaxMs: 5_000, shell: "/bin/sh", } as const; @@ -81,6 +89,8 @@ export class Runner { sweepIntervalMs: number; defaultTimeoutMs: number; heartbeatIntervalMs: number; + exitDrainMs: number; + exitDrainMaxMs: number; shell: string; now: () => number; }; @@ -98,6 +108,8 @@ export class Runner { sweepIntervalMs: init.sweepIntervalMs ?? DEFAULTS.sweepIntervalMs, defaultTimeoutMs: init.defaultTimeoutMs ?? DEFAULTS.defaultTimeoutMs, heartbeatIntervalMs: init.heartbeatIntervalMs ?? 0, + exitDrainMs: init.exitDrainMs ?? DEFAULTS.exitDrainMs, + exitDrainMaxMs: init.exitDrainMaxMs ?? DEFAULTS.exitDrainMaxMs, shell: init.shell ?? DEFAULTS.shell, now: init.now ?? Date.now, }; @@ -176,8 +188,22 @@ export class Runner { this.records.set(id, record); const onData = (name: "stdout" | "stderr") => (chunk: Buffer) => { + // A background job the command started inherits these pipes and + // can write to them after the exec has finished or been disposed. + // Its output has no subscriber and no log to go to, so it is + // dropped. This listener runs outside any caller, so a throw here + // would reach the process's uncaughtException handler, which + // exits the daemon. + if (!record.live) return; const value = new Uint8Array(chunk); - const seq = log.append(name, value); + let seq: number; + try { + seq = log.append(name, value); + } catch (err) { + record.subscriber?.error(err instanceof Error ? err : new Error(String(err))); + record.subscriber = undefined; + return; + } record.lastOutputAt = this.opts.now(); record.subscriber?.enqueue({ id, seq, name, value }); }; @@ -223,6 +249,19 @@ export class Runner { clearTimeout(record.heartbeatTimer); record.heartbeatTimer = undefined; } + if (record.drainTimer !== undefined) { + clearTimeout(record.drainTimer); + record.drainTimer = undefined; + } + if (record.drainDeadlineTimer !== undefined) { + clearTimeout(record.drainDeadlineTimer); + record.drainDeadlineTimer = undefined; + } + // Keep reading whatever a background job still writes, and drop + // it in onData. Left paused, its pipe fills and the job blocks on + // its next write. Closing the pipe instead would kill it with + // SIGPIPE. + flowStdio(record.child); const exitCode = mapExitCode(code, signal); let seq: number; try { @@ -244,7 +283,43 @@ export class Runner { record.subscriber = undefined; this.scheduleSweep(); }; - child.once("exit", finalise); + // Finish the exec on "close", which Node emits once the child has + // exited and its stdout and stderr have both ended, so every byte it + // wrote is logged before the exit event. A background job that holds + // the pipes open delays "close" for as long as it runs, so "exit" + // starts a timer that finishes the exec without waiting for it. The + // timer waits while the subscriber has paused the pipes: their + // buffered output is still to come, and a reader that is keeping up + // will ask for it shortly. A reader that has stopped reading never + // will, so a second timer finishes the exec regardless, keeping what + // Node has already read from the pipes. + const drainAfterExit = (code: number | null, signal: NodeJS.Signals | null): void => { + const finish = (): void => { + if (!record.live) return; + readBufferedStdio(child); + finalise(code, signal); + }; + const waitForReader = (): void => { + record.drainTimer = setTimeout(() => { + record.drainTimer = undefined; + if (record.subscriber?.paused === true) { + waitForReader(); + return; + } + finish(); + }, this.opts.exitDrainMs); + }; + waitForReader(); + record.drainDeadlineTimer = setTimeout( + () => { + record.drainDeadlineTimer = undefined; + finish(); + }, + Math.max(this.opts.exitDrainMs, this.opts.exitDrainMaxMs), + ); + }; + child.once("exit", drainAfterExit); + child.once("close", finalise); child.once("error", (err) => { if (!record.live) return; const value = new TextEncoder().encode(`spawn failed: ${err.message}\n`); @@ -441,6 +516,15 @@ export class Runner { clearTimeout(record.heartbeatTimer); record.heartbeatTimer = undefined; } + if (record.drainTimer !== undefined) { + clearTimeout(record.drainTimer); + record.drainTimer = undefined; + } + if (record.drainDeadlineTimer !== undefined) { + clearTimeout(record.drainDeadlineTimer); + record.drainDeadlineTimer = undefined; + } + flowStdio(record.child); try { if (record.child.exitCode === null && record.child.signalCode === null) { record.child.kill("SIGKILL"); @@ -503,6 +587,26 @@ function shellQuote(s: string): string { return `'${s.replace(/'/g, "'\\''")}'`; } +// Hand the output Node has already read from the child's pipes to the +// `data` listeners, which log it and pass it to the live subscriber. +// `read()` emits `data` for each chunk it returns, paused or not. +// Output still in the pipes is not waited for. +function readBufferedStdio(child: ChildProcess): void { + for (const stream of [child.stdout, child.stderr]) { + if (stream === null) continue; + while (stream.read() !== null) { + // Each chunk has gone to the listeners. + } + } +} + +// Resume the child's stdout and stderr after the exec is over, so a +// background job still writing to them never blocks on a full pipe. +function flowStdio(child: ChildProcess): void { + child.stdout?.resume(); + child.stderr?.resume(); +} + function mapExitCode(code: number | null, signal: NodeJS.Signals | null): number { if (code !== null) return code; if (signal === "SIGTERM") return 143; diff --git a/packages/computerd/src/exec/types.ts b/packages/computerd/src/exec/types.ts index 5e222392d..9c352b36d 100644 --- a/packages/computerd/src/exec/types.ts +++ b/packages/computerd/src/exec/types.ts @@ -62,6 +62,20 @@ export interface RunnerOptions { // Emit a heartbeat event every this many milliseconds while a child // process is alive. When unset or zero, no heartbeat events are emitted. heartbeatIntervalMs?: number; + // How long to wait, after the child exits, for its stdout and stderr to + // close before finishing the exec. They normally close right after the + // exit. A background job the command started keeps them open, and the + // exec finishes after this many milliseconds instead of waiting for the + // job. Output the job writes later is dropped. The wait is extended + // while the reader of the live stream has paused the output, up to + // exitDrainMaxMs. Default 250. + exitDrainMs?: number; + // The longest the exec waits after the child exits, however slowly its + // reader is reading. At this point it keeps the output Node has already + // read, drops the rest, and finishes, so a reader that stops without + // cancelling cannot hold the exec open. Never less than exitDrainMs. + // Default 5000. + exitDrainMaxMs?: number; // Absolute path to the interpreter each command runs under. Defaults to // /bin/sh, which on a Debian-family image is dash: bash-only syntax a // caller may reach for, notably ${PIPESTATUS[@]}, is a parse error there