VAPI-4023 fix(signaling): clean teardown on server-initiated close and stop leaking diagnostics log listeners

src/v1/diagnostics.test.ts

@@ -89,4 +89,13 @@ describe("DiagnosticsBatcher", () => {
     expect(logger.removeListener).toHaveBeenCalledWith("log", expect.any(Function));
     clearIntervalSpy.mockRestore();
   });
+
+  test("should remove the same logger listener it registered on shutdown", () => {
+    const diagnostics = new DiagnosticsBatcher();
+    const registered = (logger.on as jest.Mock).mock.calls[0][1];
+
+    diagnostics.shutdown();
+
+    expect(logger.removeListener).toHaveBeenCalledWith("log", registered);
+  });
 });

src/v1/diagnostics.ts

@@ -9,12 +9,14 @@ export type Diagnostics = {
 export class DiagnosticsBatcher extends EventEmitter {
   private logEvents: any[];
   private flushInterval: Timeout;
+  // Kept so shutdown() can remove the exact listener that was registered.
+  private logListener = this.handleLogEvent.bind(this);
 
   constructor(flushIntervalMillis = 300000) {
     super();
     this.logEvents = new Array();
 
-    logger.on("log", this.handleLogEvent.bind(this));
+    logger.on("log", this.logListener);
     this.flushInterval = setInterval(this.flushDiagnostics.bind(this), flushIntervalMillis);
   }
 
@@ -30,7 +32,7 @@ export class DiagnosticsBatcher extends EventEmitter {
   shutdown() {
     this.flushDiagnostics();
     clearInterval(this.flushInterval);
-    logger.removeListener("log", this.handleLogEvent.bind(this));
+    logger.removeListener("log", this.logListener);
   }
 
   private handleLogEvent(level: LogLevel, timestamp: any, ...args: any) {

src/v1/signaling.serverClose.test.ts

@@ -0,0 +1,67 @@
+/**
+ * @jest-environment node
+ */
+// Exercises the real rpc-websockets client (no mock) against a local server, because
+// the bug only shows up with the library's real ordering: it clears its socket and
+// "ready" flag synchronously on close, then emits "close" on the next tick.
+import { AddressInfo } from "net";
+import { Server } from "rpc-websockets";
+import Signaling from "./signaling";
+import { DiagnosticsBatcher } from "./diagnostics";
+import logger from "../logging";
+
+describe("Signaling server-initiated close", () => {
+  let server: Server;
+
+  beforeEach(async () => {
+    server = new Server({ port: 0, host: "127.0.0.1" });
+    await new Promise((resolve) => server.on("listening", resolve));
+    server.register("setMediaPreferences", () => ({}));
+  });
+
+  afterEach(async () => {
+    await server.close();
+  });
+
+  test("tears down without logging errors when the server closes the socket", async () => {
+    const errorSpy = jest.spyOn(logger, "error");
+    const port = (server.wss.address() as AddressInfo).port;
+    server.on("connection", (socket: any) => {
+      socket.send(JSON.stringify({ notification: "ready", params: { endpointId: "e-test" } }));
+      // Close the way the gateway does when the endpoint is deleted.
+      setTimeout(() => socket.close(1000), 50);
+    });
+
+    const batcher = new DiagnosticsBatcher();
+    const signaling = new Signaling(batcher);
+    await signaling.connect({ endpointToken: "t" }, { websocketUrl: `ws://127.0.0.1:${port}` });
+
+    await new Promise((resolve) => setTimeout(resolve, 200));
+
+    expect((signaling as any).ws).toBeNull();
+    expect((signaling as any).isReady).toBe(false);
+    expect(errorSpy).not.toHaveBeenCalled();
+    errorSpy.mockRestore();
+  });
+
+  test("disconnect() cancels a reconnect already scheduled after a 1001 close", async () => {
+    const port = (server.wss.address() as AddressInfo).port;
+    let connections = 0;
+    server.on("connection", (socket: any) => {
+      connections++;
+      socket.send(JSON.stringify({ notification: "ready", params: { endpointId: "e-test" } }));
+      if (connections === 1) {
+        setTimeout(() => socket.close(1001), 50);
+      }
+    });
+
+    const signaling = new Signaling(new DiagnosticsBatcher());
+    await signaling.connect({ endpointToken: "t" }, { websocketUrl: `ws://127.0.0.1:${port}` });
+    await new Promise((resolve) => setTimeout(resolve, 100));
+    signaling.disconnect();
+
+    // Past rpc-websockets' default 1s reconnect interval.
+    await new Promise((resolve) => setTimeout(resolve, 1500));
+    expect(connections).toBe(1);
+  });
+});

src/v1/signaling.test.ts

@@ -7,6 +7,9 @@ jest.mock("rpc-websockets", () => {
   return {
     Client: jest.fn().mockImplementation(() => {
       const mockClient = {
+        // An open socket, as rpc-websockets reports it while connected.
+        ready: true,
+        socket: {},
         on: jest.fn((event: string, callback: Function) => {
           // Automatically trigger 'ready' event for successful connections
           if (event === "ready") {

src/v1/signaling.ts

@@ -185,7 +185,7 @@ class Signaling extends EventEmitter {
   private _disconnect(notifyLeave: boolean) {
     logger.debug("Disconnecting websocket");
     if (this.ws) {
-      if (notifyLeave) {
+      if (notifyLeave && this.socketOpen) {
         try {
           this.ws.notify("leave");
         } catch (err) {
@@ -220,11 +220,18 @@ class Signaling extends EventEmitter {
       } catch (err) {
         logger.error("Error disabling auto-reconnect", err);
       }
+      // A reconnect may already be scheduled (e.g. after a 1001), which
+      // setAutoReconnect(false) does not cancel.
+      clearTimeout((this.ws as any).reconnect_timer_id);
       this.ws.removeAllListeners();
-      try {
-        this.ws.close();
-      } catch (err) {
-        logger.error(err);
+      // rpc-websockets drops its socket as soon as the server closes it, so there is
+      // nothing left to close on a server-initiated disconnect.
+      if ((this.ws as any).socket) {
+        try {
+          this.ws.close();
+        } catch (err) {
+          logger.error(err);
+        }
       }
       this.ws = null;
     }
@@ -277,7 +284,16 @@ class Signaling extends EventEmitter {
     }) as Promise<void>;
   }
 
+  // rpc-websockets clears its ready flag synchronously when the socket closes, but only
+  // emits "close" on the next tick, so anything sent from the close handler would fail.
+  private get socketOpen(): boolean {
+    return Boolean((this.ws as any)?.ready);
+  }
+
   private sendDiagnostics(diagnostics: Diagnostics): Promise<void> {
+    if (!this.socketOpen) {
+      return Promise.resolve();
+    }
     logger.debug(`Calling "deviceDiagnostics"`);
     return this.ws?.notify("deviceDiagnostics", diagnostics).catch((err: any) => {
       logger.error("Error sending diagnostics", err);