diff --git a/.changes/offer-ids b/.changes/offer-ids new file mode 100644 index 000000000..51d2860f7 --- /dev/null +++ b/.changes/offer-ids @@ -0,0 +1 @@ +patch type="fixed" "Offers and answers now carry offer ids, so a late answer to an offer that an ICE restart or a newer offer replaced is ignored instead of being applied to the wrong offer" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 7ac40969f..c08949f43 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -700,9 +700,9 @@ class Engine extends Disposable with EventsEmittable { } }; - publisher?.onOffer = (offer) { - logger.fine('publisher onOffer'); - signalClient.sendOffer(offer); + publisher?.onOffer = (offer, offerId) { + logger.fine('publisher onOffer (offerId: $offerId)'); + signalClient.sendOffer(offer, offerId); }; publisher?.onNegotiationError = _onPublisherNegotiationError; @@ -1541,18 +1541,18 @@ class Engine extends Disposable with EventsEmittable { final signalingState = await subscriber!.pc.getSignalingState(); logger.fine( '[$objectId] Received server offer(type: ${event.sd.type}, ' - '$signalingState)', + 'offerId: ${event.offerId}, $signalingState)', ); logger.finer('sdp: ${event.sd.sdp}'); - await subscriber!.setRemoteDescription(event.sd); + await subscriber!.setRemoteDescription(event.sd, offerId: event.offerId); try { final answer = await subscriber!.pc.createAnswer(); logger.fine('Created answer'); logger.finer('sdp: ${answer.sdp}'); await subscriber!.pc.setLocalDescription(answer); - signalClient.sendAnswer(answer); + signalClient.sendAnswer(answer, event.offerId); } catch (_) { logger.severe('[$objectId] Failed to createAnswer()'); } @@ -1561,9 +1561,9 @@ class Engine extends Disposable with EventsEmittable { if (publisher == null) { return; } - logger.fine('received answer (type: ${event.sd.type})'); + logger.fine('received answer (type: ${event.sd.type}, offerId: ${event.offerId})'); logger.finer('sdp: ${event.sd.sdp}'); - await publisher!.setRemoteDescription(event.sd); + await publisher!.setRemoteDescription(event.sd, offerId: event.offerId); events.emit(const EngineSubscribedCodecRefreshRequestedEvent()); }) ..on((event) async { diff --git a/lib/src/core/signal_client.dart b/lib/src/core/signal_client.dart index 497342e48..2f5e59194 100644 --- a/lib/src/core/signal_client.dart +++ b/lib/src/core/signal_client.dart @@ -287,10 +287,10 @@ class SignalClient extends Disposable with EventsEmittable { events.emit(SignalJoinResponseEvent(response: msg.join)); break; case lk_rtc.SignalResponse_Message.answer: - events.emit(SignalAnswerEvent(sd: msg.answer.toSDKType())); + events.emit(SignalAnswerEvent(sd: msg.answer.toSDKType(), offerId: msg.answer.id)); break; case lk_rtc.SignalResponse_Message.offer: - events.emit(SignalOfferEvent(sd: msg.offer.toSDKType())); + events.emit(SignalOfferEvent(sd: msg.offer.toSDKType(), offerId: msg.offer.id)); break; case lk_rtc.SignalResponse_Message.trickle: events.emit( @@ -474,16 +474,16 @@ class SignalClient extends Disposable with EventsEmittable { extension SignalClientRequests on SignalClient { @internal - void sendOffer(rtc.RTCSessionDescription offer) => _sendRequest( + void sendOffer(rtc.RTCSessionDescription offer, int offerId) => _sendRequest( lk_rtc.SignalRequest( - offer: offer.toPBType(), + offer: offer.toPBType(id: offerId), ), ); @internal - void sendAnswer(rtc.RTCSessionDescription answer) => _sendRequest( + void sendAnswer(rtc.RTCSessionDescription answer, int offerId) => _sendRequest( lk_rtc.SignalRequest( - answer: answer.toPBType(), + answer: answer.toPBType(id: offerId), ), ); diff --git a/lib/src/core/transport.dart b/lib/src/core/transport.dart index 929f59c5c..b17adde9e 100644 --- a/lib/src/core/transport.dart +++ b/lib/src/core/transport.dart @@ -18,6 +18,7 @@ import 'dart:math' as math; import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:meta/meta.dart'; import 'package:sdp_transform/sdp_transform.dart' as sdp_transform; +import 'package:synchronized/synchronized.dart' as sync; import '../exceptions.dart'; import '../extensions.dart'; @@ -204,7 +205,7 @@ void applyVideoStartBitrate(Map media, int codecPayload, int st media['fmtp'] = fmtpList; } -typedef TransportOnOffer = void Function(rtc.RTCSessionDescription offer); +typedef TransportOnOffer = void Function(rtc.RTCSessionDescription offer, int offerId); typedef TransportOnNegotiationError = void Function(Object error); typedef PeerConnectionCreate = Future Function(Map configuration, [Map constraints]); @@ -231,9 +232,21 @@ class Transport extends Disposable { bool restartingIce = false; bool renegotiate = false; - /// Whether an answer is being applied right now, so a copy that arrives + /// Id of the last offer created. Ids start at 1, so 0 stays free for an answer from a + /// server that does not echo offer ids. + int _latestOfferId = 0; + + /// Id of the last offer whose answer was applied. + int _answeredOfferId = 0; + + /// Id (0 if unnumbered) of the answer being applied right now, so a copy that arrives /// mid-apply is recognised as a duplicate too. - bool _answerInFlight = false; + int? _answerInFlight; + + /// Builds one offer at a time. A call that overlaps another waits for it, then finds that + /// offer waiting for its answer and defers to [renegotiate] instead of sending a second one. + /// Only an ICE restart replaces a waiting offer. + final _offerLock = sync.Lock(); TransportOnOffer? onOffer; TransportOnNegotiationError? onNegotiationError; Function? _cancelDebounce; @@ -314,30 +327,40 @@ class Transport extends Disposable { } } - Future setRemoteDescription(rtc.RTCSessionDescription sd) async { + /// [offerId] is the id of the offer an answer answers, or the server's id for a remote + /// offer. 0 from servers that send none. + Future setRemoteDescription(rtc.RTCSessionDescription sd, {int offerId = 0}) async { if (isDisposed) { logger.warning('[$objectId] setRemoteDescription() already disposed'); return; } if (sd.type == 'answer') { - // Offers are not numbered on this transport, so the signaling state is what says - // whether an answer is still wanted: in `stable` there is no local offer left to - // answer, and a copy arriving mid-apply is caught by the in-flight flag. - if (_answerInFlight || await pc.getSignalingState() == rtc.RTCSignalingState.RTCSignalingStateStable) { - logger.warning('[$objectId] ignoring duplicate answer'); + if (_latestOfferId > 0 && offerId > 0 && offerId != _latestOfferId) { + logger.warning('[$objectId] ignoring answer for old offer $offerId, latest is $_latestOfferId'); + return; + } + if (_answerInFlight == offerId) { + logger.warning('[$objectId] ignoring duplicate answer (offerId: $offerId)'); return; } - _answerInFlight = true; + // claimed before any await, so a copy arriving mid-apply is caught too + _answerInFlight = offerId; } try { + if (sd.type == 'answer' && await _isAnswered(offerId)) { + logger.warning('[$objectId] ignoring duplicate answer (offerId: $offerId)'); + return; + } await pc.setRemoteDescription(sd); + // only a successfully applied answer makes later copies duplicates + if (sd.type == 'answer' && offerId > 0) _answeredOfferId = offerId; } catch (e) { logger.warning('[$objectId] setRemoteDescription() failed with error: $e'); } finally { // released however the apply ends, so a genuine retry is still possible - if (sd.type == 'answer') _answerInFlight = false; + if (sd.type == 'answer') _answerInFlight = null; } for (final candidate in _pendingCandidates) { @@ -360,7 +383,18 @@ class Transport extends Disposable { } } - Future createAndSendOffer([RTCOfferOptions? options]) async { + /// Whether the offer [offerId] answers already has its answer applied. Servers that do not + /// echo offer ids leave only the signaling state to go by: in `stable` there is no local + /// offer left to answer. + Future _isAnswered(int offerId) async { + if (offerId > 0) return offerId == _answeredOfferId; + return await pc.getSignalingState() == rtc.RTCSignalingState.RTCSignalingStateStable; + } + + Future createAndSendOffer([RTCOfferOptions? options]) => + _offerLock.synchronized(() => _createAndSendOffer(options)); + + Future _createAndSendOffer(RTCOfferOptions? options) async { if (isDisposed) { logger.warning('[$objectId] createAndSendOffer() already disposed'); return; @@ -395,7 +429,8 @@ class Transport extends Disposable { } // actually negotiate - logger.fine('starting to negotiate'); + final offerId = ++_latestOfferId; + logger.fine('starting to negotiate (offerId: $offerId)'); final offer = await pc.createOffer(options?.toMap() ?? {}); if ((offer.sdp ?? '').contains('goog-sped-v1')) { @@ -453,7 +488,7 @@ class Transport extends Disposable { if (appliedVideoStartBitrate && offer.sdp == mungedSdp) { _hasAppliedVideoStartBitrate = true; } - onOffer?.call(offer); + onOffer?.call(offer, offerId); } Future addIceCandidate(rtc.RTCIceCandidate candidate) async { diff --git a/lib/src/extensions.dart b/lib/src/extensions.dart index 5ec6b6291..3ffc8e3da 100644 --- a/lib/src/extensions.dart +++ b/lib/src/extensions.dart @@ -128,8 +128,9 @@ extension RTCIceTransportPolicyExt on RTCIceTransportPolicy { // not so neat to directly expose protobuf types so we // define our own types (and convert methods) extension RTCSessionDescriptionExt on rtc.RTCSessionDescription { - lk_rtc.SessionDescription toPBType() { - return lk_rtc.SessionDescription(type: type, sdp: sdp); + /// [id] numbers an offer, or names the offer an answer answers. 0 means unnumbered. + lk_rtc.SessionDescription toPBType({int id = 0}) { + return lk_rtc.SessionDescription(type: type, sdp: sdp, id: id); } } diff --git a/lib/src/internal/events.dart b/lib/src/internal/events.dart index 85624644b..93a1b7a2d 100644 --- a/lib/src/internal/events.dart +++ b/lib/src/internal/events.dart @@ -342,23 +342,31 @@ class SignalDisconnectedEvent with SignalEvent, InternalEvent { @internal class SignalOfferEvent with SignalEvent, InternalEvent { final rtc.RTCSessionDescription sd; + + /// The server's id for this offer, echoed back on the answer. 0 from servers that send none. + final int offerId; const SignalOfferEvent({ required this.sd, + required this.offerId, }); @override - String toString() => '${runtimeType}(sd: ${sd})'; + String toString() => '${runtimeType}(sd: ${sd}, offerId: ${offerId})'; } @internal class SignalAnswerEvent with SignalEvent, InternalEvent { final rtc.RTCSessionDescription sd; + + /// Id of the offer this answers. 0 from servers that do not echo offer ids. + final int offerId; const SignalAnswerEvent({ required this.sd, + required this.offerId, }); @override - String toString() => '${runtimeType}(sd: ${sd})'; + String toString() => '${runtimeType}(sd: ${sd}, offerId: ${offerId})'; } @internal diff --git a/test/core/transport_negotiate_test.dart b/test/core/transport_negotiate_test.dart index 8ced88ca9..f12f7156b 100644 --- a/test/core/transport_negotiate_test.dart +++ b/test/core/transport_negotiate_test.dart @@ -22,6 +22,7 @@ import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; import 'package:livekit_client/src/core/transport.dart' show Transport; import 'package:livekit_client/src/exceptions.dart' show NegotiationError; +import 'package:livekit_client/src/internal/types.dart' show RTCOfferOptions; import 'package:livekit_client/src/options.dart' show ConnectOptions; import '../mock/peerconnection_mock.dart'; @@ -37,6 +38,24 @@ Future _createRejecting( Map? constraints, ]) async => _RejectingPeerConnection(); +class _RestartablePeerConnection extends MockPeerConnection { + @override + Future restartIce() async {} +} + +class _RejectFirstAnswerPeerConnection extends MockPeerConnection { + bool rejected = false; + + @override + Future setRemoteDescription(rtc.RTCSessionDescription description) async { + if (description.type == 'answer' && !rejected) { + rejected = true; + throw Exception('Failed to set remote answer sdp'); + } + await super.setRemoteDescription(description); + } +} + class _RejectAfterAnswerPeerConnection extends MockPeerConnection { bool rejectLocal = false; @@ -58,7 +77,7 @@ void main() { await runZonedGuarded(() async { final transport = await Transport.create(_createRejecting, connectOptions: const ConnectOptions()); addTearDown(transport.dispose); - transport.onOffer = (_) {}; + transport.onOffer = (_, _) {}; transport.onNegotiationError = reported.complete; transport.negotiate(null); @@ -76,7 +95,7 @@ void main() { final offers = []; final transport = await Transport.create(MockPeerConnection.create, connectOptions: const ConnectOptions()); addTearDown(transport.dispose); - transport.onOffer = offers.add; + transport.onOffer = (offer, _) => offers.add(offer); transport.onNegotiationError = (error) => fail('unexpected negotiation error: $error'); transport.negotiate(null); @@ -92,7 +111,7 @@ void main() { connectOptions: const ConnectOptions(), ); addTearDown(transport.dispose); - transport.onOffer = (_) {}; + transport.onOffer = (_, _) {}; final reported = Completer(); transport.onNegotiationError = reported.complete; @@ -115,7 +134,7 @@ void main() { connectOptions: const ConnectOptions(), ); addTearDown(transport.dispose); - transport.onOffer = (_) {}; + transport.onOffer = (_, _) {}; await pc.setLocalDescription(await pc.createOffer()); await transport.createAndSendOffer(); @@ -131,7 +150,7 @@ void main() { test('still throws from a direct createAndSendOffer', () async { final transport = await Transport.create(_createRejecting, connectOptions: const ConnectOptions()); addTearDown(transport.dispose); - transport.onOffer = (_) {}; + transport.onOffer = (_, _) {}; transport.onNegotiationError = (error) => fail('unexpected negotiation error: $error'); await expectLater(transport.createAndSendOffer(), throwsA(isA())); @@ -146,7 +165,7 @@ void main() { connectOptions: const ConnectOptions(), ); addTearDown(transport.dispose); - transport.onOffer = (_) {}; + transport.onOffer = (_, _) {}; return (transport, pc); } @@ -194,4 +213,91 @@ void main() { expect(remote?.sdp, 'v=0\r\nremote-offer'); }); }); + + group('Transport offer ids', () { + Future<(Transport, MockPeerConnection, List)> makeTransport([MockPeerConnection? mock]) async { + final pc = mock ?? MockPeerConnection(); + final transport = await Transport.create( + (Map configuration, [Map? constraints]) async => pc, + connectOptions: const ConnectOptions(), + ); + addTearDown(transport.dispose); + final sent = []; + transport.onOffer = (_, offerId) => sent.add(offerId); + return (transport, pc, sent); + } + + test('numbers offers from 1', () async { + final (transport, _, sent) = await makeTransport(); + + await transport.createAndSendOffer(); + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nanswer-1', 'answer'), offerId: 1); + await transport.createAndSendOffer(); + + expect(sent, [1, 2]); + }); + + test('ignores the answer to an offer that an ICE restart replaced', () async { + final (transport, pc, sent) = await makeTransport(_RestartablePeerConnection()); + + await transport.createAndSendOffer(); + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nanswer-1', 'answer'), offerId: 1); + await transport.createAndSendOffer(); + // Offer 2 is still waiting for its answer when the ICE restart replaces it. + await transport.createAndSendOffer(const RTCOfferOptions(iceRestart: true)); + expect(sent, [1, 2, 3]); + + // The answer to offer 2 lands after the restart offer was sent. Real WebRTC is in + // have-local-offer here, so the signaling state cannot tell it from the restart answer. + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nanswer-2', 'answer'), offerId: 2); + expect((await pc.getRemoteDescription())?.sdp, 'v=0\r\nanswer-1'); + + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nanswer-3', 'answer'), offerId: 3); + expect((await pc.getRemoteDescription())?.sdp, 'v=0\r\nanswer-3'); + }); + + test('ignores a second copy of a numbered answer', () async { + final (transport, pc, _) = await makeTransport(); + + await transport.createAndSendOffer(); + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nfirst', 'answer'), offerId: 1); + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nduplicate', 'answer'), offerId: 1); + + expect((await pc.getRemoteDescription())?.sdp, 'v=0\r\nfirst'); + }); + + test('ignores a copy that arrives while the first is still being applied', () async { + final (transport, pc, _) = await makeTransport(); + + await transport.createAndSendOffer(); + await Future.wait([ + transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nfirst', 'answer'), offerId: 1), + transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nduplicate', 'answer'), offerId: 1), + ]); + + expect((await pc.getRemoteDescription())?.sdp, 'v=0\r\nfirst'); + }); + + test('accepts the answer again after a failed apply', () async { + final (transport, pc, _) = await makeTransport(_RejectFirstAnswerPeerConnection()); + + await transport.createAndSendOffer(); + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nrejected', 'answer'), offerId: 1); + expect(await pc.getRemoteDescription(), isNull); + + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nretry', 'answer'), offerId: 1); + expect((await pc.getRemoteDescription())?.sdp, 'v=0\r\nretry'); + }); + + test('defers an offer that overlaps another until its answer', () async { + final (transport, _, sent) = await makeTransport(); + + await Future.wait([transport.createAndSendOffer(), transport.createAndSendOffer()]); + expect(sent, [1]); + expect(transport.renegotiate, isTrue); + + await transport.setRemoteDescription(rtc.RTCSessionDescription('v=0\r\nanswer-1', 'answer'), offerId: 1); + expect(sent, [1, 2]); + }); + }); }