chore(llc): multiple FAST reconnect attempts and improved ICE restarts (#994)

* reconnect improvements

* quality and scalability fixes

* fixed layers dimentions

* tweaks

* Calculate size of video track with actual pixels

* fixes

* add unit test

* Add test for fast reconnect

* Don't mock stateManager by default

---------

Co-authored-by: Rene Floor <[email protected]>

Author: Brazol

packages/stream_video/lib/globals.dart

@@ -3,7 +3,7 @@ import 'package:meta/meta.dart';
 const String streamSdkName = 'stream-flutter';
 const String streamVideoVersion = '0.9.6';
 const String openapiModelsVersion = '167.9.1';
-const String protocolModelsVersion = '1.35.0';
+const String protocolModelsVersion = '1.38.0';
 const String androidWebRTCVersion = '1.3.8';
 const String iosWebRTCVersion = '125.6422.65';
 

packages/stream_video/lib/protobuf/video/sfu/models/models.pb.dart

@@ -227,12 +227,14 @@ class Call {
     required SdpPolicy sdpPolicy,
     required RtcMediaDeviceNotifier rtcMediaDeviceNotifier,
     CallCredentials? credentials,
-  })  : _sessionFactory = CallSessionFactory(
-          callCid: stateManager.callState.callCid,
-          sdpEditor: sdpPolicy.spdEditingEnabled
-              ? SdpEditorImpl(sdpPolicy)
-              : NoOpSdpEditor(),
-        ),
+    CallSessionFactory? sessionFactory,
+  })  : _sessionFactory = sessionFactory ??
+            CallSessionFactory(
+              callCid: stateManager.callState.callCid,
+              sdpEditor: sdpPolicy.spdEditingEnabled
+                  ? SdpEditorImpl(sdpPolicy)
+                  : NoOpSdpEditor(),
+            ),
         _stateManager = stateManager,
         _permissionsManager = permissionManager,
         _coordinatorClient = coordinatorClient,
@@ -762,13 +764,7 @@ class Call {
           dynascaleManager: dynascaleManager,
           networkMonitor: networkMonitor,
           statsOptions: _sfuStatsOptions!,
-          onPeerConnectionFailure: (pc) async {
-            if (state.value.status is! CallStatusReconnecting) {
-              await pc.pc.restartIce().onError((_, __) {
-                _reconnect(SfuReconnectionStrategy.rejoin);
-              });
-            }
-          },
+          onReconnectionNeeded: (pc, strategy) => _reconnect(strategy),
           clientPublishOptions:
               _stateManager.callState.preferences.clientPublishOptions,
         );
