VAPI-4027 fix(signaling): never lose a JSON-RPC reply that beats the send callback

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",

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<WebSocketServer> {
+  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<RpcClient> {
+  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<unknown>, 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
+  });
+});

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<unknown> {
+    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<T extends string | symbol>(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);
+    }
+  }
+}

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<unknown>) {
+      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();

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");
       });