diff --git a/src/client/media/viewer-sfu-route.ts b/src/client/media/viewer-sfu-route.ts index 5eebcf7..d7011e7 100644 --- a/src/client/media/viewer-sfu-route.ts +++ b/src/client/media/viewer-sfu-route.ts @@ -68,6 +68,10 @@ interface ViewerSfuRouteEvents { ) => void; onSfuVideoAvailability?: (available: boolean, revision: number) => void; onSfuUpdate?: (metrics: ConnectionMetrics | null, revision: number) => void; + onSfuDecodedFrameSample?: ( + framesDecodedDelta: number | null, + revision: number, + ) => void; onSfuState?: ( state: "connected" | "reconnecting", revision: number, @@ -78,6 +82,7 @@ interface ViewerSfuRouteEvents { onStream: (stream: MediaStream | null) => void; onVideoAvailability: (available: boolean) => void; onStats: (metrics: ConnectionMetrics) => void; + onDecodedFrameSample: (framesDecodedDelta: number | null) => void; onFirstDecodedFrame: () => boolean; onState: (state: "connected" | "reconnecting") => void; onDisconnected: () => void; @@ -333,6 +338,14 @@ export class ViewerSfuRoute { this.events.onSfuUpdate?.(metrics, slot.revision); } }, + onDecodedFrameSample: (framesDecodedDelta: number | null) => { + if (this.active === slot && !slot.failed) { + this.events.onSfuDecodedFrameSample?.( + framesDecodedDelta, + slot.revision, + ); + } + }, onFirstDecodedFrame: () => this.handlePendingDecodedFrame(slot), onState: (state: "connected" | "reconnecting") => { if (this.active === slot && !slot.failed) { diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index a6ab586..f81a36a 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -694,14 +694,18 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { ); reconcileRelayChildren(previousChildPeerIds, revision); }, + onSfuDecodedFrameSample: (framesDecodedDelta, revision) => { + if (active && viewerSfuRoute === route) { + observeActiveDecodedFrames( + "sfu", + String(revision), + framesDecodedDelta, + ); + } + }, onSfuUpdate: (metrics, revision) => { if (active && viewerSfuRoute === route) { if (metrics) { - observeActiveDecodedFrames( - "sfu", - String(revision), - metrics.intervalFramesDecoded, - ); if ( currentRouteAssignment?.upstream.kind === "sfu" && currentRouteConnectionId && diff --git a/src/client/sfu/subscriber.ts b/src/client/sfu/subscriber.ts index 3e6d03c..f2013a7 100644 --- a/src/client/sfu/subscriber.ts +++ b/src/client/sfu/subscriber.ts @@ -22,6 +22,7 @@ interface SubscriberEvents { onStream: (stream: MediaStream | null) => void; onVideoAvailability?: (available: boolean) => void; onStats?: (metrics: ConnectionMetrics) => void; + onDecodedFrameSample?: (framesDecodedDelta: number | null) => void; onFirstDecodedFrame?: () => boolean; onState?: (state: "connected" | "reconnecting") => void; onDisconnected?: () => void; @@ -34,6 +35,11 @@ interface SubscribedTrack { onEnded: () => void; } +interface AudioStatsSample { + track: SubscribedTrack; + report: RTCStatsReport | null; +} + interface KnownHostPublication { publication: RemoteTrackPublication; participant: RemoteParticipant; @@ -67,6 +73,7 @@ export class SfuSubscriber { private statsAccumulator: StatsAccumulator = createStatsAccumulator(); private statsTimer: ReturnType | null = null; private statsInFlight: StatsAccumulator | null = null; + private audioStatsInFlight: SubscribedTrack | null = null; private decodedFrameProofMode: "fresh" | "progress" | null = null; private stopDecodedFrameObserver: (() => void) | null = null; private generation = 0; @@ -125,6 +132,7 @@ export class SfuSubscriber { const generation = this.generation; this.state = "active"; try { + this.startStats(); this.reconcileHostSubscriptions(room, this.sdk, generation); return true; } catch (error) { @@ -414,8 +422,10 @@ export class SfuSubscriber { this.stream.addTrack(mediaStreamTrack); if (isVideo) { this.video = subscribed; + this.resetStatsAccumulator(); } else { this.audio = subscribed; + this.resetAudioStats(); } this.emitStream(); if (!hadVideo && this.video) { @@ -440,16 +450,15 @@ export class SfuSubscriber { if (this.audio?.sid === sid && (!track || this.audio.mediaStreamTrack === track)) { this.detachTrack(this.audio); this.audio = null; + this.resetAudioStats(); } if (this.video === previousVideo && this.audio === previousAudio) { return; } this.emitStream(); - if (!this.video) { - this.stopStats(); - if (previousVideo) { - this.events.onVideoAvailability?.(false); - } + if (!this.video && previousVideo) { + this.resetStatsAccumulator(); + this.events.onVideoAvailability?.(false); } } @@ -518,41 +527,51 @@ export class SfuSubscriber { const video = this.video; const generation = this.generation; const accumulator = this.statsAccumulator; - if ( - !room || - !video || - this.state !== "active" || - this.statsInFlight === accumulator - ) { + if (!room || this.state !== "active") { return; } + if (this.statsInFlight === accumulator || !video) { + this.events.onDecodedFrameSample?.(null); + return; + } + const ownsSample = (): boolean => + this.owns(room, generation) && + this.state === "active" && + this.video === video && + this.statsAccumulator === accumulator; + const audioStatsSample: AudioStatsSample | null = this.audio + ? { track: this.audio, report: null } + : null; + if (audioStatsSample) { + void this.updateAudioStats(room, generation, audioStatsSample); + } this.statsInFlight = accumulator; try { - const reports = await Promise.all([ - video.track.getRTCStatsReport(), - this.audio?.track.getRTCStatsReport(), + const videoReport = await video.track.getRTCStatsReport(); + if (!ownsSample()) { + return; + } + const report = mergeStatsReports([ + videoReport, + audioStatsSample?.track === this.audio + ? (audioStatsSample.report ?? undefined) + : undefined, ]); - if ( - !this.owns(room, generation) || - this.state !== "active" || - this.video !== video || - this.statsAccumulator !== accumulator - ) { - return; - } - const report = mergeStatsReports(reports); if (!report) { + this.events.onDecodedFrameSample?.(null); return; } - this.events.onStats?.( - collectConnectionMetricsFromReport( - report, - "receive", - accumulator, - ), + const metrics = collectConnectionMetricsFromReport( + report, + "receive", + accumulator, ); + this.events.onDecodedFrameSample?.(metrics.intervalFramesDecoded); + this.events.onStats?.(metrics); } catch { - // Stats are observational and must never disrupt active SFU media. + if (ownsSample()) { + this.events.onDecodedFrameSample?.(null); + } } finally { if (this.statsInFlight === accumulator) { this.statsInFlight = null; @@ -560,20 +579,59 @@ export class SfuSubscriber { } } + private async updateAudioStats( + room: Room, + generation: number, + sample: AudioStatsSample, + ): Promise { + const audio = sample.track; + if (this.audioStatsInFlight === audio) { + return; + } + this.audioStatsInFlight = audio; + try { + const report = await audio.track.getRTCStatsReport(); + if ( + report && + this.owns(room, generation) && + this.state === "active" && + this.audio === audio + ) { + sample.report = report; + } + } catch { + // Audio diagnostics must not affect video liveness. + } finally { + if (this.audioStatsInFlight === audio) { + this.audioStatsInFlight = null; + } + } + } + private stopStats(): void { if (this.statsTimer !== null) { clearInterval(this.statsTimer); this.statsTimer = null; } + this.resetStatsAccumulator(); + } + + private resetStatsAccumulator(): void { this.statsInFlight = null; this.statsAccumulator = createStatsAccumulator(); } + private resetAudioStats(): void { + this.audioStatsInFlight = null; + } + private clearMedia(notify: boolean): void { const hadStream = this.streamEmitted; const hadVideo = this.video !== null; this.stopDecodedFrameProof(); - this.stopStats(); + if (this.state !== "active") { + this.stopStats(); + } if (this.video) { this.detachTrack(this.video); } @@ -582,6 +640,10 @@ export class SfuSubscriber { } this.video = null; this.audio = null; + this.resetAudioStats(); + if (this.state === "active") { + this.resetStatsAccumulator(); + } this.streamEmitted = false; if (hadVideo) { this.events.onVideoAvailability?.(false); diff --git a/tests/decoded-frame-stall.test.ts b/tests/decoded-frame-stall.test.ts index 05c1d8e..cde09f4 100644 --- a/tests/decoded-frame-stall.test.ts +++ b/tests/decoded-frame-stall.test.ts @@ -8,7 +8,7 @@ import { describe("DecodedFrameStallDetector", () => { it("reports one exact active stall and resets on progress or identity change", () => { const detector = new DecodedFrameStallDetector(); - expect(detector.observe("peer:1:a", 0, 0)).toBe(false); + expect(detector.observe("peer:1:a", null, 0)).toBe(false); expect( detector.observe("peer:1:a", 0, ROUTE_DECODED_FRAME_STALL_MS), ).toBe(true); diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index f94f8e8..bd7923e 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -423,11 +423,17 @@ describe("minimal route transition contracts", () => { const log: string[] = []; const messages: ClientMessage[] = []; const streams: MediaStream[] = []; + const decodedSamples: Array<{ + framesDecodedDelta: number | null; + revision: number; + }> = []; const subscribers: ReturnType[] = []; const route = new ViewerSfuRoute("viewer_12345678", { activatePeer: () => true, reconcileSfuChildren: () => undefined, onSfuStream: (stream) => streams.push(stream), + onSfuDecodedFrameSample: (framesDecodedDelta, revision) => + decodedSamples.push({ framesDecodedDelta, revision }), send: (message) => { messages.push(message); return true; @@ -453,6 +459,8 @@ describe("minimal route transition contracts", () => { subscribers[0]?.events.onStream(firstStream); subscribers[0]?.events.onFirstDecodedFrame(); await vi.waitFor(() => expect(streams).toEqual([firstStream])); + subscribers[0]?.events.onDecodedFrameSample(3); + expect(decodedSamples).toEqual([{ framesDecodedDelta: 3, revision: 1 }]); route.accept({ revision: 2, @@ -466,7 +474,9 @@ describe("minimal route transition contracts", () => { subscribers[1]?.events.onStats?.({ intervalFramesDecoded: 2, } as ConnectionMetrics); + subscribers[1]?.events.onDecodedFrameSample(null); expect(messages.filter((message) => message.type === "route-ready")).toEqual([]); + expect(decodedSamples).toEqual([{ framesDecodedDelta: 3, revision: 1 }]); subscribers[1]?.events.onFirstDecodedFrame(); subscribers[1]?.events.onFirstDecodedFrame(); @@ -481,6 +491,12 @@ describe("minimal route transition contracts", () => { assignment: viewerSfuAssignment([], "publication-generation-2"), }); await vi.waitFor(() => expect(streams).toEqual([firstStream, pendingStream])); + subscribers[1]?.events.onDecodedFrameSample(null); + subscribers[0]?.events.onDecodedFrameSample(9); + expect(decodedSamples).toEqual([ + { framesDecodedDelta: 3, revision: 1 }, + { framesDecodedDelta: null, revision: 2 }, + ]); expect(subscribers[0]?.deactivate).toHaveBeenCalledOnce(); expect(subscribers[0]?.disconnect).toHaveBeenCalledOnce(); }); diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index 8d4dfcb..9f8db82 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -1687,12 +1687,266 @@ describe("SfuSubscriber", () => { }, ); + it("keeps active decoded-frame sampling through unavailable receiver stats", async () => { + vi.useFakeTimers(); + const samples: Array = []; + const updates: ConnectionMetrics[] = []; + const subscriber = new SfuSubscriber({ + onStream: vi.fn(), + onStats: (metrics) => updates.push(metrics), + onDecodedFrameSample: (framesDecodedDelta) => + samples.push(framesDecodedDelta), + }); + await subscriber.connect(connection); + const room = livekit.state.rooms[0]; + const host = new livekit.FakeRemoteParticipant("host"); + const publication = new livekit.FakeRemotePublication( + "host-video", + Track.Source.ScreenShare, + ); + host.add(publication); + room.remoteParticipants.set("host", host); + + expect(subscriber.activate()).toBe(true); + expect(samples).toEqual([null]); + + let receiverStats: RTCStatsReport | Promise | Error = + statsReport([]); + const video = track("video", "video-1"); + const remoteVideo = remoteTrack(video, () => { + if (receiverStats instanceof Error) throw receiverStats; + return receiverStats; + }); + room.emit(RoomEvent.TrackSubscribed, remoteVideo, publication, host); + await vi.advanceTimersByTimeAsync(0); + expect(samples).toEqual([null, null]); + expect(updates).toEqual([]); + + receiverStats = new Error("stats unavailable"); + const samplesBeforeStatsError = samples.length; + await vi.advanceTimersByTimeAsync(2_000); + expect(samples).toHaveLength(samplesBeforeStatsError + 1); + expect(samples.at(-1)).toBeNull(); + expect(updates).toEqual([]); + + receiverStats = statsReport([ + { + id: "video-in", + type: "inbound-rtp", + timestamp: 1_000, + kind: "video", + trackIdentifier: video.id, + framesDecoded: 1, + }, + ]); + await vi.advanceTimersByTimeAsync(2_000); + expect(samples.at(-1)).toBeNull(); + expect(updates).toHaveLength(1); + + receiverStats = statsReport([ + { + id: "video-in", + type: "inbound-rtp", + timestamp: 3_000, + kind: "video", + trackIdentifier: video.id, + framesDecoded: 5, + }, + ]); + await vi.advanceTimersByTimeAsync(2_000); + expect(samples.at(-1)).toBe(4); + expect(updates).toHaveLength(2); + + let releasePendingStats!: (report: RTCStatsReport) => void; + receiverStats = new Promise((resolve) => { + releasePendingStats = resolve; + }); + const statsCallsBeforePending = remoteVideo.getRTCStatsReport.mock.calls.length; + const samplesBeforePending = samples.length; + await vi.advanceTimersByTimeAsync(2_000); + expect(remoteVideo.getRTCStatsReport).toHaveBeenCalledTimes( + statsCallsBeforePending + 1, + ); + expect(samples).toHaveLength(samplesBeforePending); + await vi.advanceTimersByTimeAsync(2_000); + expect(remoteVideo.getRTCStatsReport).toHaveBeenCalledTimes( + statsCallsBeforePending + 1, + ); + expect(samples).toHaveLength(samplesBeforePending + 1); + expect(samples.at(-1)).toBeNull(); + + const replacementVideo = track("video", "video-2"); + let replacementStats = statsReport([ + { + id: "replacement-video-in", + type: "inbound-rtp", + timestamp: 1_000, + kind: "video", + trackIdentifier: replacementVideo.id, + framesDecoded: 2, + }, + ]); + const replacementRemoteVideo = remoteTrack( + replacementVideo, + () => replacementStats, + ); + room.emit( + RoomEvent.TrackSubscribed, + replacementRemoteVideo, + publication, + host, + ); + await vi.advanceTimersByTimeAsync(0); + expect(updates.at(-1)?.trackIdentifier).toBe(replacementVideo.id); + const samplesAfterReplacement = samples.length; + const updatesAfterReplacement = updates.length; + releasePendingStats( + statsReport([ + { + id: "video-in", + type: "inbound-rtp", + timestamp: 5_000, + kind: "video", + trackIdentifier: video.id, + framesDecoded: 9, + }, + ]), + ); + await Promise.resolve(); + await Promise.resolve(); + expect(samples).toHaveLength(samplesAfterReplacement); + expect(updates).toHaveLength(updatesAfterReplacement); + + replacementStats = statsReport([ + { + id: "replacement-video-in", + type: "inbound-rtp", + timestamp: 3_000, + kind: "video", + trackIdentifier: replacementVideo.id, + framesDecoded: 7, + }, + ]); + await vi.advanceTimersByTimeAsync(2_000); + expect(samples.at(-1)).toBe(5); + + room.emit( + RoomEvent.TrackUnsubscribed, + replacementRemoteVideo, + publication, + host, + ); + const samplesBeforeMissingTrack = samples.length; + await vi.advanceTimersByTimeAsync(2_000); + expect(samples).toHaveLength(samplesBeforeMissingTrack + 1); + expect(samples.at(-1)).toBeNull(); + + expect(subscriber.deactivate()).toBe(true); + const samplesAfterDeactivate = samples.length; + await vi.advanceTimersByTimeAsync(4_000); + expect(samples).toHaveLength(samplesAfterDeactivate); + }); + + it("keeps video liveness independent from optional audio stats", async () => { + vi.useFakeTimers(); + const samples: Array = []; + const updates: ConnectionMetrics[] = []; + const subscriber = new SfuSubscriber({ + onStream: vi.fn(), + onStats: (metrics) => updates.push(metrics), + onDecodedFrameSample: (framesDecodedDelta) => + samples.push(framesDecodedDelta), + }); + await subscriber.connect(connection); + const room = livekit.state.rooms[0]; + const host = new livekit.FakeRemoteParticipant("host"); + const hostVideo = new livekit.FakeRemotePublication( + "host-video", + Track.Source.ScreenShare, + ); + const hostAudio = new livekit.FakeRemotePublication( + "host-audio", + Track.Source.ScreenShareAudio, + ); + host.add(hostVideo).add(hostAudio); + room.remoteParticipants.set("host", host); + expect(subscriber.activate()).toBe(true); + + const video = track("video", "video-1"); + let videoTimestamp = 1_000; + let videoFrames = 1; + const remoteVideo = remoteTrack(video, () => + statsReport([ + { + id: "video-in", + type: "inbound-rtp", + timestamp: videoTimestamp, + kind: "video", + trackIdentifier: video.id, + framesDecoded: videoFrames, + estimatedPlayoutTimestamp: 10_000, + }, + ]), + ); + const audio = track("audio", "audio-1"); + let audioStats: RTCStatsReport | Promise | Error = + new Error("audio stats unavailable"); + const remoteAudio = remoteTrack(audio, () => { + if (audioStats instanceof Error) throw audioStats; + return audioStats; + }); + room.emit(RoomEvent.TrackSubscribed, remoteAudio, hostAudio, host); + room.emit(RoomEvent.TrackSubscribed, remoteVideo, hostVideo, host); + await vi.advanceTimersByTimeAsync(0); + expect(samples.at(-1)).toBeNull(); + expect(updates).toHaveLength(1); + + videoTimestamp = 3_000; + videoFrames = 5; + await vi.advanceTimersByTimeAsync(2_000); + expect(samples.at(-1)).toBe(4); + expect(remoteAudio.getRTCStatsReport).toHaveBeenCalledTimes(2); + + audioStats = receiverReport("audio"); + videoTimestamp = 5_000; + videoFrames = 9; + await vi.advanceTimersByTimeAsync(2_000); + expect(samples.at(-1)).toBe(4); + expect(updates.at(-1)?.audioVideoPlayoutDeltaMs).toBe(12); + expect(remoteAudio.getRTCStatsReport).toHaveBeenCalledTimes(3); + + audioStats = new Promise(() => undefined); + videoTimestamp = 7_000; + videoFrames = 13; + const samplesBeforePendingAudio = samples.length; + await vi.advanceTimersByTimeAsync(2_000); + expect(samples).toHaveLength(samplesBeforePendingAudio + 1); + expect(samples.at(-1)).toBe(4); + expect(updates.at(-1)?.audioVideoPlayoutDeltaMs).toBeNull(); + expect(remoteAudio.getRTCStatsReport).toHaveBeenCalledTimes(4); + + videoTimestamp = 9_000; + videoFrames = 17; + const samplesBeforeSecondPendingAudioTick = samples.length; + await vi.advanceTimersByTimeAsync(2_000); + expect(samples).toHaveLength(samplesBeforeSecondPendingAudioTick + 1); + expect(samples.at(-1)).toBe(4); + expect(updates.at(-1)?.audioVideoPlayoutDeltaMs).toBeNull(); + expect(remoteAudio.getRTCStatsReport).toHaveBeenCalledTimes(4); + expect(remoteVideo.getRTCStatsReport).toHaveBeenCalledTimes(5); + + expect(subscriber.deactivate()).toBe(true); + }); + it("reports merged receiver stats and drops a stale sample", async () => { const updates: ConnectionMetrics[] = []; + const samples: Array = []; const states: string[] = []; const subscriber = new SfuSubscriber({ onStream: vi.fn(), onStats: (metrics) => updates.push(metrics), + onDecodedFrameSample: (framesDecodedDelta) => + samples.push(framesDecodedDelta), onState: (state) => states.push(state), }); await subscriber.connect(connection); @@ -1734,6 +1988,7 @@ describe("SfuSubscriber", () => { host, ); await vi.waitFor(() => expect(updates).toHaveLength(1)); + expect(samples).toEqual([null, null]); expect(updates.at(-1)).toMatchObject({ rtpStatsId: "video-in", @@ -1753,10 +2008,12 @@ describe("SfuSubscriber", () => { ); expect(subscriber.deactivate()).toBe(true); + const samplesAfterDeactivate = samples.length; releaseStats(); await Promise.resolve(); await Promise.resolve(); expect(updates).toHaveLength(1); + expect(samples).toHaveLength(samplesAfterDeactivate); }); it("notifies the controller after a terminal room disconnect", async () => {