From a7aa4aaadc20b221ac29cc3a0e14a93315109384 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Braz=CC=87ewicz?= Date: Wed, 22 Jul 2026 21:57:20 +0200 Subject: [PATCH 1/4] improve ice candidate tracing --- .../flutter/generated_plugin_registrant.cc | 4 ---- .../linux/flutter/generated_plugins.cmake | 1 - dogfooding/pubspec.yaml | 2 +- .../windows/flutter/generated_plugins.cmake | 1 - .../lib/src/call/stats/trace_tag.dart | 1 + .../lib/src/webrtc/peer_connection.dart | 24 ++++++++++++++----- .../lib/src/webrtc/rtc_manager.dart | 11 +++++---- .../src/webrtc/traced_peer_connection.dart | 23 +++++++++++------- 8 files changed, 41 insertions(+), 26 deletions(-) diff --git a/dogfooding/linux/flutter/generated_plugin_registrant.cc b/dogfooding/linux/flutter/generated_plugin_registrant.cc index aac6d2082..3bcba2534 100644 --- a/dogfooding/linux/flutter/generated_plugin_registrant.cc +++ b/dogfooding/linux/flutter/generated_plugin_registrant.cc @@ -9,7 +9,6 @@ #include #include #include -#include #include #include #include @@ -24,9 +23,6 @@ void fl_register_plugins(FlPluginRegistry* registry) { g_autoptr(FlPluginRegistrar) gtk_registrar = fl_plugin_registry_get_registrar_for_plugin(registry, "GtkPlugin"); gtk_plugin_register_with_registrar(gtk_registrar); - g_autoptr(FlPluginRegistrar) media_kit_video_registrar = - fl_plugin_registry_get_registrar_for_plugin(registry, "MediaKitVideoPlugin"); - media_kit_video_plugin_register_with_registrar(media_kit_video_registrar); g_autoptr(FlPluginRegistrar) record_linux_registrar = fl_plugin_registry_get_registrar_for_plugin(registry, "RecordLinuxPlugin"); record_linux_plugin_register_with_registrar(record_linux_registrar); diff --git a/dogfooding/linux/flutter/generated_plugins.cmake b/dogfooding/linux/flutter/generated_plugins.cmake index b8f45926a..3a0aaf4df 100644 --- a/dogfooding/linux/flutter/generated_plugins.cmake +++ b/dogfooding/linux/flutter/generated_plugins.cmake @@ -6,7 +6,6 @@ list(APPEND FLUTTER_PLUGIN_LIST desktop_drop file_selector_linux gtk - media_kit_video record_linux stream_webrtc_flutter url_launcher_linux diff --git a/dogfooding/pubspec.yaml b/dogfooding/pubspec.yaml index e006691ae..ff2a25925 100644 --- a/dogfooding/pubspec.yaml +++ b/dogfooding/pubspec.yaml @@ -38,7 +38,7 @@ dependencies: rxdart: ^0.28.0 share_plus: ^13.0.0 shared_preferences: ^2.5.3 - stream_chat_flutter: ^10.0.0 + stream_chat_flutter: ^10.2.0 stream_video_filters: ^1.4.2 stream_video_flutter: ^1.4.2 stream_video_noise_cancellation: ^1.4.2 diff --git a/dogfooding/windows/flutter/generated_plugins.cmake b/dogfooding/windows/flutter/generated_plugins.cmake index 310f70598..4f60c1cdf 100644 --- a/dogfooding/windows/flutter/generated_plugins.cmake +++ b/dogfooding/windows/flutter/generated_plugins.cmake @@ -11,7 +11,6 @@ list(APPEND FLUTTER_PLUGIN_LIST firebase_auth firebase_core gal - media_kit_video permission_handler_windows record_windows share_plus diff --git a/packages/stream_video/lib/src/call/stats/trace_tag.dart b/packages/stream_video/lib/src/call/stats/trace_tag.dart index 00727515a..767dc3a2e 100644 --- a/packages/stream_video/lib/src/call/stats/trace_tag.dart +++ b/packages/stream_video/lib/src/call/stats/trace_tag.dart @@ -88,6 +88,7 @@ abstract class TraceTag { static const String setLocalDescriptionError = 'setLocalDescription.error'; static const String addIceCandidate = 'addIceCandidate'; static const String addIceCandidateSuccess = 'addIceCandidate.success'; + static const String addIceCandidatePending = 'addIceCandidate.pending'; static const String addIceCandidateError = 'addIceCandidate.error'; // Peer connection factory diff --git a/packages/stream_video/lib/src/webrtc/peer_connection.dart b/packages/stream_video/lib/src/webrtc/peer_connection.dart index d8d659f7f..094f6355b 100644 --- a/packages/stream_video/lib/src/webrtc/peer_connection.dart +++ b/packages/stream_video/lib/src/webrtc/peer_connection.dart @@ -10,7 +10,6 @@ import '../logger/impl/tagged_logger.dart'; import '../models/call_cid.dart'; import '../sfu/data/models/sfu_error.dart'; import '../sfu/sfu_client.dart'; -import '../utils/none.dart'; import '../utils/result.dart'; import '../utils/standard.dart'; import 'model/stats/rtc_printable_stats.dart'; @@ -58,6 +57,16 @@ typedef OnTrack = rtc.RTCTrackEvent, ); +enum AddIceCandidateResult { + /// The candidate was applied to the peer connection immediately. + added, + + /// The remote description was not set yet, so the candidate was buffered + /// and will be applied when [StreamPeerConnection.setRemoteDescription] + /// completes. This is a normal, deferred-success outcome — not a drop. + buffered, +} + /// Wrapper around the WebRTC connection that contains tracks. class StreamPeerConnection extends Disposable { /// Creates [StreamPeerConnection] instance. @@ -314,17 +323,20 @@ class StreamPeerConnection extends Disposable { /// Adds an ice candidate to the peer connection. /// - /// If the peer connection is not yet ready, the candidate is added to a list - /// of pending candidates. - Future> addIceCandidate(rtc.RTCIceCandidate candidate) async { + /// If the remote description has not been set yet, the candidate cannot be + /// applied immediately, so it is buffered in [_pendingCandidates] and flushed + /// by [setRemoteDescription]. + Future> addIceCandidate( + rtc.RTCIceCandidate candidate, + ) async { try { final remoteDescription = await pc.getRemoteDescription(); if (remoteDescription == null) { _pendingCandidates.add(candidate); - return Result.error('no remoteDescription set'); + return const Result.success(AddIceCandidateResult.buffered); } await pc.addCandidate(candidate); - return const Result.success(none); + return const Result.success(AddIceCandidateResult.added); } catch (e, stk) { return Result.failure(VideoErrors.compose(e, stk)); } diff --git a/packages/stream_video/lib/src/webrtc/rtc_manager.dart b/packages/stream_video/lib/src/webrtc/rtc_manager.dart index 4531d4bcb..04dde7db6 100644 --- a/packages/stream_video/lib/src/webrtc/rtc_manager.dart +++ b/packages/stream_video/lib/src/webrtc/rtc_manager.dart @@ -217,13 +217,16 @@ class RtcManager extends Disposable { required String iceCandidate, }) async { final candidate = RtcIceCandidateParser.fromJsonString(iceCandidate); + final Result? result; if (peerType == StreamPeerType.publisher) { - return publisher?.addIceCandidate(candidate) ?? - Result.error('no publisher created'); + result = await publisher?.addIceCandidate(candidate); + if (result == null) return Result.error('no publisher created'); } else if (peerType == StreamPeerType.subscriber) { - return subscriber.addIceCandidate(candidate); + result = await subscriber.addIceCandidate(candidate); + } else { + return Result.error('unexpected peerType: $peerType'); } - return Result.error('unexpected peerType: $peerType'); + return result.map((_) => none); } void _onRemoteTrack(StreamPeerConnection pc, rtc.RTCTrackEvent event) { diff --git a/packages/stream_video/lib/src/webrtc/traced_peer_connection.dart b/packages/stream_video/lib/src/webrtc/traced_peer_connection.dart index 2e8c3117b..a2e6dde17 100644 --- a/packages/stream_video/lib/src/webrtc/traced_peer_connection.dart +++ b/packages/stream_video/lib/src/webrtc/traced_peer_connection.dart @@ -9,7 +9,6 @@ import '../call/stats/tracer.dart'; import '../sfu/data/models/sfu_codec.dart'; import '../sfu/data/models/sfu_model_mapper_extensions.dart'; import '../sfu/data/models/sfu_track_type.dart'; -import '../utils/none.dart'; import '../utils/result.dart'; import 'model/stats/rtc_codec.dart'; import 'model/stats/rtc_inbound_rtp_video_stream.dart'; @@ -471,19 +470,25 @@ class TracedStreamPeerConnection extends StreamPeerConnection { } @override - Future> addIceCandidate(rtc.RTCIceCandidate candidate) async { + Future> addIceCandidate( + rtc.RTCIceCandidate candidate, + ) async { tracer.trace(TraceTag.addIceCandidate, candidate.toMap()); final result = await super.addIceCandidate(candidate); - if (result.isSuccess) { - tracer.trace(TraceTag.addIceCandidateSuccess, null); - } else { - tracer.trace( + result.when( + success: (outcome) => tracer.trace( + outcome == AddIceCandidateResult.buffered + ? TraceTag.addIceCandidatePending + : TraceTag.addIceCandidateSuccess, + null, + ), + failure: (error) => tracer.trace( TraceTag.addIceCandidateError, - result.getErrorOrNull()?.toString(), - ); - } + error.toString(), + ), + ); return result; } From d60e48d5de9b8640f25bd3a4246754b5ef641460 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Braz=CC=87ewicz?= Date: Thu, 23 Jul 2026 11:32:31 +0200 Subject: [PATCH 2/4] serialize icecandidate and negotiate --- packages/stream_video/CHANGELOG.md | 6 + .../lib/src/webrtc/peer_connection.dart | 52 +++++--- .../peer_connection_renegotiation_test.dart | 120 ++++++++++++++++++ 3 files changed, 157 insertions(+), 21 deletions(-) diff --git a/packages/stream_video/CHANGELOG.md b/packages/stream_video/CHANGELOG.md index aa9c26323..243308903 100644 --- a/packages/stream_video/CHANGELOG.md +++ b/packages/stream_video/CHANGELOG.md @@ -1,3 +1,9 @@ +## Upcoming + +### 🐞 Fixed + +- Serialized remote-description and ICE-candidate handling so a candidate arriving mid-negotiation can no longer be dropped. + ## 1.4.2 ### ✅ Added diff --git a/packages/stream_video/lib/src/webrtc/peer_connection.dart b/packages/stream_video/lib/src/webrtc/peer_connection.dart index 094f6355b..1dbc4e909 100644 --- a/packages/stream_video/lib/src/webrtc/peer_connection.dart +++ b/packages/stream_video/lib/src/webrtc/peer_connection.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'package:stream_webrtc_flutter/stream_webrtc_flutter.dart' as rtc; +import 'package:synchronized/synchronized.dart'; import '../../protobuf/video/sfu/models/models.pbenum.dart'; import '../../protobuf/video/sfu/signal_rpc/signal.pb.dart'; @@ -130,6 +131,9 @@ class StreamPeerConnection extends Disposable { final _pendingCandidates = []; + /// Serializes [setRemoteDescription] and [addIceCandidate] + final _candidateLock = Lock(); + /// Attempts to restart ICE on the `RTCPeerConnection`. /// If the restart fails, this method will trigger onReconnectionNeeded with /// the appropriate reconnection strategy based on the error. @@ -273,17 +277,21 @@ class StreamPeerConnection extends Disposable { /// Sets the remote description and adds any pending ice candidates. Future> setRemoteDescription( rtc.RTCSessionDescription sd, - ) async { - try { - final result = await pc.setRemoteDescription(sd); - for (final candidate in _pendingCandidates) { - await pc.addCandidate(candidate); + ) { + return _candidateLock.synchronized(() async { + try { + final result = await pc.setRemoteDescription(sd); + // Flush buffered candidates. + final pending = List.of(_pendingCandidates); + for (final candidate in pending) { + await pc.addCandidate(candidate); + _pendingCandidates.remove(candidate); + } + return Result.success(result); + } catch (e, stk) { + return Result.failure(VideoErrors.compose(e, stk)); } - _pendingCandidates.clear(); - return Result.success(result); - } catch (e, stk) { - return Result.failure(VideoErrors.compose(e, stk)); - } + }); } //Sets the local description @@ -328,18 +336,20 @@ class StreamPeerConnection extends Disposable { /// by [setRemoteDescription]. Future> addIceCandidate( rtc.RTCIceCandidate candidate, - ) async { - try { - final remoteDescription = await pc.getRemoteDescription(); - if (remoteDescription == null) { - _pendingCandidates.add(candidate); - return const Result.success(AddIceCandidateResult.buffered); + ) { + return _candidateLock.synchronized(() async { + try { + final remoteDescription = await pc.getRemoteDescription(); + if (remoteDescription == null) { + _pendingCandidates.add(candidate); + return const Result.success(AddIceCandidateResult.buffered); + } + await pc.addCandidate(candidate); + return const Result.success(AddIceCandidateResult.added); + } catch (e, stk) { + return Result.failure(VideoErrors.compose(e, stk)); } - await pc.addCandidate(candidate); - return const Result.success(AddIceCandidateResult.added); - } catch (e, stk) { - return Result.failure(VideoErrors.compose(e, stk)); - } + }); } /// Adds a local [rtc.MediaStreamTrack] with audio to the current connection. diff --git a/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart b/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart index 18950860f..558dcbaab 100644 --- a/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart +++ b/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:fake_async/fake_async.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:mocktail/mocktail.dart'; @@ -91,6 +93,43 @@ class _FakeRtcPeerConnection extends Fake implements rtc.RTCPeerConnection { @override rtc.RTCIceConnectionState? get iceConnectionState => stubbedIceConnectionState; + + // --- Remote description / ICE candidate plumbing ------------------------ + + rtc.RTCSessionDescription? _remoteDescription; + + /// Every candidate that reached [addCandidate] successfully, in order. + final addedCandidates = []; + + /// When non-null, [addCandidate] throws for this exact candidate. Used to + /// simulate a partial flush failure. + rtc.RTCIceCandidate? failOnCandidate; + + /// When non-null, [setRemoteDescription] blocks on this until completed — + /// lets a test hold the operation "in flight" while another one is issued. + Completer? setRemoteDescriptionGate; + + @override + Future getRemoteDescription() async => + _remoteDescription; + + @override + Future setRemoteDescription( + rtc.RTCSessionDescription description, + ) async { + if (setRemoteDescriptionGate != null) { + await setRemoteDescriptionGate!.future; + } + _remoteDescription = description; + } + + @override + Future addCandidate(rtc.RTCIceCandidate candidate) async { + if (failOnCandidate != null && identical(candidate, failOnCandidate)) { + throw Exception('addCandidate failed for $candidate'); + } + addedCandidates.add(candidate); + } } StreamPeerConnection _build({ @@ -415,6 +454,87 @@ void main() { }); }); + group('StreamPeerConnection ICE candidate buffering', () { + rtc.RTCSessionDescription offer() => + rtc.RTCSessionDescription('sdp', 'offer'); + + test( + 'a candidate that arrives while setRemoteDescription is in flight is ' + 'applied exactly once and never stranded', + () async { + final pc = _FakeRtcPeerConnection(); + final sp = _build(pc: pc, type: StreamPeerType.subscriber); + + // Hold setRemoteDescription open so addIceCandidate is issued + // concurrently, then release it. + final gate = Completer(); + pc.setRemoteDescriptionGate = gate; + + final candidate = rtc.RTCIceCandidate('cand', 'mid', 0); + + final setFuture = sp.setRemoteDescription(offer()); + final addFuture = sp.addIceCandidate(candidate); + + gate.complete(); + final results = await Future.wait([setFuture, addFuture]); + + expect(results[0].isSuccess, isTrue); + expect(results[1].isSuccess, isTrue); + expect( + pc.addedCandidates, + [candidate], + reason: + 'the candidate must be applied exactly once regardless of which ' + 'operation acquires the lock first', + ); + }, + ); + + test( + 'a partial flush failure keeps failed and unattempted candidates ' + 'buffered without re-applying the ones already added', + () async { + final pc = _FakeRtcPeerConnection(); + final sp = _build(pc: pc, type: StreamPeerType.subscriber); + + final c1 = rtc.RTCIceCandidate('c1', 'mid', 0); + final c2 = rtc.RTCIceCandidate('c2', 'mid', 0); + final c3 = rtc.RTCIceCandidate('c3', 'mid', 0); + + // Remote description not set yet -> all three buffer. + for (final c in [c1, c2, c3]) { + final r = await sp.addIceCandidate(c); + expect(r.getDataOrNull(), AddIceCandidateResult.buffered); + } + expect(pc.addedCandidates, isEmpty); + + // Flush fails on c2: c1 applied then c2 throws. + pc.failOnCandidate = c2; + final firstFlush = await sp.setRemoteDescription(offer()); + + expect(firstFlush.isFailure, isTrue); + expect( + pc.addedCandidates, + [c1], + reason: 'only c1 was applied before c2 failed', + ); + + // Recover and flush again: c2 and c3 apply, c1 is NOT re-applied. + pc.failOnCandidate = null; + final secondFlush = await sp.setRemoteDescription(offer()); + + expect(secondFlush.isSuccess, isTrue); + expect( + pc.addedCandidates, + [c1, c2, c3], + reason: + 'each candidate is applied exactly once — the already-added c1 ' + 'must not be re-applied on retry', + ); + }, + ); + }); + group('StreamPeerConnection.dispose', () { test( 'drops every rtc-side callback and disposes the underlying pc', From 8b4d9644939e5fa4e1aa50bbb74be4d79c58fdbc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Braz=CC=87ewicz?= Date: Tue, 28 Jul 2026 21:18:50 +0200 Subject: [PATCH 3/4] tweaks --- .../lib/src/webrtc/peer_connection.dart | 25 ++++-- .../peer_connection_renegotiation_test.dart | 76 ++++++++++++++----- 2 files changed, 75 insertions(+), 26 deletions(-) diff --git a/packages/stream_video/lib/src/webrtc/peer_connection.dart b/packages/stream_video/lib/src/webrtc/peer_connection.dart index 1dbc4e909..4de6b4db2 100644 --- a/packages/stream_video/lib/src/webrtc/peer_connection.dart +++ b/packages/stream_video/lib/src/webrtc/peer_connection.dart @@ -280,17 +280,26 @@ class StreamPeerConnection extends Disposable { ) { return _candidateLock.synchronized(() async { try { - final result = await pc.setRemoteDescription(sd); - // Flush buffered candidates. - final pending = List.of(_pendingCandidates); - for (final candidate in pending) { - await pc.addCandidate(candidate); - _pendingCandidates.remove(candidate); - } - return Result.success(result); + await pc.setRemoteDescription(sd); } catch (e, stk) { return Result.failure(VideoErrors.compose(e, stk)); } + + final pending = List.of(_pendingCandidates); + _pendingCandidates.clear(); + for (final candidate in pending) { + try { + await pc.addCandidate(candidate); + } catch (e) { + _logger.w( + () => + '[setRemoteDescription] #$type; dropping candidate ' + '${candidate.candidate}; failed: $e', + ); + } + } + + return const Result.success(null); }); } diff --git a/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart b/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart index 558dcbaab..f63b16e7e 100644 --- a/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart +++ b/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart @@ -491,8 +491,8 @@ void main() { ); test( - 'a partial flush failure keeps failed and unattempted candidates ' - 'buffered without re-applying the ones already added', + 'a candidate that fails to apply during the flush does not fail ' + 'setRemoteDescription, and the rest of the buffer still applies', () async { final pc = _FakeRtcPeerConnection(); final sp = _build(pc: pc, type: StreamPeerType.subscriber); @@ -508,31 +508,71 @@ void main() { } expect(pc.addedCandidates, isEmpty); - // Flush fails on c2: c1 applied then c2 throws. + // c2 is permanently invalid: it throws on every addCandidate attempt. pc.failOnCandidate = c2; - final firstFlush = await sp.setRemoteDescription(offer()); + final flush = await sp.setRemoteDescription(offer()); - expect(firstFlush.isFailure, isTrue); expect( - pc.addedCandidates, - [c1], - reason: 'only c1 was applied before c2 failed', + flush.isSuccess, + isTrue, + reason: + 'the remote description was applied — a bad candidate must not ' + 'turn that into a Failure, or onSubscriberOffer skips the answer ' + 'and negotiation stalls', ); - - // Recover and flush again: c2 and c3 apply, c1 is NOT re-applied. - pc.failOnCandidate = null; - final secondFlush = await sp.setRemoteDescription(offer()); - - expect(secondFlush.isSuccess, isTrue); expect( pc.addedCandidates, - [c1, c2, c3], - reason: - 'each candidate is applied exactly once — the already-added c1 ' - 'must not be re-applied on retry', + [c1, c3], + reason: 'c2 failing must not stop c3 from being applied', ); }, ); + + test( + 'a candidate that failed the flush is dropped, so it cannot poison ' + 'later renegotiations', + () async { + final pc = _FakeRtcPeerConnection(); + final sp = _build(pc: pc, type: StreamPeerType.subscriber); + + final bad = rtc.RTCIceCandidate('bad', 'mid', 0); + await sp.addIceCandidate(bad); + + // `bad` keeps throwing for the whole lifetime of the connection. + pc.failOnCandidate = bad; + + expect((await sp.setRemoteDescription(offer())).isSuccess, isTrue); + expect(pc.addedCandidates, isEmpty); + + // Renegotiation: the dropped candidate must not be retried, so nothing + // throws and the answer path stays healthy. + expect((await sp.setRemoteDescription(offer())).isSuccess, isTrue); + expect((await sp.setRemoteDescription(offer())).isSuccess, isTrue); + expect(pc.addedCandidates, isEmpty); + }, + ); + + test( + 'candidates buffered before the remote description are each applied ' + 'exactly once', + () async { + final pc = _FakeRtcPeerConnection(); + final sp = _build(pc: pc, type: StreamPeerType.subscriber); + + final c1 = rtc.RTCIceCandidate('c1', 'mid', 0); + final c2 = rtc.RTCIceCandidate('c2', 'mid', 0); + + await sp.addIceCandidate(c1); + await sp.addIceCandidate(c2); + + await sp.setRemoteDescription(offer()); + expect(pc.addedCandidates, [c1, c2]); + + // A later renegotiation must not re-flush an already-drained buffer. + await sp.setRemoteDescription(offer()); + expect(pc.addedCandidates, [c1, c2]); + }, + ); }); group('StreamPeerConnection.dispose', () { From 5149275dccc30f5a24bf3ccd96385e576d36b9a9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maciej=20Braz=CC=87ewicz?= Date: Thu, 30 Jul 2026 16:01:43 +0200 Subject: [PATCH 4/4] test fix --- .../peer_connection_renegotiation_test.dart | 85 ++++++++++++++++--- 1 file changed, 74 insertions(+), 11 deletions(-) diff --git a/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart b/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart index f63b16e7e..b3b4d5127 100644 --- a/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart +++ b/packages/stream_video/test/src/webrtc/peer_connection_renegotiation_test.dart @@ -109,9 +109,23 @@ class _FakeRtcPeerConnection extends Fake implements rtc.RTCPeerConnection { /// lets a test hold the operation "in flight" while another one is issued. Completer? setRemoteDescriptionGate; + /// When non-null, [getRemoteDescription] samples the current remote + /// description immediately but only *delivers* it once this completes. + /// + /// This models the real platform-channel round trip: the native side answers + /// with whatever it sees at call time, and that answer can reach Dart long + /// after an interleaved `setRemoteDescription` has finished — so the caller + /// resumes holding a stale `null`. + Completer? getRemoteDescriptionGate; + @override - Future getRemoteDescription() async => - _remoteDescription; + Future getRemoteDescription() async { + final sampled = _remoteDescription; + if (getRemoteDescriptionGate != null) { + await getRemoteDescriptionGate!.future; + } + return sampled; + } @override Future setRemoteDescription( @@ -459,14 +473,14 @@ void main() { rtc.RTCSessionDescription('sdp', 'offer'); test( - 'a candidate that arrives while setRemoteDescription is in flight is ' - 'applied exactly once and never stranded', + 'a candidate issued while setRemoteDescription is in flight waits for it ' + 'and is then applied directly, exactly once', () async { final pc = _FakeRtcPeerConnection(); final sp = _build(pc: pc, type: StreamPeerType.subscriber); - // Hold setRemoteDescription open so addIceCandidate is issued - // concurrently, then release it. + // Hold setRemoteDescription open so addIceCandidate is issued while it + // is still in flight, then release it. final gate = Completer(); pc.setRemoteDescriptionGate = gate; @@ -476,16 +490,65 @@ void main() { final addFuture = sp.addIceCandidate(candidate); gate.complete(); - final results = await Future.wait([setFuture, addFuture]); + final setResult = await setFuture; + final addResult = await addFuture; + + expect(setResult.isSuccess, isTrue); + expect( + addResult.getDataOrNull(), + AddIceCandidateResult.added, + reason: + 'setRemoteDescription was issued first, so it runs first: by the ' + 'time the candidate is handled the remote description exists and ' + 'it needs no buffering', + ); + expect(pc.addedCandidates, [candidate]); + }, + ); + + test( + 'a candidate whose remote-description read resolves after the flush is ' + 'not stranded', + () async { + final pc = _FakeRtcPeerConnection(); + final sp = _build(pc: pc, type: StreamPeerType.subscriber); + + // This is the stall: addIceCandidate samples a null remote description, + // and that answer only comes back after setRemoteDescription has + // already drained the buffer. Unserialized, the candidate lands in a + // buffer nobody will flush again and the subscriber never connects. + final gate = Completer(); + pc.getRemoteDescriptionGate = gate; + + final candidate = rtc.RTCIceCandidate('cand', 'mid', 0); + + final addFuture = sp.addIceCandidate(candidate); + // Let the read actually be issued (and sample null) before the remote + // description is set. + await pumpEventQueue(); + + final setFuture = sp.setRemoteDescription(offer()); + // Give setRemoteDescription every chance to run to completion while the + // candidate is still parked on the read. + await pumpEventQueue(); + + gate.complete(); + final addResult = await addFuture; + final setResult = await setFuture; - expect(results[0].isSuccess, isTrue); - expect(results[1].isSuccess, isTrue); + expect(setResult.isSuccess, isTrue); + expect( + addResult.getDataOrNull(), + AddIceCandidateResult.buffered, + reason: 'the read returned null, so the candidate had to be buffered', + ); expect( pc.addedCandidates, [candidate], reason: - 'the candidate must be applied exactly once regardless of which ' - 'operation acquires the lock first', + 'a candidate buffered on a stale null read must still reach the ' + 'peer connection — setRemoteDescription and addIceCandidate are ' + 'serialized so the flush cannot run past it', ); }, );