diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index fb860fc..69211c8 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/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(); + }); +}); diff --git a/src/v1/bandwidthRtc.ts b/src/v1/bandwidthRtc.ts index e5eb2ba..b683389 100644 --- a/src/v1/bandwidthRtc.ts +++ b/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 { + 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) { @@ -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> = 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,9 +729,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); } }; @@ -718,9 +737,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); } }; @@ -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); } }; diff --git a/src/v1/signaling.test.ts b/src/v1/signaling.test.ts index 29de410..4a55f83 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 bb5d386..488a85a 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", {});