From dc41370d46e3a2b30185a2724b170dd9c4040329 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:31:28 +0800 Subject: [PATCH] Send reliable data after a full reconnect instead of dropping it sendDataPacket parked a reliable packet sent during any reconnect for the resume replay. A full reconnect never runs that replay, so the packet was reported as sent and never went out. publishData, RPC and data stream writes all hit this. A reliable send now waits the reconnect out and goes out on the connection it leaves, the way rust-sdks waits for reconnection in publish_data. It fails with UnexpectedConnectionState if the engine closes instead. All waits happen before the sequence number is taken, a send that a full reconnect interrupts while it waits for the publisher goes round again instead of failing, and nothing awaits between the last reconnect check and the send. A full reconnect restarts the reliable sequence and clears the resend buffer, as client-sdk-js, client-sdk-swift and rust-sdks do for a new session. Lossy packets are still dropped during a reconnect rather than sent late. --- .changes/reliable-sends-full-reconnect | 1 + lib/src/core/engine.dart | 137 ++++++++++------- test/core/reliable_send_reconnect_test.dart | 158 ++++++++++++++++++++ 3 files changed, 245 insertions(+), 51 deletions(-) create mode 100644 .changes/reliable-sends-full-reconnect create mode 100644 test/core/reliable_send_reconnect_test.dart diff --git a/.changes/reliable-sends-full-reconnect b/.changes/reliable-sends-full-reconnect new file mode 100644 index 000000000..7f46c2e9c --- /dev/null +++ b/.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" \ No newline at end of file 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..d3fd78925 --- /dev/null +++ b/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 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)); + }); +}