VAPI-3239 fix: wait for init() before handling a subscribe SDP offer

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);
+  });
+});

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<void>;
+  private markSubscribingPeerConnectionReady!: () => void;
+
   private localDtmfSenders: Map<string, RTCDTMFSender> = 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<void> {
+    if (this.subscribingPeerConnection) {
+      return;
+    }
+    let timeoutHandle: ReturnType<typeof setTimeout>;
+    const timeout = new Promise<never>((_, 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<void> {
     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(