fix(llc): renegotiate unacknowledged transceivers (#1288)

* renegotiate unacknowledged transcivers

* changelog

* tweaks

* format fix

* tweaks

* tweak

* tweak

Author: Brazol

packages/stream_video/CHANGELOG.md

@@ -1,7 +1,8 @@
-## Unreleased
+## Upcoming
 
 ### 🐞 Fixed
 
+- Fixed an issue where republishing could reuse a cached publisher transceiver without renegotiating.
 - Fixed a `FormatException` when sending requests if the application name (or other device/app info) contains non-ASCII characters. Values included in the `X-Stream-Client` header are now sanitized to valid header characters.
 
 ## 1.4.2

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

@@ -553,7 +553,15 @@ class CallSession extends Disposable {
       final publisher = rtcManager?.publisher;
       if (publisher != null) {
         _logger.v(() => '[fastReconnect] triggering publisher renegotiation');
-        await _onRenegotiationNeeded(publisher);
+        final renegotiation = await _onRenegotiationNeeded(publisher);
+        if (renegotiation is Failure) {
+          _logger.w(
+            () =>
+                '[fastReconnect] publisher renegotiation failed: '
+                '${renegotiation.error}',
+          );
+          result = renegotiation;
+        }
       }
 
       final remoteTracks = rtcManager!.tracks.values
@@ -620,7 +628,8 @@ class CallSession extends Disposable {
 
         // Signaling stall: local offer created but SetPublisher never completed,
         // leaving the peer connection wedged in `have-local-offer`. Negotiation
-        // won't resume on its own, so renegotiate and rejoin if recovery fails.
+        // won't resume on its own, so renegotiate and fall back to a reconnect
+        // if the recovery fails.
         final signalingState = publisher.pc.signalingState;
         if (signalingState ==
             rtc.RTCSignalingState.RTCSignalingStateHaveLocalOffer) {
@@ -640,9 +649,9 @@ class CallSession extends Disposable {
             _logger.w(
               () =>
                   '[publisherConnectionCheck] recovery renegotiation failed '
-                  '(${result.getErrorOrNull()}) — triggering rejoin',
+                  '(${result.getErrorOrNull()}) — triggering fast reconnect',
             );
-            onReconnectionNeeded(publisher, SfuReconnectionStrategy.rejoin);
+            onReconnectionNeeded(publisher, SfuReconnectionStrategy.fast);
           }
           return;
         }
@@ -1097,7 +1106,7 @@ class CallSession extends Disposable {
       return Result.error('SFU WS is not connected');
     }
 
-    await _negotiationLock.synchronized(() async {
+    return _negotiationLock.synchronized(() async {
       _logger.d(() => '[negotiate] type: ${pc.type}');
 
       final offer = await pc.createOffer();
@@ -1114,7 +1123,14 @@ class CallSession extends Disposable {
         _logger.w(
           () => '[negotiate] rejected(tracksInfo is empty): $tracksInfo',
         );
-        return pc.rollbackLocalDescription();
+
+        // Nothing to publish is a no-op, not a negotiation failure — but a
+        // failed rollback leaves the publisher wedged in `have-local-offer`,
+        // which is.
+        final rollback = await pc.rollbackLocalDescription();
+        if (rollback.isFailure) return rollback;
+
+        return const Result.success(null);
       }
 
       _logger.v(() => '[negotiate] announcing tracks: $tracksInfo');
@@ -1130,7 +1146,10 @@ class CallSession extends Disposable {
 
         if (pubResult is! Success<sfu.SetPublisherResponse>) {
           _logger.w(() => '[negotiate] #setPublisher; failed: $pubResult');
-          return pc.rollbackLocalDescription();
+          await pc.rollbackLocalDescription();
+          return Result<void>.error(
+            'SetPublisher failed: ${pubResult.getErrorOrNull()}',
+          );
         }
 
         if (pubResult.data.hasSdp()) {
@@ -1139,15 +1158,21 @@ class CallSession extends Disposable {
             _logger.w(
               () => '[negotiate] #setRemoteAnswer; failed: $ansResult',
             );
+            return Result<void>.error(
+              'Failed to set remote answer: ${ansResult.getErrorOrNull()}',
+            );
           }
         }
+
+        rtcManager?.transceiversManager.markNegotiated(tracksInfo);
+
+        return const Result.success(null);
       } catch (e, stk) {
         _logger.e(() => '[negotiate] failed: $e\n$stk');
-        return pc.rollbackLocalDescription();
+        await pc.rollbackLocalDescription();
+        return Result.failure(VideoErrors.compose(e, stk));
       }
     });
-
-    return const Result.success(null);
   }
 
   Future<void> _onRemoteTrackReceived(

packages/stream_video/lib/src/webrtc/peer_connection.dart

@@ -88,6 +88,9 @@ class StreamPeerConnection extends Disposable {
   /// Flag to indicate that a reconnect is in progress.
   bool _isReconnecting = false;
 
+  /// Whether a reconnect is currently in progress for this peer connection.
+  bool get isReconnecting => _isReconnecting;
+
   void setReconnecting(bool value) {
     _logger.v(() => '[setReconnecting] #$type; value: $value');
     _isReconnecting = value;

packages/stream_video/lib/src/webrtc/rtc_manager.dart

@@ -811,6 +811,8 @@ extension PublisherRtcManager on RtcManager {
     tracks[audioTrack.trackId] = audioTrack;
     var updatedTrack = audioTrack.copyWith(stopTrackOnMute: stopTrackOnMute);
 
+    var needsRenegotiation = false;
+
     for (final option in publishOptions) {
       if (option.trackType != audioTrack.trackType) continue;
 
@@ -819,7 +821,8 @@ extension PublisherRtcManager on RtcManager {
       final mediaTrackClone = await audioTrack.mediaTrack.clone();
       final trackToPublish = audioTrack.copyWith(mediaTrack: mediaTrackClone);
 
-      final cachedTransceiver = transceiversManager.get(option)?.transceiver;
+      final cachedBundle = transceiversManager.get(option);
+      final cachedTransceiver = cachedBundle?.transceiver;
       if (cachedTransceiver == null) {
         final transceiverResult = await _addTransceiver(
           trackToPublish,
@@ -854,6 +857,13 @@ extension PublisherRtcManager on RtcManager {
           trackPublishOptions: publishOptions,
         );
 
+        // Reusing a transceiver via replaceTrack does not fire
+        // onRenegotiationNeeded. If its previous negotiation never reached the
+        // SFU, the SFU still doesn't know about it, so force a renegotiation.
+        if (cachedBundle != null && !cachedBundle.negotiated) {
+          needsRenegotiation = true;
+        }
+
         _logger.v(
           () => '[publishAudioTrack] cached transceiver: $cachedTransceiver',
         );
@@ -864,6 +874,10 @@ extension PublisherRtcManager on RtcManager {
       );
     }
 
+    if (needsRenegotiation) {
+      _forceRenegotiation('[publishAudioTrack]');
+    }
+
     // Notify listeners.
     onLocalTrackPublished?.call(updatedTrack);
     tracks[updatedTrack.trackId] = updatedTrack;
@@ -901,6 +915,8 @@ extension PublisherRtcManager on RtcManager {
       );
     }
 
+    var needsRenegotiation = false;
+
     for (final option in publishOptions) {
       if (option.trackType != videoTrack.trackType) continue;
 
@@ -909,7 +925,8 @@ extension PublisherRtcManager on RtcManager {
       final mediaTrackClone = await videoTrack.mediaTrack.clone();
       final trackToPublish = videoTrack.copyWith(mediaTrack: mediaTrackClone);
 
-      final cachedTransceiver = transceiversManager.get(option)?.transceiver;
+      final cachedBundle = transceiversManager.get(option);
+      final cachedTransceiver = cachedBundle?.transceiver;
       if (cachedTransceiver == null) {
         final transceiverResult = await _addTransceiver(
           trackToPublish,
@@ -936,6 +953,13 @@ extension PublisherRtcManager on RtcManager {
 
         transceiversManager.update(option, track: trackToPublish);
 
+        // Reusing a transceiver via replaceTrack does not fire
+        // onRenegotiationNeeded. If its previous negotiation never reached the
+        // SFU, the SFU still doesn't know about it, so force a renegotiation.
+        if (cachedBundle != null && !cachedBundle.negotiated) {
+          needsRenegotiation = true;
+        }
+
         _logger.v(
           () => '[publishVideoTrack] cached transceiver: $cachedTransceiver',
         );
@@ -946,6 +970,10 @@ extension PublisherRtcManager on RtcManager {
       );
     }
 
+    if (needsRenegotiation) {
+      _forceRenegotiation('[publishVideoTrack]');
+    }
+
     // Notify listeners.
     onLocalTrackPublished?.call(updatedTrack);
     tracks[updatedTrack.trackId] = updatedTrack;
@@ -1002,6 +1030,39 @@ extension PublisherRtcManager on RtcManager {
     ];
   }
 
+  /// Explicitly triggers a publisher renegotiation.
+  void _forceRenegotiation(String tag) {
+    final pub = publisher;
+    if (pub == null) return;
+
+    if (pub.isReconnecting) {
+      _logger.v(
+        () => '$tag skipping forced renegotiation — reconnect in progress',
+      );
+      return;
+    }
+
+    _logger.v(
+      () =>
+          '$tag forcing renegotiation for a reused transceiver the SFU never '
+          'acknowledged',
+    );
+
+    pub.onRenegotiationNeeded?.call(pub);
+  }
+
+  /// Forces a publisher renegotiation if any transceiver sending [trackId]
+  /// was never acknowledged by the SFU.
+  void _renegotiateIfUnacknowledged(String trackId, String tag) {
+    final hasUnacknowledged = transceiversManager
+        .findAll((t) => t.track.trackId == trackId)
+        .any((t) => t.transceiver.sender.track != null && !t.negotiated);
+
+    if (hasUnacknowledged) {
+      _forceRenegotiation(tag);
+    }
+  }
+
   Future<Result<rtc.RTCRtpTransceiver>> _addTransceiver(
     RtcLocalTrack track,
     SfuPublishOptions publishOptions,
@@ -1181,13 +1242,15 @@ extension PublisherRtcManager on RtcManager {
       tracks[trackId] = updatedTrack;
       onLocalTrackMuted?.call(updatedTrack, false);
 
+      _renegotiateIfUnacknowledged(trackId, '[unmuteTrack]');
       return Result.success(updatedTrack);
     }
 
     // Otherwise simply enable it again
     track.enable();
     onLocalTrackMuted?.call(track, false);
 
+    _renegotiateIfUnacknowledged(trackId, '[unmuteTrack]');
     return Result.success(track);
   }
 

packages/stream_video/lib/src/webrtc/transceiver_cache.dart

@@ -3,6 +3,7 @@ import 'package:stream_webrtc_flutter/stream_webrtc_flutter.dart';
 
 import '../sfu/data/models/sfu_publish_options.dart';
 import '../sfu/data/models/sfu_track_type.dart';
+import 'model/rtc_tracks_info.dart';
 import 'rtc_track/rtc_track.dart';
 import 'rtc_track/rtc_track_publish_options.dart';
 
@@ -12,16 +13,21 @@ class TransceiverCache {
     required this.publishOption,
     required this.transceiver,
     required this.trackPublishOptions,
+    this.negotiated = false,
   });
 
   RtcLocalTrack track;
   SfuPublishOptions publishOption;
   RTCRtpTransceiver transceiver;
   RtcTrackPublishOptions trackPublishOptions;
 
+  /// Whether the SFU has acknowledged this transceiver through a completed
+  /// publisher negotiation.
+  bool negotiated;
+
   @override
   String toString() {
-    return 'TransceiverCache{mediaTrackId: ${track.mediaTrack.id}, publishOption: ${publishOption.id},${publishOption.codec}, sender.track.enabled: ${transceiver.sender.track?.enabled}}';
+    return 'TransceiverCache{mediaTrackId: ${track.mediaTrack.id}, publishOption: ${publishOption.id},${publishOption.codec}, sender.track.enabled: ${transceiver.sender.track?.enabled}, negotiated: $negotiated}';
   }
 }
 
@@ -110,6 +116,20 @@ class TransceiverManager {
     return _transceivers;
   }
 
+  /// Marks the cached transceivers that were part of [announced] as negotiated,
+  /// i.e. acknowledged by the SFU after a completed negotiation.
+  void markNegotiated(Iterable<RtcTrackInfo> announced) {
+    for (final info in announced) {
+      final item = find(
+        (c) =>
+            c.publishOption.id == info.publishOptionId &&
+            c.track.mediaTrack.id == info.trackId,
+      );
+
+      item?.negotiated = true;
+    }
+  }
+
   /// Init index of the transceiver in the cache.
   int indexOf(RTCRtpTransceiver transceiver) {
     return _transceiverOrder.indexOf(transceiver);

packages/stream_video/test/src/call/session/call_session_reconnect_safety_test.dart

@@ -246,8 +246,8 @@ void main() {
     });
 
     test(
-      'renegotiates and escalates to rejoin when the publisher is wedged in '
-      '"have-local-offer" after the delay',
+      'renegotiates and asks for a fast reconnect when the publisher is wedged '
+      'in "have-local-offer" after the delay',
       () {
         fakeAsync((async) {
           final reconnects =
@@ -273,10 +273,13 @@ void main() {
             ..flushMicrotasks();
 
           // The recovery renegotiation cannot complete (no SFU connection in
-          // the test), so the watchdog falls back to a full rejoin.
+          // the test), so the watchdog falls back to a fast reconnect. It does
+          // not ask for a rejoin directly — the reconnect loop escalates to one
+          // once the fast attempts are exhausted or a peer connection is
+          // unhealthy.
           expect(reconnects, hasLength(1));
           expect(reconnects.single.$1, same(wires.publisher));
-          expect(reconnects.single.$2, SfuReconnectionStrategy.rejoin);
+          expect(reconnects.single.$2, SfuReconnectionStrategy.fast);
         });
       },
     );