fix(llc): fixes for fast reconnect flow (#918)

* fixes for fast reconnect flow

* tweaks to timeouts

* tweak

* using unique participant key

* tweaks

* fix: don't get member participants on reconnect

* fix

* small refactor + ignoring normal closure of ws

* fixes

* better handling of disconnect state

* cleanup

* tweak

* updated webrtc dependencies

* fix for muting call on Android

---------

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

Author: Brazol

melos.yaml

@@ -18,10 +18,10 @@ command:
 
     # Dependencies used in the project.
     dependencies:
-      dart_webrtc: ^1.4.10
+      dart_webrtc: ^1.5.3
       device_info_plus: ">=10.1.2 <12.0.0"
       stream_chat_flutter: ^9.6.0
-      stream_webrtc_flutter: ^1.0.0-dev.1
+      stream_webrtc_flutter: ^1.0.1
 
 scripts:
   postclean:

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

@@ -80,8 +80,8 @@ const _idSessionEvents = 4;
 const _idSessionStats = 5;
 const _idConnect = 6;
 const _idAwait = 7;
-const _idReconnect = 7;
 const _idFastReconnectTimeout = 8;
+const _idReconnect = 9;
 
 const _tag = 'SV:Call';
 int _callSeq = 1;
@@ -95,6 +95,7 @@ class Call {
     required StreamCallCid callCid,
     required CoordinatorClient coordinatorClient,
     required StreamVideo streamVideo,
+    required InternetConnection networkMonitor,
     RetryPolicy? retryPolicy,
     SdpPolicy? sdpPolicy,
     CallPreferences? preferences,
@@ -104,6 +105,7 @@ class Call {
       callCid: callCid,
       coordinatorClient: coordinatorClient,
       streamVideo: streamVideo,
+      networkMonitor: networkMonitor,
       retryPolicy: retryPolicy,
       sdpPolicy: sdpPolicy,
       preferences: preferences,
@@ -117,6 +119,7 @@ class Call {
     required CallCreatedData data,
     required CoordinatorClient coordinatorClient,
     required StreamVideo streamVideo,
+    required InternetConnection networkMonitor,
     RetryPolicy? retryPolicy,
     SdpPolicy? sdpPolicy,
     CallPreferences? preferences,
@@ -126,6 +129,7 @@ class Call {
       callCid: data.callCid,
       coordinatorClient: coordinatorClient,
       streamVideo: streamVideo,
+      networkMonitor: networkMonitor,
       retryPolicy: retryPolicy,
       sdpPolicy: sdpPolicy,
       preferences: preferences,
@@ -144,6 +148,7 @@ class Call {
     required CallRingingData data,
     required CoordinatorClient coordinatorClient,
     required StreamVideo streamVideo,
+    required InternetConnection networkMonitor,
     RetryPolicy? retryPolicy,
     SdpPolicy? sdpPolicy,
     CallPreferences? preferences,
@@ -153,6 +158,7 @@ class Call {
       callCid: data.callCid,
       coordinatorClient: coordinatorClient,
       streamVideo: streamVideo,
+      networkMonitor: networkMonitor,
       retryPolicy: retryPolicy,
       sdpPolicy: sdpPolicy,
       preferences: preferences,
@@ -163,6 +169,7 @@ class Call {
     required StreamCallCid callCid,
     required CoordinatorClient coordinatorClient,
     required StreamVideo streamVideo,
+    required InternetConnection networkMonitor,
     RetryPolicy? retryPolicy,
     SdpPolicy? sdpPolicy,
     CallPreferences? preferences,
@@ -189,6 +196,7 @@ class Call {
     return Call._(
       coordinatorClient: coordinatorClient,
       streamVideo: streamVideo,
+      networkMonitor: networkMonitor,
       stateManager: stateManager,
       credentials: credentials,
       retryPolicy: finalRetryPolicy,
@@ -202,6 +210,7 @@ class Call {
     required StreamVideo streamVideo,
     required CallStateNotifier stateManager,
     required PermissionsManager permissionManager,
+    required this.networkMonitor,
     required RetryPolicy retryPolicy,
     required SdpPolicy sdpPolicy,
     CallCredentials? credentials,
@@ -240,6 +249,7 @@ class Call {
   final CallStateNotifier _stateManager;
   final PermissionsManager _permissionsManager;
   final DynascaleManager dynascaleManager;
+  final InternetConnection networkMonitor;
 
   CallCredentials? _credentials;
   CallSession? _session;
@@ -253,6 +263,7 @@ class Call {
   SfuReconnectionStrategy _reconnectStrategy =
       SfuReconnectionStrategy.unspecified;
   Future<InternetStatus>? _awaitNetworkAvailableFuture;
+  Future<Result<None>>? _awaitMigrationCompleteFuture;
   bool _initialized = false;
 
   final List<Timer> _reactionTimers = [];
@@ -352,11 +363,10 @@ class Call {
   void _observeReconnectEvents() {
     _subscriptions.add(
       _idReconnect,
-      InternetConnection.createInstance().onStatusChange.listen(
+      networkMonitor.onStatusChange.listen(
         (status) {
           if (status == InternetStatus.disconnected) {
             _logger.d(() => '[observeReconnectEvents] network disconnected');
-            _awaitNetworkAvailableFuture = _awaitNetworkAvailable();
             _reconnect(SfuReconnectionStrategy.fast);
           }
         },
@@ -685,17 +695,15 @@ class Call {
       _credentials = joinedResult.data;
       _previousSession = _session;
 
-      final isWsHealthy = _previousSession?.sfuWS.isConnected ?? false;
-
       final reconnectDetails =
           _reconnectStrategy == SfuReconnectionStrategy.unspecified
               ? null
               : await _previousSession?.getReconnectDetails(_reconnectStrategy);
 
-      if (performingRejoin || performingMigration || !isWsHealthy) {
+      if (!performingFastReconnect) {
         _logger.v(
           () =>
-              '[join] creating new sfu session (rejoin: $performingRejoin, migration: $performingMigration, wsHealthy: $isWsHealthy)',
+              '[join] creating new sfu session (rejoin: $performingRejoin, migration: $performingMigration)',
         );
 
         _session = await _sessionFactory.makeCallSession(
@@ -705,6 +713,7 @@ class Call {
           credentials: _credentials!,
           stateManager: _stateManager,
           dynascaleManager: dynascaleManager,
+          networkMonitor: networkMonitor,
           onPeerConnectionFailure: (pc) async {
             if (state.value.status is! CallStatusReconnecting) {
               await pc.pc.restartIce().onError((_, __) {
@@ -716,20 +725,15 @@ class Call {
               _stateManager.callState.preferences.clientPublishOptions,
         );
 
+        if (performingMigration) {
+          _awaitMigrationCompleteFuture = _session!.waitForMigrationComplete();
+        }
+
         dynascaleManager.init(
           sfuClient: _session!.sfuClient,
           sessionId: _session!.sessionId,
         );
-      } else {
-        _logger.v(
-          () =>
-              '[join] reusing previous sfu session (rejoin: $performingRejoin, migration: $performingMigration, wsHealthy: $isWsHealthy)',
-        );
 
-        _session = _previousSession;
-      }
-
-      if (_session?.sessionSeq != _previousSession?.sessionSeq) {
         _logger.d(() => '[join] starting sfu session');
 
         final sessionResult = await _startSession(
@@ -744,23 +748,27 @@ class Call {
           _stateManager.lifecycleCallConnectFailed(error: error);
           return sessionResult;
         }
-      }
+      } else {
+        _logger.v(
+          () =>
+              '[join] reusing previous sfu session (rejoin: $performingRejoin, migration: $performingMigration)',
+        );
+
+        _session = _previousSession;
 
-      if (performingFastReconnect && isWsHealthy) {
         _logger.d(() => '[join] fast reconnecting');
+        final result = await _session!.fastReconnect();
 
-        final result = await _session?.fastReconnect();
+        if (result.isFailure) {
+          _logger.e(() => '[join] fast reconnecting failed: $result');
+          _reconnectStrategy = SfuReconnectionStrategy.rejoin;
+          return Result.error('fast reconnecting failed');
+        }
 
-        result?.fold(
-          success: (success) {
-            _logger.v(() => '[join] fast reconnecting success');
-            _fastReconnectDeadline = success.data.fastReconnectDeadline;
-          },
-          failure: (failure) {
-            _logger.e(() => '[join] fast reconnecting failed: $failure');
-            return failure;
-          },
-        );
+        _logger.v(() => '[join] fast reconnecting success');
+        _fastReconnectDeadline =
+            result.getDataOrNull()?.fastReconnectDeadline ??
+                _fastReconnectDeadline;
       }
 
       // make sure we only track connection timing if we are not calling this method as part of a migration flow
@@ -779,16 +787,11 @@ class Call {
               'Closing previous WS after reconnect with strategy: ${_reconnectStrategy.name}',
         );
         await _previousSession?.dispose();
-      } else if (!isWsHealthy) {
-        _logger.v(() => '[join] closing unhealthy WS');
-        await _previousSession?.close(
-          StreamWebSocketCloseCode.disposeOldSocket,
-          closeReason: 'Closing unhealthy WS after reconnect',
-        );
       }
 
       // For migration we have to wait for confirmation before we can complete the flow
       if (_reconnectStrategy != SfuReconnectionStrategy.migrate) {
+        _logger.v(() => '[join] connected');
         _previousSession = null;
         _stateManager.lifecycleCallConnected();
       }
@@ -1061,9 +1064,15 @@ class Call {
     }
 
     if (sfuEvent is SfuSocketDisconnected) {
-      _logger.w(() => '[onSfuEvent] socket disconnected');
+      if (!StreamWebSocketCloseCode.isIntentionalClosure(
+        sfuEvent.reason.closeCode,
+      )) {
+        _logger.w(() => '[onSfuEvent] socket disconnected');
+        await _reconnect(SfuReconnectionStrategy.fast);
+      }
     } else if (sfuEvent is SfuSocketFailed) {
       _logger.w(() => '[onSfuEvent] socket failed');
+      await _reconnect(SfuReconnectionStrategy.fast);
     } else if (sfuEvent is SfuGoAwayEvent) {
       _logger.w(() => '[onSfuEvent] go away, migrating sfu');
       await _reconnect(SfuReconnectionStrategy.migrate);
@@ -1094,6 +1103,11 @@ class Call {
   }
 
   Future<void> _reconnect(SfuReconnectionStrategy strategy) async {
+    if (state.value.status is CallStatusDisconnected) {
+      _logger.w(() => '[reconnect] rejected (call is already disconnected)');
+      return;
+    }
+
     if (_callReconnectLock.locked) {
       _logger.w(
         () =>
@@ -1104,6 +1118,7 @@ class Call {
 
     await _callReconnectLock.synchronized(() async {
       _reconnectStrategy = strategy;
+      _awaitNetworkAvailableFuture = _awaitNetworkAvailable();
 
       do {
         if (strategy != SfuReconnectionStrategy.migrate) {
@@ -1183,25 +1198,42 @@ class Call {
     final migrateTimeStopwatch = Stopwatch()..start();
 
     _reconnectStrategy = SfuReconnectionStrategy.migrate;
-    await _join();
-    final result = await _session?.waitForMigrationComplete();
+    final joinResult = await _join();
+
+    if (joinResult.isFailure) {
+      _logger.e(() => '[reconnectMigrate] join failed: $joinResult');
+      _reconnectStrategy = SfuReconnectionStrategy.rejoin;
+      return;
+    }
 
     await _previousSession?.close(StreamWebSocketCloseCode.disposeOldSocket);
 
-    result?.fold(
+    final migrationResult = await _awaitMigrationCompleteFuture;
+    if (migrationResult == null) {
+      _logger.e(() => '[reconnectMigrate] migration failed');
+      _reconnectStrategy = SfuReconnectionStrategy.rejoin;
+      return;
+    }
+
+    migrationResult.fold(
       success: (_) {
         _stateManager.lifecycleCallConnected();
       },
       failure: (_) {
+        _logger.e(
+          () => '[reconnectMigrate] migration did not complete correctly',
+        );
         _reconnectStrategy = SfuReconnectionStrategy.rejoin;
       },
     );
 
-    migrateTimeStopwatch.stop();
-    await _sfuStatsReporter?.sendSfuStats(
-      connectionTimeMs: migrateTimeStopwatch.elapsedMilliseconds,
-      reconnectionStrategy: _reconnectStrategy,
-    );
+    if (migrationResult.isSuccess) {
+      migrateTimeStopwatch.stop();
+      await _sfuStatsReporter?.sendSfuStats(
+        connectionTimeMs: migrateTimeStopwatch.elapsedMilliseconds,
+        reconnectionStrategy: _reconnectStrategy,
+      );
+    }
   }
 
   Future<InternetStatus> _awaitNetworkAvailable() async {
@@ -1213,10 +1245,14 @@ class Call {
       }
     });
 
+    final previousCheckInterval = networkMonitor.checkInterval;
+    networkMonitor.setIntervalAndResetTimer(const Duration(seconds: 1));
+
     final connectionStatus = await InternetConnection.createInstance(
       checkInterval: const Duration(seconds: 1),
     )
         .onStatusChange
+        .startWithFuture(networkMonitor.internetStatus)
         .firstWhere((status) => status == InternetStatus.connected)
         .timeout(
           _retryPolicy.config.callRejoinTimeout,
@@ -1230,6 +1266,8 @@ class Call {
         .valueOrDefault(InternetStatus.disconnected);
 
     fastReconnectTimer.cancel();
+    networkMonitor.setIntervalAndResetTimer(previousCheckInterval);
+
     return connectionStatus;
   }
 
@@ -1280,6 +1318,7 @@ class Call {
     }
 
     _stateManager.lifecycleCallDisconnected(reason: reason);
+
     _logger.v(() => '[leave] finished');
 
     return const Result.success(none);
@@ -2393,6 +2432,11 @@ class Call {
     required SfuTrackTypeVideo trackType,
     RtcVideoDimension? videoDimension,
   }) async {
+    if (state.value.status.isDisconnected) {
+      _logger.d(() => '[updateSubscription] rejected (disconnected)');
+      return const Result.success(none);
+    }
+
     final result = await dynascaleManager.updateSubscription(
       SubscriptionChange.update(
         userId: userId,
@@ -2423,6 +2467,11 @@ class Call {
     required SfuTrackTypeVideo trackType,
     RtcVideoDimension? videoDimension,
   }) async {
+    if (state.value.status.isDisconnected) {
+      _logger.d(() => '[removeSubscription] rejected (disconnected)');
+      return const Result.success(none);
+    }
+
     final result = await dynascaleManager.updateSubscription(
       SubscriptionChange.update(
         userId: userId,
@@ -2626,3 +2675,10 @@ enum TrackType {
     }
   }
 }
+
+extension FutureStartWithEx<T> on Stream<T> {
+  Stream<T> startWithFuture(Future<T> futureValue) async* {
+    yield await futureValue;
+    yield* this;
+  }
+}