VAPI-3917 fix(signaling): rebuild peer connections and re-publish media on signaling reconnect

src/v1/bandwidthRtc.test.ts

@@ -303,3 +303,155 @@ describe("bandwidthRtcV1 connect method", () => {
     expect(signaling.on).toHaveBeenCalledWith("init", expect.any(Function));
   });
 });
+
+describe("bandwidthRtcV1 retryIceOnFailed", () => {
+  beforeAll(() => {
+    setupNavigatorMocks();
+    setupMocks();
+  });
+
+  beforeEach(() => {
+    jest.useFakeTimers();
+  });
+
+  afterEach(() => {
+    jest.useRealTimers();
+  });
+
+  function makePc(connectionState: string) {
+    return { connectionState } as any as RTCPeerConnection;
+  }
+
+  test("does nothing when shouldRetry is false", async () => {
+    const brtc = new BandwidthRtc();
+    const offerPublishSdp = jest.spyOn(brtc as any, "offerPublishSdp");
+
+    await (brtc as any).retryIceOnFailed(makePc("failed"), "publish", false);
+
+    expect(offerPublishSdp).not.toHaveBeenCalled();
+  });
+
+  test("does not restart the publish connection when the subscribe connection failed", async () => {
+    const brtc = new BandwidthRtc();
+    const offerPublishSdp = jest.spyOn(brtc as any, "offerPublishSdp");
+
+    await (brtc as any).retryIceOnFailed(makePc("failed"), "subscribe", true);
+
+    expect(offerPublishSdp).not.toHaveBeenCalled();
+  });
+
+  test("re-offers once and stops once the publish connection recovers", async () => {
+    const brtc = new BandwidthRtc();
+    const pc = makePc("failed");
+    jest.spyOn(brtc as any, "offerPublishSdp").mockImplementation(async () => {
+      (pc as any).connectionState = "connected";
+      return {} as any;
+    });
+
+    await (brtc as any).retryIceOnFailed(pc, "publish", true);
+
+    expect((brtc as any).offerPublishSdp).toHaveBeenCalledTimes(1);
+  });
+
+  test("retries every 5s until the timeout elapses if still failed", async () => {
+    const brtc = new BandwidthRtc();
+    const pc = makePc("failed");
+    jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue({} as any);
+
+    const done = (brtc as any).retryIceOnFailed(pc, "publish", true);
+    // Initial offer, then retries at 5s/10s/... up to the 30s timeout.
+    for (let i = 0; i < 6; i++) {
+      await Promise.resolve();
+      await jest.advanceTimersByTimeAsync(5_000);
+    }
+    await done;
+
+    expect((brtc as any).offerPublishSdp).toHaveBeenCalledTimes(7);
+  });
+
+  test("does not throw when a retry's offerPublishSdp rejects", async () => {
+    const brtc = new BandwidthRtc();
+    const pc = makePc("failed");
+    jest.spyOn(brtc as any, "offerPublishSdp").mockRejectedValue(new Error("signaling down"));
+
+    const done = (brtc as any).retryIceOnFailed(pc, "publish", true);
+    for (let i = 0; i < 6; i++) {
+      await Promise.resolve();
+      await jest.advanceTimersByTimeAsync(5_000);
+    }
+
+    await expect(done).resolves.not.toThrow();
+  });
+});
+
+describe("bandwidthRtcV1 init on signaling reconnect", () => {
+  beforeAll(() => {
+    setupNavigatorMocks();
+    setupMocks();
+  });
+
+  function makeMockStream(id: string) {
+    return { id, getTracks: () => [] };
+  }
+
+  function makePreferencesResponse() {
+    return {
+      publishSdpOffer: { sdpOffer: "publish-offer" },
+      subscribeSdpOffer: { sdpOffer: "subscribe-offer" },
+    } as any;
+  }
+
+  test("first init does not close old peer connections or re-publish", async () => {
+    const brtc = new BandwidthRtc();
+    const setupPeerConnection = jest.spyOn(brtc as any, "setupPeerConnection").mockResolvedValue({ close: jest.fn() });
+    const addStream = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection");
+    const offerPublishSdp = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue({});
+    (brtc as any).publishedStreams.set("stream-1", { mediaStream: makeMockStream("stream-1") });
+
+    await brtc.init(makePreferencesResponse());
+
+    expect(setupPeerConnection).toHaveBeenCalledTimes(2);
+    expect(addStream).not.toHaveBeenCalled();
+    expect(offerPublishSdp).not.toHaveBeenCalled();
+  });
+
+  test("reconnect closes stale peer connections and re-publishes existing streams", async () => {
+    const brtc = new BandwidthRtc();
+    const oldPublishPc = { close: jest.fn() };
+    const oldSubscribePc = { close: jest.fn() };
+    (brtc as any).publishingPeerConnection = oldPublishPc;
+    (brtc as any).subscribingPeerConnection = oldSubscribePc;
+    (brtc as any).subscribingPeerConnectionSdpRevision = 5;
+    (brtc as any).subscribeTrackMetadata.set("track-1", { from: "someone" });
+    (brtc as any).localDtmfSenders.set("stream-1", { insertDTMF: jest.fn() });
+
+    const stream = makeMockStream("stream-1");
+    (brtc as any).publishedStreams.set("stream-1", { mediaStream: stream });
+
+    jest.spyOn(brtc as any, "setupPeerConnection").mockResolvedValue({ close: jest.fn() });
+    const addStream = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {});
+    const offerPublishSdp = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue({});
+
+    await brtc.init(makePreferencesResponse(), true);
+
+    expect(oldPublishPc.close).toHaveBeenCalledTimes(1);
+    expect(oldSubscribePc.close).toHaveBeenCalledTimes(1);
+    expect(addStream).toHaveBeenCalledWith(stream);
+    expect(offerPublishSdp).toHaveBeenCalledTimes(1);
+    expect((brtc as any).subscribingPeerConnectionSdpRevision).toBe(0);
+    expect((brtc as any).subscribeTrackMetadata.size).toBe(0);
+    expect((brtc as any).localDtmfSenders.size).toBe(0);
+  });
+
+  test("reconnect with no published streams does not re-offer", async () => {
+    const brtc = new BandwidthRtc();
+    (brtc as any).publishingPeerConnection = { close: jest.fn() };
+    (brtc as any).subscribingPeerConnection = { close: jest.fn() };
+    jest.spyOn(brtc as any, "setupPeerConnection").mockResolvedValue({ close: jest.fn() });
+    const offerPublishSdp = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue({});
+
+    await brtc.init(makePreferencesResponse(), true);
+
+    expect(offerPublishSdp).not.toHaveBeenCalled();
+  });
+});

