Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 115 additions & 0 deletions extensions/imessage/src/client.test.ts
Original file line number Diff line number Diff line change
@@ -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<typeof vi.fn>;
};

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<typeof import("node:child_process")>("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();
}
});
});
41 changes: 29 additions & 12 deletions extensions/imessage/src/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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));
});
}

Expand Down Expand Up @@ -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(
Expand Down
Loading