From 4927236bdc75201b9af314490a9e6cac90760d83 Mon Sep 17 00:00:00 2001 From: smoghe-bw Date: Wed, 23 Sep 2026 15:14:18 -0400 Subject: [PATCH 1/3] fix(v1): harden unpublish against races, unknown ids, and disconnect - Run unpublish's transceiver cleanup and renegotiation under one publishMutex section; do the same for publish's attach + negotiate. - unpublish with ids that match no published stream is now a no-op instead of unpublishing every stream. - Stop local tracks first, wait for the publish peer to be connected, and reject with a clear error if renegotiation still fails. - unpublish after disconnect stops tracks locally instead of throwing. - republishStreams skips streams unpublished mid-reconnect and stops any tracks it reacquired for them. - AudioLevelDetector gains stop(); unpublish releases its AudioContext and sampling interval. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/audioLevelDetector.test.ts | 57 +++++++++++ src/audioLevelDetector.ts | 25 ++++- src/v1/bandwidthRtc.test.ts | 166 ++++++++++++++++++++++++++++++++- src/v1/bandwidthRtc.ts | 158 ++++++++++++++++++++----------- src/v1/types.ts | 3 + 5 files changed, 354 insertions(+), 55 deletions(-) diff --git a/src/audioLevelDetector.test.ts b/src/audioLevelDetector.test.ts index 4cdebf7..45c8d74 100644 --- a/src/audioLevelDetector.test.ts +++ b/src/audioLevelDetector.test.ts @@ -102,3 +102,60 @@ test("test emit silent after time threshold", async () => { audioLevelDetector.emitCurrentAudioLevel(); expect(spy).toHaveBeenLastCalledWith(AudioLevel.SILENT); }); + +describe("AudioLevelDetector.stop", () => { + function withMockAudioContext(closeImpl: () => Promise) { + const disconnect = jest.fn(); + const close = jest.fn(closeImpl); + //@ts-ignore + global.AudioContext = class { + createMediaStreamSource() { + return { connect: () => {}, disconnect }; + } + createAnalyser() { + return {}; + } + close() { + return close(); + } + }; + return { disconnect, close }; + } + + afterEach(() => { + //@ts-ignore + global.AudioContext = MockAudioContext; + }); + + test("clears the interval, disconnects the source, and closes the context", () => { + const { disconnect, close } = withMockAudioContext(() => Promise.resolve()); + const clearIntervalSpy = jest.spyOn(global, "clearInterval"); + const audioLevelDetector = new AudioLevelDetector({ mediaStream: {} as fakeMediaStream }); + const removeAllListenersSpy = jest.spyOn(audioLevelDetector, "removeAllListeners"); + + audioLevelDetector.stop(); + + expect(clearIntervalSpy).toHaveBeenCalledWith(mockInterval); + expect(disconnect).toHaveBeenCalledTimes(1); + expect(close).toHaveBeenCalledTimes(1); + expect(removeAllListenersSpy).toHaveBeenCalledTimes(1); + }); + + test("is idempotent: a second call does nothing", () => { + const { disconnect, close } = withMockAudioContext(() => Promise.resolve()); + const audioLevelDetector = new AudioLevelDetector({ mediaStream: {} as fakeMediaStream }); + + audioLevelDetector.stop(); + audioLevelDetector.stop(); + + expect(disconnect).toHaveBeenCalledTimes(1); + expect(close).toHaveBeenCalledTimes(1); + }); + + test("does not throw when context.close() rejects", () => { + withMockAudioContext(() => Promise.reject(new Error("already closed"))); + const audioLevelDetector = new AudioLevelDetector({ mediaStream: {} as fakeMediaStream }); + + expect(() => audioLevelDetector.stop()).not.toThrow(); + }); +}); diff --git a/src/audioLevelDetector.ts b/src/audioLevelDetector.ts index d8d6533..6ffd1ea 100644 --- a/src/audioLevelDetector.ts +++ b/src/audioLevelDetector.ts @@ -21,6 +21,10 @@ export default class AudioLevelDetector extends EventEmitter { private analyserNode: AnalyserNode; private currentAudioLevel: AudioLevel = AudioLevel.SILENT; private previousAudioLevel: AudioLevel | undefined; + private audioContext: AudioContext; + private sourceNode: MediaStreamAudioSourceNode; + private intervalHandle: ReturnType; + private stopped = false; constructor(config: AudioLevelDetectorOptions) { super(); @@ -55,8 +59,27 @@ export default class AudioLevelDetector extends EventEmitter { analyser.smoothingTimeConstant = 0.85; source.connect(analyser); this.analyserNode = analyser; + this.audioContext = context; + this.sourceNode = source; - setInterval(this.analyse.bind(this), this.sampleInterval); + this.intervalHandle = setInterval(this.analyse.bind(this), this.sampleInterval); + } + + /** + * Releases the AudioContext, source node, and sampling interval. Idempotent + * so callers don't need to track whether they already stopped this detector. + */ + stop(): void { + if (this.stopped) { + return; + } + this.stopped = true; + clearInterval(this.intervalHandle); + this.sourceNode.disconnect(); + // close() can reject if the context is already closed; that's fine, we're + // tearing down either way. + this.audioContext.close().catch(() => {}); + this.removeAllListeners(); } analyse() { diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index aba1bbc..746d9cb 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 { BandwidthRtcError } from "../types"; // Mock Signaling class jest.mock("./signaling", () => { @@ -272,6 +273,132 @@ describe("bandwidthRtcV1 addStreamToPublishingPeerConnection", () => { }); }); +describe("bandwidthRtcV1 unpublish", () => { + function makeTrack(id: string) { + return { id, stop: jest.fn() }; + } + + function makeStream(id: string, tracks: any[]) { + return { id, getTracks: () => tracks } as any; + } + + function makeTransceiverFor(track: any) { + return { sender: { track }, stop: jest.fn() }; + } + + function makePublishingPeerConnection(transceivers: any[], connectionState: string = "connected") { + return { + getTransceivers: jest.fn().mockReturnValue(transceivers), + removeTrack: jest.fn(), + createOffer: jest.fn().mockResolvedValue({ sdp: "v=0" }), + setLocalDescription: jest.fn().mockResolvedValue(undefined), + setRemoteDescription: jest.fn().mockResolvedValue(undefined), + connectionState, + }; + } + + function stubOfferSdp(brtc: BandwidthRtc, impl?: () => Promise) { + const offerSdp = jest.fn(impl ?? (() => Promise.resolve({ sdpAnswer: "sdp", peerType: "publish" }))); + (brtc as any).signaling.offerSdp = offerSdp; + return offerSdp; + } + + test("unpublish with an unknown id leaves other published streams in place and does not renegotiate", async () => { + const brtc = new BandwidthRtc(); + const track = makeTrack("stream-1-track"); + const stream = makeStream("stream-1", [track]); + const transceiver = makeTransceiverFor(track); + const pc = makePublishingPeerConnection([transceiver]); + (brtc as any).publishingPeerConnection = pc; + (brtc as any).publishedStreams.set("stream-1", { mediaStream: stream }); + const offerSdp = stubOfferSdp(brtc); + + await brtc.unpublish("unknown-id"); + + expect((brtc as any).publishedStreams.has("stream-1")).toBe(true); + expect(pc.removeTrack).not.toHaveBeenCalled(); + expect(track.stop).not.toHaveBeenCalled(); + expect(offerSdp).not.toHaveBeenCalled(); + }); + + test("unpublish(stream) stops that stream's audio level detector", async () => { + const brtc = new BandwidthRtc(); + const track = makeTrack("stream-1-track"); + const stream = makeStream("stream-1", [track]); + const transceiver = makeTransceiverFor(track); + const pc = makePublishingPeerConnection([transceiver]); + (brtc as any).publishingPeerConnection = pc; + const audioLevelDetector = { stop: jest.fn() }; + (brtc as any).publishedStreams.set("stream-1", { mediaStream: stream, audioLevelDetector }); + stubOfferSdp(brtc); + + await brtc.unpublish({ mediaStream: stream } as any); + + expect(audioLevelDetector.stop).toHaveBeenCalledTimes(1); + }); + + test("unpublish after disconnect does not throw, stops the tracks, and does not call offerSdp", async () => { + const brtc = new BandwidthRtc(); + const track = makeTrack("stream-1-track"); + const stream = makeStream("stream-1", [track]); + (brtc as any).publishingPeerConnection = undefined; + (brtc as any).publishedStreams.set("stream-1", { mediaStream: stream }); + const offerSdp = stubOfferSdp(brtc); + + await expect(brtc.unpublish("stream-1")).resolves.toBeUndefined(); + + expect(track.stop).toHaveBeenCalledTimes(1); + expect((brtc as any).publishedStreams.has("stream-1")).toBe(false); + expect(offerSdp).not.toHaveBeenCalled(); + }); + + test("rejects with BandwidthRtcError when renegotiation fails, but still cleans up locally", async () => { + const brtc = new BandwidthRtc(); + const track = makeTrack("stream-1-track"); + const stream = makeStream("stream-1", [track]); + const transceiver = makeTransceiverFor(track); + const pc = makePublishingPeerConnection([transceiver]); + (brtc as any).publishingPeerConnection = pc; + (brtc as any).publishedStreams.set("stream-1", { mediaStream: stream }); + stubOfferSdp(brtc, () => Promise.reject(new Error("gateway rejected offer"))); + + await expect(brtc.unpublish("stream-1")).rejects.toThrow(BandwidthRtcError); + + expect(track.stop).toHaveBeenCalledTimes(1); + expect(pc.removeTrack).toHaveBeenCalledWith(transceiver.sender); + expect((brtc as any).publishedStreams.has("stream-1")).toBe(false); + }); + + test("does not remove transceivers until an in-flight publish negotiation completes", async () => { + const brtc = new BandwidthRtc(); + const track = makeTrack("stream-1-track"); + const stream = makeStream("stream-1", [track]); + const transceiver = makeTransceiverFor(track); + const pc = makePublishingPeerConnection([transceiver]); + (brtc as any).publishingPeerConnection = pc; + (brtc as any).publishedStreams.set("stream-1", { mediaStream: stream }); + + let resolveOfferSdp!: (value: any) => void; + const deferredOfferSdp = new Promise((resolve) => { + resolveOfferSdp = resolve; + }); + (brtc as any).signaling.offerSdp = jest.fn().mockReturnValueOnce(deferredOfferSdp).mockResolvedValue({ sdpAnswer: "sdp", peerType: "publish" }); + + // Simulate an in-flight publish negotiation holding publishMutex, blocked on the gateway's answer. + const inFlightPublish = (brtc as any).publishMutex.runExclusive(() => (brtc as any).negotiatePublishSdp()); + + const unpublishPromise = brtc.unpublish("stream-1"); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(pc.removeTrack).not.toHaveBeenCalled(); + + resolveOfferSdp({ sdpAnswer: "sdp", peerType: "publish" }); + await inFlightPublish; + await unpublishPromise; + + expect(pc.removeTrack).toHaveBeenCalledWith(transceiver.sender); + }); +}); + describe("bandwidthRtcV1 init reconnect replay", () => { // init() only needs a stand-in RTCPeerConnection; the real negotiation performed by // setupPeerConnection is exercised elsewhere. Defaults to already connected so the @@ -405,7 +532,10 @@ describe("bandwidthRtcV1 init reconnect replay", () => { const brtc = new BandwidthRtc(); (brtc as any).publishingPeerConnection = {}; jest.spyOn(brtc as any, "addStreamToPublishingPeerConnection").mockImplementation(() => {}); - jest.spyOn(brtc as any, "offerPublishSdp").mockResolvedValue(undefined); + // publish() now negotiates via negotiatePublishSdp directly (under its own publishMutex + // section), not the mutex-acquiring offerPublishSdp wrapper - see the unpublish/publish + // race fix in bandwidthRtc.ts. + jest.spyOn(brtc as any, "negotiatePublishSdp").mockResolvedValue(undefined); const mediaStream = makeLiveStream("stream-1"); mockGetUserMedia.mockResolvedValue(mediaStream); @@ -548,6 +678,40 @@ describe("bandwidthRtcV1 init reconnect replay", () => { expect(freshTrack.enabled).toBe(false); }); + test("skips a stream that was unpublished while its getUserMedia reacquire was pending", 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 endedTrack = makeTrack("audio", "ended"); + const mediaStream = makeLiveStream("stream-1", [endedTrack]); + (brtc as any).publishedStreams.set(mediaStream.id, { mediaStream }); + + const freshTrack = makeTrack("audio"); + let resolveGetUserMedia!: () => void; + mockGetUserMedia.mockReturnValue( + new Promise((resolve) => { + resolveGetUserMedia = () => resolve({ getTracks: () => [freshTrack] }); + }), + ); + + const initPromise = brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + // Let reacquireEndedTracks start (and reach its getUserMedia await) before unpublishing + // the stream out from under it. + await new Promise((resolve) => setTimeout(resolve, 10)); + (brtc as any).publishedStreams.delete(mediaStream.id); + resolveGetUserMedia(); + + await initPromise; + + // The freshly acquired track is live and would otherwise leak, so it must still be stopped. + expect(freshTrack.stop).toHaveBeenCalledTimes(1); + expect(addSpy).not.toHaveBeenCalled(); + expect(offerSpy).not.toHaveBeenCalled(); + }); + test("one stream's reacquisition failure does not block another stream's replay", async () => { const { mockGetUserMedia } = setupNavigatorMocks(); const brtc = new BandwidthRtc(); diff --git a/src/v1/bandwidthRtc.ts b/src/v1/bandwidthRtc.ts index 418f393..c2ecb0c 100644 --- a/src/v1/bandwidthRtc.ts +++ b/src/v1/bandwidthRtc.ts @@ -243,28 +243,34 @@ export class BandwidthRtc { } logger.info(`Publishing mediaStream ${mediaStream.id} (${alias})`); - this.addStreamToPublishingPeerConnection(mediaStream, codecPreferences); const publishMetadata: StreamPublishMetadata = {}; if (alias) { publishMetadata.alias = alias; } - this.publishedStreams.set(mediaStream.id, { - mediaStream: mediaStream, - metadata: publishMetadata, - codecPreferences: codecPreferences, - constraints: constraints, - }); + let audioLevelDetector: AudioLevelDetector | undefined; if (audioLevelChangeHandler) { - const audioLevelDetector = new AudioLevelDetector({ + audioLevelDetector = new AudioLevelDetector({ mediaStream: mediaStream, }); audioLevelDetector.on("audioLevelChange", audioLevelChangeHandler); } - // Perform SDP negotiation with Bandwidth WebRTC - const remoteSdpAnswer = await this.offerPublishSdp(); + // addStreamToPublishingPeerConnection, the publishedStreams entry, and the negotiation all + // have to happen atomically: an unpublish() racing in between would otherwise remove a + // transceiver mid-offer, or renegotiate before this stream's track is even attached. + const remoteSdpAnswer = await this.publishMutex.runExclusive(async () => { + this.addStreamToPublishingPeerConnection(mediaStream, codecPreferences); + this.publishedStreams.set(mediaStream.id, { + mediaStream: mediaStream, + metadata: publishMetadata, + codecPreferences: codecPreferences, + constraints: constraints, + audioLevelDetector: audioLevelDetector, + }); + return this.negotiatePublishSdp(); + }); // TODO: // const remoteStreamMetadata = remoteSdpAnswer.streamMetadata[mediaStream.id]; @@ -290,14 +296,39 @@ export class BandwidthRtc { publishedStreams.push(s); } } else { - publishedStreams.push({ - mediaStream: stream.mediaStream!, - }); + // Look up the stored entry first so its audioLevelDetector gets cleaned up too; + // fall back to a synthetic entry for a stream the SDK never tracked. + publishedStreams.push(this.publishedStreams.get(stream.mediaStream!.id) ?? { mediaStream: stream.mediaStream! }); } } - this.cleanupPublishedStreams(...publishedStreams); - await this.offerPublishSdp(); + // An empty list here only means "unpublish everything" when the caller passed zero + // arguments. If arguments were given but none resolved to a known stream (e.g. an + // unknown id), cleaning up with an empty list would otherwise unpublish everything. + if (streams.length > 0 && publishedStreams.length === 0) { + logger.warn("unpublish: none of the given streams are currently published", streams); + return; + } + + if (!this.publishingPeerConnection) { + // Nothing to renegotiate with the gateway; just release local resources. + this.cleanupPublishedStreams(...publishedStreams); + return; + } + + await this.publishMutex.runExclusive(async () => { + // Stop the local tracks first regardless of what happens next - the user's intent + // (stop sending this media) must take effect immediately. + this.cleanupPublishedStreams(...publishedStreams); + try { + // The gateway rejects offers unless the publish peer is "connected"; wait it out + // rather than sending an offer doomed to be rejected during a brief ICE blip. + await this.waitForPublishConnected(); + await this.negotiatePublishSdp(); + } catch (err) { + throw new BandwidthRtcError(`Stream(s) were unpublished locally, but renegotiation with the gateway failed: ${err}`); + } + }); } /** @@ -455,47 +486,53 @@ export class BandwidthRtc { throw new BandwidthRtcError("No publishing RTCPeerConnection, cannot offer SDP"); } - return await this.publishMutex.runExclusive(async () => { - const localSdpOffer = await this.publishingPeerConnection!.createOffer({ - offerToReceiveVideo: false, - offerToReceiveAudio: false, - iceRestart: restartIce, - }); + return this.publishMutex.runExclusive(() => this.negotiatePublishSdp(restartIce)); + } - // Diagnostic only: if an audio m-line is offered without telephone-event, DTMF - // can never negotiate for this session regardless of how long sendDtmf waits. - if (localSdpOffer.sdp?.includes("m=audio") && !localSdpOffer.sdp.includes(TELEPHONE_EVENT_MIME_TYPE.split("/")[1])) { - logger.warn("Publish SDP offer has an audio track but no telephone-event codec; DTMF will not be able to negotiate for this session"); - } + // Does the actual createOffer/setLocalDescription/setRemoteDescription dance. Callers are + // responsible for holding publishMutex - this does not take it itself, so a caller that + // needs to mutate transceivers (e.g. unpublish's cleanup) can do so under the same lock + // as the negotiation, instead of racing it. + private async negotiatePublishSdp(restartIce: boolean = false): Promise { + const localSdpOffer = await this.publishingPeerConnection!.createOffer({ + offerToReceiveVideo: false, + offerToReceiveAudio: false, + iceRestart: restartIce, + }); - let publishMetadata = { - mediaStreams: {}, - dataChannels: {}, - }; - publishMetadata.mediaStreams = Object.fromEntries(new Map([...this.publishedStreams].map(([streamId, stream]) => [streamId, stream.metadata || {}]))); - publishMetadata.dataChannels = Object.fromEntries( - new Map( - [...this.publishedDataChannels].map(([label, dataChannel]) => [ - label, - { - label: dataChannel.label, - streamId: dataChannel.id, - }, - ]), - ), - ); - logger.debug("publish metadata", publishMetadata); - const remoteSdpAnswer = await this.signaling.offerSdp(PEER_CONNECTION_TYPE_PUBLISH, localSdpOffer.sdp!); - - await this.publishingPeerConnection!.setLocalDescription(localSdpOffer); - logger.debug("remoteSdpAnswer", remoteSdpAnswer); - await this.publishingPeerConnection!.setRemoteDescription({ - type: "answer", - sdp: remoteSdpAnswer.sdpAnswer, - }); + // Diagnostic only: if an audio m-line is offered without telephone-event, DTMF + // can never negotiate for this session regardless of how long sendDtmf waits. + if (localSdpOffer.sdp?.includes("m=audio") && !localSdpOffer.sdp.includes(TELEPHONE_EVENT_MIME_TYPE.split("/")[1])) { + logger.warn("Publish SDP offer has an audio track but no telephone-event codec; DTMF will not be able to negotiate for this session"); + } - return remoteSdpAnswer; + let publishMetadata = { + mediaStreams: {}, + dataChannels: {}, + }; + publishMetadata.mediaStreams = Object.fromEntries(new Map([...this.publishedStreams].map(([streamId, stream]) => [streamId, stream.metadata || {}]))); + publishMetadata.dataChannels = Object.fromEntries( + new Map( + [...this.publishedDataChannels].map(([label, dataChannel]) => [ + label, + { + label: dataChannel.label, + streamId: dataChannel.id, + }, + ]), + ), + ); + logger.debug("publish metadata", publishMetadata); + const remoteSdpAnswer = await this.signaling.offerSdp(PEER_CONNECTION_TYPE_PUBLISH, localSdpOffer.sdp!); + + await this.publishingPeerConnection!.setLocalDescription(localSdpOffer); + logger.debug("remoteSdpAnswer", remoteSdpAnswer); + await this.publishingPeerConnection!.setRemoteDescription({ + type: "answer", + sdp: remoteSdpAnswer.sdpAnswer, }); + + return remoteSdpAnswer; } private async handleReady(readyMetadata: ReadyMetadata): Promise { @@ -692,8 +729,20 @@ export class BandwidthRtc { const reacquireErrors: unknown[] = []; let attachedCount = 0; for (const publishedStream of [...this.publishedStreams.values()]) { + // An unpublish() can race this loop: it may have already removed the stream (and + // stopped its tracks) before we get to it, or while reacquireEndedTracks' getUserMedia + // below is pending. + if (!this.publishedStreams.has(publishedStream.mediaStream.id)) { + continue; + } try { await this.reacquireEndedTracks(publishedStream); + if (!this.publishedStreams.has(publishedStream.mediaStream.id)) { + // Unpublished while getUserMedia was pending - the freshly acquired tracks are + // live and would otherwise leak, but the stream itself must not be re-attached. + publishedStream.mediaStream.getTracks().forEach((track) => track.stop()); + continue; + } this.addStreamToPublishingPeerConnection(publishedStream.mediaStream, publishedStream.codecPreferences); attachedCount++; } catch (err) { @@ -986,7 +1035,9 @@ export class BandwidthRtc { for (const stream of streams) { stream.mediaStream.getTracks().forEach((track) => { - this.publishingPeerConnection!.getTransceivers() + // No publishing peer connection (e.g. after disconnect()) means nothing to remove + // a transceiver from, but the track still needs to be stopped below. + (this.publishingPeerConnection?.getTransceivers() ?? []) .filter((transceiver) => transceiver.sender.track === track) .forEach((transceiver) => { this.publishingPeerConnection!.removeTrack(transceiver.sender); @@ -995,6 +1046,7 @@ export class BandwidthRtc { track.stop(); }); + stream.audioLevelDetector?.stop(); this.localDtmfSenders.delete(stream.mediaStream.id); this.publishedStreams.delete(stream.mediaStream.id); } diff --git a/src/v1/types.ts b/src/v1/types.ts index 0c2f898..4c34abc 100644 --- a/src/v1/types.ts +++ b/src/v1/types.ts @@ -1,4 +1,5 @@ import { MediaType } from "../types"; +import type AudioLevelDetector from "../audioLevelDetector"; export interface SetMediaPreferencesWebRtcResponse { endpointId: string; @@ -78,6 +79,8 @@ export interface PublishedStream { * same devices. Undefined when the application supplied its own MediaStream. */ constraints?: MediaStreamConstraints; + /** The audio level detector attached to this stream, if any, so it can be stopped on unpublish. */ + audioLevelDetector?: AudioLevelDetector; } export interface PublishMetadata { From 96a4dc256293dc6547da88c5bc5cabcb2f2a4a3c Mon Sep 17 00:00:00 2001 From: smoghe-bw Date: Wed, 23 Sep 2026 15:32:40 -0400 Subject: [PATCH 2/3] test(v1): pass isReconnect in the unpublish-during-reacquire test Keeps the test valid once init() republishes only on a reconnect (#18). Co-Authored-By: Claude Opus 5.5 (1M context) --- src/v1/bandwidthRtc.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index 746d9cb..2893f85 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/src/v1/bandwidthRtc.test.ts @@ -697,7 +697,8 @@ describe("bandwidthRtcV1 init reconnect replay", () => { }), ); - const initPromise = brtc.init({ publishSdpOffer: {}, subscribeSdpOffer: {} } as any); + // isReconnect=true keeps this valid once init() only republishes on a reconnect (#18). + const initPromise = (brtc as any).init({ publishSdpOffer: {}, subscribeSdpOffer: {} }, true); // Let reacquireEndedTracks start (and reach its getUserMedia await) before unpublishing // the stream out from under it. await new Promise((resolve) => setTimeout(resolve, 10)); From cadfcb2b1662fb0d68a674185d142ba230bf1f5c Mon Sep 17 00:00:00 2001 From: smoghe-bw Date: Wed, 23 Sep 2026 15:45:47 -0400 Subject: [PATCH 3/3] fix(v1): release publishMutex while unpublish waits for the publish peer The wait can last 10 s and would otherwise block publish() and gateway ICE-restart offers. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/v1/bandwidthRtc.test.ts | 28 ++++++++++++++++++++++++++++ src/v1/bandwidthRtc.ts | 22 ++++++++++++---------- 2 files changed, 40 insertions(+), 10 deletions(-) diff --git a/src/v1/bandwidthRtc.test.ts b/src/v1/bandwidthRtc.test.ts index 2893f85..a7de589 100644 --- a/src/v1/bandwidthRtc.test.ts +++ b/src/v1/bandwidthRtc.test.ts @@ -397,6 +397,34 @@ describe("bandwidthRtcV1 unpublish", () => { expect(pc.removeTrack).toHaveBeenCalledWith(transceiver.sender); }); + + test("does not hold publishMutex while waiting for the publish peer to reach connected", async () => { + const brtc = new BandwidthRtc(); + const track = makeTrack("stream-1-track"); + const stream = makeStream("stream-1", [track]); + const transceiver = makeTransceiverFor(track); + const pc = makePublishingPeerConnection([transceiver], "connecting"); + (brtc as any).publishingPeerConnection = pc; + (brtc as any).publishedStreams.set("stream-1", { mediaStream: stream }); + const offerSdp = stubOfferSdp(brtc); + + const unpublishPromise = brtc.unpublish("stream-1"); + + // Give the wait loop a couple of polls to prove unpublish is actually waiting, not racing ahead. + await new Promise((resolve) => setTimeout(resolve, 250)); + expect(offerSdp).not.toHaveBeenCalled(); + + // A concurrent publish-side task must be able to acquire and release publishMutex while + // unpublish is still waiting for "connected" - proving the wait doesn't hold the mutex. + const otherTask = jest.fn().mockResolvedValue(undefined); + await (brtc as any).publishMutex.runExclusive(otherTask); + expect(otherTask).toHaveBeenCalledTimes(1); + + pc.connectionState = "connected"; + await unpublishPromise; + + expect(offerSdp).toHaveBeenCalledTimes(1); + }); }); describe("bandwidthRtcV1 init reconnect replay", () => { diff --git a/src/v1/bandwidthRtc.ts b/src/v1/bandwidthRtc.ts index c2ecb0c..9b71d52 100644 --- a/src/v1/bandwidthRtc.ts +++ b/src/v1/bandwidthRtc.ts @@ -316,19 +316,21 @@ export class BandwidthRtc { return; } + // Stop local media first so unpublish takes effect even if renegotiation fails. The + // mutex keeps transceiver removal out of an in-flight negotiation. await this.publishMutex.runExclusive(async () => { - // Stop the local tracks first regardless of what happens next - the user's intent - // (stop sending this media) must take effect immediately. this.cleanupPublishedStreams(...publishedStreams); - try { - // The gateway rejects offers unless the publish peer is "connected"; wait it out - // rather than sending an offer doomed to be rejected during a brief ICE blip. - await this.waitForPublishConnected(); - await this.negotiatePublishSdp(); - } catch (err) { - throw new BandwidthRtcError(`Stream(s) were unpublished locally, but renegotiation with the gateway failed: ${err}`); - } }); + + try { + // The gateway rejects offers until the publish peer is connected. Wait outside the + // mutex so a gateway-initiated ICE restart is not blocked; a renegotiation in between + // already carries the removed transceivers, which makes ours a no-op. + await this.waitForPublishConnected(); + await this.offerPublishSdp(); + } catch (err) { + throw new BandwidthRtcError(`Stream(s) were unpublished locally, but renegotiation with the gateway failed: ${err}`); + } } /**