src/v1/bandwidthRtc.ts

@@ -57,8 +57,7 @@ const CONNECTION_STATE_FAILED = "failed";
 const CONNECTION_STATE_DISCONNECTED = "disconnected";
 
 // When true, automatically trigger an ICE restart (via offerPublishSdp(true)) on connection failure.
-// Disabled by default until the retry loop is production-hardened with a proper timeout/backoff.
-const RETRY_ICE_ON_FAILED = false;
+const RETRY_ICE_ON_FAILED = true;
 
 export class BandwidthRtc {
   private options?: RtcOptions;
@@ -396,16 +395,32 @@ export class BandwidthRtc {
   }
 
   // Re-publishes the SDP with iceRestart=true to trigger ICE renegotiation after a connection failure.
-  private async retryIceOnFailed(pc: RTCPeerConnection, shouldRetry: boolean): Promise<void> {
+  private async retryIceOnFailed(pc: RTCPeerConnection, peerConnectionType: string, shouldRetry: boolean): Promise<void> {
     if (!shouldRetry) {
       return;
     }
+    if (peerConnectionType !== PEER_CONNECTION_TYPE_PUBLISH) {
+      // The subscribing peer connection never creates its own SDP offer - the gateway always
+      // initiates that renegotiation - so there's no client-side offer to re-send with
+      // iceRestart=true here. offerPublishSdp() only ever acts on publishingPeerConnection,
+      // so calling it here would incorrectly restart the *other* (unfailed) connection.
+      logger.warn(`ICE restart on the ${peerConnectionType} peer connection requires the gateway to re-offer; client cannot initiate`);
+      return;
+    }
 
     const ICE_RESTART_TIMEOUT_MS = 30_000;
     const ICE_RESTART_RETRY_INTERVAL_MS = 5_000;
     const startTime = Date.now();
 
-    await this.offerPublishSdp(true);
+    const retryOffer = async () => {
+      try {
+        await this.offerPublishSdp(true);
+      } catch (err) {
+        logger.warn("ICE restart offer failed", err);
+      }
+    };
+
+    await retryOffer();
     let connectionState = pc.connectionState;
     while (connectionState === CONNECTION_STATE_FAILED) {
       if (Date.now() - startTime >= ICE_RESTART_TIMEOUT_MS) {
@@ -414,7 +429,7 @@ export class BandwidthRtc {
       }
       await new Promise((resolve) => setTimeout(resolve, ICE_RESTART_RETRY_INTERVAL_MS));
       // Don't block on this, we should try multiple times
-      this.offerPublishSdp(true);
+      retryOffer();
       connectionState = pc.connectionState;
     }
   }
@@ -519,7 +534,20 @@ export class BandwidthRtc {
     }
   }
 
-  public async init(setMediaPreferencesResponse: SetMediaPreferencesWebRtcResponse) {
+  public async init(setMediaPreferencesResponse: SetMediaPreferencesWebRtcResponse, isReconnect: boolean = false) {
+    if (isReconnect) {
+      // The signaling websocket reconnected (e.g. a gateway-initiated 1001 that expects the
+      // same endpoint to keep going, not a fresh connect()). The old peer connections are
+      // stale, so rebuild them and re-publish whatever was already being sent instead of
+      // silently losing media.
+      logger.info("Signaling reconnected; rebuilding peer connections and re-publishing existing streams");
+      this.publishingPeerConnection?.close();
+      this.subscribingPeerConnection?.close();
+      this.subscribingPeerConnectionSdpRevision = 0;
+      this.subscribeTrackMetadata.clear();
+      this.localDtmfSenders.clear();
+    }
+
     const publishOnTrackHandler = (event: RTCTrackEvent) => {
       logger.debug("publish ontrack event", event);
     };
@@ -529,6 +557,16 @@ export class BandwidthRtc {
       setMediaPreferencesResponse.publishSdpOffer.sdpOffer,
     );
 
+    if (isReconnect && this.publishedStreams.size > 0) {
+      // The new publishing peer connection starts with no tracks; re-add whatever was
+      // already published so the gateway (and far end) keep receiving media instead of
+      // silence. Per-stream codecPreferences aren't retained across a reconnect.
+      for (const publishedStream of this.publishedStreams.values()) {
+        this.addStreamToPublishingPeerConnection(publishedStream.mediaStream);
+      }
+      await this.offerPublishSdp();
+    }
+
     let streamTracks: Map<MediaStream, Set<MediaStreamTrack>> = new Map();
 
     const subscriptionOnTrackHandler = (event: RTCTrackEvent) => {
@@ -629,14 +667,14 @@ export class BandwidthRtc {
         const pc = event.target as RTCPeerConnection;
         const connectionState = pc.connectionState;
         logger.debug("onconnectionstatechange", connectionState, pc);
-        if (connectionState === CONNECTION_STATE_FAILED) {
+        if (connectionState === CONNECTION_STATE_DISCONNECTED) {
+          logger.warn("Peer disconnected, connection may be reestablished");
+        } else if (connectionState === CONNECTION_STATE_FAILED) {
           logger.warn("Connection failed, ICE restart required");
-          await this.retryIceOnFailed(pc, RETRY_ICE_ON_FAILED);
+          await this.retryIceOnFailed(pc, peerConnectionType, RETRY_ICE_ON_FAILED);
         }
       } catch (err) {
-        if (globalThis.window) {
-          logger.warn("onconnectionstatechange error", err);
-        }
+        logger.warn("onconnectionstatechange error", err);
       }
     };
     logger.debug("Initial SDP offer", initialSdpOffer);
@@ -677,29 +715,12 @@ export class BandwidthRtc {
       }
     };
 
-    peerConnection.onconnectionstatechange = (event) => {
-      try {
-        const pc = event.target as RTCPeerConnection;
-        logger.debug("onconnectionstatechange", pc.connectionState, pc);
-        const connectionState = pc.connectionState;
-        if (connectionState === CONNECTION_STATE_DISCONNECTED) {
-          logger.warn("Peer disconnected, connection may be reestablished");
-        }
-      } catch (err) {
-        if (globalThis.window) {
-          logger.warn("onconnectionstatechange error", err);
-        }
-      }
-    };
-
     peerConnection.oniceconnectionstatechange = (event) => {
       try {
         const pc = event.target as RTCPeerConnection;
         logger.debug("oniceconnectionstatechange", pc.iceConnectionState, pc);
       } catch (err) {
-        if (globalThis.window) {
-          logger.warn("oniceconnectionstatechange error", err);
-        }
+        logger.warn("oniceconnectionstatechange error", err);
       }
     };
 
@@ -708,19 +729,15 @@ export class BandwidthRtc {
         const pc = event.target as RTCPeerConnection;
         logger.debug("onicegatheringstatechange", pc.iceGatheringState, pc);
       } catch (err) {
-        if (globalThis.window) {
-          logger.warn("onicegatheringstatechange error", err);
-        }
+        logger.warn("onicegatheringstatechange error", err);
       }
     };
 
     peerConnection.onnegotiationneeded = (event) => {
       try {
         logger.debug("onnegotiationneeded", event.target);
       } catch (err) {
-        if (globalThis.window) {
-          logger.warn("onnegotiationneeded error", err);
-        }
+        logger.warn("onnegotiationneeded error", err);
       }
     };
 
@@ -729,9 +746,7 @@ export class BandwidthRtc {
         const pc = event.target as RTCPeerConnection;
         logger.debug("onsignalingstatechange", pc.signalingState, pc);
       } catch (err) {
-        if (globalThis.window) {
-          logger.warn("onsignalingstatechange error", err);
-        }
+        logger.warn("onsignalingstatechange error", err);
       }
     };
 

src/v1/signaling.test.ts

@@ -136,15 +136,19 @@ describe("Signaling websocket event handlers", () => {
     return ws.on.mock.calls.find((call: any) => call[0] === event)?.[1];
   }
 
-  test("should emit init and set up ping interval on open", async () => {
+  test("should emit init with isReconnect false on the first open, then true on subsequent opens", async () => {
     const emitSpy = jest.spyOn(signaling, "emit");
     const openCallback = getWsCallback("open");
     expect(openCallback).toBeDefined();
 
     await openCallback();
 
-    expect(emitSpy).toHaveBeenCalledWith("init", expect.anything());
+    expect(emitSpy).toHaveBeenCalledWith("init", expect.anything(), false);
     expect((signaling as any).pingInterval).toBeDefined();
+
+    await openCallback();
+
+    expect(emitSpy).toHaveBeenCalledWith("init", expect.anything(), true);
   });
 
   test("should reject with error and disconnect on 403 error", async () => {

src/v1/signaling.ts

@@ -59,6 +59,10 @@ class Signaling extends EventEmitter {
 
   connect(authParams: RtcAuthParams, options?: RtcOptions) {
     let rpc_id = 1;
+    // rpc-websockets auto-reconnects with a brand new underlying WebSocket (same
+    // JsonRpcClient instance), so "open" fires again on every reconnect. Scoped to
+    // this connect() call so a fresh top-level connect() always starts as false.
+    let hasConnectedOnce = false;
 
     return new Promise<void>((resolve, reject) => {
       if (this.ws) {
@@ -102,16 +106,18 @@ class Signaling extends EventEmitter {
 
       ws.on("open", async () => {
         logger.debug("Websocket open");
-        if (globalThis.addEventListener) {
+        const isReconnect = hasConnectedOnce;
+        hasConnectedOnce = true;
+        if (!isReconnect && globalThis.addEventListener) {
           globalThis.addEventListener("beforeunload", (event) => {
             this.disconnect();
           });
         }
-        // TODO: handle reconnections
         let preferencesResponse = await this.setMediaPreferences();
         // logger.debug(`Media preferences set`, preferencesResponse);
-        // Setup Peers
-        this.emit("init", 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", {});