diff --git a/.changes/reliable-sends-full-reconnect b/.changes/reliable-sends-full-reconnect new file mode 100644 index 000000000..543e25642 --- /dev/null +++ b/.changes/reliable-sends-full-reconnect @@ -0,0 +1 @@ +patch type="fixed" "Reliable data sent during a reconnect now goes out once the reconnect finishes, instead of being reported as sent and dropped. Such sends wait for the reconnect and fail with UnexpectedConnectionState if the room disconnects instead" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index af32afa34..d060e0b7b 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -179,7 +179,7 @@ class Engine extends Disposable with EventsEmittable { 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 _sendStateChanged = ReusableCompleter(); @@ -454,6 +454,12 @@ class Engine extends Disposable with EventsEmittable { } } + // 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 _waitForReconnection() async { + if (_isReconnecting) await _waitUntil(() => !_isReconnecting); + } + Future _resendReliableMessagesForResume(int lastMessageSeq) async { logger.fine('Resending reliable messages from sequence $lastMessageSeq'); @@ -491,67 +497,91 @@ class Engine extends Disposable with EventsEmittable { 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( - 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( + 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 { ); } - // 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 { _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 diff --git a/test/core/reliable_send_reconnect_test.dart b/test/core/reliable_send_reconnect_test.dart new file mode 100644 index 000000000..e11a3ebec --- /dev/null +++ b/test/core/reliable_send_reconnect_test.dart @@ -0,0 +1,208 @@ +// 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:flutter_webrtc/flutter_webrtc.dart' show RTCPeerConnectionState; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/internal/events.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 recordNewChannels() { + final sent = []; + 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 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.delayed(const Duration(milliseconds: 10)); + } + expect(identical(ws.handlers, previousHandlers), isFalse, reason: 'the SDK never re-opened the signal connection'); + await Future.delayed(const Duration(milliseconds: 1)); + ws.onData(joinResponse.writeToBuffer()); + await Future.delayed(const Duration(milliseconds: 20)); + expect(container.engine.isFullReconnectInProgress, isTrue); + } + + Future finishFullReconnect() async { + final reconnected = container.room.events.waitFor(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.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())); + await Future.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 = []; + 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.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.delayed(const Duration(milliseconds: 50)); + + final reconnected = container.room.events.waitFor(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.delayed(const Duration(milliseconds: 50)); + + expect(sent.where(_isMarker), hasLength(1)); + }); + + test('a reliable send after the reconnect response but before the resume ends goes out', () async { + final sent = []; + for (final dc in mockDataChannels) { + dc.onMessageSend = (message) => sent.add(lk_models.DataPacket.fromBuffer(message.binary)); + } + await container.room.localParticipant!.publishData([1], reliable: true); + final acknowledged = sent.last.sequence; + // The first send negotiates the publisher after a debounce. Let that finish before the resume + // starts, the mock cannot run two offers at once. + await Future.delayed(const Duration(milliseconds: 100)); + + // Hold the primary connection in connecting so the resume is still running after its reconnect + // response has replayed the buffer. + final primary = container.engine.subscriber!.pc as MockPeerConnection; + primary.connectionState = RTCPeerConnectionState.RTCPeerConnectionStateConnecting; + + final previousHandlers = ws.handlers; + ws.onDispose(); + for (var i = 0; i < 200 && identical(ws.handlers, previousHandlers); i++) { + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(ws.uri?.queryParameters['reconnect'], '1', reason: 'this test needs a resume'); + + final responded = container.engine.events.waitFor(duration: const Duration(seconds: 2)); + ws.onData( + lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse(lastMessageSeq: acknowledged)).writeToBuffer(), + ); + await responded; + // The resume sets restartingIce right before it waits for the primary connection. + for (var i = 0; i < 200 && container.engine.subscriber?.restartingIce != true; i++) { + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(container.engine.subscriber?.restartingIce, isTrue, reason: 'the resume never waited for the primary'); + + // The replay is over and the resume is waiting for the primary. A send here used to be parked + // in the resend buffer, which nothing replayed again. + final send = container.engine.sendDataPacket(_marker(), reliability: Reliability.reliable); + await Future.delayed(const Duration(milliseconds: 50)); + + final reconnected = container.room.events.waitFor(duration: const Duration(seconds: 5)); + primary.connectionState = RTCPeerConnectionState.RTCPeerConnectionStateConnected; + await reconnected; + await send.timeout(const Duration(seconds: 2)); + await Future.delayed(const Duration(milliseconds: 50)); + + expect(sent.where(_isMarker), hasLength(1), reason: 'the packet was buffered after the replay and never sent'); + }); +}