diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index aba1bbc..9f0e022 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/src/v1/bandwidthRtc.test.ts @@ -317,7 +317,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); // Simulate the websocket "open" handler re-emitting "init" after a reconnect. - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); expect(addSpy).toHaveBeenCalledWith(mediaStream, undefined); expect(offerSpy).toHaveBeenCalledTimes(1); @@ -333,7 +333,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { (brtc as any).publishedStreams.set(id, { mediaStream: makeLiveStream(id) }); } - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); expect(addSpy).toHaveBeenCalledTimes(3); expect(offerSpy).toHaveBeenCalledTimes(1); @@ -350,7 +350,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { (brtc as any).publishedStreams.set("stream-1", { mediaStream: makeLiveStream("stream-1") }); - const initPromise = brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + const initPromise = brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); // Give the wait loop a couple of polls to prove it is actually waiting, not racing ahead. await new Promise((resolve) => setTimeout(resolve, 250)); expect(addSpy).not.toHaveBeenCalled(); @@ -373,7 +373,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { (brtc as any).publishedStreams.set("stream-1", { mediaStream: makeLiveStream("stream-1") }); jest.useFakeTimers({ doNotFake: ["nextTick"] }); - const initPromise = brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + const initPromise = brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); await jest.advanceTimersByTimeAsync(15_000); await initPromise; @@ -394,7 +394,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { const mediaStream = makeLiveStream("stream-1"); (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream, codecPreferences }); - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); expect(addSpy).toHaveBeenCalledWith(mediaStream, codecPreferences); }); @@ -434,7 +434,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { const freshTrack = makeTrack("audio"); mockGetUserMedia.mockResolvedValue({ getTracks: () => [freshTrack] }); - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); // Re-acquired using the stored constraints' audio settings, but only for the kind // that actually ended - video is left out entirely rather than requested as false. @@ -455,7 +455,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { const mediaStream = makeLiveStream("stream-1"); (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); expect(mockGetUserMedia).not.toHaveBeenCalled(); }); @@ -472,7 +472,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); mockGetUserMedia.mockResolvedValue({ getTracks: () => [makeTrack("audio")] }); - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); expect(mockGetUserMedia).toHaveBeenCalledWith({ audio: true }); }); @@ -487,7 +487,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); (brtc as any).localDtmfSenders.set(mediaStream.id, { insertDTMF: jest.fn(), canInsertDTMF: true }); - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); expect((brtc as any).localDtmfSenders.size).toBe(0); }); @@ -504,7 +504,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { const mediaStream = makeLiveStream("stream-1"); (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); - await expect(brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any)).resolves.toBeUndefined(); + await expect(brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true)).resolves.toBeUndefined(); expect(errorHandler).toHaveBeenCalledTimes(1); expect(errorHandler.mock.calls[0][0].message).toContain("gateway said no"); @@ -523,7 +523,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); mockGetUserMedia.mockRejectedValue(new Error("NotAllowedError")); - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); expect(errorHandler).toHaveBeenCalledTimes(1); expect(offerSpy).not.toHaveBeenCalled(); @@ -543,7 +543,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { const freshTrack = makeTrack("audio"); mockGetUserMedia.mockResolvedValue({ getTracks: () => [freshTrack] }); - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); expect(freshTrack.enabled).toBe(false); }); @@ -564,7 +564,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { (brtc as any).publishedStreams.set(healthyStream.id, { mediaStream: healthyStream }); mockGetUserMedia.mockRejectedValue(new Error("NotAllowedError")); - await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true); // The healthy stream still gets attached and renegotiated... expect(addSpy).toHaveBeenCalledWith(healthyStream, undefined); @@ -587,7 +587,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { const mediaStream = makeLiveStream("stream-1"); (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); - await expect(brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any)).resolves.toBeUndefined(); + await expect(brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any, true)).resolves.toBeUndefined(); }); test("concurrent init() calls are serialized rather than interleaved", async () => { @@ -657,3 +657,155 @@ describe("bandwidthRtcV1 connect method", () => { expect(errorHandler).toHaveBeenCalledWith(error); }); }); + +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(), connectionState: "connected" }); + 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, undefined); + 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(); + }); +}); diff --git a/src/v1/bandwidthRtc.ts b/src/v1/bandwidthRtc.ts index 418f393..0263c99 100644 --- a/src/v1/bandwidthRtc.ts +++ b/src/v1/bandwidthRtc.ts @@ -64,8 +64,7 @@ const PUBLISH_ICE_CONNECT_TIMEOUT_MS = 10_000; const PUBLISH_ICE_CONNECT_POLL_INTERVAL_MS = 100; // 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; @@ -427,16 +426,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 { + private async retryIceOnFailed(pc: RTCPeerConnection, peerConnectionType: string, shouldRetry: boolean): Promise { 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) { @@ -445,7 +460,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; } } @@ -550,11 +565,23 @@ export class BandwidthRtc { } } - public async init(setMediaPreferencesResponse: SetMediaPreferencesWebRtcResponse) { - return this.initMutex.runExclusive(() => this.doInit(setMediaPreferencesResponse)); + public async init(setMediaPreferencesResponse: SetMediaPreferencesWebRtcResponse, isReconnect: boolean = false) { + return this.initMutex.runExclusive(() => this.doInit(setMediaPreferencesResponse, isReconnect)); } - private async doInit(setMediaPreferencesResponse: SetMediaPreferencesWebRtcResponse) { + private async doInit(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 subscribing side's SDP + // revision counter and pending track metadata are scoped to the peer connection torn + // down below, so they must reset alongside it. republishStreams() (at the end of this + // method) rebuilds the publish side, including re-adding every currently published + // stream so the far end keeps receiving media instead of silence. + logger.info("Signaling reconnected; rebuilding peer connections and re-publishing existing streams"); + this.subscribingPeerConnectionSdpRevision = 0; + this.subscribeTrackMetadata.clear(); + } + const publishOnTrackHandler = (event: RTCTrackEvent) => { logger.debug("publish ontrack event", event); }; @@ -654,17 +681,18 @@ export class BandwidthRtc { setMediaPreferencesResponse.subscribeSdpOffer.sdpOffer, ); - await this.republishStreams(); + if (isReconnect) { + await this.republishStreams(); + } } /** * Re-attach every previously published stream to the new publishing peer connection. * - * On a fresh connect nothing has been published yet and this is a no-op. On a - * reconnect (the websocket re-opened and re-emitted "init") the peer connection - * built above is trackless: without this the session comes back fully connected - * but silent, and the gateway never sees media so the endpoint stays ineligible - * for calls. + * Only called from doInit's isReconnect branch: the websocket re-opened and + * re-emitted "init", and the peer connection built above is trackless - without + * this the session comes back fully connected but silent, and the gateway never + * sees media so the endpoint stays ineligible for calls. * * init() is driven by a signaling event, so a throw here would only become an * unhandled rejection. Report it to the application instead: the session is up @@ -815,14 +843,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); @@ -863,29 +891,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); } }; @@ -894,9 +905,7 @@ 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); } }; @@ -904,9 +913,7 @@ export class BandwidthRtc { try { logger.debug("onnegotiationneeded", event.target); } catch (err) { - if (globalThis.window) { - logger.warn("onnegotiationneeded error", err); - } + logger.warn("onnegotiationneeded error", err); } }; @@ -915,9 +922,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); } }; diff --git a/src/v1/signaling.test.ts b/src/v1/signaling.test.ts index 95b66e7..3f6caea 100644 --- a/src/v1/signaling.test.ts +++ b/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 () => { diff --git a/src/v1/signaling.ts b/src/v1/signaling.ts index 88a0c47..9c9f1d8 100644 --- a/src/v1/signaling.ts +++ b/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((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", {});