-
Notifications
You must be signed in to change notification settings - Fork 1
fix(cli): flush NDJSON before forced exit #95
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| import { Log } from "../util/log" | ||
| import { Stdout } from "./stdout" | ||
|
|
||
| export namespace Shutdown { | ||
| export async function flush() { | ||
| await Stdout.flush().catch((error) => { | ||
| Log.Default.error("stdout flush failed", { | ||
| error: error instanceof Error ? error.message : error, | ||
| }) | ||
| process.exitCode = 1 | ||
| }) | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| const state = { | ||
| bound: false, | ||
| closed: false, | ||
| pending: new Set<Promise<void>>(), | ||
| } | ||
| let failure: unknown | ||
|
|
||
| function epipe(error: unknown) { | ||
| return error instanceof Error && "code" in error && error.code === "EPIPE" | ||
| } | ||
|
|
||
| function fail(error: unknown) { | ||
| if (epipe(error)) { | ||
| state.closed = true | ||
| return false | ||
| } | ||
| failure = error | ||
| return true | ||
| } | ||
|
|
||
| function bind() { | ||
| if (state.bound) return | ||
| state.bound = true | ||
| process.stdout.on("error", fail) | ||
| } | ||
|
|
||
| export namespace Stdout { | ||
| export function write(chunk: string) { | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 write() never rejects while flush() throws. 🤖 Fix with your agentWhy this mattersStdout.write returns a Promise that ALWAYS resolves: the write callback stores any error into state.error via fail() and then resolves unconditionally. Errors only surface when flush() is later called and throws. This is asymmetric with flush()'s throw-on-error contract and breaks the usual Promise intuition — callers who export function write(chunk: string) {
bind()
if (state.closed || process.stdout.destroyed || process.stdout.writableEnded) return Promise.resolve()
const pending = new Promise<void>((resolve) => {
process.stdout.write(chunk, (error) => {
if (error) fail(error)
resolve()
})
})
state.pending.add(pending)
pending.then(() => state.pending.delete(pending))
return pending
} |
||
| bind() | ||
| if (failure !== undefined) { | ||
| const pending = Promise.reject(failure) | ||
| pending.catch(() => {}) | ||
| return pending | ||
| } | ||
| if (state.closed || process.stdout.destroyed || process.stdout.writableEnded) return Promise.resolve() | ||
| const pending = new Promise<void>((resolve, reject) => { | ||
| process.stdout.write(chunk, (error) => { | ||
| if (error && fail(error)) { | ||
| reject(error) | ||
| return | ||
| } | ||
| resolve() | ||
| }) | ||
| }) | ||
| state.pending.add(pending) | ||
| pending.catch(() => {}) | ||
| pending.then( | ||
| () => state.pending.delete(pending), | ||
| () => state.pending.delete(pending), | ||
| ) | ||
| return pending | ||
| } | ||
|
|
||
| export async function flush() { | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟠 Signal exits bypass flush(); NDJSON still truncates. 🤖 Fix with your agentWhy this matters
export namespace Stdout {
export function write(chunk: string) {
state.queue = state.queue.then(() => output(chunk))
return state.queue
}
export async function flush() {
await state.queue
await output("", true)
}
}There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 flush() throwing makes it unsafe for finally blocks. 🤖 Fix with your agentWhy this mattersflush() is intended to be awaited in the finally of both entrypoints before process.exit(). By throwing export async function flush() {
while (state.pending.size) await Promise.all(state.pending)
if (state.error) throw state.error
} |
||
| while (state.pending.size) await Promise.allSettled(state.pending) | ||
| if (failure !== undefined) throw failure | ||
| } | ||
|
|
||
| export function isClosed() { | ||
| return state.closed | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -29,6 +29,7 @@ import path from "path" | |
| import { Global } from "./global" | ||
| import { JsonMigration } from "./storage/json-migration" | ||
| import { Database } from "./storage/db" | ||
| import { Shutdown } from "./cli/shutdown" | ||
|
|
||
| process.on("unhandledRejection", (e) => { | ||
| Log.Default.error("rejection", { | ||
|
|
@@ -212,5 +213,6 @@ try { | |
| // Most notably, some docker-container-based MCP servers don't handle such signals unless | ||
| // run using `docker run --init`. | ||
| // Explicitly exit to avoid any hanging subprocesses. | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 Flush-catch block duplicated across both entrypoints. 🤖 Fix with your agentWhy this mattersThe exact 6-line // run using `docker run --init`.
// Explicitly exit to avoid any hanging subprocesses.
await Stdout.flush().catch((error) => {
Log.Default.error("stdout flush failed", {
error: error instanceof Error ? error.message : error,
})
process.exitCode = 1
})
process.exit()
} |
||
| await Shutdown.flush() | ||
| process.exit() | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| import { Stdout } from "../../../src/cli/stdout" | ||
| import { Shutdown } from "../../../src/cli/shutdown" | ||
|
|
||
| if (process.argv.includes("--idle")) process.exit(process.stdout.listenerCount("error")) | ||
| if (process.argv.includes("--shutdown-error")) { | ||
| Stdout.write("") | ||
| await Stdout.flush() | ||
| process.stdout.emit("error", Object.assign(new Error("broken stdout"), { code: "EIO" })) | ||
| await Shutdown.flush() | ||
| const status = process.argv.find((arg) => arg.startsWith("--status="))?.slice("--status=".length) | ||
| if (status) await Bun.write(status, String(process.exitCode)) | ||
| process.exit() | ||
| } | ||
| if (process.argv.includes("--error")) { | ||
| Stdout.write("") | ||
| await Stdout.flush() | ||
| process.stdout.emit("error", Object.assign(new Error("broken stdout"), { code: "EIO" })) | ||
| const write = await Stdout.write("").then( | ||
| () => false, | ||
| (error) => error instanceof Error && error.message === "broken stdout", | ||
| ) | ||
| const flush = await Stdout.flush().then( | ||
| () => false, | ||
| (error) => error instanceof Error && error.message === "broken stdout", | ||
| ) | ||
| process.exit(write && flush ? 0 : 2) | ||
| } | ||
|
|
||
| const records = Array.from({ length: 1024 }, (_, index) => | ||
| JSON.stringify({ | ||
| type: "data", | ||
| index, | ||
| value: "x".repeat(2048), | ||
| }), | ||
| ) | ||
|
|
||
| Array.from({ length: process.argv.includes("--epipe") ? 8 : 1 }).forEach(() => | ||
| records.forEach((record) => Stdout.write(record + "\n")), | ||
| ) | ||
| Stdout.write(JSON.stringify({ type: "complete" }) + "\n") | ||
| await Stdout.flush() | ||
| if (process.argv.includes("--epipe")) { | ||
| const status = process.argv.find((arg) => arg.startsWith("--status="))?.slice("--status=".length) | ||
| if (status) await Bun.write(status, Stdout.isClosed() ? "closed" : "open") | ||
| process.exit(Stdout.isClosed() ? 0 : 2) | ||
| } | ||
| process.exit() |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| import { describe, expect, test } from "bun:test" | ||
| import { $ } from "bun" | ||
| import path from "path" | ||
| import { tmpdir } from "../fixture/fixture" | ||
|
|
||
| describe("stdout", () => { | ||
| test("flushes NDJSON beyond pipe capacity before forced exit", async () => { | ||
| const child = Bun.spawn([process.execPath, path.join(import.meta.dir, "fixture", "stdout.ts")], { | ||
| cwd: path.join(import.meta.dir, "../.."), | ||
| stdout: "pipe", | ||
| stderr: "pipe", | ||
| }) | ||
| const output = new Response(child.stdout).text() | ||
| const error = new Response(child.stderr).text() | ||
|
|
||
| expect(await child.exited).toBe(0) | ||
| expect(await error).toBe("") | ||
|
|
||
| const lines = (await output) | ||
| .trim() | ||
| .split("\n") | ||
| .map((line) => JSON.parse(line)) | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 EPIPE test shells out to bash+head, unlike sibling tests. 🤖 Fix with your agentWhy this mattersThe "treats a closed output pipe as controlled" test wraps the fixture in test("treats a closed output pipe as controlled", async () => {
const child = Bun.spawn(
[
"bash",
"-c",
'set -o pipefail; "$1" "$2" --epipe | head -c 1 >/dev/null',
"--",
process.execPath,
path.join(import.meta.dir, "fixture", "stdout.ts"),
],
{
cwd: path.join(import.meta.dir, "../.."),
stdout: "ignore",
stderr: "pipe",
},
) |
||
| expect(lines).toHaveLength(1025) | ||
| expect(lines.at(-1)).toEqual({ type: "complete" }) | ||
| }) | ||
|
|
||
| test("treats a closed output pipe as controlled", async () => { | ||
| await using tmp = await tmpdir() | ||
| const status = path.join(tmp.path, "status") | ||
| const child = await $`${process.execPath} ${path.join( | ||
| import.meta.dir, | ||
| "fixture", | ||
| "stdout.ts", | ||
| )} --epipe --status=${status} | ${process.execPath} -e ${"await Bun.stdin.stream().getReader().read(); process.exit()"}` | ||
| .cwd(path.join(import.meta.dir, "../..")) | ||
| .quiet() | ||
| .nothrow() | ||
|
|
||
| expect(child.exitCode).toBe(0) | ||
| expect(child.stderr.toString()).toBe("") | ||
| expect(await Bun.file(status).text()).toBe("closed") | ||
| }) | ||
|
|
||
| test("does not bind an error handler until output is written", async () => { | ||
| const child = Bun.spawn([process.execPath, path.join(import.meta.dir, "fixture", "stdout.ts"), "--idle"], { | ||
| cwd: path.join(import.meta.dir, "../.."), | ||
| stdout: "pipe", | ||
| stderr: "pipe", | ||
| }) | ||
|
|
||
| expect(await child.exited).toBe(0) | ||
| expect(await new Response(child.stderr).text()).toBe("") | ||
| }) | ||
|
|
||
| test("surfaces unexpected output errors from write and flush", async () => { | ||
| const child = Bun.spawn([process.execPath, path.join(import.meta.dir, "fixture", "stdout.ts"), "--error"], { | ||
| cwd: path.join(import.meta.dir, "../.."), | ||
| stdout: "pipe", | ||
| stderr: "pipe", | ||
| }) | ||
|
|
||
| expect(await child.exited).toBe(0) | ||
| expect(await new Response(child.stderr).text()).toBe("") | ||
| }) | ||
|
|
||
| test("flush failure sets exit status without blocking forced exit", async () => { | ||
| await using tmp = await tmpdir() | ||
| const status = path.join(tmp.path, "status") | ||
| const child = Bun.spawn( | ||
| [process.execPath, path.join(import.meta.dir, "fixture", "stdout.ts"), "--shutdown-error", `--status=${status}`], | ||
| { | ||
| cwd: path.join(import.meta.dir, "../.."), | ||
| stdout: "pipe", | ||
| stderr: "pipe", | ||
| }, | ||
| ) | ||
|
|
||
| expect(await child.exited).toBe(1) | ||
| expect(await Bun.file(status).text()).toBe("1") | ||
| expect(await new Response(child.stderr).text()).toBe("") | ||
| }) | ||
| }) | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟡 state.queue grows unbounded under backpressure.
🤖 Fix with your agent
Why this matters
Stdout.write()reassignsstate.queue = state.queue.then(() => output(chunk))on every call; each.thenclosure captureschunk.output()is fully serialized (one in-flight write at a time), so when the downstream pipe applies backpressure (slow consumer, e.g.aictrl run --format=json | pv -L 1k), writes accumulate as pending closures instate.queue, each retaining its full chunk string. There is no high-watermark, no drop policy, and no backpressure signal back toemit()callers —emit()returnstruesynchronously and the producer keeps calling. Long runs against a slow pipe can grow memory without bound beforeflush()runs. The previous directprocess.stdout.writehad the same fundamental issue but resided in libuv's buffer; this PR duplicates the retention in a JS Promise chain on top.