diff --git a/package-lock.json b/package-lock.json index a90474d..67d7e63 100644 --- a/package-lock.json +++ b/package-lock.json @@ -23,6 +23,7 @@ "@babel/preset-typescript": "^7.27.1", "@types/jest": "^30.0.0", "@types/node": "^24.7.0", + "@types/ws": "8.18.1", "babel-jest": "^30.2.0", "jest": "^30.2.0", "jest-environment-jsdom": "^30.2.0", @@ -33,7 +34,8 @@ "typescript": "^5.9.3", "webpack": "^5.102.1", "webpack-cli": "^6.0.1", - "webpack-merge": "^6.0.1" + "webpack-merge": "^6.0.1", + "ws": "8.19.0" } }, "node_modules/@asamuzakjp/css-color": { @@ -2871,6 +2873,16 @@ "integrity": "sha512-7gqG38EyHgyP1S+7+xomFtL+ZNHcKv6DwNaCZmJmo1vgMugyF3TCnXVg4t1uk89mLNwnLtnY3TpOpCOyp1/xHQ==", "license": "MIT" }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@types/yargs": { "version": "17.0.35", "resolved": "https://registry.npmjs.org/@types/yargs/-/yargs-17.0.35.tgz", diff --git a/package.json b/package.json index 61fe4fb..3f517a8 100644 --- a/package.json +++ b/package.json @@ -37,6 +37,7 @@ "@babel/preset-typescript": "^7.27.1", "@types/jest": "^30.0.0", "@types/node": "^24.7.0", + "@types/ws": "8.18.1", "babel-jest": "^30.2.0", "jest": "^30.2.0", "jest-environment-jsdom": "^30.2.0", @@ -47,7 +48,8 @@ "typescript": "^5.9.3", "webpack": "^5.102.1", "webpack-cli": "^6.0.1", - "webpack-merge": "^6.0.1" + "webpack-merge": "^6.0.1", + "ws": "8.19.0" }, "dependencies": { "@types/uuid": "^10.0.0", diff --git a/src/v1/rpcClient.test.ts b/src/v1/rpcClient.test.ts new file mode 100644 index 0000000..2ca470c --- /dev/null +++ b/src/v1/rpcClient.test.ts @@ -0,0 +1,103 @@ +/** + * @jest-environment node + */ +import { AddressInfo } from "net"; +import { WebSocketServer } from "ws"; +import { RpcClient, RpcTimeoutError } from "./rpcClient"; + +// Echo server that answers every call immediately, like the gateway's jrpc2 server. +function startServer(onMessage?: (msg: any, socket: any) => void): Promise { + return new Promise((resolve) => { + const server: WebSocketServer = new WebSocketServer({ port: 0 }, () => resolve(server)); + server.on("connection", (socket) => + socket.on("message", (data) => { + const msg = JSON.parse(data.toString()); + if (onMessage) return onMessage(msg, socket); + socket.send(JSON.stringify({ jsonrpc: "2.0", id: msg.id, result: { accepted: true } })); + }), + ); + }); +} + +function connect(server: WebSocketServer): Promise { + const client = new RpcClient(`ws://localhost:${(server.address() as AddressInfo).port}`, { reconnect: false }); + return new Promise((resolve) => client.on("open", () => resolve(client))); +} + +// Holds back the send callback until after the reply has been processed: the +// ordering Node's TLS stack produces when the event loop stalls after a send. +function delaySendCallback(client: RpcClient) { + const socket = (client as any).socket; + const send = socket.send.bind(socket); + socket.send = (data: any, opts: any, cb: (err?: Error) => void) => send(data, opts, (err?: Error) => setTimeout(() => cb(err), 50)); +} + +const settlesWithin = (p: Promise, ms: number) => + Promise.race([ + p.then( + () => "settled", + () => "settled", + ), + new Promise((r) => setTimeout(() => r("pending"), ms)), + ]); + +describe("RpcClient", () => { + let server: WebSocketServer; + let client: RpcClient | undefined; + + afterEach(async () => { + client?.close(); + server.clients.forEach((socket) => socket.terminate()); + await new Promise((r) => server.close(r)); + }); + + test("resolves a reply that beats the send callback", async () => { + server = await startServer(); + client = await connect(server); + delaySendCallback(client); + await expect(client.call("requestOutboundConnection", {})).resolves.toEqual({ accepted: true }); + }); + + test("rejects when no reply arrives before the timeout", async () => { + server = await startServer(() => {}); + client = await connect(server); + await expect(client.call("requestOutboundConnection", {}, 100)).rejects.toThrow( + new RpcTimeoutError('"requestOutboundConnection" reply timeout after 100ms'), + ); + }); + + test.each([null, 0])("never times out when the timeout is %p, like the stock call()", async (timeout) => { + server = await startServer(() => {}); + client = await connect(server); + expect(await settlesWithin(client.call("requestOutboundConnection", {}, timeout as any), 100)).toBe("pending"); + const pending: any[] = Object.values((client as any).queue); + expect(pending).toHaveLength(1); + expect(pending[0].timeout).toBeUndefined(); + }); + + test("accepts ws options as the third argument, like the stock call()", async () => { + server = await startServer(); + client = await connect(server); + const send = jest.spyOn((client as any).socket, "send"); + await expect(client.call("requestOutboundConnection", {}, { binary: true })).resolves.toEqual({ accepted: true }); + expect(send).toHaveBeenCalledWith(expect.anything(), { binary: true }, expect.any(Function)); + }); + + test("rejects and forgets the call when send throws synchronously", async () => { + server = await startServer(); + client = await connect(server); + (client as any).socket.send = () => { + throw new Error("send failed"); + }; + await expect(client.call("requestOutboundConnection", {})).rejects.toThrow("send failed"); + expect((client as any).queue).toEqual({}); + }); + + test("rejects pending calls when the socket closes, even with no listeners", async () => { + server = await startServer((_msg, socket) => socket.close()); + client = await connect(server); + client.removeAllListeners(); + await expect(client.call("requestOutboundConnection", {})).rejects.toThrow("websocket closed before reply"); + client = undefined; // already closed by the server + }); +}); diff --git a/src/v1/rpcClient.ts b/src/v1/rpcClient.ts new file mode 100644 index 0000000..3948bdd --- /dev/null +++ b/src/v1/rpcClient.ts @@ -0,0 +1,81 @@ +import { Client } from "rpc-websockets"; + +/** + * Upper bound on how long any gateway RPC waits for its reply. It must stay + * above the longest the gateway may hold a call open: requestOutboundConnection + * is planned to block for up to ~30s waiting for the application's accept/deny, + * then reply with a deny. If the gateway ever holds a call longer, raise this + * too, or the client gives up on a call the gateway goes on to complete. + */ +export const DEFAULT_CALL_TIMEOUT_MS = 45000; + +/** Rejection for a call whose reply did not arrive within its timeout. */ +export class RpcTimeoutError extends Error {} + +/** + * rpc-websockets' Client with call() fixed to never lose a reply. + * + * The stock call() registers the pending call only inside the socket's send + * callback. Under Node over TLS that callback is deferred to setImmediate, so if + * the event loop stalls for a few milliseconds after the send (GC, busy timers) + * the gateway's reply is read first, finds no pending entry, and is silently + * dropped; the entry is registered afterwards and the promise never settles. + * This is still the case in the latest rpc-websockets release. + * + * This version registers the call before sending, applies a reply timeout by + * default, and fails every pending call when the socket closes: the gateway + * handles each connection independently, so a reply can never arrive on a + * reconnected socket. + */ +export class RpcClient extends Client { + /** + * Same signature as the stock call(): ws_opts may be passed as the third + * argument, and a falsy timeout (null, 0) means no timeout. + */ + call(method: string, params?: object, timeout: number | object | null = DEFAULT_CALL_TIMEOUT_MS, ws_opts?: object): Promise { + if (!ws_opts && typeof timeout === "object" && timeout !== null) { + ws_opts = timeout; + timeout = DEFAULT_CALL_TIMEOUT_MS; + } + const replyTimeout = timeout as number | null; + // The fields below are private in rpc-websockets' typings but are what its + // own message handler resolves replies against. + const self = this as any; + return new Promise((resolve, reject) => { + if (!self.ready) return reject(new Error("socket not ready")); + const rpc_id = self.generate_request_id(method, params); + const fail = (error: Error) => { + if (!self.queue[rpc_id]) return; + clearTimeout(self.queue[rpc_id].timeout); + delete self.queue[rpc_id]; + reject(error); + }; + self.queue[rpc_id] = { promise: [resolve, reject] }; + if (replyTimeout) { + self.queue[rpc_id].timeout = setTimeout(() => fail(new RpcTimeoutError(`"${method}" reply timeout after ${replyTimeout}ms`)), replyTimeout); + } + try { + const message = { jsonrpc: "2.0", method, params: params || undefined, id: rpc_id }; + self.socket.send(self.dataPack.encode(message), ws_opts, (error?: Error) => error && fail(error)); + } catch (error) { + fail(error as Error); + } + }); + } + + // Hooked here rather than via on("close") so it survives removeAllListeners(). + emit(event: T, ...args: any[]): boolean { + if (event === "close") this.failPendingCalls(new Error("websocket closed before reply")); + return super.emit(event, ...args); + } + + private failPendingCalls(error: Error) { + const queue = (this as any).queue; + for (const id of Object.keys(queue)) { + const pending = queue[id]; + delete queue[id]; + clearTimeout(pending.timeout); + pending.promise[1](error); + } + } +} diff --git a/src/v1/signaling.test.ts b/src/v1/signaling.test.ts index eddb00d..9fe0aca 100644 --- a/src/v1/signaling.test.ts +++ b/src/v1/signaling.test.ts @@ -1,6 +1,7 @@ import Signaling from "./signaling"; import { DiagnosticsBatcher } from "./diagnostics"; import { EndpointType } from "../types"; +import { RpcTimeoutError } from "./rpcClient"; // Mock rpc-websockets jest.mock("rpc-websockets", () => { @@ -154,6 +155,63 @@ describe("Signaling websocket event handlers", () => { expect(emitSpy).toHaveBeenCalledWith("init", expect.anything(), true); }); + test("should tear down and emit fatalError when setMediaPreferences fails", async () => { + const emitSpy = jest.spyOn(signaling, "emit"); + const ws = (signaling as any).ws; + ws.call.mockRejectedValueOnce({ code: -32000, message: "boom" }); + + await getWsCallback("open")(); + + expect(emitSpy).toHaveBeenCalledWith("fatalError", new Error("setMediaPreferences failed: boom")); + expect(emitSpy).not.toHaveBeenCalledWith("init", expect.anything(), expect.anything()); + expect(ws.setAutoReconnect).toHaveBeenCalledWith(false); + expect((signaling as any).ws).toBeNull(); + }); + + // The socket closed mid-call: the close handler decides whether to reconnect. + test("should leave setMediaPreferences failures on a closed socket to the close handler", async () => { + const emitSpy = jest.spyOn(signaling, "emit"); + const ws = (signaling as any).ws; + ws.call.mockImplementationOnce(() => { + ws.ready = false; + return Promise.reject(new Error("websocket closed before reply")); + }); + + await getWsCallback("open")(); + + expect(emitSpy).not.toHaveBeenCalledWith("fatalError", expect.anything()); + expect((signaling as any).ws).toBe(ws); + }); + + describe("ping", () => { + beforeEach(() => jest.useFakeTimers()); + afterEach(() => jest.useRealTimers()); + + async function ping(result: Promise) { + await getWsCallback("open")(); + (signaling as any).ws.call.mockReturnValueOnce(result); + await jest.advanceTimersByTimeAsync(60000); + } + + test("should tear down and emit fatalError when a ping gets no reply", async () => { + const emitSpy = jest.spyOn(signaling, "emit"); + + await ping(Promise.reject(new RpcTimeoutError("timeout"))); + + expect(emitSpy).toHaveBeenCalledWith("fatalError", new Error("Connection lost: ping timed out")); + expect((signaling as any).ws).toBeNull(); + }); + + test("should keep the session when a ping fails for another reason", async () => { + const emitSpy = jest.spyOn(signaling, "emit"); + + await ping(Promise.reject(new Error("websocket closed before reply"))); + + expect(emitSpy).not.toHaveBeenCalledWith("fatalError", expect.anything()); + expect((signaling as any).ws).not.toBeNull(); + }); + }); + test("should reject with error and disconnect on 403 error", async () => { const errorCallback = getWsCallback("error"); expect(errorCallback).toBeDefined(); diff --git a/src/v1/signaling.ts b/src/v1/signaling.ts index 0ba59df..06eb334 100644 --- a/src/v1/signaling.ts +++ b/src/v1/signaling.ts @@ -1,7 +1,7 @@ const sdkVersion = require("../../package.json").version; import { v4 as uuid } from "uuid"; import { EventEmitter } from "events"; -import { Client as JsonRpcClient } from "rpc-websockets"; +import { RpcClient as JsonRpcClient, RpcTimeoutError } from "./rpcClient"; import logger from "../logging"; import { EndpointType, HangupResult, OutboundConnectionResult, RtcAuthParams, RtcOptions } from "../types"; import { PublishSdpAnswer, PublishMetadata, ReadyMetadata, SetMediaPreferencesWebRtcResponse, SdpAnswer } from "./types"; @@ -96,6 +96,16 @@ class Signaling extends EventEmitter { logger.debug(`Connected to ${websocketUrl}`); this.ws = ws; + // The session cannot continue on this socket: tear it down and tell the + // application. On a reconnect the connect() promise has already settled, so + // the event is the only thing that reaches it. + const failSession = (error: Error) => { + logger.error(error.message); + reject(error); + this.emit("fatalError", error); + this._disconnect(false); + }; + ws.on("sdpOffer", (event: any) => { this.emit("sdpOffer", event); }); @@ -113,14 +123,30 @@ class Signaling extends EventEmitter { this.disconnect(); }); } - let preferencesResponse = await this.setMediaPreferences(); + let preferencesResponse; + try { + preferencesResponse = await this.setMediaPreferences(); + } catch (err: any) { + // A close fails this call too, and the close handler owns what happens next. + if (this.ws !== ws || !this.socketOpen) return; + failSession(new Error(`setMediaPreferences failed: ${err?.message ?? err}`)); + return; + } // logger.debug(`Media preferences set`, preferencesResponse); // Setup Peers. isReconnect tells the caller whether existing peer connections/media // need to be rebuilt and re-published, rather than created for the first time. this.emit("init", preferencesResponse, isReconnect); this.pingInterval = setInterval(() => { - ws.call("ping", {}); + ws.call("ping", {}).catch((err) => { + // No reply means the connection is dead even though the socket still + // looks open, e.g. a NAT or load balancer silently dropped it. + if (err instanceof RpcTimeoutError && this.ws === ws) { + failSession(new Error("Connection lost: ping timed out")); + } else { + logger.debug("ping failed", err); + } + }); }, 60000); logger.debug("Websocket configured"); });