@@ -1209,6 +1205,7 @@ class Call {
       _reconnectStrategy = strategy;
       _awaitNetworkAvailableFuture = _awaitNetworkAvailable();
 
+      var attempt = 0;
       do {
         _stateManager.lifecycleCallConnecting(
           attempt: _reconnectAttempts,
@@ -1262,7 +1259,28 @@ class Call {
               await Future<void>.delayed(
                 _retryPolicy.backoff(_reconnectAttempts),
               );
-              _reconnectStrategy = SfuReconnectionStrategy.rejoin;
+
+              final wasMigrating =
+                  _reconnectStrategy == SfuReconnectionStrategy.migrate;
+
+              // don't immediately switch to the REJOIN strategy, but instead attempt
+              // to reconnect with the FAST strategy for a few times before switching.
+              // in some cases, we immediately switch to the REJOIN strategy.
+              final shouldRejoin =
+                  wasMigrating || // if we were migrating, but the migration failed
+                      attempt >= 3 || // after 3 failed attempts
+                      !(_session?.rtcManager?.publisher?.isHealthy() ??
+                          true) || // if the publisher is not healthy
+                      !(_session?.rtcManager?.subscriber.isHealthy() ??
+                          true); // if the subscriber is not healthy
+
+              attempt++;
+
+              final nextStrategy = shouldRejoin
+                  ? SfuReconnectionStrategy.rejoin
+                  : SfuReconnectionStrategy.fast;
+
+              _reconnectStrategy = nextStrategy;
           }
         }
       } while (state.value.status is! CallStatusConnected &&
@@ -1337,10 +1355,7 @@ class Call {
     final previousCheckInterval = networkMonitor.checkInterval;
     networkMonitor.setIntervalAndResetTimer(const Duration(seconds: 1));
 
-    final connectionStatus = await InternetConnection.createInstance(
-      checkInterval: const Duration(seconds: 1),
-    )
-        .onStatusChange
+    final connectionStatus = await networkMonitor.onStatusChange
         .startWithFuture(networkMonitor.internetStatus)
         .firstWhere((status) => status == InternetStatus.connected)
         .timeout(
@@ -2903,3 +2918,34 @@ extension FutureStartWithEx<T> on Stream<T> {
     yield* this;
   }
 }
+
+// ignore: avoid_classes_with_only_static_members
+/// Call factory to create a [Call] instance.
+/// Only meant for testing purposes.
+@visibleForTesting
+class BaseCallFactory {
+  static Call makeCall({
+    required CoordinatorClient coordinatorClient,
+    required StreamVideo streamVideo,
+    required CallStateNotifier stateManager,
+    required PermissionsManager permissionManager,
+    required InternetConnection networkMonitor,
+    required RetryPolicy retryPolicy,
+    required SdpPolicy sdpPolicy,
+    required RtcMediaDeviceNotifier rtcMediaDeviceNotifier,
+    required CallCredentials? credentials,
+    required CallSessionFactory? sessionFactory,
+  }) =>
+      Call._(
+        coordinatorClient: coordinatorClient,
+        streamVideo: streamVideo,
+        stateManager: stateManager,
+        permissionManager: permissionManager,
+        networkMonitor: networkMonitor,
+        retryPolicy: retryPolicy,
+        sdpPolicy: sdpPolicy,
+        rtcMediaDeviceNotifier: rtcMediaDeviceNotifier,
+        credentials: credentials,
+        sessionFactory: sessionFactory,
+      );
+}

packages/stream_video/lib/protobuf/video/sfu/models/models.pbenum.dart

@@ -55,8 +55,6 @@ const _tag = 'SV:CallSession';
 const _debounceDuration = Duration(milliseconds: 200);
 const _migrationCompleteEventTimeout = Duration(seconds: 7);
 
-typedef OnPeerConnectionIssue = void Function(StreamPeerConnection pc);
-
 class CallSession extends Disposable {
   CallSession({
     required this.callCid,
@@ -65,7 +63,7 @@ class CallSession extends Disposable {
     required this.config,
     required this.stateManager,
     required this.dynascaleManager,
-    required this.onPeerConnectionIssue,
+    required this.onReconnectionNeeded,
     required SdpEditor sdpEditor,
     required this.networkMonitor,
     required this.statsOptions,
@@ -107,7 +105,7 @@ class CallSession extends Disposable {
   final SfuClient sfuClient;
   final SfuWebSocket sfuWS;
   final RtcManagerFactory rtcManagerFactory;
-  final OnPeerConnectionIssue onPeerConnectionIssue;
+  final OnReconnectionNeeded onReconnectionNeeded;
   final ClientPublishOptions? clientPublishOptions;
   final InternetConnection networkMonitor;
   final StatsOptions statsOptions;
@@ -358,13 +356,14 @@ class CallSession extends Disposable {
 
       if (isAnonymousUser) {
         rtcManager = await rtcManagerFactory.makeRtcManager(
+          sfuClient: sfuClient,
           clientDetails: _clientDetails,
           sessionSequence: sessionSeq,
           statsOptions: statsOptions,
         )
           ..onSubscriberIceCandidate = _onLocalIceCandidate
-          ..onSubscriberIssue = onPeerConnectionIssue
           ..onRenegotiationNeeded = _onRenegotiationNeeded
+          ..onReconnectionNeeded = onReconnectionNeeded
           ..onRemoteTrackReceived = _onRemoteTrackReceived;
       } else {
         final currentUserId = stateManager.callState.currentUserId;
@@ -376,6 +375,7 @@ class CallSession extends Disposable {
         _logger.v(() => '[start] localTrackId: $localTrackId');
 
         rtcManager = await rtcManagerFactory.makeRtcManager(
+          sfuClient: sfuClient,
           publisherId: localTrackId,
           publishOptions: event.publishOptions,
           clientDetails: _clientDetails,
@@ -385,10 +385,9 @@ class CallSession extends Disposable {
         )
           ..onPublisherIceCandidate = _onLocalIceCandidate
           ..onSubscriberIceCandidate = _onLocalIceCandidate
-          ..onPublisherIssue = onPeerConnectionIssue
-          ..onSubscriberIssue = onPeerConnectionIssue
           ..onLocalTrackMuted = _onLocalTrackMuted
           ..onLocalTrackPublished = _onLocalTrackPublished
+          ..onReconnectionNeeded = onReconnectionNeeded
           ..onRenegotiationNeeded = _onRenegotiationNeeded
           ..onRemoteTrackReceived = _onRemoteTrackReceived;
       }
@@ -1071,6 +1070,7 @@ extension RtcTracksInfoMapper on List<RtcTrackInfo> {
             ),
             bitrate: layer.parameters.encoding.maxBitrate,
             fps: layer.parameters.encoding.maxFramerate,
+            quality: layer.parameters.encoding.quality.toDTO(),
           );
         }),
       );

packages/stream_video/lib/protobuf/video/sfu/models/models.pbjson.dart

@@ -8,6 +8,7 @@ import '../../models/call_cid.dart';
 import '../../models/call_client_publish_options.dart';
 import '../../models/call_credentials.dart';
 import '../../types/other.dart';
+import '../../webrtc/peer_connection.dart';
 import '../../webrtc/sdp/editor/sdp_editor.dart';
 import '../state/call_state_notifier.dart';
 import '../stats/tracer.dart';
@@ -32,7 +33,7 @@ class CallSessionFactory {
     required CallCredentials credentials,
     required CallStateNotifier stateManager,
     required DynascaleManager dynascaleManager,
-    required OnPeerConnectionIssue onPeerConnectionFailure,
+    required OnReconnectionNeeded onReconnectionNeeded,
     required InternetConnection networkMonitor,
     required StatsOptions statsOptions,
     ClientPublishOptions? clientPublishOptions,
@@ -68,7 +69,7 @@ class CallSessionFactory {
       stateManager: stateManager,
       dynascaleManager: dynascaleManager,
       sdpEditor: sdpEditor,
-      onPeerConnectionIssue: onPeerConnectionFailure,
+      onReconnectionNeeded: onReconnectionNeeded,
       clientPublishOptions: clientPublishOptions,
       networkMonitor: networkMonitor,
       statsOptions: statsOptions,

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

@@ -2,6 +2,9 @@ import 'package:tart/tart.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
 
 import '../../open_api/video/coordinator/api.dart';
+import '../../protobuf/video/sfu/models/models.pb.dart' as sfu_models;
+import '../sfu/data/events/sfu_event_mapper_extensions.dart';
+import '../sfu/data/models/sfu_error.dart';
 import 'video_error.dart';
 
 /// TODO
@@ -13,6 +16,17 @@ mixin VideoErrors {
         message: exception,
         stackTrace: stackTrace,
       );
+    } else if (exception is sfu_models.Error) {
+      return VideoErrorWithCause(
+        message: exception.message,
+        cause: SfuError(
+          message: exception.message,
+          code: exception.code.toDomain(),
+          shouldRetry: exception.shouldRetry,
+          reconnectStrategy: SfuReconnectionStrategy.unspecified,
+        ),
+        stackTrace: stackTrace,
+      );
     } else if (exception is TwirpError) {
       return VideoErrorWithCause(
         message: exception.getMsg,