diff --git a/fluxer_app/pkgs/livekit-client/VENDORING.md b/fluxer_app/pkgs/livekit-client/VENDORING.md index 04b0ae433..b9ec4fab6 100644 --- a/fluxer_app/pkgs/livekit-client/VENDORING.md +++ b/fluxer_app/pkgs/livekit-client/VENDORING.md @@ -61,6 +61,41 @@ source edits in this package. `gainNode` branch; the `el.volume` branch would throw `IndexSizeError`. `webAudioMix` must stay unconditional. +11. **Processor teardown before source stop** (`src/room/track/LocalTrack.ts`) + `stop()` called `super.stop()` first, killing the source `MediaStreamTrack` + and closing the readable feeding a track processor before `processor.destroy()` + ran. A camera-effect worker therefore saw input EOF before its owner's stop + command and reported an operational failure during an ordinary camera-off. + The processor is now captured, detached, and its teardown initiated before + `super.stop()`. + +12. **Transactional source and processor swaps** (`src/room/track/LocalTrack.ts`, + `LocalVideoTrack.ts`, `LocalAudioTrack.ts`) + `setMediaStreamTrack()` applied the new source, restarted the processor and + re-armed the sender with no unwind path, so a failure anywhere in the middle + left a half-applied track: listeners moved, elements detached, sender pointing + at a dead track. It now takes `SetMediaStreamTrackOptions` + (`force`, `deferEndedListener`, `preservePreviousTrack`) and, on failure, + restores the previous source, constraints, `enabled` state, listeners, + processor and sender, throwing `TrackInvalidError` when the previous source is + no longer `live` because an ended track cannot be restored. Both errors are + surfaced together as an `AggregateError` when the unwind itself fails. + `stageTrackReplacement()` / `commitStagedTrackReplacement()` expose a two-phase + swap: the candidate becomes the active source with its `ended` listener + deferred and the previous source preserved, and only the commit adopts the + `ended` listener and clears the staged identity, so a caller can validate its + publication before the swap is observable. `replaceTrack()` and `restart()` + guard the `providedByUser` flip behind a `replacementCommitted` flag. + `restart()` still detaches and stops the previous source before calling + `getUserMedia()`, as upstream does, because Safari ends a freshly acquired + track with a capture failure while the old track for the same device is + live. `setSimulcastTrackSender()` routes an already-installed processor's + `processedTrack` to a newly registered secondary sender so a backup codec + never publishes raw frames while the primary is processed. + Processor install and teardown in all three classes roll the processed/raw + sender track back, including `LocalVideoTrack`'s secondary simulcast senders, + and aggregate every cleanup failure instead of discarding it. + ## Updating from upstream 1. Check the upstream changelog for the target version. diff --git a/fluxer_app/pkgs/livekit-client/src/room/track/LocalAudioTrack.ts b/fluxer_app/pkgs/livekit-client/src/room/track/LocalAudioTrack.ts index 4ff1fa71f..4bb9d9fd0 100644 --- a/fluxer_app/pkgs/livekit-client/src/room/track/LocalAudioTrack.ts +++ b/fluxer_app/pkgs/livekit-client/src/room/track/LocalAudioTrack.ts @@ -159,21 +159,69 @@ export default class LocalAudioTrack extends LocalTrack { audioContext: this.audioContext as AudioContext, }; this.log.debug(`setting up audio processor ${processor.name}`, this.logContext); - - await processor.init(processorOptions); - this.processor = processor; - if (this.processor.processedTrack) { - await this.sender?.replaceTrack(this.processor.processedTrack); - this.processor.processedTrack.addEventListener( - 'enable-lk-krisp-noise-filter', - this.handleKrispNoiseFilterEnable, - ); - this.processor.processedTrack.addEventListener( - 'disable-lk-krisp-noise-filter', - this.handleKrispNoiseFilterDisable, - ); + try { + await processor.init(processorOptions); + } catch (error) { + try { + await processor.destroy(); + } catch (cleanupError) { + throw new AggregateError( + [error, cleanupError], + 'Audio track processor setup and candidate cleanup both failed', + ); + } + throw error; + } + const processedTrack = processor.processedTrack; + try { + if (processedTrack) await this.sender?.replaceTrack(processedTrack); + this.processor = processor; + if (processedTrack) { + processedTrack.addEventListener('enable-lk-krisp-noise-filter', this.handleKrispNoiseFilterEnable); + processedTrack.addEventListener('disable-lk-krisp-noise-filter', this.handleKrispNoiseFilterDisable); + } + this.emit(TrackEvent.TrackProcessorUpdate, processor); + } catch (error) { + const cleanupErrors: Array = []; + if (this.processor === processor) this.processor = undefined; + processedTrack?.removeEventListener('enable-lk-krisp-noise-filter', this.handleKrispNoiseFilterEnable); + processedTrack?.removeEventListener('disable-lk-krisp-noise-filter', this.handleKrispNoiseFilterDisable); + try { + await processor.destroy(); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + if (processedTrack && processedTrack.readyState !== 'ended') { + processedTrack.enabled = false; + try { + processedTrack.stop(); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + } + const sender = this.sender; + if (sender && sender.transport?.state !== 'closed') { + const rawSenderTrack = this._mediaStreamTrack.readyState === 'live' ? this._mediaStreamTrack : null; + if (sender.track !== rawSenderTrack) { + try { + await sender.replaceTrack(rawSenderTrack); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + if (sender.track?.readyState === 'ended') { + try { + await sender.replaceTrack(null); + } catch (failCloseError) { + cleanupErrors.push(failCloseError); + } + } + } + } + } + if (cleanupErrors.length > 0) { + throw new AggregateError([error, ...cleanupErrors], 'Audio track processor install rollback was incomplete'); + } + throw error; } - this.emit(TrackEvent.TrackProcessorUpdate, this.processor); } finally { unlock(); } diff --git a/fluxer_app/pkgs/livekit-client/src/room/track/LocalTrack.ts b/fluxer_app/pkgs/livekit-client/src/room/track/LocalTrack.ts index de11982b0..2217a1fdb 100644 --- a/fluxer_app/pkgs/livekit-client/src/room/track/LocalTrack.ts +++ b/fluxer_app/pkgs/livekit-client/src/room/track/LocalTrack.ts @@ -19,6 +19,12 @@ import type {ReplaceTrackOptions} from './types.ts'; const DEFAULT_DIMENSIONS_TIMEOUT = 1000; const PRE_CONNECT_BUFFER_TIMEOUT = 10_000; +interface SetMediaStreamTrackOptions { + force?: boolean; + deferEndedListener?: boolean; + preservePreviousTrack: boolean; +} + export default abstract class LocalTrack extends Track { protected _sender?: RTCRtpSender; @@ -66,6 +72,8 @@ export default abstract class LocalTrack { try { - await this.setMediaStreamTrack(mediaTrack, true); + await this.setMediaStreamTrack(mediaTrack, { + force: true, + preservePreviousTrack: userProvidedTrack, + }); } finally { unlock(); } @@ -134,58 +145,131 @@ export default abstract class LocalTrack { - detachTrack(this._mediaStreamTrack, el); - }); - this.debouncedTrackMuteHandler.cancel('new-track'); - this._mediaStreamTrack.removeEventListener('ended', this.handleEnded); - this._mediaStreamTrack.removeEventListener('mute', this.handleTrackMuteEvent); - this._mediaStreamTrack.removeEventListener('unmute', this.handleTrackUnmuteEvent); + private addMediaStreamTrackListeners(track: MediaStreamTrack, includeEndedListener = true) { + if (includeEndedListener) { + track.addEventListener('ended', this.handleEnded); } + track.addEventListener('mute', this.handleTrackMuteEvent); + track.addEventListener('unmute', this.handleTrackUnmuteEvent); + } - this.mediaStream = new MediaStream([newTrack]); - if (newTrack) { - newTrack.addEventListener('ended', this.handleEnded); - newTrack.addEventListener('mute', this.handleTrackMuteEvent); - newTrack.addEventListener('unmute', this.handleTrackUnmuteEvent); - this._constraints = newTrack.getConstraints(); + private removeMediaStreamTrackListeners(track: MediaStreamTrack) { + track.removeEventListener('ended', this.handleEnded); + track.removeEventListener('mute', this.handleTrackMuteEvent); + track.removeEventListener('unmute', this.handleTrackUnmuteEvent); + } + + private async restoreMediaStreamTrackAfterFailure( + previousTrack: MediaStreamTrack, + previousConstraints: MediaTrackConstraints, + previousEnabled: boolean, + failedTrack: MediaStreamTrack, + failedProcessedTrack: MediaStreamTrack | undefined, + previousTrackEndedListenerDeferred: boolean, + ) { + this.removeMediaStreamTrackListeners(failedTrack); + for (const element of this.attachedElements) { + detachTrack(failedTrack, element); + if (failedProcessedTrack) { + detachTrack(failedProcessedTrack, element); + } } - let processedTrack: MediaStreamTrack | undefined; - if (this.processor && newTrack) { - this.log.debug('restarting processor', this.logContext); + if (previousTrack.readyState !== 'live') { + throw new TrackInvalidError('unable to restore an ended track after replacement failure'); + } + this.mediaStream = new MediaStream([previousTrack]); + this._mediaStreamTrack = previousTrack; + this._constraints = previousConstraints; + previousTrack.enabled = previousEnabled; + this.addMediaStreamTrackListeners(previousTrack, !previousTrackEndedListenerDeferred); + let restoredProcessedTrack: MediaStreamTrack | undefined; + if (this.processor) { if (this.kind === 'unknown') { - throw TypeError('cannot set processor on track of unknown kind'); + throw TypeError('cannot restore processor on track of unknown kind'); } - if (this.processorElement) { - attachToElement(newTrack, this.processorElement); + attachToElement(previousTrack, this.processorElement); this.processorElement.muted = true; } await this.processor.restart({ - track: newTrack, + track: previousTrack, kind: this.kind, element: this.processorElement, }); - processedTrack = this.processor.processedTrack; + restoredProcessedTrack = this.processor.processedTrack; } if (this.sender && this.sender.transport?.state !== 'closed') { - await this.sender.replaceTrack(processedTrack ?? newTrack); + await this.sender.replaceTrack(restoredProcessedTrack ?? previousTrack); } - if (!this.providedByUser && this._mediaStreamTrack !== newTrack) { - this._mediaStreamTrack.stop(); + await this.resumeUpstream(); + for (const element of this.attachedElements) { + attachToElement(restoredProcessedTrack ?? previousTrack, element); } - this._mediaStreamTrack = newTrack; - if (newTrack) { + } + + private async setMediaStreamTrack(newTrack: MediaStreamTrack, options: SetMediaStreamTrackOptions) { + const {deferEndedListener = false, force = false, preservePreviousTrack} = options; + if (newTrack === this._mediaStreamTrack && !force) { + return; + } + const previousTrack = this._mediaStreamTrack; + const previousConstraints = this._constraints; + const previousEnabled = previousTrack.enabled; + const previousTrackEndedListenerDeferred = this.stagedReplacementTrack === previousTrack; + const nextTrackEndedListenerDeferred = deferEndedListener || this.stagedReplacementTrack === newTrack; + let processedTrack: MediaStreamTrack | undefined; + try { + this.attachedElements.forEach((el) => { + detachTrack(previousTrack, el); + }); + this.debouncedTrackMuteHandler.cancel('new-track'); + this.removeMediaStreamTrackListeners(previousTrack); + this.mediaStream = new MediaStream([newTrack]); + this.addMediaStreamTrackListeners(newTrack, !nextTrackEndedListenerDeferred); + this._constraints = newTrack.getConstraints(); + if (this.processor) { + this.log.debug('restarting processor', this.logContext); + if (this.kind === 'unknown') { + throw TypeError('cannot set processor on track of unknown kind'); + } + + if (this.processorElement) { + attachToElement(newTrack, this.processorElement); + this.processorElement.muted = true; + } + await this.processor.restart({ + track: newTrack, + kind: this.kind, + element: this.processorElement, + }); + processedTrack = this.processor.processedTrack; + } + if (this.sender && this.sender.transport?.state !== 'closed') { + await this.sender.replaceTrack(processedTrack ?? newTrack); + } + this._mediaStreamTrack = newTrack; this._mediaStreamTrack.enabled = !this.isMuted; await this.resumeUpstream(); this.attachedElements.forEach((el) => { attachToElement(processedTrack ?? newTrack, el); }); + if (!preservePreviousTrack && previousTrack !== newTrack) { + previousTrack.stop(); + } + } catch (error) { + try { + await this.restoreMediaStreamTrackAfterFailure( + previousTrack, + previousConstraints, + previousEnabled, + newTrack, + processedTrack, + previousTrackEndedListenerDeferred, + ); + } catch (rollbackError) { + throw new AggregateError([error, rollbackError], 'Track replacement and internal rollback both failed'); + } + throw error; } } @@ -255,7 +339,12 @@ export default abstract class LocalTrack; async replaceTrack(track: MediaStreamTrack, userProvidedOrOptions: boolean | ReplaceTrackOptions | undefined) { const unlock = await this.trackChangeLock.lock(); + const previousProvidedByUser = this.providedByUser; + let replacementCommitted = false; try { + if (this.stagedReplacementTrack) { + throw new TrackInvalidError('unable to replace a track while a staged replacement is active'); + } if (!this.sender) { throw new TrackInvalidError('unable to replace an unpublished track'); } @@ -273,12 +362,88 @@ export default abstract class LocalTrack(operation: () => Promise): Promise { + const unlock = await this.trackChangeLock.lock(); + try { + return await operation(); + } finally { + unlock(); + } + } + + async stageTrackReplacement(track: MediaStreamTrack): Promise { + const unlock = await this.trackChangeLock.lock(); + const previousProvidedByUser = this.providedByUser; + const previousStagedReplacementTrack = this.stagedReplacementTrack; + try { + if (!this.sender) { + throw new TrackInvalidError('unable to stage a replacement for an unpublished track'); + } + if (previousStagedReplacementTrack && previousStagedReplacementTrack !== this._mediaStreamTrack) { + throw new TrackInvalidError('staged replacement identity does not match the active source track'); + } + if (track === this._mediaStreamTrack) { + throw new TrackInvalidError('unable to stage the active source track as its own replacement'); + } + if (track.readyState !== 'live') { + throw new TrackInvalidError('unable to stage an ended replacement track'); + } + + this.providedByUser = true; + this.log.debug('stage MediaStreamTrack replacement', this.logContext); + await this.setMediaStreamTrack(track, { + deferEndedListener: true, + preservePreviousTrack: true, + }); + this.stagedReplacementTrack = track; + return this; + } catch (error) { + this.providedByUser = previousProvidedByUser; + this.stagedReplacementTrack = previousStagedReplacementTrack; + throw error; + } finally { + unlock(); + } + } + + async commitStagedTrackReplacement(track: MediaStreamTrack, userProvidedTrack: boolean): Promise { + const unlock = await this.trackChangeLock.lock(); + try { + if (this.stagedReplacementTrack !== track || this._mediaStreamTrack !== track) { + throw new TrackInvalidError('unable to commit a replacement that is not the active staged track'); + } + if (!this.sender) { + throw new TrackInvalidError('unable to commit a replacement for an unpublished track'); + } + if (track.readyState !== 'live') { + throw new TrackInvalidError('unable to commit an ended staged track'); + } + + track.addEventListener('ended', this.handleEnded); + if (track.readyState !== 'live') { + track.removeEventListener('ended', this.handleEnded); + throw new TrackInvalidError('staged track ended while its replacement was committed'); + } + this.providedByUser = userProvidedTrack; + this.stagedReplacementTrack = undefined; + return this; } finally { unlock(); } @@ -287,8 +452,14 @@ export default abstract class LocalTrack { + this.log.error('failed to destroy track processor during stop', {...this.logContext, error}); + }); + } catch (error) { + this.log.error('failed to destroy track processor during stop', {...this.logContext, error}); + } super.stop(); this._mediaStreamTrack.removeEventListener('ended', this.handleEnded); this._mediaStreamTrack.removeEventListener('mute', this.handleTrackMuteEvent); this._mediaStreamTrack.removeEventListener('unmute', this.handleTrackUnmuteEvent); - this.processor?.destroy(); - this.processor = undefined; } async pauseUpstream() { @@ -459,56 +648,159 @@ export default abstract class LocalTrack = []; + try { + await processor.destroy(); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + processorElement.remove(); + const sender = this.sender; + if (this.processor !== previousProcessor && sender && sender.transport?.state !== 'closed') { + const rawSenderTrack = this._mediaStreamTrack.readyState === 'live' ? this._mediaStreamTrack : null; + if (sender.track !== rawSenderTrack) { + try { + await sender.replaceTrack(rawSenderTrack); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + if (sender.track?.readyState === 'ended') { + try { + await sender.replaceTrack(null); + } catch (failCloseError) { + cleanupErrors.push(failCloseError); + } + } + } + } + } + if (cleanupErrors.length > 0) { + throw new AggregateError( + [error, ...cleanupErrors], + 'Existing track processor removal and candidate cleanup both failed', + ); + } + throw error; + } } if (this.kind === 'unknown') { - throw TypeError('cannot set processor on track of unknown kind'); - } - - attachToElement(this._mediaStreamTrack, processorElement); - processorElement.muted = true; - - processorElement.play().catch((error) => { - if (error instanceof DOMException && error.name === 'AbortError') { - this.log.warn('failed to play processor element, retrying', { - ...this.logContext, - error, - }); - setTimeout(() => { - processorElement.play().catch((err) => { - this.log.error('failed to play processor element', {...this.logContext, err}); - }); - }, 100); - } else { - this.log.error('failed to play processor element', {...this.logContext, error}); + let cleanupError: unknown; + try { + await processor.destroy(); + } catch (error) { + cleanupError = error; } - }); - - this.processor = processor; - this.processorElement = processorElement; - if (this.processor.processedTrack) { - for (const el of this.attachedElements) { - if (el !== this.processorElement && showProcessedStreamLocally) { - detachTrack(this._mediaStreamTrack, el); - attachToElement(this.processor.processedTrack, el); + processorElement.remove(); + const kindError = new TypeError('cannot set processor on track of unknown kind'); + if (cleanupError !== undefined) { + throw new AggregateError([kindError, cleanupError], 'Invalid track processor kind and cleanup both failed'); + } + throw kindError; + } + const processedTrack = processor.processedTrack; + try { + attachToElement(this._mediaStreamTrack, processorElement); + processorElement.muted = true; + processorElement.play().catch((error) => { + if (error instanceof DOMException && error.name === 'AbortError') { + this.log.warn('failed to play processor element, retrying', { + ...this.logContext, + error, + }); + setTimeout(() => { + processorElement.play().catch((err) => { + this.log.error('failed to play processor element', {...this.logContext, err}); + }); + }, 100); + } else { + this.log.error('failed to play processor element', {...this.logContext, error}); + } + }); + if (processedTrack) { + for (const el of this.attachedElements) { + if (showProcessedStreamLocally) { + detachTrack(this._mediaStreamTrack, el); + attachToElement(processedTrack, el); + } + } + await this.sender?.replaceTrack(processedTrack); + } + this.processor = processor; + this.processorElement = processorElement; + this.emit(TrackEvent.TrackProcessorUpdate, processor); + } catch (error) { + const cleanupErrors: Array = []; + if (this.processor === processor) this.processor = undefined; + if (this.processorElement === processorElement) this.processorElement = undefined; + if (processedTrack) { + for (const el of this.attachedElements) { + try { + detachTrack(processedTrack, el); + if (this._mediaStreamTrack.readyState === 'live') attachToElement(this._mediaStreamTrack, el); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } } } - await this.sender?.replaceTrack(this.processor.processedTrack); + processorElement.remove(); + try { + await processor.destroy(); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + if (processedTrack && processedTrack.readyState !== 'ended') { + processedTrack.enabled = false; + try { + processedTrack.stop(); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + } + const sender = this.sender; + if (sender && sender.transport?.state !== 'closed') { + const rawSenderTrack = this._mediaStreamTrack.readyState === 'live' ? this._mediaStreamTrack : null; + if (sender.track !== rawSenderTrack) { + try { + await sender.replaceTrack(rawSenderTrack); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + if (sender.track?.readyState === 'ended') { + try { + await sender.replaceTrack(null); + } catch (failCloseError) { + cleanupErrors.push(failCloseError); + } + } + } + } + } + if (cleanupErrors.length > 0) { + throw new AggregateError([error, ...cleanupErrors], 'Track processor install rollback was incomplete'); + } + throw error; } - this.emit(TrackEvent.TrackProcessorUpdate, this.processor); } finally { unlock(); } @@ -527,18 +819,88 @@ export default abstract class LocalTrack, keepElement = true): Promise { + const unlock = await this.trackChangeLock.lock(); + try { + if (this.processor !== processor) { + return false; + } + await this.internalStopProcessor(keepElement); + return true; + } finally { + unlock(); + } + } + protected async internalStopProcessor(keepElement = true) { if (!this.processor) return; this.log.debug('stopping processor', this.logContext); - this.processor.processedTrack?.stop(); - await this.processor.destroy(); + const processor = this.processor; + const processedTrack = processor.processedTrack; + const constraints = this._constraints; this.processor = undefined; + if (processedTrack) { + for (const element of this.attachedElements) { + detachTrack(processedTrack, element); + } + } if (!keepElement) { this.processorElement?.remove(); this.processorElement = undefined; } - await this._mediaStreamTrack.applyConstraints(this._constraints); - await this.setMediaStreamTrack(this._mediaStreamTrack, true); + const cleanupErrors: Array = []; + if (this._mediaStreamTrack.readyState === 'live') { + try { + await this.setMediaStreamTrack(this._mediaStreamTrack, { + force: true, + preservePreviousTrack: this.providedByUser, + }); + } catch (error) { + cleanupErrors.push(error); + } + if (this._mediaStreamTrack.readyState === 'live') { + try { + await this._mediaStreamTrack.applyConstraints(constraints); + } catch (error) { + cleanupErrors.push(error); + } + } + } + this._constraints = constraints; + if (processedTrack && processedTrack.readyState !== 'ended') { + processedTrack.enabled = false; + try { + processedTrack.stop(); + } catch (error) { + cleanupErrors.push(error); + } + } + try { + await processor.destroy(); + } catch (error) { + cleanupErrors.push(error); + } + const sender = this.sender; + if (sender && sender.transport?.state !== 'closed') { + const rawSenderTrack = this._mediaStreamTrack.readyState === 'live' ? this._mediaStreamTrack : null; + if (sender.track !== rawSenderTrack) { + try { + await sender.replaceTrack(rawSenderTrack); + } catch (error) { + cleanupErrors.push(error); + if (sender.track?.readyState === 'ended') { + try { + await sender.replaceTrack(null); + } catch (failCloseError) { + cleanupErrors.push(failCloseError); + } + } + } + } + } + if (cleanupErrors.length > 0) { + throw new AggregateError(cleanupErrors, 'Failed to stop track processor cleanly'); + } this.emit(TrackEvent.TrackProcessorUpdate); } diff --git a/fluxer_app/pkgs/livekit-client/src/room/track/LocalVideoTrack.ts b/fluxer_app/pkgs/livekit-client/src/room/track/LocalVideoTrack.ts index 459372919..48dff855d 100644 --- a/fluxer_app/pkgs/livekit-client/src/room/track/LocalVideoTrack.ts +++ b/fluxer_app/pkgs/livekit-client/src/room/track/LocalVideoTrack.ts @@ -39,6 +39,33 @@ export class SimulcastTrackInfo { const refreshSubscribedCodecAfterNewCodec = 5000; +function restoreSecondarySenderTrack( + sender: RTCRtpSender | undefined, + track: MediaStreamTrack | null, +): Promise | undefined { + if (!sender) return undefined; + if (sender.track === track) return undefined; + if (track != null && track.readyState !== 'live') return undefined; + return sender.replaceTrack(track); +} + +function createProcessorRecoveryError( + primaryError: unknown, + cleanupErrors: ReadonlyArray, + rollbackErrors: ReadonlyArray, +): AggregateError { + const recoveryErrors: Array = []; + if (cleanupErrors.length > 0) { + recoveryErrors.push(new AggregateError(cleanupErrors, 'Video processor candidate cleanup failed')); + } + if (rollbackErrors.length > 0) { + recoveryErrors.push(new AggregateError(rollbackErrors, 'Video processor secondary sender rollback failed')); + } + return new AggregateError(recoveryErrors, 'Video processor apply failed and recovery was incomplete', { + cause: primaryError, + }); +} + export default class LocalVideoTrack extends LocalTrack { signalClient?: SignalClient; @@ -251,21 +278,86 @@ export default class LocalVideoTrack extends LocalTrack { this.isCpuConstrained = false; + const processedTrack = this.processor?.processedTrack; 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); + const previousTrack = sc.mediaStreamTrack; + const nextTrack = this._mediaStreamTrack.clone(); + try { + await sc.sender.replaceTrack(processedTrack ?? nextTrack); + sc.mediaStreamTrack = nextTrack; + previousTrack.stop(); + } catch (error) { + nextTrack.stop(); + throw error; + } } } } override async setProcessor(processor: TrackProcessor, showProcessedStreamLocally = true) { - await super.setProcessor(processor, showProcessedStreamLocally); - - if (this.processor?.processedTrack) { - for await (const sc of this.simulcastCodecs.values()) { - await sc.sender?.replaceTrack(this.processor.processedTrack); + const secondarySenderSnapshots = Array.from(this.simulcastCodecs.values(), (trackInfo) => ({ + sender: trackInfo.sender, + track: trackInfo.sender?.track ?? null, + })); + try { + await super.setProcessor(processor, showProcessedStreamLocally); + if (this.processor?.processedTrack) { + for await (const sc of this.simulcastCodecs.values()) { + await sc.sender?.replaceTrack(this.processor.processedTrack); + } } + } catch (error) { + const cleanupErrors: Array = []; + if (this.processor === processor) { + try { + await this.stopProcessor(false); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + } + const rollbackResults = await Promise.allSettled( + secondarySenderSnapshots.map(({sender, track}) => restoreSecondarySenderTrack(sender, track)), + ); + const rollbackErrors = rollbackResults.flatMap((result) => (result.status === 'rejected' ? [result.reason] : [])); + if (cleanupErrors.length === 0 && rollbackErrors.length === 0) { + throw error; + } + throw createProcessorRecoveryError(error, cleanupErrors, rollbackErrors); + } + } + + protected override async internalStopProcessor(keepElement = true) { + const processor = this.processor; + if (!processor) { + await super.internalStopProcessor(keepElement); + return; + } + const secondarySenderSnapshots = Array.from(this.simulcastCodecs.values(), (trackInfo) => ({ + sender: trackInfo.sender, + track: trackInfo.sender?.track ?? null, + replacement: trackInfo.mediaStreamTrack, + })); + try { + for (const {sender, replacement} of secondarySenderSnapshots) { + await sender?.replaceTrack(replacement); + } + await super.internalStopProcessor(keepElement); + } catch (error) { + const rollbackResults = await Promise.allSettled( + secondarySenderSnapshots.map(({sender, track}) => restoreSecondarySenderTrack(sender, track)), + ); + const rollbackErrors = rollbackResults.flatMap((result) => (result.status === 'rejected' ? [result.reason] : [])); + if (rollbackErrors.length === 0) { + throw error; + } + throw new AggregateError( + rollbackErrors, + 'Video processor stop failed and secondary sender recovery was incomplete', + { + cause: error, + }, + ); } } @@ -290,7 +382,7 @@ export default class LocalVideoTrack extends LocalTrack { } const simulcastCodecInfo: SimulcastTrackInfo = { codec, - mediaStreamTrack: this.mediaStreamTrack.clone(), + mediaStreamTrack: this._mediaStreamTrack.clone(), sender: undefined, encodings, }; @@ -304,6 +396,12 @@ export default class LocalVideoTrack extends LocalTrack { return; } simulcastCodecInfo.sender = sender; + const processedTrack = this.processor?.processedTrack; + if (processedTrack) { + void sender.replaceTrack(processedTrack).catch((error: unknown) => { + this.log.warn('failed to route processed track to secondary sender', {...this.logContext, error}); + }); + } setTimeout(() => { if (this.subscribedCodecs) {