diff --git a/extensions/imessage/src/client.test.ts b/extensions/imessage/src/client.test.ts new file mode 100644 index 00000000000..50c49c341f4 --- /dev/null +++ b/extensions/imessage/src/client.test.ts @@ -0,0 +1,115 @@ +// iMessage tests cover the RPC client child-process stream error handling. +import { EventEmitter } from "node:events"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const spawnMock = vi.hoisted(() => vi.fn()); + +vi.mock("node:child_process", () => ({ + spawn: spawnMock, +})); + +// A dead imsg helper can emit an async `error` on any of its stdio streams. On +// a raw EventEmitter an unhandled `error` throws synchronously, which in the +// real gateway surfaces as an uncaughtException and crashes the process (#75438 +// covered stdin only). The mock child mirrors that stdio shape so we can assert +// each stream's `error` is caught and routed to failAll. +type MockChild = EventEmitter & { + stdout: EventEmitter; + stderr: EventEmitter; + stdin: EventEmitter & { + write: (line: string, cb?: (err?: Error | null) => void) => boolean; + end: () => void; + }; + killed: boolean; + kill: ReturnType; +}; + +function createMockChild(): MockChild { + const child = new EventEmitter() as MockChild; + child.stdout = new EventEmitter(); + child.stderr = new EventEmitter(); + const stdin = new EventEmitter() as MockChild["stdin"]; + // Resolve every write cleanly so the pending request only settles via the + // stream error path under test. + stdin.write = (_line, cb) => { + cb?.(null); + return true; + }; + stdin.end = () => {}; + child.stdin = stdin; + child.killed = false; + child.kill = vi.fn(() => { + child.killed = true; + return true; + }); + return child; +} + +describe("IMessageRpcClient child stream error handling", () => { + let child: MockChild; + + beforeEach(() => { + // start() refuses to spawn under a test env; clear the markers so the real + // spawn/listener wiring runs against the mock child. + vi.stubEnv("NODE_ENV", "development"); + vi.stubEnv("VITEST", ""); + child = createMockChild(); + spawnMock.mockReset().mockReturnValue(child); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + vi.resetModules(); + }); + + it.each(["stdout", "stderr", "stdin"] as const)( + "catches a %s stream error and rejects in-flight requests instead of crashing", + async (streamName) => { + const { IMessageRpcClient } = await import("./client.js"); + const client = new IMessageRpcClient({ cliPath: "imsg" }); + await client.start(); + + const pending = client.request("ping", {}, { timeoutMs: 0 }); + // Keep the rejection from surfacing as an unhandled rejection before we + // assert on it. + pending.catch(() => {}); + + const streamError = new Error(`${streamName} broke`); + expect(() => child[streamName].emit("error", streamError)).not.toThrow(); + + await expect(pending).rejects.toThrow(`${streamName} broke`); + await expect(client.waitForClose()).resolves.toBeUndefined(); + expect(child.kill).toHaveBeenCalledOnce(); + expect(child.kill).toHaveBeenCalledWith("SIGTERM"); + + await client.stop(); + }, + ); + + it("settles the client after a real child stdout stream failure", async () => { + const childProcess = + await vi.importActual("node:child_process"); + const realChild = childProcess.spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { + stdio: ["pipe", "pipe", "pipe"], + }); + spawnMock.mockReturnValueOnce(realChild); + const { IMessageRpcClient } = await import("./client.js"); + const client = new IMessageRpcClient({ cliPath: "imsg" }); + await client.start(); + + try { + const pending = client.request("ping", {}, { timeoutMs: 0 }); + pending.catch(() => {}); + realChild.stdout.destroy(new Error("real stdout failure")); + + await expect(pending).rejects.toThrow("real stdout failure"); + await expect(client.waitForClose()).resolves.toBeUndefined(); + expect(realChild.killed).toBe(true); + } finally { + if (!realChild.killed) { + realChild.kill("SIGTERM"); + } + await client.stop(); + } + }); +}); diff --git a/extensions/imessage/src/client.ts b/extensions/imessage/src/client.ts index ff55c822a3e..285e3577ec6 100644 --- a/extensions/imessage/src/client.ts +++ b/extensions/imessage/src/client.ts @@ -118,23 +118,29 @@ export class IMessageRpcClient { } }); - child.on("error", (err) => { - this.failAll(err instanceof Error ? err : new Error(String(err))); - this.closedResolve?.(); - }); - - // Without this listener, async EPIPE from a dead child crashes the - // gateway via uncaughtException. (#75438) - child.stdin.on("error", (err) => { - this.failAll(err instanceof Error ? err : new Error(String(err))); - }); + // Every process/stdio error is terminal for this RPC transport. Settle the + // client once and terminate a helper whose pipe failed; otherwise the + // monitor can wait forever on an unusable child. #75438 covered stdin only. + const failFromProcessError = (err: unknown) => { + if (!this.finish(err instanceof Error ? err : new Error(String(err)))) { + return; + } + try { + child.kill("SIGTERM"); + } catch { + // The helper may already be gone. + } + }; + child.on("error", failFromProcessError); + child.stdin.on("error", failFromProcessError); + child.stdout.on("error", failFromProcessError); + child.stderr.on("error", failFromProcessError); child.on("close", (code, signal) => { if (this.child === child) { this.flushStdoutBuffer(); } - this.failAll(this.buildCloseError(code, signal)); - this.closedResolve?.(); + this.finish(this.buildCloseError(code, signal)); }); } @@ -328,6 +334,17 @@ export class IMessageRpcClient { this.pending.delete(key); } } + + private finish(err: Error): boolean { + const resolve = this.closedResolve; + if (!resolve) { + return false; + } + this.closedResolve = null; + this.failAll(err); + resolve(); + return true; + } } export async function createIMessageRpcClient(