chore(llc): improved sfu stats (#1122)

* improved sfu stats

* tweaks

* tweaks

* tweak

* fix

* tweak

* fix

Author: Brazol

packages/stream_video/CHANGELOG.md

@@ -4,6 +4,8 @@
 * Fixed incoming call timeout handling.
   - Use `streamVideo.observeCoreRingingEventsForBackground()` instead of `streamVideo.observeCallDeclinedRingingEvent()` in `firebaseMessagingBackgroundHandler` to support all necessary events.
 
+* Improved SFU stats implementation.
+
 ## 1.0.1
 
 ### ✅ Added

packages/stream_video/lib/src/call/call.dart

@@ -967,7 +967,12 @@ class Call {
         networkMonitor: networkMonitor,
         streamVideo: _streamVideo,
         statsOptions: _sfuStatsOptions!,
-        leftoverTraceRecords: _previousSession?.getTrace().snapshot ?? [],
+        leftoverTraceRecords:
+            _previousSession
+                ?.getTrace()
+                .expand((slice) => slice.snapshot)
+                .toList() ??
+            const [],
         onReconnectionNeeded: (pc, strategy) {
           _session?.trace('pc_reconnection_needed', {
             'peerConnectionId': pc.type.name,
@@ -1481,8 +1486,8 @@ class Call {
           return;
         }
 
-        _session?.trace('call_reconnect', {
-          'strategy': _reconnectStrategy.name,
+        _session?.trace('callReconnect', {
+          'strategy': strategy.name,
         });
 
         _stateManager.lifecycleCallConnecting(
@@ -1531,8 +1536,8 @@ class Call {
               await _reconnectMigrate();
           }
 
-          _session?.trace('call_reconnect_success', {
-            'strategy': _reconnectStrategy.name,
+          _session?.trace('callReconnectSuccess', {
+            'strategy': strategy.name,
           });
         } catch (error) {
           switch (error) {
@@ -1541,8 +1546,8 @@ class Call {
               _logger.w(() => '[reconnect] unrecoverable error');
               _stateManager.lifecycleCallReconnectingFailed();
 
-              _session?.trace('call_reconnect_failed', {
-                'strategy': _reconnectStrategy.name,
+              _session?.trace('callReconnectFailed', {
+                'strategy': strategy.name,
                 'error': error.toString(),
               });
 

packages/stream_video/lib/src/call/session/call_session.dart

@@ -104,6 +104,7 @@ class CallSession extends Disposable {
   final InternetConnection networkMonitor;
   final StatsOptions statsOptions;
   final Tracer _tracer;
+  final Tracer _zonedTracer = Tracer(null);
   final StreamVideo _streamVideo;
 
   final Duration joinResponseTimeout;
@@ -121,6 +122,7 @@ class CallSession extends Disposable {
   StatsReporter? statsReporter;
 
   Timer? _peerConnectionCheckTimer;
+  bool _isLeavingOrClosed = false;
 
   sfu_models.ClientDetails? _clientDetails;
 
@@ -132,12 +134,13 @@ class CallSession extends Disposable {
     onCancel: () => Result.error('UpdateViewportVisibility cancelled'),
   );
 
-  TraceSlice getTrace() {
-    return _tracer.take();
+  List<TraceSlice> getTrace() {
+    return [_tracer.take(), _zonedTracer.take()];
   }
 
   void setTraceEnabled(bool enabled) {
     _tracer.setEnabled(enabled);
+    _zonedTracer.setEnabled(enabled);
   }
 
   void trace(String tag, dynamic data) {
@@ -459,6 +462,12 @@ class CallSession extends Disposable {
           fastReconnectDeadline: event.fastReconnectDeadline,
         ),
       );
+    } on TimeoutException catch (e, stk) {
+      final message =
+          'Waiting for "joinResponse" has timed out after ${joinResponseTimeout.inMilliseconds}ms';
+      _tracer.trace('joinRequestTimeout', message);
+      _logger.e(() => '[start] failed: $e');
+      return Result.failure(VideoErrors.compose(e, stk));
     } catch (e, stk) {
       _logger.e(() => '[start] failed: $e');
       return Result.failure(VideoErrors.compose(e, stk));
@@ -585,6 +594,7 @@ class CallSession extends Disposable {
 
   void leave({String? reason}) {
     _logger.d(() => '[leave] no args');
+    _isLeavingOrClosed = true;
     sfuWS.leave(sessionId: sessionId, reason: reason);
   }
 
@@ -593,6 +603,7 @@ class CallSession extends Disposable {
     String? closeReason,
   }) async {
     _logger.d(() => '[close] code: $code, closeReason: $closeReason');
+    _isLeavingOrClosed = true;
 
     await _eventsSubscription?.cancel();
     await _networkStatusSubscription?.cancel();
@@ -614,6 +625,7 @@ class CallSession extends Disposable {
   @override
   Future<void> dispose() async {
     _logger.d(() => '[dispose] no args');
+    _isLeavingOrClosed = true;
 
     await close(StreamWebSocketCloseCode.normalClosure);
     return await super.dispose();
@@ -683,6 +695,7 @@ class CallSession extends Disposable {
       } else if (event is SfuParticipantLeftEvent) {
         stateManager.sfuParticipantLeft(event);
       } else if (event is SfuConnectionQualityChangedEvent) {
+        _tracer.trace('ConnectionQualityChanged', event.toJson());
         stateManager.sfuConnectionQualityChanged(event);
       } else if (event is SfuAudioLevelChangedEvent) {
         stateManager.sfuUpdateAudioLevelChanged(event);
@@ -905,7 +918,7 @@ class CallSession extends Disposable {
   }
 
   Future<void> _onRenegotiationNeeded(StreamPeerConnection pc) async {
-    if (stateManager.callState.status.isDisconnected) {
+    if (_isLeavingOrClosed || stateManager.callState.status.isDisconnected) {
       _logger.w(() => '[negotiate] call is disconnected');
       return;
     }
@@ -1037,7 +1050,7 @@ class CallSession extends Disposable {
     }
 
     final result = TracerZone.run(
-      _tracer,
+      _zonedTracer,
       ++zonedTracerSeq,
       () async {
         return rtcManager.setCameraEnabled(
@@ -1060,7 +1073,7 @@ class CallSession extends Disposable {
     }
 
     final result = TracerZone.run(
-      _tracer,
+      _zonedTracer,
       ++zonedTracerSeq,
       () async {
         return rtcManager.setMicrophoneEnabled(

packages/stream_video/lib/src/call/session/call_session_factory.dart

@@ -72,18 +72,17 @@ class CallSessionFactory {
 
     _logger.v(() => '[makeCallSession] sfuName: $sfuName, sfuUrl: $sfuUrl');
 
-    final tracer = Tracer('$sessionSeq')
+    final tracer = Tracer('$sessionSeq-$sfuName')
       ..setEnabled(statsOptions.enableRtcStats)
       ..traceMultiple(
         leftoverTraceRecords
             .map(
               (r) => r.copyWith(
-                id: '${max(0, sessionSeq - 1)}',
+                id: '${max(0, sessionSeq - 1)}-$sfuName',
               ),
             )
             .toList(),
-      )
-      ..trace('create', {'url': sfuName});
+      );
 
     return CallSession(
       sessionSeq: sessionSeq,

packages/stream_video/lib/src/call/stats/sfu_stats_reporter.dart

@@ -188,11 +188,14 @@ class SfuStatsReporter {
           final publisherTrace = callSession.rtcManager?.publisher?.tracer
               .take();
           final sessionTrace = callSession.getTrace();
+          final mediaDeviceNotifierTrace = RtcMediaDeviceNotifier.instance
+              .getTrace();
 
           traces.addAll([
             if (subscriberTrace != null) subscriberTrace,
             if (publisherTrace != null) publisherTrace,
-            sessionTrace,
+            ...sessionTrace,
+            mediaDeviceNotifierTrace,
           ]);
         }
 

packages/stream_video/lib/src/call/stats/stats_reporter.dart

@@ -46,6 +46,8 @@ class StatsReporter extends StateNotifier<CallMetrics?> {
     return Stream.periodic(interval, (tick) => (collectStats(), tick)).asyncMap(
       (data) async {
         final stats = await data.$1;
+        if (!mounted) return stats;
+
         unawaited(_processStats(stats, data.$2));
         return stats;
       },
@@ -92,6 +94,8 @@ class StatsReporter extends StateNotifier<CallMetrics?> {
     stats,
     int tick,
   ) async {
+    if (!mounted) return;
+
     var publisherStats = state?.publisher ?? PeerConnectionStats.empty();
     var subscriberStats = state?.subscriber ?? PeerConnectionStats.empty();