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 1/2] 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)); + }); +} From 81900e7cdf6955729a4d7089ea81bebfa4a2e33d Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:23:28 +0800 Subject: [PATCH 2/2] Test the resume window and state the behavior change in the changeset A reliable send made after the reconnect response had replayed the resend buffer, but before the resume finished, was parked in that buffer and never sent, because nothing replayed it again. This branch already fixes it (such sends wait for the reconnect to end), but no unit test covered it. The new test holds the primary connection in connecting so the resume is still running after its response, sends, and checks the packet goes out once the resume ends. It fails on the branch below. The changeset now says that reliable sends wait for the reconnect and fail with UnexpectedConnectionState if the room disconnects instead. --- .changes/reliable-sends-full-reconnect | 2 +- test/core/reliable_send_reconnect_test.dart | 50 +++++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/.changes/reliable-sends-full-reconnect b/.changes/reliable-sends-full-reconnect index 7f46c2e9c..543e25642 100644 --- a/.changes/reliable-sends-full-reconnect +++ b/.changes/reliable-sends-full-reconnect @@ -1 +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 +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/test/core/reliable_send_reconnect_test.dart b/test/core/reliable_send_reconnect_test.dart index d3fd78925..e11a3ebec 100644 --- a/test/core/reliable_send_reconnect_test.dart +++ b/test/core/reliable_send_reconnect_test.dart @@ -16,8 +16,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'; @@ -155,4 +157,52 @@ void main() { 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'); + }); }