Match publisher answers to offers by offer id

.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"

lib/src/core/engine.dart

@@ -700,9 +700,9 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
       }
     };
 
-    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<EngineEvent> {
       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<EngineEvent> {
       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<SignalTrickleEvent>((event) async {

lib/src/core/signal_client.dart

@@ -287,10 +287,10 @@ class SignalClient extends Disposable with EventsEmittable<SignalEvent> {
         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<SignalEvent> {
 
 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),
     ),
   );
 

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<String, dynamic> 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<rtc.RTCPeerConnection> Function(Map<String, dynamic> configuration, [Map<String, dynamic> 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<void> 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<void> 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<void> 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<bool> _isAnswered(int offerId) async {
+    if (offerId > 0) return offerId == _answeredOfferId;
+    return await pc.getSignalingState() == rtc.RTCSignalingState.RTCSignalingStateStable;
+  }
+
+  Future<void> createAndSendOffer([RTCOfferOptions? options]) =>
+      _offerLock.synchronized(() => _createAndSendOffer(options));
+
+  Future<void> _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() ?? <String, dynamic>{});
 
     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<void> addIceCandidate(rtc.RTCIceCandidate candidate) async {

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);
   }
 }
 

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