fix: memory leaks, bugs and logic fixes (#46)

Author: evgmel

src/NetworkScoresCalculator.ts

@@ -6,7 +6,7 @@ import {
   WebRTCStatsParsed,
   NetworkQualityStatsSample,
 } from './types';
-import { scheduleTask } from './utils/tasks';
+import { createTaskScheduler } from './utils/tasks';
 import { CLEANUP_PREV_STATS_TTL_MS } from './utils/constants';
 
 type MosCalculatorResult = {
@@ -17,13 +17,15 @@ type MosCalculatorResult = {
 class NetworkScoresCalculator implements INetworkScoresCalculator {
   #lastProcessedStats: { [connectionId: string]: WebRTCStatsParsed } = {};
 
+  readonly #scheduleTask = createTaskScheduler();
+
   calculate(data: WebRTCStatsParsed): NetworkScores {
     const { connection: { id: connectionId } } = data;
     const { mos: outbound, stats: outboundStatsSample } = this.calculateOutboundScore(data) || {};
     const { mos: inbound, stats: inboundStatsSample } = this.calculateInboundScore(data) || {};
     this.#lastProcessedStats[connectionId] = data;
 
-    scheduleTask({
+    this.#scheduleTask({
       taskId: connectionId,
       delayMs: CLEANUP_PREV_STATS_TTL_MS,
       callback: () => (delete this.#lastProcessedStats[connectionId]),

src/detectors/BaseIssueDetector.ts

@@ -5,7 +5,7 @@ import {
   WebRTCStatsParsed,
   WebRTCStatsParsedWithNetworkScores,
 } from '../types';
-import { scheduleTask } from '../utils/tasks';
+import { createTaskScheduler } from '../utils/tasks';
 import { CLEANUP_PREV_STATS_TTL_MS, MAX_PARSED_STATS_STORAGE_SIZE } from '../utils/constants';
 
 export interface PrevStatsCleanupPayload {
@@ -25,6 +25,8 @@ abstract class BaseIssueDetector implements IssueDetector {
 
   readonly #maxParsedStatsStorageSize: number;
 
+  readonly #scheduleTask = createTaskScheduler();
+
   constructor(params: BaseIssueDetectorParams = {}) {
     this.#statsCleanupDelayMs = params.statsCleanupTtlMs ?? CLEANUP_PREV_STATS_TTL_MS;
     this.#maxParsedStatsStorageSize = params.maxParsedStatsStorageSize ?? MAX_PARSED_STATS_STORAGE_SIZE;
@@ -57,7 +59,7 @@ abstract class BaseIssueDetector implements IssueDetector {
       return;
     }
 
-    scheduleTask({
+    this.#scheduleTask({
       taskId: connectionId,
       delayMs: this.#statsCleanupDelayMs,
       callback: () => {

src/detectors/InboundNetworkIssueDetector.ts

@@ -23,7 +23,7 @@ class InboundNetworkIssueDetector extends BaseIssueDetector {
   readonly #highRttThresholdMs: number;
 
   constructor(params: InboundNetworkIssueDetectorParams = {}) {
-    super();
+    super(params);
     this.#highPacketLossThresholdPct = params.highPacketLossThresholdPct ?? 5;
     this.#highJitterThreshold = params.highJitterThreshold ?? 200;
     this.#highJitterBufferDelayThresholdMs = params.highJitterBufferDelayThresholdMs ?? 500;

src/detectors/NetworkMediaSyncIssueDetector.ts

@@ -14,7 +14,7 @@ class NetworkMediaSyncIssueDetector extends BaseIssueDetector {
   readonly #correctedSamplesThresholdPct: number;
 
   constructor(params: NetworkMediaSyncIssueDetectorParams = {}) {
-    super();
+    super(params);
     this.#correctedSamplesThresholdPct = params.correctedSamplesThresholdPct ?? 5;
   }
 
@@ -47,6 +47,11 @@ class NetworkMediaSyncIssueDetector extends BaseIssueDetector {
       }
 
       const deltaSamplesReceived = stats.track.totalSamplesReceived - previousStreamStats.track.totalSamplesReceived;
+
+      if (deltaSamplesReceived === 0) {
+        return;
+      }
+
       const deltaCorrectedSamples = nowCorrectedSamples - lastCorrectedSamples;
       const correctedSamplesPct = Math.round((deltaCorrectedSamples * 100) / deltaSamplesReceived);
       const statsSample = {

src/detectors/OutboundNetworkIssueDetector.ts

@@ -17,7 +17,7 @@ class OutboundNetworkIssueDetector extends BaseIssueDetector {
   readonly #highJitterThreshold: number;
 
   constructor(params: OutboundNetworkIssueDetectorParams = {}) {
-    super();
+    super(params);
     this.#highPacketLossThresholdPct = params.highPacketLossThresholdPct ?? 5;
     this.#highJitterThreshold = params.highJitterThreshold ?? 200;
   }
@@ -79,7 +79,7 @@ class OutboundNetworkIssueDetector extends BaseIssueDetector {
     const isHighPacketsLoss = packetLossPct > this.#highPacketLossThresholdPct;
     const isHighJitter = avgJitter >= this.#highJitterThreshold;
     const isNetworkMediaLatencyIssue = isHighPacketsLoss && isHighJitter;
-    const isNetworkIssue = (!isHighPacketsLoss && isHighJitter) || isHighJitter || isHighPacketsLoss;
+    const isNetworkIssue = isHighJitter || isHighPacketsLoss;
 
     const statsSample = {
       rtt,

src/helpers/calc.ts

@@ -15,6 +15,11 @@ export const calculateVolatility = (values: number[]) => {
   }
 
   const mean = calculateMean(values);
+
+  if (mean === 0) {
+    return 0;
+  }
+
   const meanAbsoluteDeviationFps = values.reduce((acc, val) => acc + Math.abs(val - mean), 0) / values.length;
   return (meanAbsoluteDeviationFps * 100) / mean;
 };

src/helpers/streams.ts

@@ -7,7 +7,8 @@ export const isDtxLikeBehavior = (
   stdDevThreshold = 30,
 ): boolean => {
   const frameIntervals: number[] = [];
-  for (let i = 1; i < allProcessedStats.length - 1; i += 1) {
+
+  for (let i = 1; i < allProcessedStats.length; i += 1) {
     const videoStreamStats = allProcessedStats[i]?.video?.inbound.find(
       (stream) => stream.ssrc === ssrc,
     );

src/parser/RTCStatsParser.ts

@@ -15,7 +15,7 @@ import {
   Logger,
 } from '../types';
 import { checkIsConnectionClosed, calcBitrate } from './utils';
-import { scheduleTask } from '../utils/tasks';
+import { createTaskScheduler } from '../utils/tasks';
 import { CLEANUP_PREV_STATS_TTL_MS } from '../utils/constants';
 
 interface PrevStatsItem {
@@ -31,6 +31,8 @@ interface WebRTCStatsParserParams {
 class RTCStatsParser implements StatsParser {
   private readonly prevStats = new Map<string, PrevStatsItem | undefined>();
 
+  private readonly scheduleTask = createTaskScheduler();
+
   private readonly allowedReportTypes: Set<RTCStatsType> = new Set<RTCStatsType>([
     'candidate-pair',
     'inbound-rtp',
@@ -133,7 +135,7 @@ class RTCStatsParser implements StatsParser {
       ts: Date.now(),
     });
 
-    scheduleTask({
+    this.scheduleTask({
       taskId: connectionId,
       delayMs: CLEANUP_PREV_STATS_TTL_MS,
       callback: () => (this.prevStats.delete(connectionId)),

src/utils/tasks.ts

@@ -26,5 +26,3 @@ export const createTaskScheduler = () => {
     scheduledTasks.set(taskId, newTimer);
   };
 };
-
-export const scheduleTask = createTaskScheduler();