diff --git a/.changeset/paused-upstream-republish.md b/.changeset/paused-upstream-republish.md new file mode 100644 index 0000000000..8ff32801f1 --- /dev/null +++ b/.changeset/paused-upstream-republish.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +A track whose upstream is paused stays paused when it is republished, e.g. after a full reconnect: the new sender starts without the track and the server is told the track is muted. diff --git a/src/room/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index 23195d8592..ab9278b687 100644 --- a/src/room/RTCEngine.test.ts +++ b/src/room/RTCEngine.test.ts @@ -16,6 +16,7 @@ import { roomOptionDefaults } from './defaults'; import { ConnectionError, PublishDataError, UnexpectedConnectionState } from './errors'; import { EngineEvent } from './events'; import CriticalTimers from './timers'; +import type LocalVideoTrack from './track/LocalVideoTrack'; import { Future } from './utils'; describe('RTCEngine', () => { @@ -623,6 +624,68 @@ describe('RTCEngine', () => { }); }); + describe('createSender', () => { + // a full reconnect republishes every track over a new sender, including one whose upstream + // the application paused + function stubPublisher(engine: RTCEngine) { + vi.stubGlobal( + 'RTCPeerConnection', + class { + addTransceiver() {} + }, + ); + const sender = { replaceTrack: vi.fn(async () => {}) }; + Object.assign(engine as unknown as Record, { + pcManager: { addPublisherTransceiver: vi.fn(async () => ({ sender })) }, + }); + return sender; + } + + function makeTrack(isUpstreamPaused: boolean) { + return { + kind: 'video', + isUpstreamPaused, + mediaStreamTrack: { id: 'camera' }, + setSimulcastTrackSender: vi.fn(async () => {}), + } as unknown as LocalVideoTrack; + } + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('keeps a paused track off the new sender', async () => { + const engine = new RTCEngine(roomOptionDefaults); + const sender = stubPublisher(engine); + + await engine.createSender(makeTrack(true), {}); + + expect(sender.replaceTrack).toHaveBeenCalledExactlyOnceWith(null); + }); + + it('keeps a paused track off a new backup codec sender', async () => { + const engine = new RTCEngine(roomOptionDefaults); + const sender = stubPublisher(engine); + + await engine.createSimulcastSender( + makeTrack(true), + { codec: 'vp8', mediaStreamTrack: { id: 'camera-clone' } as MediaStreamTrack }, + { videoCodec: 'vp8' }, + ); + + expect(sender.replaceTrack).toHaveBeenCalledExactlyOnceWith(null); + }); + + it('leaves a track that is not paused on the new sender', async () => { + const engine = new RTCEngine(roomOptionDefaults); + const sender = stubPublisher(engine); + + await engine.createSender(makeTrack(false), {}); + + expect(sender.replaceTrack).not.toHaveBeenCalled(); + }); + }); + describe('local connection quality Lost handling', () => { // The engine reacts to the server's own verdict: a sustained local `LOST` while // connected and publishing means our media isn't reaching the server, so it forces diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index 8fd42ad9f6..96920cd570 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -1095,6 +1095,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit throw new UnexpectedConnectionState('Required webRTC APIs not supported on this device'); } this.setupFrameMetadataSender(sender, opts); + await this.keepPausedTrackOffSender(track, sender); return sender; } @@ -1115,10 +1116,19 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } if (sender) { this.setupFrameMetadataSender(sender, opts); + await this.keepPausedTrackOffSender(track, sender); } return sender; } + // a track republished while its upstream is paused (e.g. after a full reconnect) gets a new + // sender, which pauseUpstream() never saw: the sender has to start without the track + private async keepPausedTrackOffSender(track: LocalTrack, sender: RTCRtpSender) { + if (track.isUpstreamPaused) { + await sender.replaceTrack(null); + } + } + private get frameMetadataWorker(): Worker | undefined { return this.frameMetadataOptions?.worker; } diff --git a/src/room/participant/LocalParticipant.test.ts b/src/room/participant/LocalParticipant.test.ts index 0a082000c0..0878654878 100644 --- a/src/room/participant/LocalParticipant.test.ts +++ b/src/room/participant/LocalParticipant.test.ts @@ -217,6 +217,16 @@ describe('LocalParticipant.publishAdditionalCodecForTrack', () => { expect(req.simulcastCodecs[0].layers).toEqual(req.layers); }); + it('announces the backup codec of a paused track as muted', async () => { + const track = makeBackupCodecTrack({ width: 1280, height: 720 }); + Object.assign(track, { isUpstreamPaused: true }); + const participant = makeBackupCodecParticipant(track); + + await publishBackupCodec(participant, track); + + expect(capturedRequest(participant).muted).toBe(true); + }); + it('omits layers when the capture dimensions are unknown', async () => { const track = makeBackupCodecTrack({}); const participant = makeBackupCodecParticipant(track); @@ -231,3 +241,51 @@ describe('LocalParticipant.publishAdditionalCodecForTrack', () => { expect(participant.log.warn).toHaveBeenCalledOnce(); }); }); + +describe('LocalParticipant republishing a track whose upstream is paused', () => { + // a full reconnect unpublishes every track and publishes it again over a new sender, while + // the track keeps its paused upstream state + function makePausedMicrophone() { + return { + isLocal: true, + kind: Track.Kind.Audio, + source: Track.Source.Microphone, + isMuted: false, + isUpstreamPaused: true, + mediaStreamTrack: { id: 'microphone-cid' }, + on: vi.fn(), + getSourceTrackSettings: () => ({}), + } as unknown as LocalTrack; + } + + async function capturePublishRequest(track: LocalTrack) { + const addTrack = vi.fn(async () => { + throw new Error('stop after the add track request'); + }); + const participant = Object.create(LocalParticipant.prototype) as Record; + Object.assign(participant, { + encryptionType: Encryption_Type.NONE, + trackPublications: new Map(), + enabledPublishVideoCodecs: [], + hasPermissionsToPublish: () => true, + canPublishFrameMetadata: () => false, + engine: { isClosed: false, addTrack }, + log: { warn: vi.fn(), debug: vi.fn(), info: vi.fn() }, + }); + const publish = ( + participant as unknown as { + publish: (track: LocalTrack, opts: TrackPublishOptions, isStereo: boolean) => Promise; + } + ).publish.bind(participant); + + await expect(publish(track, {}, false)).rejects.toThrow('stop after the add track request'); + expect(addTrack).toHaveBeenCalledOnce(); + return addTrack.mock.calls[0][0] as AddTrackRequest; + } + + it('announces the track as muted to the server', async () => { + const req = await capturePublishRequest(makePausedMicrophone()); + + expect(req.muted).toBe(true); + }); +}); diff --git a/src/room/participant/LocalParticipant.ts b/src/room/participant/LocalParticipant.ts index 5061a966d6..b3cd351753 100644 --- a/src/room/participant/LocalParticipant.ts +++ b/src/room/participant/LocalParticipant.ts @@ -929,7 +929,8 @@ export class LocalParticipant extends Participant { cid: track.mediaStreamTrack.id, name: opts.name, type: Track.kindToProto(track.kind), - muted: track.isMuted, + // the server sees a paused upstream as muted, see onTrackUpstreamPaused + muted: track.isMuted || track.isUpstreamPaused, source: Track.sourceToProto(track.source), disableDtx, encryption: this.encryptionType, diff --git a/src/room/video/publisher.ts b/src/room/video/publisher.ts index 5e9c3914ee..6122bd43a0 100644 --- a/src/room/video/publisher.ts +++ b/src/room/video/publisher.ts @@ -403,7 +403,8 @@ export async function publishAdditionalCodecForTrack( const req = new AddTrackRequest({ cid: simulcastTrack.mediaStreamTrack.id, type: Track.kindToProto(track.kind), - muted: track.isMuted, + // match the primary publish request, the backup sender starts without a paused track + muted: track.isMuted || track.isUpstreamPaused, source: Track.sourceToProto(track.source), sid: track.sid, width: dims?.width,