mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-27 11:42:52 +00:00
fix(server): keep daemon alive when JSONL RPC stdin hits EPIPE (#4048)
* fix(server): keep daemon alive when JSONL RPC stdin hits EPIPE A dead Claude/OMP child currently turns stdin.write() EPIPE into uncaughtException and kills the daemon. Catch the write, listen for stdin errors, and close that transport instead. Closes #4047 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * test(server): reproduce JSONL RPC stdin EPIPE * test(server): gate native EPIPE fixture on POSIX --------- Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com> Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>
This commit is contained in:
co-authored by
Claude Sonnet 5
Mohamed Boudra
parent
63923fdd07
commit
74a377ff6a
@@ -54,6 +54,14 @@ readline.createInterface({ input: process.stdin, crlfDelay: Infinity }).on("line
|
||||
});
|
||||
`;
|
||||
|
||||
const CLOSED_STDIN_CHILD_SOURCE = String.raw`
|
||||
const fs = require("node:fs");
|
||||
|
||||
fs.closeSync(0);
|
||||
process.stdout.write(JSON.stringify({ type: "stdin_closed" }) + "\n");
|
||||
setInterval(() => {}, 1_000);
|
||||
`;
|
||||
|
||||
interface InMemoryChildProcess extends ChildProcessWithoutNullStreams {
|
||||
stdin: PassThrough;
|
||||
stdout: PassThrough;
|
||||
@@ -63,6 +71,7 @@ interface InMemoryChildProcess extends ChildProcessWithoutNullStreams {
|
||||
interface StartProcessOptions {
|
||||
child?: ChildProcessWithoutNullStreams;
|
||||
defaultRequestTimeoutMs?: number;
|
||||
source?: string;
|
||||
}
|
||||
|
||||
function createInMemoryChildProcess(): InMemoryChildProcess {
|
||||
@@ -85,7 +94,7 @@ function startProcess(options: StartProcessOptions = {}): JsonlRpcProcess {
|
||||
return new JsonlRpcProcess({
|
||||
launch: {
|
||||
command: process.execPath,
|
||||
args: ["-e", CHILD_SOURCE, "--", "resolved-arg"],
|
||||
args: ["-e", options.source ?? CHILD_SOURCE, "--", "resolved-arg"],
|
||||
cwd: process.cwd(),
|
||||
env: { JSONL_RPC_TEST_VALUE: "resolved-env" },
|
||||
},
|
||||
@@ -222,6 +231,70 @@ describe("JsonlRpcProcess", () => {
|
||||
await rejection;
|
||||
});
|
||||
|
||||
test("closes the transport when a direct send synchronously throws EPIPE", async () => {
|
||||
const child = createInMemoryChildProcess();
|
||||
const transport = startProcess({ child });
|
||||
|
||||
child.stdin.write = () => {
|
||||
throw Object.assign(new Error("write EPIPE"), { code: "EPIPE" });
|
||||
};
|
||||
|
||||
expect(() => transport.send({ type: "notice" })).not.toThrow();
|
||||
await expect(transport.request({ type: "echo", value: "after" })).rejects.toThrow(
|
||||
"JSONL RPC process is closed",
|
||||
);
|
||||
});
|
||||
|
||||
// Closing fd 0 does not sever the inherited pipe on Windows, so this fixture cannot produce EPIPE.
|
||||
test.skipIf(process.platform === "win32")(
|
||||
"keeps the parent alive when a real child closes its stdin pipe",
|
||||
async () => {
|
||||
const transport = startProcess({ source: CLOSED_STDIN_CHILD_SOURCE });
|
||||
const stdinClosed = new Promise<void>((resolve) => {
|
||||
const unsubscribe = transport.onMessage((message) => {
|
||||
if (message.type !== "stdin_closed") return;
|
||||
unsubscribe();
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
|
||||
await stdinClosed;
|
||||
await expect(transport.request({ type: "echo", value: "after" })).rejects.toThrow(
|
||||
/EPIPE|stdin is not writable/,
|
||||
);
|
||||
await expect(transport.request({ type: "echo", value: "later" })).rejects.toThrow(
|
||||
"JSONL RPC process is closed",
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
test("stdin error events close the transport instead of becoming uncaught exceptions", async () => {
|
||||
const child = createInMemoryChildProcess();
|
||||
const transport = startProcess({ child });
|
||||
const request = transport.request({ type: "hang" });
|
||||
const err = Object.assign(new Error("write EPIPE"), { code: "EPIPE" });
|
||||
|
||||
child.stdin.emit("error", err);
|
||||
|
||||
await expect(request).rejects.toThrow("write EPIPE");
|
||||
await expect(transport.request({ type: "echo", value: "after" })).rejects.toThrow(
|
||||
"JSONL RPC process is closed",
|
||||
);
|
||||
});
|
||||
|
||||
test("a non-writable stdin closes the transport instead of hanging the request", async () => {
|
||||
const child = createInMemoryChildProcess();
|
||||
const transport = startProcess({ child });
|
||||
child.stdin.end();
|
||||
|
||||
await expect(transport.request({ type: "hang" })).rejects.toThrow(
|
||||
"JSONL RPC stdin is not writable",
|
||||
);
|
||||
await expect(transport.request({ type: "echo", value: "after" })).rejects.toThrow(
|
||||
"JSONL RPC process is closed",
|
||||
);
|
||||
});
|
||||
|
||||
test("reassembles protocol v2 chunked responses into the logical frame", async () => {
|
||||
const child = createInMemoryChildProcess();
|
||||
const transport = startProcess({ child });
|
||||
|
||||
@@ -104,6 +104,9 @@ export class JsonlRpcProcess {
|
||||
this.stderrBuffer = this.stderrBuffer.slice(-STDERR_BUFFER_LIMIT);
|
||||
}
|
||||
});
|
||||
this.child.stdin.on("error", (error) => {
|
||||
this.handleStdinError(error);
|
||||
});
|
||||
this.child.on("error", (error) => {
|
||||
this.failAll(error instanceof Error ? error : new Error(String(error)));
|
||||
});
|
||||
@@ -173,10 +176,18 @@ export class JsonlRpcProcess {
|
||||
}
|
||||
|
||||
send(message: Record<string, unknown>): void {
|
||||
if (this.disposed || this.child.stdin.destroyed || !this.child.stdin.writable) {
|
||||
if (this.disposed) {
|
||||
return;
|
||||
}
|
||||
this.child.stdin.write(`${JSON.stringify(message)}\n`);
|
||||
if (this.child.stdin.destroyed || !this.child.stdin.writable) {
|
||||
this.handleStdinError(new Error(`${this.diagnosticName} stdin is not writable`));
|
||||
return;
|
||||
}
|
||||
try {
|
||||
this.child.stdin.write(`${JSON.stringify(message)}\n`);
|
||||
} catch (error) {
|
||||
this.handleStdinError(error);
|
||||
}
|
||||
}
|
||||
|
||||
async close(error = new Error(`${this.diagnosticName} process is closed`)): Promise<void> {
|
||||
@@ -243,6 +254,15 @@ export class JsonlRpcProcess {
|
||||
pending.resolve(response.data);
|
||||
}
|
||||
|
||||
private handleStdinError(error: unknown): void {
|
||||
if (this.disposed) {
|
||||
return;
|
||||
}
|
||||
const err = error instanceof Error ? error : new Error(String(error));
|
||||
this.options.logger.warn({ err }, `${this.diagnosticName} stdin write failed`);
|
||||
void this.close(err);
|
||||
}
|
||||
|
||||
private failAll(error: Error): void {
|
||||
if (this.disposed) {
|
||||
return;
|
||||
|
||||
Reference in New Issue
Block a user