diff --git a/.changeset/backup-codec-track-clones.md b/.changeset/backup-codec-track-clones.md new file mode 100644 index 0000000000..46cca99af8 --- /dev/null +++ b/.changeset/backup-codec-track-clones.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +Backup codecs now follow the track after `replaceTrack`, an unmute of an SDK managed camera or a processor change. Before this fix they kept sending the clone taken at publish time, so a backup codec subscriber could see the old track or no video at all, and the clone was never stopped, which kept the camera on. diff --git a/src/room/participant/LocalParticipant.test.ts b/src/room/participant/LocalParticipant.test.ts index 909f398027..693c33b845 100644 --- a/src/room/participant/LocalParticipant.test.ts +++ b/src/room/participant/LocalParticipant.test.ts @@ -214,3 +214,37 @@ describe('LocalParticipant.publishAdditionalCodecForTrack', () => { expect(participant.log.warn).toHaveBeenCalledOnce(); }); }); + +describe('LocalParticipant.unpublishTrack', () => { + // a disconnect closes the engine, and so drops the peer connections, before unpublishing + it('stops and drops the backup codec clones when the peer connection is gone', async () => { + const clone = { stop: vi.fn() }; + const track = { + isLocal: true, + kind: Track.Kind.Video, + source: Track.Source.Camera, + mediaStreamTrack: { enabled: true, id: 'media-track-id' }, + sender: {}, + simulcastCodecs: new Map([['vp8', { codec: 'vp8', mediaStreamTrack: clone, sender: {} }]]), + off: vi.fn(), + stopMonitor: vi.fn(), + } as unknown as LocalVideoTrack; + const publication = { track, trackSid: 'track-sid', kind: Track.Kind.Video, setTrack: vi.fn() }; + const participant = Object.create(LocalParticipant.prototype) as LocalParticipant; + Object.assign(participant, { + log: { info: vi.fn(), warn: vi.fn(), debug: vi.fn() }, + pendingPublishPromises: new Map(), + getPublicationForTrack: () => publication, + engine: { pcManager: undefined }, + trackPublications: new Map([['track-sid', publication]]), + videoTrackPublications: new Map([['track-sid', publication]]), + emit: vi.fn(), + }); + + // kept for republishing, so the track itself is not stopped + await participant.unpublishTrack(track, false); + + expect(clone.stop).toHaveBeenCalledOnce(); + expect(track.simulcastCodecs.size).toBe(0); + }); +}); diff --git a/src/room/participant/LocalParticipant.ts b/src/room/participant/LocalParticipant.ts index 3407d126b1..69c22e278c 100644 --- a/src/room/participant/LocalParticipant.ts +++ b/src/room/participant/LocalParticipant.ts @@ -1674,13 +1674,19 @@ export default class LocalParticipant extends Participant { trackInfo.sender = undefined; } } - track.simulcastCodecs.clear(); } } catch (e) { this.log.warn('failed to unpublish track', { ...pubLogContext, error: e }); } } + if (isLocalVideoTrack(track)) { + // the backup codec clones are dropped with the map, so they have to be stopped even when the + // track itself is kept for republishing, or the peer connection is already gone + track.simulcastCodecs.forEach((trackInfo) => trackInfo.mediaStreamTrack.stop()); + track.simulcastCodecs.clear(); + } + // remove from our maps this.trackPublications.delete(publication.trackSid); switch (publication.kind) { diff --git a/src/room/track/LocalVideoTrack.test.ts b/src/room/track/LocalVideoTrack.test.ts index 03c2e63afb..fc666146a6 100644 --- a/src/room/track/LocalVideoTrack.test.ts +++ b/src/room/track/LocalVideoTrack.test.ts @@ -1,8 +1,10 @@ -import { describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { Mutex } from '@livekit/mutex'; import LocalVideoTrack, { videoLayersFromEncodings } from './LocalVideoTrack'; import type { SimulcastTrackInfo } from './LocalVideoTrack'; -import { VideoQuality } from './Track'; +import { Track, VideoQuality } from './Track'; import type { VideoCodec } from './options'; +import type { TrackProcessor } from './processor/types'; describe('videoLayersFromEncodings', () => { it('returns single layer for no encoding', () => { @@ -234,3 +236,272 @@ describe('setDegradationPreference', () => { expect(events).toEqual(['primary:start', 'primary:done', 'backup:start', 'backup:done']); }); }); + +describe('backup codec tracks', () => { + // happy-dom's user agent reads as an old Safari, which refuses pauseUpstream + beforeEach(() => { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue( + 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36', + ); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + type FakeTrack = MediaStreamTrack & { clones: FakeTrack[] }; + + function makeMediaStreamTrack(id: string): FakeTrack { + const track = { + id, + kind: 'video', + enabled: true, + muted: false, + readyState: 'live', + clones: [] as FakeTrack[], + clone: () => { + const clone = makeMediaStreamTrack(`${id}-clone${track.clones.length}`); + track.clones.push(clone); + return clone; + }, + stop: () => { + track.readyState = 'ended'; + }, + getConstraints: () => ({}), + getSettings: () => ({}), + addEventListener: () => {}, + removeEventListener: () => {}, + }; + return track as unknown as FakeTrack; + } + + function makeTrackSender() { + const sender = { + track: null as MediaStreamTrack | null, + replaceTrack: vi.fn(async (track: MediaStreamTrack | null) => { + sender.track = track; + }), + }; + return sender; + } + + async function publishWithBackupCodec(source: MediaStreamTrack, userProvidedTrack = true) { + const track = new LocalVideoTrack(source, undefined, userProvidedTrack); + // the constructor sets the track up under this lock, wait for it like a real publish would + const { trackChangeLock } = track as unknown as { trackChangeLock: Mutex }; + (await trackChangeLock.lock())(); + const primary = makeTrackSender(); + // bypass the sender setter, which would apply the degradation preference + Object.assign(track, { _sender: primary }); + await primary.replaceTrack(track.mediaStreamTrack); + + const info = track.addSimulcastTrack('vp8', [])!; + const backup = makeTrackSender(); + info.sender = backup as unknown as RTCRtpSender; + await backup.replaceTrack(info.mediaStreamTrack); + return { track, info, backup }; + } + + // the backup sender rejects the first swap to a clone of the track `id` + function rejectFirstSwapTo(backup: ReturnType, id: string) { + let rejected = false; + backup.replaceTrack.mockImplementation(async (next: MediaStreamTrack | null) => { + if (!rejected && next?.id.startsWith(id)) { + rejected = true; + throw new Error('rejected'); + } + backup.track = next; + }); + } + + // lets queued work, such as a retry waiting for the track change lock, run + const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); + + // a processor whose init() waits until the test rejects it + function makeFailingProcessor() { + let rejectInit: (error: Error) => void = () => {}; + const processor = { + name: 'failing', + init: () => + new Promise((_, reject) => { + rejectInit = reject; + }), + restart: async () => {}, + destroy: async () => {}, + } as TrackProcessor; + return { processor, rejectInit: (error: Error) => rejectInit(error) }; + } + + it('sends a clone of the replacement track on the backup codec and stops the old clone', async () => { + const camera = makeMediaStreamTrack('camera'); + const { track, info, backup } = await publishWithBackupCodec(camera); + const oldClone = info.mediaStreamTrack; + expect(backup.track).toBe(oldClone); + + // turning a user provided camera off and on again + await track.pauseUpstream(); + expect(backup.track).toBeNull(); + camera.stop(); + const newCamera = makeMediaStreamTrack('new-camera'); + await track.replaceTrack(newCamera); + + expect(track.mediaStreamTrack).toBe(newCamera); + expect(info.mediaStreamTrack).toBe(newCamera.clones[0]); + expect(backup.track).toBe(newCamera.clones[0]); + expect(oldClone.readyState).toBe('ended'); + + // resuming after another pause sends the current clone, not the first one + await track.pauseUpstream(); + await track.resumeUpstream(); + expect(backup.track).toBe(newCamera.clones[0]); + }); + + it('stops the backup codec when its sender rejects the new track, and retries it with the same track', async () => { + const cameraA = makeMediaStreamTrack('camera'); + const { track, info, backup } = await publishWithBackupCodec(cameraA); + const cloneA = info.mediaStreamTrack; + rejectFirstSwapTo(backup, 'camera-b'); + + // 1. replacing A with B fails on the backup sender + const cameraB = makeMediaStreamTrack('camera-b'); + await expect(track.replaceTrack(cameraB)).rejects.toThrow('rejected'); + + // 2. the backup sender stops sending, 3. and A's clone is stopped: the application replaced A + expect(backup.replaceTrack).toHaveBeenLastCalledWith(null); + expect(backup.track).toBeNull(); + expect(cloneA.readyState).toBe('ended'); + expect(cameraB.clones[0].readyState).toBe('ended'); + + // 4. retrying with the same track sends a fresh clone of it + await track.replaceTrack(cameraB); + expect(cameraB.clones).toHaveLength(2); + expect(backup.track).toBe(cameraB.clones[1]); + expect(info.mediaStreamTrack).toBe(cameraB.clones[1]); + expect(cameraB.clones[1].readyState).toBe('live'); + }); + + it('retries a backup codec whose sender rejected the track when upstream resumes', async () => { + const { track, info, backup } = await publishWithBackupCodec(makeMediaStreamTrack('camera')); + rejectFirstSwapTo(backup, 'camera-b'); + const cameraB = makeMediaStreamTrack('camera-b'); + await expect(track.replaceTrack(cameraB)).rejects.toThrow('rejected'); + + await track.pauseUpstream(); + await track.resumeUpstream(); + + // a fresh clone of B, not the clone stopped when the swap failed + expect(cameraB.clones).toHaveLength(2); + expect(backup.track).toBe(cameraB.clones[1]); + expect(info.mediaStreamTrack).toBe(cameraB.clones[1]); + expect(cameraB.clones[1].readyState).toBe('live'); + }); + + it('clones a replacement track once for a backup codec whose sender rejected the previous one', async () => { + const { track, backup } = await publishWithBackupCodec(makeMediaStreamTrack('camera')); + rejectFirstSwapTo(backup, 'camera-b'); + await expect(track.replaceTrack(makeMediaStreamTrack('camera-b'))).rejects.toThrow('rejected'); + + const cameraC = makeMediaStreamTrack('camera-c'); + await track.replaceTrack(cameraC); + await flush(); + + // the track change refreshes every backup codec, so resuming upstream on the way must not + // swap the rejected one, neither before nor after + expect(cameraC.clones).toHaveLength(1); + expect(backup.track).toBe(cameraC.clones[0]); + expect(cameraC.clones[0].readyState).toBe('live'); + }); + + it('retries a rejected backup codec after a track change that fails before refreshing it', async () => { + const { track, info, backup } = await publishWithBackupCodec(makeMediaStreamTrack('camera')); + rejectFirstSwapTo(backup, 'camera-b'); + const cameraB = makeMediaStreamTrack('camera-b'); + await expect(track.replaceTrack(cameraB)).rejects.toThrow('rejected'); + await track.pauseUpstream(); + + // resuming while setProcessor() holds the track change lock, then the processor fails + const { processor, rejectInit } = makeFailingProcessor(); + const setup = track.setProcessor(processor); + await flush(); + await track.resumeUpstream(); + rejectInit(new Error('init failed')); + await expect(setup).rejects.toThrow('init failed'); + await flush(); + + expect(cameraB.clones).toHaveLength(2); + expect(backup.track).toBe(cameraB.clones[1]); + expect(info.mediaStreamTrack).toBe(cameraB.clones[1]); + expect(cameraB.clones[1].readyState).toBe('live'); + }); + + it('keeps a newer backup codec track when an older swap is rejected after it', async () => { + const { track, info, backup } = await publishWithBackupCodec(makeMediaStreamTrack('camera')); + let rejectRetry: ((error: Error) => void) | undefined; + backup.replaceTrack.mockImplementation(async (next: MediaStreamTrack | null) => { + if (next?.id === 'camera-b-clone0') { + throw new Error('rejected'); + } + if (next?.id === 'camera-b-clone1' && !rejectRetry) { + // the retry's swap is still pending when the next camera comes in, then fails + return new Promise((_, reject) => { + rejectRetry = reject; + }); + } + backup.track = next; + }); + await expect(track.replaceTrack(makeMediaStreamTrack('camera-b'))).rejects.toThrow('rejected'); + + // a retry is queued behind a failing track change, and starts its swap + const { processor, rejectInit } = makeFailingProcessor(); + const setup = track.setProcessor(processor); + await flush(); + await track.resumeUpstream(); + rejectInit(new Error('init failed')); + await expect(setup).rejects.toThrow('init failed'); + await flush(); + expect(rejectRetry).toBeDefined(); + + const cameraC = makeMediaStreamTrack('camera-c'); + await track.replaceTrack(cameraC); + rejectRetry!(new Error('rejected')); + await flush(); + + // the failed retry must not take the newer camera off the backup codec + expect(backup.track).toBe(cameraC.clones[0]); + expect(info.mediaStreamTrack).toBe(cameraC.clones[0]); + expect(cameraC.clones[0].readyState).toBe('live'); + expect(info.needsRefresh).toBe(false); + }); + + it('drops a pending backup codec retry when the track is stopped first', async () => { + const { track, backup } = await publishWithBackupCodec(makeMediaStreamTrack('camera')); + rejectFirstSwapTo(backup, 'camera-b'); + const cameraB = makeMediaStreamTrack('camera-b'); + await expect(track.replaceTrack(cameraB)).rejects.toThrow('rejected'); + + const { processor, rejectInit } = makeFailingProcessor(); + const setup = track.setProcessor(processor); + await flush(); + await track.resumeUpstream(); + track.stop(); + rejectInit(new Error('init failed')); + await expect(setup).rejects.toThrow('init failed'); + await flush(); + + // a retry now would open the capture again through a new clone + expect(cameraB.clones).toHaveLength(1); + expect(backup.track).toBeNull(); + }); + + it('stops the backup codec clone when an SDK managed camera is muted', async () => { + const camera = makeMediaStreamTrack('camera'); + const { track, info } = await publishWithBackupCodec(camera, false); + Object.defineProperty(track, 'source', { get: () => Track.Source.Camera }); + + await track.mute(); + + // the clone shares the capture device, so leaving it live keeps the camera on + expect(camera.readyState).toBe('ended'); + expect(info.mediaStreamTrack.readyState).toBe('ended'); + }); +}); diff --git a/src/room/track/LocalVideoTrack.ts b/src/room/track/LocalVideoTrack.ts index 65f72f2647..3dc901d99b 100644 --- a/src/room/track/LocalVideoTrack.ts +++ b/src/room/track/LocalVideoTrack.ts @@ -33,6 +33,9 @@ export class SimulcastTrackInfo { encodings?: RTCRtpEncodingParameters[]; + /** @internal set when the sender rejected the current track, so that the next swap or resume retries it */ + needsRefresh?: boolean; + constructor(codec: VideoCodec, mediaStreamTrack: MediaStreamTrack) { this.codec = codec; this.mediaStreamTrack = mediaStreamTrack; @@ -145,10 +148,39 @@ export default class LocalVideoTrack extends LocalTrack { async resumeUpstream() { await super.resumeUpstream(); for await (const sc of this.simulcastCodecs.values()) { - await sc.sender?.replaceTrack(sc.mediaStreamTrack); + if (!sc.needsRefresh) { + await sc.sender?.replaceTrack(sc.mediaStreamTrack); + } else if (this.trackChangeLock.isLocked()) { + this.retryAfterTrackChange(sc); + } else { + // its clone was stopped when the sender rejected it + await this.refreshSimulcastTrack(sc); + } } } + /** + * A track change in progress refreshes every backup codec when it succeeds, but it can fail + * before that. Not awaited: this can run from within the track change itself. + */ + private retryAfterTrackChange(sc: SimulcastTrackInfo) { + this.trackChangeLock + .lock() + .then(async (unlock) => { + unlock(); + // still pending, and not dropped by an unpublish or a stop in the meantime + if (sc.needsRefresh && !this.isStopped && this.simulcastCodecs.get(sc.codec) === sc) { + await this.refreshSimulcastTrack(sc); + } + }) + .catch((error) => { + this.log.warn('failed to retry backup codec after a track change', { + ...this.logContext, + error, + }); + }); + } + async mute(): Promise { const unlock = await this.muteLock.lock(); try { @@ -161,6 +193,8 @@ export default class LocalVideoTrack extends LocalTrack { this.log.debug('stopping camera track', this.logContext); // also stop the track, so that camera indicator is turned off this._mediaStreamTrack.stop(); + // backup codecs hold their own clones of the camera, which keep it on as well + this.simulcastCodecs.forEach((sc) => sc.mediaStreamTrack.stop()); } await super.mute(); return this; @@ -280,13 +314,6 @@ export default class LocalVideoTrack extends LocalTrack { // reset cpu constrained state after track is restarted this.isCpuConstrained = false; - for await (const sc of this.simulcastCodecs.values()) { - if (sc.sender && sc.sender.transport?.state !== 'closed') { - sc.mediaStreamTrack = this.mediaStreamTrack.clone(); - await sc.sender.replaceTrack(sc.mediaStreamTrack); - } - } - // The new MediaStreamTrack may have different dimensions than the previous one // (e.g. switching between cameras with different native resolutions), which would // leave the sender's encoding parameters (scaleResolutionDownBy, maxBitrate, etc.) @@ -296,9 +323,56 @@ export default class LocalVideoTrack extends LocalTrack { } protected override async onSenderTrackSwapped(): Promise { + // replaceTrack() with the same track skips setMediaStreamTrack(), so a backup codec whose + // sender rejected that track is retried here + for (const sc of this.simulcastCodecs.values()) { + if (sc.needsRefresh) { + await this.refreshSimulcastTrack(sc); + } + } await this.refreshSenderEncodings(); } + protected override async onMediaStreamTrackChanged(): Promise { + await this.refreshSimulcastTracks(); + } + + /** + * Each backup codec sends its own clone of the track, which `resumeUpstream` hands back to its + * sender. Whenever the track (or the processed track) changes, every backup codec needs a fresh + * clone of it, and the clone it replaces has to be stopped, or it would keep the capture alive. + */ + private async refreshSimulcastTracks() { + for (const sc of this.simulcastCodecs.values()) { + await this.refreshSimulcastTrack(sc); + } + } + + private async refreshSimulcastTrack(sc: SimulcastTrackInfo) { + const previousTrack = sc.mediaStreamTrack; + const newTrack = this.mediaStreamTrack.clone(); + sc.mediaStreamTrack = newTrack; + sc.needsRefresh = false; + if (sc.sender && sc.sender.transport?.state !== 'closed' && !this.isUpstreamPaused) { + try { + await sc.sender.replaceTrack(newTrack); + } catch (e) { + newTrack.stop(); + previousTrack.stop(); + // unless a newer refresh took the codec over while this swap was pending: the application + // replaced the previous track, so it must not keep being published, and this codec sends + // nothing until a swap or a resume retries it + if (sc.mediaStreamTrack === newTrack) { + sc.needsRefresh = true; + // a sender that rejects null as well is not sending anything anymore + await sc.sender.replaceTrack(null).catch(() => undefined); + } + throw e; + } + } + previousTrack.stop(); + } + /** * Recomputes encoding parameters for this track's senders based on the current * MediaStreamTrack dimensions and reapplies them via setParameters. This is a no-op @@ -415,9 +489,7 @@ export default class LocalVideoTrack extends LocalTrack { await super.setProcessor(processor, showProcessedStreamLocally); if (this.processor?.processedTrack) { - for await (const sc of this.simulcastCodecs.values()) { - await sc.sender?.replaceTrack(this.processor.processedTrack); - } + await this.refreshSimulcastTracks(); } }