From 41b650b4721be8854dad02facbe6375c6f1a14fa Mon Sep 17 00:00:00 2001 From: stampercasey Date: Thu, 24 Sep 2026 13:47:12 -0400 Subject: [PATCH 1/2] VAPI-4027 fix(signaling): never lose a JSON-RPC reply that beats the send callback rpc-websockets' Client.call() registers the pending call only inside the socket's send callback. Under Node over TLS that callback is deferred to setImmediate, so an event-loop stall of a few milliseconds after a send lets the gateway's reply be read first. The reply finds no pending entry and is silently dropped, then the entry is registered and the promise never settles. This is what hung the endpoint-to-endpoint monitor on requestOutboundConnection about 0.3% of the time. The latest rpc-websockets release (10.0.1) has the same code, and browsers are unaffected because their send callback is synchronous. Signaling now uses RpcClient, a subclass whose call(): - registers the pending call before sending, - always applies a reply timeout (45s default), and - fails every pending call when the socket closes, since a reply can never arrive on a reconnected socket. The fire-and-forget ping and the setMediaPreferences call on open now catch rejections, which would otherwise be unhandled now that calls can time out. --- package-lock.json | 14 ++++++- package.json | 4 +- src/v1/rpcClient.test.ts | 82 ++++++++++++++++++++++++++++++++++++++++ src/v1/rpcClient.ts | 65 +++++++++++++++++++++++++++++++ src/v1/signaling.ts | 12 ++++-- 5 files changed, 172 insertions(+), 5 deletions(-) create mode 100644 src/v1/rpcClient.test.ts create mode 100644 src/v1/rpcClient.ts 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..639214c --- /dev/null +++ b/src/v1/rpcClient.test.ts @@ -0,0 +1,82 @@ +/** + * @jest-environment node + */ +import { AddressInfo } from "net"; +import { WebSocketServer } from "ws"; +import { Client } from "rpc-websockets"; +import { RpcClient } 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(ClientType: typeof Client, server: WebSocketServer): Promise { + const client = new ClientType(`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: Client) { + 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: Client | undefined; + + afterEach(async () => { + client?.close(); + server.clients.forEach((socket) => socket.terminate()); + await new Promise((r) => server.close(r)); + }); + + test("stock rpc-websockets loses a reply that beats the send callback", async () => { + server = await startServer(); + client = await connect(Client, server); + delaySendCallback(client); + expect(await settlesWithin(client.call("requestOutboundConnection", {}), 300)).toBe("pending"); + }); + + test("resolves a reply that beats the send callback", async () => { + server = await startServer(); + client = await connect(RpcClient, 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(RpcClient, server); + await expect(client.call("requestOutboundConnection", {}, 100)).rejects.toThrow('"requestOutboundConnection" reply timeout after 100ms'); + }); + + test("rejects pending calls when the socket closes, even with no listeners", async () => { + server = await startServer((_msg, socket) => socket.close()); + client = await connect(RpcClient, 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..b7ebd8e --- /dev/null +++ b/src/v1/rpcClient.ts @@ -0,0 +1,65 @@ +import { Client } from "rpc-websockets"; + +/** + * Upper bound on how long any gateway RPC waits for its reply. Generous on + * purpose: requestOutboundConnection is expected to block on the gateway for up + * to ~30s once it waits for the application's accept/deny. + */ +export const DEFAULT_CALL_TIMEOUT_MS = 45000; + +/** + * 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, always applies a reply + * timeout, 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 { + call(method: string, params?: object, timeout: number = DEFAULT_CALL_TIMEOUT_MS, ws_opts?: object): Promise { + // 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); + self.queue[rpc_id] = { + promise: [resolve, reject], + timeout: setTimeout(() => { + delete self.queue[rpc_id]; + reject(new Error(`"${method}" reply timeout after ${timeout}ms`)); + }, timeout), + }; + const message = { jsonrpc: "2.0", method, params: params || undefined, id: rpc_id }; + self.socket.send(self.dataPack.encode(message), ws_opts, (error?: Error) => { + if (!error || !self.queue[rpc_id]) return; + clearTimeout(self.queue[rpc_id].timeout); + delete self.queue[rpc_id]; + reject(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.ts b/src/v1/signaling.ts index 9c9f1d8..c7a2775 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 } from "./rpcClient"; import logger from "../logging"; import { EndpointType, HangupResult, OutboundConnectionResult, RtcAuthParams, RtcOptions } from "../types"; import { PublishSdpAnswer, PublishMetadata, ReadyMetadata, SetMediaPreferencesWebRtcResponse, SdpAnswer } from "./types"; @@ -113,14 +113,20 @@ class Signaling extends EventEmitter { this.disconnect(); }); } - let preferencesResponse = await this.setMediaPreferences(); + let preferencesResponse; + try { + preferencesResponse = await this.setMediaPreferences(); + } catch (err) { + logger.error("setMediaPreferences failed", 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) => logger.debug("ping failed", err)); }, 60000); logger.debug("Websocket configured"); }); From 21e7d8bede6e8b42f82a7c4b701127a45f7874ba Mon Sep 17 00:00:00 2001 From: stampercasey Date: Fri, 25 Sep 2026 09:22:49 -0400 Subject: [PATCH 2/2] VAPI-4027 fix(signaling): surface dead sessions and match stock call() semantics - Tear down and emit fatalError when setMediaPreferences fails or a ping gets no reply, instead of leaving a silent zombie session - Restore stock call() behavior: ws_opts as third argument, falsy timeout means no timeout - Clean up the pending entry and timer when send throws synchronously - Drop the timing-dependent test that pinned the stock rpc-websockets bug --- src/v1/rpcClient.test.ts | 55 +++++++++++++++++++++++++------------ src/v1/rpcClient.ts | 50 ++++++++++++++++++++++------------ src/v1/signaling.test.ts | 58 ++++++++++++++++++++++++++++++++++++++++ src/v1/signaling.ts | 28 ++++++++++++++++--- 4 files changed, 153 insertions(+), 38 deletions(-) diff --git a/src/v1/rpcClient.test.ts b/src/v1/rpcClient.test.ts index 639214c..2ca470c 100644 --- a/src/v1/rpcClient.test.ts +++ b/src/v1/rpcClient.test.ts @@ -3,8 +3,7 @@ */ import { AddressInfo } from "net"; import { WebSocketServer } from "ws"; -import { Client } from "rpc-websockets"; -import { RpcClient } from "./rpcClient"; +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 { @@ -20,14 +19,14 @@ function startServer(onMessage?: (msg: any, socket: any) => void): Promise { - const client = new ClientType(`ws://localhost:${(server.address() as AddressInfo).port}`, { reconnect: false }); +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: Client) { +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)); @@ -44,7 +43,7 @@ const settlesWithin = (p: Promise, ms: number) => describe("RpcClient", () => { let server: WebSocketServer; - let client: Client | undefined; + let client: RpcClient | undefined; afterEach(async () => { client?.close(); @@ -52,29 +51,51 @@ describe("RpcClient", () => { await new Promise((r) => server.close(r)); }); - test("stock rpc-websockets loses a reply that beats the send callback", async () => { - server = await startServer(); - client = await connect(Client, server); - delaySendCallback(client); - expect(await settlesWithin(client.call("requestOutboundConnection", {}), 300)).toBe("pending"); - }); - test("resolves a reply that beats the send callback", async () => { server = await startServer(); - client = await connect(RpcClient, server); + 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(RpcClient, server); - await expect(client.call("requestOutboundConnection", {}, 100)).rejects.toThrow('"requestOutboundConnection" reply timeout after 100ms'); + 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(RpcClient, server); + 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 index b7ebd8e..3948bdd 100644 --- a/src/v1/rpcClient.ts +++ b/src/v1/rpcClient.ts @@ -1,12 +1,17 @@ import { Client } from "rpc-websockets"; /** - * Upper bound on how long any gateway RPC waits for its reply. Generous on - * purpose: requestOutboundConnection is expected to block on the gateway for up - * to ~30s once it waits for the application's accept/deny. + * 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. * @@ -17,33 +22,44 @@ export const DEFAULT_CALL_TIMEOUT_MS = 45000; * 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, always applies a reply - * timeout, and fails every pending call when the socket closes: the gateway + * 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 { - call(method: string, params?: object, timeout: number = DEFAULT_CALL_TIMEOUT_MS, ws_opts?: object): Promise { + /** + * 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); - self.queue[rpc_id] = { - promise: [resolve, reject], - timeout: setTimeout(() => { - delete self.queue[rpc_id]; - reject(new Error(`"${method}" reply timeout after ${timeout}ms`)); - }, timeout), - }; - const message = { jsonrpc: "2.0", method, params: params || undefined, id: rpc_id }; - self.socket.send(self.dataPack.encode(message), ws_opts, (error?: Error) => { - if (!error || !self.queue[rpc_id]) return; + 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); + } }); } 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 715f0de..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 { RpcClient as JsonRpcClient } from "./rpcClient"; +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); }); @@ -116,8 +126,10 @@ class Signaling extends EventEmitter { let preferencesResponse; try { preferencesResponse = await this.setMediaPreferences(); - } catch (err) { - logger.error("setMediaPreferences failed", err); + } 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); @@ -126,7 +138,15 @@ class Signaling extends EventEmitter { this.emit("init", preferencesResponse, isReconnect); this.pingInterval = setInterval(() => { - ws.call("ping", {}).catch((err) => logger.debug("ping failed", err)); + 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"); });