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 520cebc9f..7a4b84c8a 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/CHANGELOG.md b/packages/stream_video/CHANGELOG.md index a212c0b5a..9c38f4be5 100644 --- a/packages/stream_video/CHANGELOG.md +++ b/packages/stream_video/CHANGELOG.md @@ -2,6 +2,7 @@ ### 🐞 Fixed +- Serialized remote-description and ICE-candidate handling so a candidate arriving mid-negotiation can no longer be dropped. - 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. 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 71e47fe9c..8e99204db 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'; @@ -10,7 +11,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 +58,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. @@ -124,6 +134,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. @@ -267,17 +280,30 @@ 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 { + await pc.setRemoteDescription(sd); + } catch (e, stk) { + return Result.failure(VideoErrors.compose(e, stk)); } + + final pending = List.of(_pendingCandidates); _pendingCandidates.clear(); - return Result.success(result); - } catch (e, stk) { - return Result.failure(VideoErrors.compose(e, stk)); - } + 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); + }); } //Sets the local description @@ -317,20 +343,25 @@ 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 { - try { - final remoteDescription = await pc.getRemoteDescription(); - if (remoteDescription == null) { - _pendingCandidates.add(candidate); - return Result.error('no remoteDescription set'); + /// 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, + ) { + 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(none); - } 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/lib/src/webrtc/rtc_manager.dart b/packages/stream_video/lib/src/webrtc/rtc_manager.dart index 1e850f7b2..fa6914323 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; } 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..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 @@ -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,57 @@ 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; + + /// 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 { + final sampled = _remoteDescription; + if (getRemoteDescriptionGate != null) { + await getRemoteDescriptionGate!.future; + } + return sampled; + } + + @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 +468,176 @@ void main() { }); }); + group('StreamPeerConnection ICE candidate buffering', () { + rtc.RTCSessionDescription offer() => + rtc.RTCSessionDescription('sdp', 'offer'); + + test( + '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 while it + // is still in flight, 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 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(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: + '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', + ); + }, + ); + + test( + '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); + + 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); + + // c2 is permanently invalid: it throws on every addCandidate attempt. + pc.failOnCandidate = c2; + final flush = await sp.setRemoteDescription(offer()); + + expect( + 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', + ); + expect( + pc.addedCandidates, + [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', () { test( 'drops every rtc-side callback and disposes the underlying pc',