From 787165c9bf7986da342e828cef4ad4a2618eb153 Mon Sep 17 00:00:00 2001 From: stampercasey Date: Wed, 9 Sep 2026 14:11:17 -0400 Subject: [PATCH 1/4] VAPI-3929 fix: republish retained streams after a websocket reconnect When the gateway closes a device's websocket (deploy drain, instance shutdown, media-server loss, heartbeat death) the SDK auto-reconnects and re-emits "init", which builds a fresh, trackless publishing peer connection. Nothing replayed the already-published local streams onto it, so the session came back fully connected but silent: the gateway never saw RTP, never populated publishedTracks, and every subsequent requestOutboundConnection was rejected as "endpoint not eligible". Retain each published stream's codec preferences and acquisition constraints, and on reconnect re-attach every retained stream to the new peer connection followed by a single renegotiation. Tracks that ended while the websocket was down are re-acquired first: an ended track attaches happily and produces valid-looking SDP, but its sender never emits RTP, which leaves the endpoint stuck in exactly the same way. On a first connect nothing is retained and the replay is a no-op. Failures that leave the session unable to publish are reported through a new onError callback rather than leaving the application believing it is healthy. A fatal handshake error on a reconnect (403/409) is surfaced the same way, since by then the connect() promise it used to reject has long since resolved. --- src/v1/bandwidthRtc.test.ts | 222 ++++++++++++++++++++++++++++++++++++ src/v1/bandwidthRtc.ts | 100 +++++++++++++++- src/v1/signaling.test.ts | 11 ++ src/v1/signaling.ts | 4 + src/v1/types.ts | 11 ++ 5 files changed, 347 insertions(+), 1 deletion(-) diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index fb860fc..831937d 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/src/v1/bandwidthRtc.test.ts @@ -272,6 +272,216 @@ describe("bandwidthRtcV1 addStreamToPublishingPeerConnection", () => { }); }); +describe("bandwidthRtcV1 init reconnect replay", () => { + function stubSetupPeerConnection(brtc: BandwidthRtc) { + // init() only needs a stand-in RTCPeerConnection; the real + // negotiation performed by setupPeerConnection is exercised elsewhere. + (brtc as any).setupPeerConnection = jest.fn().mockResolvedValue({}); + } + + function makeTrack(kind: string, readyState: string = "live") { + return { kind, id: `${kind}-track`, readyState, stop: jest.fn() }; + } + + function makeLiveStream(id: string, tracks: any[] = [makeTrack("audio")]) { + return { + id, + getTracks: () => tracks, + addTrack: jest.fn((track: any) => tracks.push(track)), + removeTrack: jest.fn((track: any) => tracks.splice(tracks.indexOf(track), 1)), + } as any; + } + + test("does not replay when no streams were previously published (first connect)", async () => { + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + const addSpy = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection"); + const offerSpy = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + expect(addSpy).not.toHaveBeenCalled(); + expect(offerSpy).not.toHaveBeenCalled(); + }); + + test("re-attaches previously published streams to the new publishing peer connection on reconnect", async () => { + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + const addSpy = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + const offerSpy = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const mediaStream = makeLiveStream("stream-1"); + (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); + + expect(addSpy).toHaveBeenCalledWith(mediaStream, undefined); + expect(offerSpy).toHaveBeenCalledTimes(1); + }); + + test("renegotiates exactly once for all replayed streams", async () => { + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + const addSpy = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + const offerSpy = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + for (const id of ["stream-1", "stream-2", "stream-3"]) { + (brtc as any).publishedStreams.set(id, { mediaStream: makeLiveStream(id) }); + } + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + expect(addSpy).toHaveBeenCalledTimes(3); + expect(offerSpy).toHaveBeenCalledTimes(1); + }); + + test("replays with the codec preferences the stream was originally published with", async () => { + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + const addSpy = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const codecPreferences = { audio: [{ mimeType: "audio/opus", clockRate: 48000 }] }; + const mediaStream = makeLiveStream("stream-1"); + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream, codecPreferences }); + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + expect(addSpy).toHaveBeenCalledWith(mediaStream, codecPreferences); + }); + + test("publish retains codec preferences and constraints for a later replay", async () => { + const { mockGetUserMedia } = setupNavigatorMocks(); + setupMocks(); + const brtc = new BandwidthRtc(); + (brtc as any).publishingPeerConnection = {}; + jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const mediaStream = makeLiveStream("stream-1"); + mockGetUserMedia.mockResolvedValue(mediaStream); + const constraints = { audio: true, video: false }; + const codecPreferences = { audio: [{ mimeType: "audio/opus", clockRate: 48000 }] }; + + await brtc.publish(constraints as any, undefined, undefined, codecPreferences as any); + + const published = (brtc as any).publishedStreams.get("stream-1"); + expect(published.codecPreferences).toBe(codecPreferences); + expect(published.constraints).toBe(constraints); + }); + + test("re-acquires tracks that ended while disconnected instead of re-attaching them dead", async () => { + const { mockGetUserMedia } = setupNavigatorMocks(); + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + const addSpy = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const endedTrack = makeTrack("audio", "ended"); + const mediaStream = makeLiveStream("stream-1", [endedTrack]); + const constraints = { audio: true, video: false }; + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream, constraints }); + + const freshTrack = makeTrack("audio"); + mockGetUserMedia.mockResolvedValue({ getTracks: () => [freshTrack] }); + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + // Re-acquired from the same constraints, swapped into the same MediaStream the + // application already holds, and the dead track dropped. + expect(mockGetUserMedia).toHaveBeenCalledWith(constraints); + expect(mediaStream.removeTrack).toHaveBeenCalledWith(endedTrack); + expect(mediaStream.addTrack).toHaveBeenCalledWith(freshTrack); + expect(mediaStream.getTracks()).toEqual([freshTrack]); + expect(addSpy).toHaveBeenCalledWith(mediaStream, undefined); + }); + + test("does not re-acquire when every track is still live", async () => { + const { mockGetUserMedia } = setupNavigatorMocks(); + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const mediaStream = makeLiveStream("stream-1"); + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + expect(mockGetUserMedia).not.toHaveBeenCalled(); + }); + + test("derives re-acquisition constraints from track kinds when the application supplied the stream", async () => { + const { mockGetUserMedia } = setupNavigatorMocks(); + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + // No constraints retained: the stream came from the application, not getUserMedia. + const mediaStream = makeLiveStream("stream-1", [makeTrack("audio", "ended")]); + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); + mockGetUserMedia.mockResolvedValue({ getTracks: () => [makeTrack("audio")] }); + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + expect(mockGetUserMedia).toHaveBeenCalledWith({ audio: true, video: false }); + }); + + test("drops DTMF senders from the closed peer connection before replaying", async () => { + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const mediaStream = makeLiveStream("stream-1"); + (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); + + expect((brtc as any).localDtmfSenders.size).toBe(0); + }); + + test("reports a republish failure to the application instead of failing silently", async () => { + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockRejectedValue(new Error("gateway said no")); + + const errorHandler = jest.fn(); + brtc.onError(errorHandler); + + const mediaStream = makeLiveStream("stream-1"); + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); + + await expect(brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any)).resolves.toBeUndefined(); + + expect(errorHandler).toHaveBeenCalledTimes(1); + expect(errorHandler.mock.calls[0][0].message).toContain("gateway said no"); + }); + + test("reports a re-acquisition failure to the application", async () => { + const { mockGetUserMedia } = setupNavigatorMocks(); + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + const offerSpy = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const errorHandler = jest.fn(); + brtc.onError(errorHandler); + + const mediaStream = makeLiveStream("stream-1", [makeTrack("audio", "ended")]); + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); + mockGetUserMedia.mockRejectedValue(new Error("NotAllowedError")); + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + expect(errorHandler).toHaveBeenCalledTimes(1); + expect(offerSpy).not.toHaveBeenCalled(); + }); +}); + describe("bandwidthRtcV1 connect method", () => { beforeAll(() => { setupNavigatorMocks(); @@ -301,5 +511,17 @@ describe("bandwidthRtcV1 connect method", () => { expect(signaling.on).toHaveBeenCalledWith("ready", expect.any(Function)); expect(signaling.on).toHaveBeenCalledWith("sdpOffer", expect.any(Function)); expect(signaling.on).toHaveBeenCalledWith("init", expect.any(Function)); + expect(signaling.on).toHaveBeenCalledWith("fatalError", expect.any(Function)); + }); + + test("forwards a fatal signaling error to the application error handler", () => { + const bandwidthRtc = new BandwidthRtc("debug"); + const errorHandler = jest.fn(); + bandwidthRtc.onError(errorHandler); + + const error = new Error("Endpoint already has an active connection"); + (bandwidthRtc as any).handleError(error); + + expect(errorHandler).toHaveBeenCalledWith(error); }); }); diff --git a/src/v1/bandwidthRtc.ts b/src/v1/bandwidthRtc.ts index e5eb2ba..99389af 100644 --- a/src/v1/bandwidthRtc.ts +++ b/src/v1/bandwidthRtc.ts @@ -48,6 +48,7 @@ const PEER_CONNECTION_TYPE_SUBSCRIBE = "subscribe"; const TRACK_KIND_AUDIO = "audio"; const TRACK_KIND_VIDEO = "video"; const TELEPHONE_EVENT_MIME_TYPE = "audio/telephone-event"; +const TRACK_STATE_ENDED = "ended"; const HEARTBEAT_PING = "PING"; const HEARTBEAT_PONG = "PONG"; @@ -93,6 +94,7 @@ export class BandwidthRtc { private streamUnavailableHandler?: { (event: RtcStream): void }; private readyHandler?: { (readyMetadata: ReadyMetadata): void }; private dtmfSentHandler?: DtmfSentHandler; + private errorHandler?: { (error: Error): void }; // Caller identity for pending subscribe tracks, keyed by track id. The gateway // negotiates a fresh subscribe track per call and rides the call's metadata on @@ -131,6 +133,7 @@ export class BandwidthRtc { this.signaling.on("ready", this.handleReady.bind(this)); this.signaling.on("sdpOffer", this.handleSubscribeSdpOffer.bind(this)); this.signaling.on("init", this.init.bind(this)); + this.signaling.on("fatalError", this.handleError.bind(this)); await this.signaling.connect(authParams, options); logger.info("Successfully connected"); @@ -178,6 +181,18 @@ export class BandwidthRtc { this.dtmfSentHandler = callback; } + /** + * Set the function that will be called when the SDK hits an error that leaves the + * session unusable and cannot be recovered from internally, such as a failure to + * republish media after the websocket reconnected, or a reconnect the gateway + * refused. Without this the application has no way to tell a healthy session from + * one that is connected but can no longer send or receive media. + * @param callback callback function + */ + onError(callback: { (error: Error): void }): void { + this.errorHandler = callback; + } + /** * Publish media to the Bandwidth WebRTC platform * @@ -196,11 +211,13 @@ export class BandwidthRtc { ): Promise { // Cast or create a MediaStream from the input let mediaStream: MediaStream; + // Only set when we acquired the stream ourselves; retained so ended tracks can be re-acquired. + let constraints: MediaStreamConstraints | undefined; if (input && this.isMediaStream(input)) { // @ts-ignore mediaStream = input; } else { - let constraints: MediaStreamConstraints = { audio: true, video: true }; + constraints = { audio: true, video: true }; if (typeof input === "object") { constraints = input as MediaStreamConstraints; } @@ -223,6 +240,8 @@ export class BandwidthRtc { this.publishedStreams.set(mediaStream.id, { mediaStream: mediaStream, metadata: publishMetadata, + codecPreferences: codecPreferences, + constraints: constraints, }); if (audioLevelChangeHandler) { @@ -614,6 +633,85 @@ export class BandwidthRtc { subscriptionOnTrackHandler, setMediaPreferencesResponse.subscribeSdpOffer.sdpOffer, ); + + 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. + * + * 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 + * but cannot publish, and only the application can decide what to do about that. + */ + private async republishStreams(): Promise { + if (this.publishedStreams.size === 0) { + return; + } + + try { + // The senders these were taken from belong to the closed peer connection. + this.localDtmfSenders.clear(); + + for (const publishedStream of [...this.publishedStreams.values()]) { + await this.reacquireEndedTracks(publishedStream); + this.addStreamToPublishingPeerConnection(publishedStream.mediaStream, publishedStream.codecPreferences); + } + + // One renegotiation covers every re-attached stream. + await this.offerPublishSdp(); + } catch (err) { + logger.error("Failed to republish streams after reconnect", err); + this.handleError(new BandwidthRtcError(`Failed to republish streams after reconnect: ${err}`)); + } + } + + /** + * Replace any track of a published stream that ended while the websocket was down + * (device unplugged, OS revoked the mic, the browser released it). + * + * An ended track can still be attached to a peer connection and will produce a + * perfectly valid-looking SDP offer, but its sender never emits RTP - so the far + * end never sees media, which is indistinguishable from never having republished + * at all. Re-acquire instead, and swap the fresh tracks into the same MediaStream + * so the object the application already holds stays valid. + */ + private async reacquireEndedTracks(publishedStream: PublishedStream): Promise { + const mediaStream = publishedStream.mediaStream; + const tracks = mediaStream.getTracks(); + if (!tracks.some((track) => track.readyState === TRACK_STATE_ENDED)) { + return; + } + + // Fall back to the kinds we had when the application supplied the stream itself. + const constraints: MediaStreamConstraints = publishedStream.constraints ?? { + audio: tracks.some((track) => track.kind === TRACK_KIND_AUDIO), + video: tracks.some((track) => track.kind === TRACK_KIND_VIDEO), + }; + logger.info(`Re-acquiring ended tracks for stream ${mediaStream.id}`, constraints); + + const replacement = await navigator.mediaDevices.getUserMedia(constraints); + for (const track of tracks) { + track.stop(); + mediaStream.removeTrack(track); + } + for (const track of replacement.getTracks()) { + mediaStream.addTrack(track); + } + } + + private handleError(error: Error): void { + if (this.errorHandler) { + this.errorHandler(error); + } else { + logger.error("Unhandled SDK error (no onError handler registered)", error); + } } private async setupPeerConnection( diff --git a/src/v1/signaling.test.ts b/src/v1/signaling.test.ts index a993a41..ef2d581 100644 --- a/src/v1/signaling.test.ts +++ b/src/v1/signaling.test.ts @@ -172,6 +172,17 @@ describe("Signaling websocket event handlers", () => { expect(ws.setAutoReconnect).toHaveBeenCalledWith(false); }); + // On a reconnect the connect() promise has already resolved, so the reject is a + // no-op: the event is the only thing that reaches the application. + test("should emit fatalError on a fatal handshake error", async () => { + const emitSpy = jest.spyOn(signaling, "emit"); + const errorCallback = getWsCallback("error"); + + errorCallback({ message: "Unexpected server response: 409" }); + + expect(emitSpy).toHaveBeenCalledWith("fatalError", expect.any(Error)); + }); + test("should handle non-fatal error without throwing", async () => { const errorCallback = getWsCallback("error"); expect(errorCallback).toBeDefined(); diff --git a/src/v1/signaling.ts b/src/v1/signaling.ts index cd369bf..850ff00 100644 --- a/src/v1/signaling.ts +++ b/src/v1/signaling.ts @@ -127,6 +127,10 @@ class Signaling extends EventEmitter { ws.close(fatal.status); ws.setAutoReconnect(false); reject(new Error(fatal.error)); + // On a reconnect the connect() promise has long since resolved, so the + // reject above goes nowhere and the application is left holding a session + // that will never come back. Surface it as an event too. + this.emit("fatalError", new Error(fatal.error)); // Disconnect without calling leave since we are not connected this._disconnect(false); return; diff --git a/src/v1/types.ts b/src/v1/types.ts index 410a83a..0c2f898 100644 --- a/src/v1/types.ts +++ b/src/v1/types.ts @@ -67,6 +67,17 @@ export interface DataChannelPublishMetadata { export interface PublishedStream { mediaStream: MediaStream; metadata?: StreamPublishMetadata; + /** + * Codec preferences the stream was originally published with. Retained so a + * republish after a reconnect negotiates the same codecs as the first publish. + */ + codecPreferences?: CodecPreferences; + /** + * Constraints the stream was acquired with, when the SDK acquired it. Retained + * so tracks that ended while the websocket was down can be re-acquired from the + * same devices. Undefined when the application supplied its own MediaStream. + */ + constraints?: MediaStreamConstraints; } export interface PublishMetadata { From e662eaf459f58efa20267b6c2e4432810f03a0c5 Mon Sep 17 00:00:00 2001 From: stampercasey Date: Thu, 10 Sep 2026 16:13:04 -0400 Subject: [PATCH 2/4] fix(reconnect): wait for the publish peer connection before offering the replay Verified against a real lab drain: the republish offer fired the instant init() finished creating the peer connections, before the publish side's own ICE handshake had reached connected. The gateway requires PeerConnectionState == Connected before it will accept an offer (pkg/device/webrtc_device_handler.go's isPeerReady), so it rejected every attempt with "peer not ready for sdp offers" and the endpoint never became eligible. publish() gets away without this wait because an application always calls it well after connect() resolves - by the time a user clicks a button, ICE has long since finished. A reconnect's republish has no such delay; it runs immediately inside init(). Swift and Kotlin already wait for ICE before offering for the same reason; JS never did, on either path, because the gap only mattered on the one path that never had a natural delay. waitForPublishConnected() polls connectionState for up to 10s, matching the timeout the other two SDKs already use, before touching any retained stream. Re-verified end to end against two more lab drains after this fix: eviction to eligible again in under 2 seconds each time, several calls placed and answered cleanly across both reconnects, no further occurrences of "peer not ready." Co-Authored-By: Claude Sonnet 5 --- src/v1/bandwidthRtc.test.ts | 56 ++++++++++++++++++++++++++++++++++--- src/v1/bandwidthRtc.ts | 30 ++++++++++++++++++++ 2 files changed, 82 insertions(+), 4 deletions(-) diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index 831937d..c8f699d 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/src/v1/bandwidthRtc.test.ts @@ -273,10 +273,13 @@ describe("bandwidthRtcV1 addStreamToPublishingPeerConnection", () => { }); describe("bandwidthRtcV1 init reconnect replay", () => { - function stubSetupPeerConnection(brtc: BandwidthRtc) { - // init() only needs a stand-in RTCPeerConnection; the real - // negotiation performed by setupPeerConnection is exercised elsewhere. - (brtc as any).setupPeerConnection = jest.fn().mockResolvedValue({}); + // init() only needs a stand-in RTCPeerConnection; the real negotiation performed by + // setupPeerConnection is exercised elsewhere. Defaults to already connected so the + // republish path's ICE wait resolves immediately; pass a mutable object with a different + // connectionState to exercise that wait itself. + function stubSetupPeerConnection(brtc: BandwidthRtc, pc: any = { connectionState: "connected" }) { + (brtc as any).setupPeerConnection = jest.fn().mockResolvedValue(pc); + return pc; } function makeTrack(kind: string, readyState: string = "live") { @@ -336,6 +339,51 @@ describe("bandwidthRtcV1 init reconnect replay", () => { expect(offerSpy).toHaveBeenCalledTimes(1); }); + test("waits for the publish peer connection to reach connected before offering", async () => { + const brtc = new BandwidthRtc(); + // The gateway rejects an offer with "peer not ready for sdp offers" until its own side of + // the publish peer connection reaches connected - starting the offer immediately after + // init() creates the peer connections raced that and failed against a live gateway. + const pc = stubSetupPeerConnection(brtc, { connectionState: "connecting" }); + const addSpy = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + const offerSpy = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + (brtc as any).publishedStreams.set("stream-1", { mediaStream: makeLiveStream("stream-1") }); + + const initPromise = brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + // 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(); + expect(offerSpy).not.toHaveBeenCalled(); + + pc.connectionState = "connected"; + await initPromise; + + expect(addSpy).toHaveBeenCalledTimes(1); + expect(offerSpy).toHaveBeenCalledTimes(1); + }); + + test("reports an error rather than offering into a peer connection that never connects", async () => { + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc, { connectionState: "connecting" }); + const addSpy = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + const errorHandler = jest.fn(); + brtc.onError(errorHandler); + + (brtc as any).publishedStreams.set("stream-1", { mediaStream: makeLiveStream("stream-1") }); + jest.useFakeTimers({ doNotFake: ["nextTick"] }); + + const initPromise = brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + await jest.advanceTimersByTimeAsync(15_000); + await initPromise; + + expect(addSpy).not.toHaveBeenCalled(); + expect(errorHandler).toHaveBeenCalledTimes(1); + expect(errorHandler.mock.calls[0][0].message).toMatch(/did not reach "connected"/); + + jest.useRealTimers(); + }); + test("replays with the codec preferences the stream was originally published with", async () => { const brtc = new BandwidthRtc(); stubSetupPeerConnection(brtc); diff --git a/src/v1/bandwidthRtc.ts b/src/v1/bandwidthRtc.ts index 99389af..94c9518 100644 --- a/src/v1/bandwidthRtc.ts +++ b/src/v1/bandwidthRtc.ts @@ -56,6 +56,11 @@ const DATA_CHANNEL_STATE_OPEN = "open"; const CONNECTION_STATE_FAILED = "failed"; const CONNECTION_STATE_DISCONNECTED = "disconnected"; +const CONNECTION_STATE_CONNECTED = "connected"; + +/** How long to wait for the publish peer connection's ICE handshake before giving up. */ +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. @@ -656,6 +661,13 @@ export class BandwidthRtc { } try { + // The publishing peer connection built by init() moments ago is still negotiating ICE; + // the gateway rejects an offer with "peer not ready for sdp offers" until its own side + // of that connection reaches connected. publish() gets away without this wait because an + // application always calls it well after connect() resolves, but a reconnect's republish + // has no such delay - it runs immediately inside init(), so it has to wait explicitly. + await this.waitForPublishConnected(); + // The senders these were taken from belong to the closed peer connection. this.localDtmfSenders.clear(); @@ -672,6 +684,24 @@ export class BandwidthRtc { } } + /** Poll until the publish peer connection reaches "connected", or throw after the timeout. */ + private async waitForPublishConnected(): Promise { + const pc = this.publishingPeerConnection; + if (!pc) { + throw new BandwidthRtcError("No publishing RTCPeerConnection, cannot republish streams"); + } + + const startTime = Date.now(); + while (pc.connectionState !== CONNECTION_STATE_CONNECTED) { + if (Date.now() - startTime >= PUBLISH_ICE_CONNECT_TIMEOUT_MS) { + throw new BandwidthRtcError( + `Publish peer connection did not reach "connected" within ${PUBLISH_ICE_CONNECT_TIMEOUT_MS}ms (state: ${pc.connectionState})`, + ); + } + await new Promise((resolve) => setTimeout(resolve, PUBLISH_ICE_CONNECT_POLL_INTERVAL_MS)); + } + } + /** * Replace any track of a published stream that ended while the websocket was down * (device unplugged, OS revoked the mic, the browser released it). From 91fd253c9682069e9c12b339e6c7acaed8028198 Mon Sep 17 00:00:00 2001 From: stampercasey Date: Thu, 10 Sep 2026 16:55:06 -0400 Subject: [PATCH 3/4] Address code review findings on republish-on-reconnect - Serialize init() behind a mutex (init() is signaling-driven and fires on every socket open, so two closely-spaced reconnects could otherwise run it concurrently: the second overwrites publishingPeerConnection out from under the first, which is still polling the old one in waitForPublishConnected). Also close the outgoing peer connections instead of leaking them on every reconnect. - Isolate each stream's reacquire/attach in republishStreams so one stream's getUserMedia rejection no longer skips every other stream and the renegotiation entirely - only skip the renegotiation if nothing was actually attached. - Guard the application's onError handler so a handler that throws can't propagate out through the signaling "fatalError" listener and skip the disconnect() that follows it. - reacquireEndedTracks now replaces only the track kinds that actually ended (not healthy siblings in the same stream) and carries over enabled=false onto the replacement track instead of silently un-muting it. - waitForPublishConnected bails immediately on "failed"/"closed" instead of waiting out the full timeout on a peer connection that will never reach "connected". --- src/v1/bandwidthRtc.test.ts | 99 ++++++++++++++++++++++++++++++++++--- src/v1/bandwidthRtc.ts | 86 ++++++++++++++++++++++++++------ 2 files changed, 165 insertions(+), 20 deletions(-) diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index c8f699d..a4b1744 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/src/v1/bandwidthRtc.test.ts @@ -282,8 +282,8 @@ describe("bandwidthRtcV1 init reconnect replay", () => { return pc; } - function makeTrack(kind: string, readyState: string = "live") { - return { kind, id: `${kind}-track`, readyState, stop: jest.fn() }; + function makeTrack(kind: string, readyState: string = "live", enabled: boolean = true) { + return { kind, id: `${kind}-track`, readyState, enabled, stop: jest.fn() }; } function makeLiveStream(id: string, tracks: any[] = [makeTrack("audio")]) { @@ -436,9 +436,9 @@ describe("bandwidthRtcV1 init reconnect replay", () => { await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); - // Re-acquired from the same constraints, swapped into the same MediaStream the - // application already holds, and the dead track dropped. - expect(mockGetUserMedia).toHaveBeenCalledWith(constraints); + // 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. + expect(mockGetUserMedia).toHaveBeenCalledWith({ audio: true }); expect(mediaStream.removeTrack).toHaveBeenCalledWith(endedTrack); expect(mediaStream.addTrack).toHaveBeenCalledWith(freshTrack); expect(mediaStream.getTracks()).toEqual([freshTrack]); @@ -474,7 +474,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); - expect(mockGetUserMedia).toHaveBeenCalledWith({ audio: true, video: false }); + expect(mockGetUserMedia).toHaveBeenCalledWith({ audio: true }); }); test("drops DTMF senders from the closed peer connection before replaying", async () => { @@ -528,6 +528,93 @@ describe("bandwidthRtcV1 init reconnect replay", () => { expect(errorHandler).toHaveBeenCalledTimes(1); expect(offerSpy).not.toHaveBeenCalled(); }); + + test("carries over a muted track's enabled=false onto its replacement", async () => { + const { mockGetUserMedia } = setupNavigatorMocks(); + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const mutedEndedTrack = makeTrack("audio", "ended", false); + const mediaStream = makeLiveStream("stream-1", [mutedEndedTrack]); + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); + + const freshTrack = makeTrack("audio"); + mockGetUserMedia.mockResolvedValue({ getTracks: () => [freshTrack] }); + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + expect(freshTrack.enabled).toBe(false); + }); + + test("one stream's reacquisition failure does not block another stream's replay", async () => { + const { mockGetUserMedia } = setupNavigatorMocks(); + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + const addSpy = jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + const offerSpy = jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + const errorHandler = jest.fn(); + brtc.onError(errorHandler); + + const brokenStream = makeLiveStream("stream-broken", [makeTrack("audio", "ended")]); + const healthyStream = makeLiveStream("stream-healthy"); + (brtc as any).publishedStreams.set(brokenStream.id, { mediaStream: brokenStream }); + (brtc as any).publishedStreams.set(healthyStream.id, { mediaStream: healthyStream }); + mockGetUserMedia.mockRejectedValue(new Error("NotAllowedError")); + + await brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + + // The healthy stream still gets attached and renegotiated... + expect(addSpy).toHaveBeenCalledWith(healthyStream, undefined); + expect(offerSpy).toHaveBeenCalledTimes(1); + // ...and the broken one is skipped, not attached dead. + expect(addSpy).not.toHaveBeenCalledWith(brokenStream, undefined); + expect(errorHandler).toHaveBeenCalledTimes(1); + }); + + test("an onError handler that throws does not become an unhandled rejection", async () => { + const brtc = new BandwidthRtc(); + stubSetupPeerConnection(brtc); + jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockRejectedValue(new Error("gateway said no")); + + brtc.onError(() => { + throw new Error("app handler blew up"); + }); + + const mediaStream = makeLiveStream("stream-1"); + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); + + await expect(brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any)).resolves.toBeUndefined(); + }); + + test("concurrent init() calls are serialized rather than interleaved", async () => { + const brtc = new BandwidthRtc(); + const order: string[] = []; + let callNum = 0; + (brtc as any).setupPeerConnection = jest.fn().mockImplementation(async () => { + const n = ++callNum; + order.push(`start-${n}`); + await new Promise((resolve) => setTimeout(resolve, 10)); + order.push(`end-${n}`); + return { connectionState: "connected", close: jest.fn() }; + }); + jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); + jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + + await Promise.all([ + brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any), + brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any), + ]); + + // setupPeerConnection is called twice per init() call (publish, then subscribe). + // If the two init() calls ran concurrently, a later call's "start" could land + // between an earlier call's "start" and "end". Serialized, every start/end pair + // is contiguous regardless of which init() call it belongs to. + expect(order).toEqual(["start-1", "end-1", "start-2", "end-2", "start-3", "end-3", "start-4", "end-4"]); + }); }); describe("bandwidthRtcV1 connect method", () => { diff --git a/src/v1/bandwidthRtc.ts b/src/v1/bandwidthRtc.ts index 94c9518..418f393 100644 --- a/src/v1/bandwidthRtc.ts +++ b/src/v1/bandwidthRtc.ts @@ -57,6 +57,7 @@ const DATA_CHANNEL_STATE_OPEN = "open"; const CONNECTION_STATE_FAILED = "failed"; const CONNECTION_STATE_DISCONNECTED = "disconnected"; const CONNECTION_STATE_CONNECTED = "connected"; +const CONNECTION_STATE_CLOSED = "closed"; /** How long to wait for the publish peer connection's ICE handshake before giving up. */ const PUBLISH_ICE_CONNECT_TIMEOUT_MS = 10_000; @@ -86,6 +87,12 @@ export class BandwidthRtc { // Prevents concurrent modification to RTCPeerConnection state (can cause race conditions) private publishMutex: Mutex = new Mutex(); private subscribeMutex: Mutex = new Mutex(); + // init() is signaling-driven ("init" fires on every (re)open), so two closely-spaced + // reconnects can otherwise run init() concurrently: the second overwrites + // publishingPeerConnection/subscribingPeerConnection out from under the first, which is + // still polling the old one in waitForPublishConnected. Serializing init() end-to-end means + // there is only ever one publishing/subscribing peer connection in flight at a time. + private initMutex: Mutex = new Mutex(); private publishedStreams: Map = new Map(); private subscribedStreams: Map = new Map(); @@ -544,9 +551,16 @@ export class BandwidthRtc { } public async init(setMediaPreferencesResponse: SetMediaPreferencesWebRtcResponse) { + return this.initMutex.runExclusive(() => this.doInit(setMediaPreferencesResponse)); + } + + private async doInit(setMediaPreferencesResponse: SetMediaPreferencesWebRtcResponse) { const publishOnTrackHandler = (event: RTCTrackEvent) => { logger.debug("publish ontrack event", event); }; + // A reconnect re-runs init() against a fresh offer; the previous peer connection (if any) + // is already dead on the far end, but nothing local closes it, so it leaks otherwise. + this.publishingPeerConnection?.close(); this.publishingPeerConnection = await this.setupPeerConnection( PEER_CONNECTION_TYPE_PUBLISH, publishOnTrackHandler, @@ -633,6 +647,7 @@ export class BandwidthRtc { tags: metadata?.tags, }); }; + this.subscribingPeerConnection?.close(); this.subscribingPeerConnection = await this.setupPeerConnection( PEER_CONNECTION_TYPE_SUBSCRIBE, subscriptionOnTrackHandler, @@ -671,20 +686,42 @@ export class BandwidthRtc { // The senders these were taken from belong to the closed peer connection. this.localDtmfSenders.clear(); + // Each stream is reacquired/attached independently: one stream's getUserMedia + // rejection (an unplugged device, a revoked permission) must not prevent every + // other stream from being attached, nor skip the renegotiation entirely. + const reacquireErrors: unknown[] = []; + let attachedCount = 0; for (const publishedStream of [...this.publishedStreams.values()]) { - await this.reacquireEndedTracks(publishedStream); - this.addStreamToPublishingPeerConnection(publishedStream.mediaStream, publishedStream.codecPreferences); + try { + await this.reacquireEndedTracks(publishedStream); + this.addStreamToPublishingPeerConnection(publishedStream.mediaStream, publishedStream.codecPreferences); + attachedCount++; + } catch (err) { + logger.error(`Failed to reacquire ended tracks for stream ${publishedStream.mediaStream.id}`, err); + reacquireErrors.push(err); + } } - // One renegotiation covers every re-attached stream. - await this.offerPublishSdp(); + // One renegotiation covers every re-attached stream. Skip it entirely if nothing + // was actually attached - there is nothing to renegotiate. + if (attachedCount > 0) { + await this.offerPublishSdp(); + } + + if (reacquireErrors.length > 0) { + throw new BandwidthRtcError(`Failed to reacquire tracks for ${reacquireErrors.length} stream(s): ${reacquireErrors.join(", ")}`); + } } catch (err) { logger.error("Failed to republish streams after reconnect", err); this.handleError(new BandwidthRtcError(`Failed to republish streams after reconnect: ${err}`)); } } - /** Poll until the publish peer connection reaches "connected", or throw after the timeout. */ + /** + * Poll until the publish peer connection reaches "connected", or throw after the timeout. + * "failed"/"closed" are unrecoverable - bail immediately instead of waiting out the full + * timeout for a peer connection that will never reach "connected". + */ private async waitForPublishConnected(): Promise { const pc = this.publishingPeerConnection; if (!pc) { @@ -693,6 +730,9 @@ export class BandwidthRtc { const startTime = Date.now(); while (pc.connectionState !== CONNECTION_STATE_CONNECTED) { + if (pc.connectionState === CONNECTION_STATE_FAILED || pc.connectionState === CONNECTION_STATE_CLOSED) { + throw new BandwidthRtcError(`Publish peer connection reached unrecoverable state "${pc.connectionState}"`); + } if (Date.now() - startTime >= PUBLISH_ICE_CONNECT_TIMEOUT_MS) { throw new BandwidthRtcError( `Publish peer connection did not reach "connected" within ${PUBLISH_ICE_CONNECT_TIMEOUT_MS}ms (state: ${pc.connectionState})`, @@ -714,20 +754,34 @@ export class BandwidthRtc { */ private async reacquireEndedTracks(publishedStream: PublishedStream): Promise { const mediaStream = publishedStream.mediaStream; - const tracks = mediaStream.getTracks(); - if (!tracks.some((track) => track.readyState === TRACK_STATE_ENDED)) { + const endedTracks = mediaStream.getTracks().filter((track) => track.readyState === TRACK_STATE_ENDED); + if (endedTracks.length === 0) { return; } - // Fall back to the kinds we had when the application supplied the stream itself. - const constraints: MediaStreamConstraints = publishedStream.constraints ?? { - audio: tracks.some((track) => track.kind === TRACK_KIND_AUDIO), - video: tracks.some((track) => track.kind === TRACK_KIND_VIDEO), - }; + // Re-acquire only the kinds that actually ended - a healthy track of the other kind + // (e.g. video still fine, only the mic dropped) is left attached rather than replaced. + // Reuse the application's original per-kind constraints (device id, resolution, etc.) + // where available, falling back to a bare boolean. + const storedConstraints = publishedStream.constraints; + const constraints: MediaStreamConstraints = {}; + if (endedTracks.some((track) => track.kind === TRACK_KIND_AUDIO)) { + constraints.audio = storedConstraints?.audio ?? true; + } + if (endedTracks.some((track) => track.kind === TRACK_KIND_VIDEO)) { + constraints.video = storedConstraints?.video ?? true; + } logger.info(`Re-acquiring ended tracks for stream ${mediaStream.id}`, constraints); const replacement = await navigator.mediaDevices.getUserMedia(constraints); - for (const track of tracks) { + for (const track of endedTracks) { + const freshTrack = replacement.getTracks().find((t) => t.kind === track.kind); + // Carry over mute state: a fresh getUserMedia track always starts enabled, which would + // silently undo a setMicEnabled(false)/setCameraEnabled(false) the application made + // before the track ended. + if (freshTrack) { + freshTrack.enabled = track.enabled; + } track.stop(); mediaStream.removeTrack(track); } @@ -738,7 +792,11 @@ export class BandwidthRtc { private handleError(error: Error): void { if (this.errorHandler) { - this.errorHandler(error); + try { + this.errorHandler(error); + } catch (err) { + logger.error("onError handler threw", err); + } } else { logger.error("Unhandled SDK error (no onError handler registered)", error); } From bde91026acca85c6f6f022baf83c221a6c1519ea Mon Sep 17 00:00:00 2001 From: stampercasey Date: Fri, 11 Sep 2026 12:02:08 -0400 Subject: [PATCH 4/4] style: run prettier on bandwidthRtc.test.ts Co-Authored-By: Claude Opus 5 (1M context) --- src/v1/bandwidthRtc.test.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index a4b1744..aba1bbc 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/src/v1/bandwidthRtc.test.ts @@ -604,10 +604,7 @@ describe("bandwidthRtcV1 init reconnect replay", () => { jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); - await Promise.all([ - brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any), - brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any), - ]); + await Promise.all([brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any), brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any)]); // setupPeerConnection is called twice per init() call (publish, then subscribe). // If the two init() calls ran concurrently, a later call's "start" could land