feat: rework MissingStreamDataDetector to check last 3 received stats

Author: eugeny-dementev

src/detectors/MissingStreamDataDetector.ts

@@ -9,17 +9,20 @@ import {
 import BaseIssueDetector from './BaseIssueDetector';
 
 interface MissingStreamDetectorParams {
-  timeoutMs?: number;
+  steps?: number; // number of last stats to check
 }
 
 export default class MissingStreamDataDetector extends BaseIssueDetector {
   readonly #lastMarkedAt = new Map<string, number>();
 
   readonly #timeoutMs: number;
 
+  readonly #steps: number;
+
   constructor(params: MissingStreamDetectorParams = {}) {
     super();
-    this.#timeoutMs = params.timeoutMs ?? 10_000;
+    this.#timeoutMs = 5_000;
+    this.#steps = params.steps ?? 3;
   }
 
   performDetection(data: WebRTCStatsParsed): IssueDetectorResult {
@@ -32,31 +35,27 @@ export default class MissingStreamDataDetector extends BaseIssueDetector {
   private processData(data: WebRTCStatsParsed): IssueDetectorResult {
     const issues: IssueDetectorResult = [];
 
-    const prevData = this.getLastProcessedStats(data.connection.id);
+    const allLastProcessedStats = [...this.getAllLastProcessedStats(data.connection.id), data];
+    if (allLastProcessedStats.length < this.#steps) {
+      return issues;
+    }
 
-    const { video: { inbound: newVideoInbound } } = data;
-    const { audio: { inbound: newAudioInbound } } = data;
-    const prevVideoInbound = prevData?.video.inbound;
-    const prevAudioInbound = prevData?.audio.inbound;
+    const lastThreeProcessedStats = allLastProcessedStats.slice(-this.#steps);
 
+    const lastThreeVideoInbound = lastThreeProcessedStats.map((stats) => stats.video.inbound);
+    const lastThreeAudioInbound = lastThreeProcessedStats.map((stats) => stats.audio.inbound);
 
-    if (prevAudioInbound) {
-      issues.push(...this.detectMissingData(
-        newAudioInbound as unknown as CommonParsedInboundStreamStats[],
-        prevAudioInbound as unknown as CommonParsedInboundStreamStats[],
-        IssueType.Stream,
-        IssueReason.MissingAudioStreamData,
-      ));
-    }
+    issues.push(...this.detectMissingData(
+      lastThreeAudioInbound as unknown as CommonParsedInboundStreamStats[][],
+      IssueType.Stream,
+      IssueReason.MissingAudioStreamData,
+    ));
 
-    if (prevVideoInbound) {
-      issues.push(...this.detectMissingData(
-        newVideoInbound,
-        prevVideoInbound,
-        IssueType.Stream,
-        IssueReason.MissingVideoStreamData,
-      ));
-    }
+    issues.push(...this.detectMissingData(
+      lastThreeVideoInbound,
+      IssueType.Stream,
+      IssueReason.MissingVideoStreamData,
+    ));
 
     const unvisitedTrackIds = new Set(this.#lastMarkedAt.keys());
 
@@ -71,43 +70,47 @@ export default class MissingStreamDataDetector extends BaseIssueDetector {
   }
 
   private detectMissingData(
-    currentCommonInboundStats: CommonParsedInboundStreamStats[],
-    previousCommonInboundStats: CommonParsedInboundStreamStats[],
-    // commonStreamStats: CommonParsedInboundStreamStats[],
+    lastThreeInboundStats: CommonParsedInboundStreamStats[][],
     type: IssueType,
     reason: IssueReason,
   ): IssueDetectorResult {
     const issues: IssuePayload[] = [];
 
-    const mapStatsByTrackId = (items: CommonParsedInboundStreamStats[]) => new Map<string, CommonParsedInboundStreamStats>(
-      items.map((item) => [item.track.trackIdentifier, item] as const),
-    );
+    const firstInboundStats = lastThreeInboundStats[0];
+    const secondInboundStats = lastThreeInboundStats[1];
+    const currentInboundStats = lastThreeInboundStats[2];
 
-    const prevInboundItemsByTrackId = mapStatsByTrackId(previousCommonInboundStats);
+    const firstInboundItemsByTrackId = MissingStreamDataDetector.mapStatsByTrackId(firstInboundStats);
+    const secondInboundItemsByTrackId = MissingStreamDataDetector.mapStatsByTrackId(secondInboundStats);
 
-    currentCommonInboundStats.forEach((inboundItem) => {
+    currentInboundStats.forEach((inboundItem) => {
       const trackId = inboundItem.track.trackIdentifier;
 
-      const prevInboundItem = prevInboundItemsByTrackId.get(trackId);
-      if (!prevInboundItem) {
+      const firstInboundItem = firstInboundItemsByTrackId.get(trackId);
+      const secondInboundItem = secondInboundItemsByTrackId.get(trackId);
+      if (!firstInboundItem || !secondInboundItem) {
         return;
       }
 
-      const bytesReceivedDelta = inboundItem.bytesReceived - prevInboundItem.bytesReceived;
-
+      if (inboundItem.track.detached || inboundItem.track.ended) {
+        return;
+      }
 
-      if (bytesReceivedDelta === 0 && !inboundItem.track.detached && !inboundItem.track.ended) {
+      if (
+        firstInboundItem.bytesReceived === secondInboundItem.bytesReceived
+        && secondInboundItem.bytesReceived === inboundItem.bytesReceived
+      ) {
         const hasIssue = this.markIssue(trackId);
 
         if (!hasIssue) {
           return;
         }
 
         const statsSample = {
-          bytesReceivedDelta,
+          bytesReceivedDelta: 0,
           bytesReceived: inboundItem.bytesReceived,
-          trackDetached: inboundItem.track.detached,
-          trackEnded: inboundItem.track.ended,
+          trackDetached: Boolean(inboundItem.track.detached),
+          trackEnded: Boolean(inboundItem.track.ended),
         };
 
         issues.push({
@@ -124,6 +127,11 @@ export default class MissingStreamDataDetector extends BaseIssueDetector {
     return issues;
   }
 
+  private static mapStatsByTrackId(items: CommonParsedInboundStreamStats[]) {
+    return new Map<string, CommonParsedInboundStreamStats>(items
+      .map((item) => [item.track.trackIdentifier, item] as const));
+  }
+
   private markIssue(trackId: string): boolean {
     const now = Date.now();
     const lastMarkedAt = this.#lastMarkedAt.get(trackId);