diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index 0e6f5d7..e77597b 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/src/v1/bandwidthRtc.test.ts @@ -1,5 +1,6 @@ import { BandwidthRtc } from "./bandwidthRtc"; import { setupMocks, setupNavigatorMocks } from "../mocks"; +import logger from "../logging"; // Mock Signaling class jest.mock("./signaling", () => { @@ -280,3 +281,88 @@ describe("bandwidthRtcV1 connect method", () => { expect(signaling.on).toHaveBeenCalledWith("init", expect.any(Function)); }); }); + +describe("bandwidthRtcV1 handleSubscribeSdpOffer / init race", () => { + beforeAll(() => { + setupNavigatorMocks(); + setupMocks(); + }); + + function makeMockPeerConnection() { + return { + setRemoteDescription: jest.fn().mockResolvedValue(undefined), + createAnswer: jest.fn().mockResolvedValue({ sdp: "answer-sdp" }), + setLocalDescription: jest.fn().mockResolvedValue(undefined), + }; + } + + test("processes a subscribe SDP offer that arrives before init() finishes, once the peer connection becomes available", async () => { + const brtc = new BandwidthRtc(); + const privateBrtc = brtc as any; + const mockPc = makeMockPeerConnection(); + privateBrtc.signaling.answerSdp = jest.fn().mockResolvedValue(undefined); + + // subscribingPeerConnection is intentionally unset here to simulate the SDP + // offer signaling event arriving before init() has finished creating it. + const handlePromise = privateBrtc.handleSubscribeSdpOffer({ + sdpOffer: "offer-sdp", + sdpRevision: 1, + streamSourceMetadata: {}, + }); + + // Let the pending call run past the point where it used to throw immediately. + await Promise.resolve(); + await Promise.resolve(); + expect(mockPc.setRemoteDescription).not.toHaveBeenCalled(); + + // init() finishes afterward and marks the peer connection ready. + privateBrtc.subscribingPeerConnection = mockPc; + privateBrtc.markSubscribingPeerConnectionReady(); + + await handlePromise; + + expect(mockPc.setRemoteDescription).toHaveBeenCalledWith({ type: "offer", sdp: "offer-sdp" }); + expect(mockPc.setLocalDescription).toHaveBeenCalledWith({ sdp: "answer-sdp" }); + expect(privateBrtc.signaling.answerSdp).toHaveBeenCalledWith("answer-sdp", "subscribe"); + expect(privateBrtc.subscribingPeerConnectionSdpRevision).toBe(1); + }); + + test("times out waiting for the subscribing peer connection instead of hanging forever if init() never runs", async () => { + jest.useFakeTimers(); + const debugSpy = jest.spyOn(logger, "debug").mockImplementation(() => {}); + const brtc = new BandwidthRtc(); + const privateBrtc = brtc as any; + + const handlePromise = privateBrtc.handleSubscribeSdpOffer({ + sdpOffer: "offer-sdp", + sdpRevision: 1, + streamSourceMetadata: {}, + }); + + await jest.advanceTimersByTimeAsync(10_000); + await handlePromise; + + expect(debugSpy).toHaveBeenCalledWith("error in handleSubscribeSdpOffer", expect.any(Error)); + + debugSpy.mockRestore(); + jest.useRealTimers(); + }); + + test("processes immediately when the subscribing peer connection is already available", async () => { + const brtc = new BandwidthRtc(); + const privateBrtc = brtc as any; + const mockPc = makeMockPeerConnection(); + privateBrtc.subscribingPeerConnection = mockPc; + privateBrtc.markSubscribingPeerConnectionReady(); + privateBrtc.signaling.answerSdp = jest.fn().mockResolvedValue(undefined); + + await privateBrtc.handleSubscribeSdpOffer({ + sdpOffer: "offer-sdp", + sdpRevision: 1, + streamSourceMetadata: {}, + }); + + expect(mockPc.setRemoteDescription).toHaveBeenCalledWith({ type: "offer", sdp: "offer-sdp" }); + expect(privateBrtc.subscribingPeerConnectionSdpRevision).toBe(1); + }); +}); diff --git a/src/v1/bandwidthRtc.ts b/src/v1/bandwidthRtc.ts index b0cae0c..57c08b7 100644 --- a/src/v1/bandwidthRtc.ts +++ b/src/v1/bandwidthRtc.ts @@ -58,6 +58,12 @@ const CONNECTION_STATE_DISCONNECTED = "disconnected"; // Disabled by default until the retry loop is production-hardened with a proper timeout/backoff. const RETRY_ICE_ON_FAILED = false; +// The server can push a subscribe SDP offer before init() finishes creating the +// subscribing RTCPeerConnection (the "ready" and "init" signaling events are not +// sequenced against each other). Cap how long handleSubscribeSdpOffer will wait +// for init() rather than hanging the subscribeMutex forever if init never runs. +const SUBSCRIBING_PEER_CONNECTION_READY_TIMEOUT_MS = 10_000; + export class BandwidthRtc { private options?: RtcOptions; @@ -85,6 +91,12 @@ export class BandwidthRtc { // Current SDP revision for the subscribing peer; used to reject outdated SDP offers private subscribingPeerConnectionSdpRevision = 0; + // Resolves once init() has finished creating subscribingPeerConnection; lets + // handleSubscribeSdpOffer wait out the race instead of failing when a subscribe + // SDP offer arrives before init() completes. + private subscribingPeerConnectionReady: Promise; + private markSubscribingPeerConnectionReady!: () => void; + private localDtmfSenders: Map = new Map(); private streamAvailableHandler?: { (event: RtcStream): void }; @@ -103,6 +115,10 @@ export class BandwidthRtc { this.diagnosticsBatcher = new DiagnosticsBatcher(); this.signaling = new Signaling(this.diagnosticsBatcher); + this.subscribingPeerConnectionReady = new Promise((resolve) => { + this.markSubscribingPeerConnectionReady = resolve; + }); + this.setMicEnabled = this.setMicEnabled.bind(this); this.setCameraEnabled = this.setCameraEnabled.bind(this); @@ -445,8 +461,30 @@ export class BandwidthRtc { } } + // Waits for init() to have created subscribingPeerConnection. init() and the SDP + // offer that triggers this handler arrive as two independent signaling events, so + // the offer can otherwise show up first and find no peer connection to negotiate on. + private async waitForSubscribingPeerConnection(): Promise { + if (this.subscribingPeerConnection) { + return; + } + let timeoutHandle: ReturnType; + const timeout = new Promise((_, reject) => { + timeoutHandle = setTimeout( + () => reject(new BandwidthRtcError("Timed out waiting for subscribing RTCPeerConnection to be initialized")), + SUBSCRIBING_PEER_CONNECTION_READY_TIMEOUT_MS, + ); + }); + try { + await Promise.race([this.subscribingPeerConnectionReady, timeout]); + } finally { + clearTimeout(timeoutHandle!); + } + } + private async handleSubscribeSdpOffer(subscribeSdpOffer: SubscribeSdpOffer): Promise { try { + await this.waitForSubscribingPeerConnection(); await this.subscribeMutex.runExclusive(async () => { logger.info("Received SDP offer", subscribeSdpOffer); logger.debug("Current SDP revision", this.subscribingPeerConnectionSdpRevision); @@ -558,6 +596,7 @@ export class BandwidthRtc { subscriptionOnTrackHandler, setMediaPreferencesResponse.subscribeSdpOffer.sdpOffer, ); + this.markSubscribingPeerConnectionReady(); } private async setupPeerConnection(