Send reliable data after a full reconnect instead of dropping it

.changes/reliable-sends-full-reconnect

@@ -0,0 +1 @@
+patch type="fixed" "Reliable data sent while the room does a full reconnect now goes out once the reconnect finishes, instead of being reported as sent and dropped"

lib/src/core/engine.dart

@@ -179,7 +179,7 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     Reliability.lossy: false,
   };
 
-  // Completed whenever something a buffer wait depends on changes: a buffer status, the reconnect
+  // Completed whenever something a waiting send depends on changes: a buffer status, the reconnect
   // state or closure. A wait checks its condition again each time it wakes.
   final ReusableCompleter<void> _sendStateChanged = ReusableCompleter<void>();
 
@@ -454,6 +454,12 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     }
   }
 
+  // Holds a reliable send until a reconnect in progress finishes. Fails if the engine closes
+  // instead, which is how a reconnect that gives up ends.
+  Future<void> _waitForReconnection() async {
+    if (_isReconnecting) await _waitUntil(() => !_isReconnecting);
+  }
+
   Future<void> _resendReliableMessagesForResume(int lastMessageSeq) async {
     logger.fine('Resending reliable messages from sequence $lastMessageSeq');
 
@@ -491,67 +497,91 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     lk_models.DataPacket packet, {
     Reliability reliability = Reliability.lossy,
   }) async {
-    // Add sequence number for reliable packets
-    if (reliability == Reliability.reliable) {
-      packet.sequence = _reliableDataSequence++;
-    }
-
-    // construct the data channel message
-    var message = rtc.RTCDataChannelMessage.fromBinary(packet.writeToBuffer());
+    final isReliable = reliability == Reliability.reliable;
+    rtc.RTCDataChannel? channel;
+    lk_models.EncryptedPacket? encryptedPacket;
 
-    if (_subscriberPrimary) {
-      // make sure publisher transport is connected
-      await ensurePublisherConnected();
-
-      // wait for data channel to open (if not already)
-      if (_publisherDataChannelState(reliability) != rtc.RTCDataChannelState.RTCDataChannelOpen) {
-        logger.fine('Waiting for data channel ${reliability} to open...');
-        await events.waitFor<PublisherDataChannelStateUpdatedEvent>(
-          filter: (event) => event.type == reliability,
-          duration: connectOptions.timeouts.connection,
-        );
+    // A reliable packet sent during a full reconnect has no replay to wait for, so a reliable send
+    // waits the reconnect out, the way rust-sdks does. Every wait happens before the sequence
+    // number is taken, and the loop runs again if a reconnect started during them.
+    while (true) {
+      if (isReliable) {
+        await _waitForReconnection();
       }
-    }
-
-    // chose data channel
-    final rtc.RTCDataChannel? channel = _publisherDataChannel(reliability);
 
-    if (channel == null) {
-      throw UnexpectedStateException('Data channel for ${packet.kind.toSDKType()} is null');
-    }
+      try {
+        if (_subscriberPrimary) {
+          // make sure publisher transport is connected
+          await ensurePublisherConnected();
+
+          // wait for data channel to open (if not already)
+          if (_publisherDataChannelState(reliability) != rtc.RTCDataChannelState.RTCDataChannelOpen) {
+            logger.fine('Waiting for data channel ${reliability} to open...');
+            await events.waitFor<PublisherDataChannelStateUpdatedEvent>(
+              filter: (event) => event.type == reliability,
+              duration: connectOptions.timeouts.connection,
+            );
+          }
+        }
 
-    if (_e2eeManager != null && _e2eeManager!.isDataChannelEncryptionEnabled) {
-      final encryptablePacket = asEncryptablePacket(packet);
-      if (encryptablePacket != null) {
-        final encryptedData = await _e2eeManager?.encryptData(data: encryptablePacket.writeToBuffer());
+        // chose data channel
+        channel = _publisherDataChannel(reliability);
 
-        if (encryptedData == null) {
-          logger.warning('Failed to encrypt data packet');
-          return;
+        if (channel == null) {
+          throw UnexpectedStateException('Data channel for ${packet.kind.toSDKType()} is null');
         }
+      } catch (_) {
+        // A full reconnect that starts during these waits resets the publisher connection and drops
+        // the channels. A reliable send waits it out and tries again.
+        if (isReliable && _isReconnecting) continue;
+        rethrow;
+      }
 
-        final encryptedPacket = lk_models.EncryptedPacket(
-          encryptionType: lk_models.Encryption_Type.GCM,
-          encryptedValue: encryptedData.data,
-          iv: encryptedData.iv,
-          keyIndex: encryptedData.keyIndex,
-        );
+      encryptedPacket = null;
+      if (_e2eeManager != null && _e2eeManager!.isDataChannelEncryptionEnabled) {
+        final encryptablePacket = asEncryptablePacket(packet);
+        if (encryptablePacket != null) {
+          final encryptedData = await _e2eeManager?.encryptData(data: encryptablePacket.writeToBuffer());
 
-        final dataToSend = lk_models.DataPacket(
-          participantIdentity: packet.participantIdentity,
-          kind: packet.kind,
-          encryptedPacket: encryptedPacket,
-          destinationIdentities: packet.destinationIdentities,
-          sequence: packet.hasSequence() ? packet.sequence : null,
-          participantSid: packet.hasParticipantSid() ? packet.participantSid : null,
-        );
+          if (encryptedData == null) {
+            logger.warning('Failed to encrypt data packet');
+            return;
+          }
 
-        message = rtc.RTCDataChannelMessage.fromBinary(dataToSend.writeToBuffer());
+          encryptedPacket = lk_models.EncryptedPacket(
+            encryptionType: lk_models.Encryption_Type.GCM,
+            encryptedValue: encryptedData.data,
+            iv: encryptedData.iv,
+            keyIndex: encryptedData.keyIndex,
+          );
+        }
       }
+
+      if (!isReliable || !_isReconnecting) break;
     }
 
+    // Nothing below awaits before the send, so a reconnect cannot start in between.
+
+    // Add sequence number for reliable packets
+    if (isReliable) {
+      packet.sequence = _reliableDataSequence++;
+    }
+
+    // construct the data channel message
+    final dataToSend = encryptedPacket == null
+        ? packet
+        : lk_models.DataPacket(
+            participantIdentity: packet.participantIdentity,
+            kind: packet.kind,
+            encryptedPacket: encryptedPacket,
+            destinationIdentities: packet.destinationIdentities,
+            sequence: packet.hasSequence() ? packet.sequence : null,
+            participantSid: packet.hasParticipantSid() ? packet.participantSid : null,
+          );
+    final message = rtc.RTCDataChannelMessage.fromBinary(dataToSend.writeToBuffer());
+
     // Buffer reliable packets for potential resending
-    if (reliability == Reliability.reliable) {
+    if (isReliable) {
       _reliableMessageBuffer.push(
         BufferedDataPacket(
           packet: packet,
@@ -561,9 +591,9 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
       );
     }
 
-    // Don't send during reconnection, but keep message buffered for resending
+    // A lossy packet is not held through a reconnect, it would only arrive stale.
     if (_isReconnecting) {
-      logger.fine('Deferring data packet send during reconnection (will resend when resumed)');
+      logger.fine('Dropping lossy data packet during reconnection');
       return;
     }
 
@@ -1404,6 +1434,11 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
       _lossyDCSub = null;
       _lossyDCPub = null;
 
+      // The new session starts its reliable sequence over, and nothing sent on the old one is
+      // replayed into it.
+      _reliableDataSequence = 1;
+      _reliableMessageBuffer.clear();
+
       await _signalListener.cancelAll();
 
       // disconnect() can run during the teardown above. It closes the engine

test/core/reliable_send_reconnect_test.dart

@@ -0,0 +1,158 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+@Timeout(Duration(seconds: 10))
+library;
+
+import 'package:flutter_test/flutter_test.dart';
+
+import 'package:livekit_client/livekit_client.dart';
+import 'package:livekit_client/src/proto/livekit_models.pb.dart' as lk_models;
+import 'package:livekit_client/src/proto/livekit_rtc.pb.dart' as lk_rtc;
+import '../mock/e2e_container.dart';
+import '../mock/peerconnection_mock.dart';
+import '../mock/websocket_mock.dart';
+import 'signal_client_test.dart' show joinResponse, offerResponse;
+
+lk_models.DataPacket _marker() => lk_models.DataPacket(
+  kind: lk_models.DataPacket_Kind.RELIABLE,
+  user: lk_models.UserPacket(payload: [42]),
+);
+
+bool _isMarker(lk_models.DataPacket packet) =>
+    packet.hasUser() && packet.user.payload.length == 1 && packet.user.payload.first == 42;
+
+void main() {
+  late E2EContainer container;
+  late MockWebSocketConnector ws;
+
+  setUp(() async {
+    resetMockDataChannels();
+    container = E2EContainer();
+    ws = container.wsConnector;
+    await container.connectRoom();
+    // The mock loops published data back into this room as if a remote sent it, which the room
+    // rejects. These tests record what is sent instead.
+    for (final dc in mockDataChannels) {
+      dc.onMessageSend = null;
+    }
+  });
+
+  tearDown(() async {
+    await container.dispose();
+  });
+
+  /// Records every packet sent on the reliable channels a full reconnect creates.
+  List<lk_models.DataPacket> recordNewChannels() {
+    final sent = <lk_models.DataPacket>[];
+    onMockReliableChannelCreated = (dc) => dc.onMessageSend = (message) {
+      sent.add(lk_models.DataPacket.fromBuffer(message.binary));
+    };
+    return sent;
+  }
+
+  /// Starts a full reconnect and answers its join, leaving it to wait for the subscriber offer.
+  /// The new publisher channels exist from here on, but the reconnect has not finished.
+  Future<void> startFullReconnect() async {
+    final previousHandlers = ws.handlers;
+    container.engine.fullReconnectOnNext = true;
+    ws.onDispose();
+    for (var i = 0; i < 200 && identical(ws.handlers, previousHandlers); i++) {
+      await Future<void>.delayed(const Duration(milliseconds: 10));
+    }
+    expect(identical(ws.handlers, previousHandlers), isFalse, reason: 'the SDK never re-opened the signal connection');
+    await Future<void>.delayed(const Duration(milliseconds: 1));
+    ws.onData(joinResponse.writeToBuffer());
+    await Future<void>.delayed(const Duration(milliseconds: 20));
+    expect(container.engine.isFullReconnectInProgress, isTrue);
+  }
+
+  Future<void> finishFullReconnect() async {
+    final reconnected = container.room.events.waitFor<RoomReconnectedEvent>(duration: const Duration(seconds: 5));
+    ws.onData(offerResponse.writeToBuffer());
+    await reconnected;
+  }
+
+  test('a reliable send during a full reconnect goes out on the new channel once it finishes', () async {
+    final sent = recordNewChannels();
+    await startFullReconnect();
+
+    var completed = false;
+    final send = container.engine
+        .sendDataPacket(_marker(), reliability: Reliability.reliable)
+        .whenComplete(() => completed = true);
+
+    await Future<void>.delayed(const Duration(milliseconds: 300));
+    expect(completed, isFalse, reason: 'the send finished while the reconnect was still running');
+
+    await finishFullReconnect();
+    await send.timeout(const Duration(seconds: 2));
+    expect(sent.where(_isMarker), hasLength(1), reason: 'the packet never reached the new channel');
+  });
+
+  test('a full reconnect restarts reliable sequence numbers', () async {
+    for (var i = 0; i < 3; i++) {
+      await container.room.localParticipant!.publishData([i], reliable: true);
+    }
+
+    final sent = recordNewChannels();
+    await startFullReconnect();
+    await finishFullReconnect();
+
+    await container.engine.sendDataPacket(_marker(), reliability: Reliability.reliable);
+    final marker = sent.singleWhere(_isMarker);
+    expect(marker.sequence, 1, reason: 'the new session carried on the old session\'s sequence numbers');
+  });
+
+  test('a send waiting out a reconnect fails when the engine closes', () async {
+    await startFullReconnect();
+
+    final send = container.engine.sendDataPacket(_marker(), reliability: Reliability.reliable);
+    final failed = expectLater(send, throwsA(isA<UnexpectedConnectionState>()));
+    await Future<void>.delayed(const Duration(milliseconds: 50));
+    await container.engine.disconnect();
+
+    await failed.timeout(const Duration(seconds: 2));
+  });
+
+  test('a reliable send during a resume goes out once', () async {
+    final sent = <lk_models.DataPacket>[];
+    for (final dc in mockDataChannels) {
+      dc.onMessageSend = (message) => sent.add(lk_models.DataPacket.fromBuffer(message.binary));
+    }
+    // The server acknowledges this one in its reconnect response.
+    await container.room.localParticipant!.publishData([1], reliable: true);
+    final acknowledged = sent.last.sequence;
+
+    final previousHandlers = ws.handlers;
+    ws.onDispose();
+    for (var i = 0; i < 200 && identical(ws.handlers, previousHandlers); i++) {
+      await Future<void>.delayed(const Duration(milliseconds: 10));
+    }
+    expect(ws.uri?.queryParameters['reconnect'], '1', reason: 'this test needs a resume');
+
+    final send = container.engine.sendDataPacket(_marker(), reliability: Reliability.reliable);
+    await Future<void>.delayed(const Duration(milliseconds: 50));
+
+    final reconnected = container.room.events.waitFor<RoomReconnectedEvent>(duration: const Duration(seconds: 5));
+    ws.onData(
+      lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse(lastMessageSeq: acknowledged)).writeToBuffer(),
+    );
+    await reconnected;
+    await send.timeout(const Duration(seconds: 2));
+    await Future<void>.delayed(const Duration(milliseconds: 50));
+
+    expect(sent.where(_isMarker), hasLength(1));
+  });
+}