fix(route): keep active SFU liveness cadence

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) {

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 &&

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<typeof setInterval> | 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<void> {
+    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);

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);

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<typeof createFakeSubscriber>[] = [];
     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();
   });

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<number | null> = [];
+    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<RTCStatsReport> | 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<RTCStatsReport>((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<number | null> = [];
+    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<RTCStatsReport> | 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<RTCStatsReport>(() => 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<number | null> = [];
     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 () => {