From 86ebcb446c9349f36884e2558eabfaaf6b684d8e Mon Sep 17 00:00:00 2001 From: shijing xian Date: Thu, 10 Sep 2026 13:46:34 -0700 Subject: [PATCH 1/6] fix: resume the session on a migration Leave instead of full reconnecting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit For a node migration the server sends `LeaveRequest{Action: RESUME, Reason: MIGRATION}`, which asks the client to reconnect with `reconnect=1` and keep its session. The engine's leave handler did exactly that, but `attemptReconnect` then unconditionally escalated any `leaveReconnect` into a full reconnect: if (... || [ClientDisconnectReason.leaveReconnect, ...].contains(reason)) { fullReconnectOnNext = true; } That list predates protocol v13 (#439), when a leave with `can_reconnect` could only mean a full reconnect. The v13 RESUME branch ported in #574 never updated it, so the resume branch has been dead code since: every RESUME leave ran `restartConnection()`, emitting `RoomReconnectingEvent`, dropping every `RemoteParticipant` and re-joining. Drop `leaveReconnect` from the escalation list — the callers that do need a full reconnect (the RECONNECT leave branch, the connection check) set `fullReconnectOnNext` themselves. Also stop forcing the flag to false in the RESUME branch: client-sdk-js and rust-sdks both treat an escalation as sticky, so a resume that already failed at the media level is not downgraded back into a resume loop. Adds `test/core/leave_action_test.dart` covering both leave actions, and implements `setConfiguration` on the mock peer connection (the resume path applies the `ReconnectResponse` ICE servers). Co-Authored-By: Claude Opus 5 (1M context) --- ...on-leave-resumes-instead-of-full-reconnect | 1 + lib/src/core/engine.dart | 9 +- test/core/leave_action_test.dart | 124 ++++++++++++++++++ test/mock/peerconnection_mock.dart | 9 +- 4 files changed, 138 insertions(+), 5 deletions(-) create mode 100644 .changes/migration-leave-resumes-instead-of-full-reconnect create mode 100644 test/core/leave_action_test.dart diff --git a/.changes/migration-leave-resumes-instead-of-full-reconnect b/.changes/migration-leave-resumes-instead-of-full-reconnect new file mode 100644 index 000000000..89325f7d2 --- /dev/null +++ b/.changes/migration-leave-resumes-instead-of-full-reconnect @@ -0,0 +1 @@ +patch type="fixed" "Session migration (server `Leave{action: RESUME}`) now resumes the session instead of escalating to a full reconnect, so remote participants are no longer dropped and re-added" \ No newline at end of file diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 7c668acfe..8ab16fbc9 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -1126,9 +1126,12 @@ class Engine extends Disposable with EventsEmittable { return; } + // `leaveReconnect` is intentionally not escalated here: since protocol v13 a server + // Leave carries an action, and `RESUME` (what the server sends for a node migration) + // must stay a resume. The callers that need a full reconnect (`RECONNECT` leave, + // connection check) set `fullReconnectOnNext` themselves before handing over. if (_clientConfiguration?.resumeConnection == lk_models.ClientConfigSetting.DISABLED || [ - ClientDisconnectReason.leaveReconnect, ClientDisconnectReason.negotiationFailed, ClientDisconnectReason.peerConnectionFailed, ].contains(reason)) { @@ -1526,7 +1529,9 @@ class Engine extends Disposable with EventsEmittable { // canReconnect is still checked for backward compatibility with v12 servers // (where action defaults to DISCONNECT=0 since it's unset). if (event.action == lk_rtc.LeaveRequest_Action.RESUME) { - fullReconnectOnNext = false; + // The server (e.g. a node migration) expects us to resume the session, so + // fullReconnectOnNext is deliberately left alone rather than forced to false: + // an escalation from an already-failed resume must not be downgraded here. // reconnect immediately instead of waiting for next attempt await handleReconnect(ClientDisconnectReason.leaveReconnect); } else if (event.action == lk_rtc.LeaveRequest_Action.RECONNECT || event.canReconnect) { diff --git a/test/core/leave_action_test.dart b/test/core/leave_action_test.dart new file mode 100644 index 000000000..8204eab64 --- /dev/null +++ b/test/core/leave_action_test.dart @@ -0,0 +1,124 @@ +// 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'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + late E2EContainer container; + late Room room; + late MockWebSocketConnector ws; + + setUp(() async { + resetMockDataChannels(); + container = E2EContainer(); + room = container.room; + ws = container.wsConnector; + await container.connectRoom(); + }); + + tearDown(() async { + await container.dispose(); + }); + + /// Feed a server-initiated `LeaveRequest` into the signal connection. + void sendLeave(lk_rtc.LeaveRequest_Action action, lk_models.DisconnectReason reason) { + ws.onData( + lk_rtc.SignalResponse( + leave: lk_rtc.LeaveRequest(action: action, reason: reason), + ).writeToBuffer(), + ); + } + + /// Wait until the SDK has opened a *new* websocket (the reconnect attempt), + /// then answer it the way the receiving node would. + Future answerReconnectAttempt(Object? previousHandlers) async { + 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: 'SDK never re-opened the signal connection'); + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + } + + test('Leave{RESUME} (node migration) resumes and keeps remote participants', () async { + await container.simulateRemoteParticipantJoin('bob'); + expect(room.remoteParticipants, hasLength(1)); + + final roomEvents = []; + final sub = room.events.listen(roomEvents.add); + final previousHandlers = ws.handlers; + + sendLeave(lk_rtc.LeaveRequest_Action.RESUME, lk_models.DisconnectReason.MIGRATION); + + await answerReconnectAttempt(previousHandlers); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await sub(); + + expect( + roomEvents.whereType(), + isNotEmpty, + reason: 'a migration must resume the session', + ); + expect( + roomEvents.whereType(), + isEmpty, + reason: 'RoomReconnectingEvent signals a full reconnect, which drops session state', + ); + expect( + roomEvents.whereType(), + isEmpty, + reason: 'a migration must not kick out remote participants', + ); + expect(room.remoteParticipants, hasLength(1)); + expect(container.engine.fullReconnectOnNext, isFalse); + }); + + test('Leave{RECONNECT} performs a full reconnect', () async { + await container.simulateRemoteParticipantJoin('bob'); + expect(room.remoteParticipants, hasLength(1)); + + final roomEvents = []; + final sub = room.events.listen(roomEvents.add); + final previousHandlers = ws.handlers; + + sendLeave(lk_rtc.LeaveRequest_Action.RECONNECT, lk_models.DisconnectReason.SERVER_SHUTDOWN); + + // a full reconnect re-joins, so it is answered with a JoinResponse + 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: 'SDK never re-opened the signal connection'); + + await sub(); + + expect( + roomEvents.whereType(), + isNotEmpty, + reason: 'a RECONNECT leave must trigger a full reconnect', + ); + expect(roomEvents.whereType(), isEmpty); + expect(roomEvents.whereType(), hasLength(1)); + }); +} diff --git a/test/mock/peerconnection_mock.dart b/test/mock/peerconnection_mock.dart index 80fa01093..1dc3c8755 100644 --- a/test/mock/peerconnection_mock.dart +++ b/test/mock/peerconnection_mock.dart @@ -285,10 +285,13 @@ a=rtpmap:32 MPV/90000 @override Future removeTrack(RTCRtpSender sender) async => true; + /// Last configuration applied via [setConfiguration] (set on resume, when the + /// server hands out new ICE servers in the `ReconnectResponse`). + Map? appliedConfiguration; + @override - Future setConfiguration(Map configuration) { - // TODO: implement setConfiguration - throw UnimplementedError(); + Future setConfiguration(Map configuration) async { + appliedConfiguration = configuration; } static Future create( From 53e077d4f0eacd67caf5f4027d4a4f9c973a9413 Mon Sep 17 00:00:00 2001 From: shijing xian Date: Thu, 10 Sep 2026 14:02:07 -0700 Subject: [PATCH 2/6] test: assert the resume re-opens the signal socket with reconnect=1 `reconnect=1` is the query parameter the server actually keys off to distinguish a resume from a re-join, and it was the only part of the resume contract the test wasn't checking. Also documents why the socket close that follows the Leave is not simulated: a bare socket drop reconnects with reason `signal`, which resumes on its own, so delivering the close before the leave-driven attempt runs makes the test pass even when the leave action is ignored. In production the close arrives a round-trip later and never wins that race, which is why the reported bug reproduced. Co-Authored-By: Claude Opus 5 (1M context) --- test/core/leave_action_test.dart | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/test/core/leave_action_test.dart b/test/core/leave_action_test.dart index 8204eab64..a68a7875f 100644 --- a/test/core/leave_action_test.dart +++ b/test/core/leave_action_test.dart @@ -70,12 +70,22 @@ void main() { final sub = room.events.listen(roomEvents.add); final previousHandlers = ws.handlers; + // The server also drops the socket right after the Leave, but that is + // deliberately not simulated here: a bare socket drop reconnects with reason + // `signal`, which resumes on its own. Delivering it before the leave-driven + // attempt runs (in production it arrives a round-trip later, so it never + // wins) makes this test pass even when the leave action is ignored entirely. sendLeave(lk_rtc.LeaveRequest_Action.RESUME, lk_models.DisconnectReason.MIGRATION); await answerReconnectAttempt(previousHandlers); await room.events.waitFor(duration: const Duration(seconds: 5)); await sub(); + expect( + ws.uri?.queryParameters['reconnect'], + '1', + reason: 'a resume must re-open the signal connection with reconnect=1', + ); expect( roomEvents.whereType(), isNotEmpty, From 643847eb13e174e6f108b7d709fbdc742227bc0d Mon Sep 17 00:00:00 2001 From: shijing xian Date: Thu, 10 Sep 2026 14:39:56 -0700 Subject: [PATCH 3/6] fix: don't drop reconnect requests or lose their escalation Three ways a reconnect request could be lost or altered: 1. `handleReconnect()` clears the pending timer and reschedules with its own reason, so a later caller (the socket close following a server Leave) overrode an earlier one and the escalation implied by the first reason was silently dropped. The reason -> escalation mapping now happens in `handleReconnect`, where the request originates, so it is captured as state instead of being re-derived later from a reason that may have been replaced. 2. `attemptReconnect()` early-returns while an attempt is in flight, so a full-reconnect request arriving mid-attempt was never acted on. The flag is now consumed at the start of an attempt and a request that arrives during it is dispatched from the finally block, as client-sdk-js does. 3. A successful attempt calls `_clearPendingReconnect()`, cancelling the queued escalation and leaving `fullReconnectOnNext` stuck true, which also suppressed the next legitimate `RoomDisconnectedEvent`. Fixed by the same consume-and-redispatch. Consuming the flag up front means it no longer describes the running attempt, which `Room` relied on to skip fast-connect republishing during a full reconnect's re-join. Added `Engine.isFullReconnectInProgress` for that question and pointed `Room` at it. Also aligns the failure path with js/rust: a failed full reconnect stays a full reconnect. Co-Authored-By: Claude Opus 5 (1M context) --- .changes/reconnect-request-dispatch | 1 + lib/src/core/engine.dart | 59 ++++++-- lib/src/core/room.dart | 11 +- .../core/reconnect_request_dispatch_test.dart | 133 ++++++++++++++++++ 4 files changed, 189 insertions(+), 15 deletions(-) create mode 100644 .changes/reconnect-request-dispatch create mode 100644 test/core/reconnect_request_dispatch_test.dart diff --git a/.changes/reconnect-request-dispatch b/.changes/reconnect-request-dispatch new file mode 100644 index 000000000..bc1a1f7e8 --- /dev/null +++ b/.changes/reconnect-request-dispatch @@ -0,0 +1 @@ +patch type="fixed" "Reconnect requests are no longer dropped when one attempt is already running, and a reason that requires a full reconnect is no longer lost when a later request replaces it" \ No newline at end of file diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 8ab16fbc9..e7370ad8c 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -114,8 +114,18 @@ class Engine extends Disposable with EventsEmittable { String? _connectedServerAddress; String? get connectedServerAddress => _connectedServerAddress; + /// A *pending* full-reconnect request. Consumed at the start of each + /// reconnect attempt, so it is false while an attempt runs unless a new + /// request arrived mid-attempt — use [isFullReconnectInProgress] to ask + /// what the running attempt is doing. bool fullReconnectOnNext = false; + bool _attemptIsFullReconnect = false; + + /// Whether the reconnect attempt currently running is a full reconnect + /// (as opposed to a resume). False when no attempt is in flight. + bool get isFullReconnectInProgress => _attemptIsFullReconnect; + // server-provided ice servers List _serverProvidedIceServers = []; @@ -1064,6 +1074,17 @@ class Engine extends Disposable with EventsEmittable { logger.info('onDisconnected state:${connectionState} reason:${reason.name}'); + // Capture the escalation the moment the request is made. A later + // handleReconnect (e.g. the socket close that follows a server Leave) + // replaces the pending timer and with it the reason, so deciding this + // later — when attemptReconnect finally runs — can silently lose it. + if ([ + ClientDisconnectReason.negotiationFailed, + ClientDisconnectReason.peerConnectionFailed, + ].contains(reason)) { + fullReconnectOnNext = true; + } + _isReconnecting = true; if (_reconnectAttempts == 0) { @@ -1126,18 +1147,21 @@ class Engine extends Disposable with EventsEmittable { return; } - // `leaveReconnect` is intentionally not escalated here: since protocol v13 a server - // Leave carries an action, and `RESUME` (what the server sends for a node migration) - // must stay a resume. The callers that need a full reconnect (`RECONNECT` leave, - // connection check) set `fullReconnectOnNext` themselves before handing over. - if (_clientConfiguration?.resumeConnection == lk_models.ClientConfigSetting.DISABLED || - [ - ClientDisconnectReason.negotiationFailed, - ClientDisconnectReason.peerConnectionFailed, - ].contains(reason)) { + // Reason-driven escalation is captured in handleReconnect, where the + // request originates. This is config, not a request, so it belongs here. + if (_clientConfiguration?.resumeConnection == lk_models.ClientConfigSetting.DISABLED) { fullReconnectOnNext = true; } + // Consume the flag up front: this attempt's mode is now fixed, and from + // here a `true` value unambiguously means a *new* full-reconnect request + // arrived while we were running (e.g. a server RECONNECT leave during a + // resume), which the finally block dispatches. Mirrors client-sdk-js. + final fullReconnect = fullReconnectOnNext; + fullReconnectOnNext = false; + _attemptIsFullReconnect = fullReconnect; + + var succeeded = false; try { _attemptingReconnect = true; @@ -1153,7 +1177,7 @@ class Engine extends Disposable with EventsEmittable { ); } - if (fullReconnectOnNext) { + if (fullReconnect) { await restartConnection(); } else { await resumeConnection( @@ -1164,11 +1188,13 @@ class Engine extends Disposable with EventsEmittable { _clearPendingReconnect(); _attemptingReconnect = false; _isReconnecting = false; + succeeded = true; } catch (e) { _reconnectAttempts = _reconnectAttempts + 1; bool recoverable = true; - if (e is WebSocketException || e is MediaConnectException) { - // cannot resume connection, need to do full reconnect + if (fullReconnect || e is WebSocketException || e is MediaConnectException) { + // a failed full reconnect stays a full reconnect; a resume that failed + // at the transport or media layer cannot be resumed again fullReconnectOnNext = true; } @@ -1196,6 +1222,15 @@ class Engine extends Disposable with EventsEmittable { } } finally { _attemptingReconnect = false; + _attemptIsFullReconnect = false; + + // A full reconnect requested while this attempt was running that a + // successful attempt didn't act on — dispatch it now. The failure path + // already retries, so only the success path needs this. + if (succeeded && fullReconnectOnNext && !_isClosed) { + logger.fine('attemptReconnect: full reconnect requested mid-attempt, dispatching'); + unawaited(handleReconnect(ClientDisconnectReason.reconnectRetry)); + } } } diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 13a1079f0..7fe4e4735 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -507,7 +507,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { info: event.response.participant, ); - if (engine.fullReconnectOnNext) { + if (engine.isFullReconnectInProgress) { await _localParticipant!.updateFromInfo(event.response.participant); } @@ -522,7 +522,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { if (connectOptions.protocolVersion.index >= ProtocolVersion.v8.index && engine.fastConnectOptions != null && - !engine.fullReconnectOnNext) { + !engine.isFullReconnectInProgress) { final options = engine.fastConnectOptions!; final audio = options.microphone; @@ -651,7 +651,12 @@ class Room extends DisposableChangeNotifier with EventsEmittable { notifyListeners(); }) ..on((event) async { - if (!engine.fullReconnectOnNext || event.reason == DisconnectReason.clientInitiated) { + // Suppress while a full reconnect is either pending or running — the + // engine is going to re-establish the session, this is not a real + // disconnect. Both flags are needed since the attempt consumes the + // pending one when it starts. + if ((!engine.fullReconnectOnNext && !engine.isFullReconnectInProgress) || + event.reason == DisconnectReason.clientInitiated) { await _cleanUp(disposeLocalParticipant: false); events.emit(RoomDisconnectedEvent(reason: event.reason)); notifyListeners(); diff --git a/test/core/reconnect_request_dispatch_test.dart b/test/core/reconnect_request_dispatch_test.dart new file mode 100644 index 000000000..34530250a --- /dev/null +++ b/test/core/reconnect_request_dispatch_test.dart @@ -0,0 +1,133 @@ +// 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. + +// Reconnect-request bookkeeping: a request must not be dropped because another +// attempt was running, and the escalation implied by a request's reason must +// not be lost when a later request replaces it. +// +// The first test ports rust-sdks' `test_resume_escalation_sticks_across_cycles` +// (livekit/tests/peer_connection_signaling_test.rs). That test needs a live SFU, +// two participants and a published sine track, and observes the escalation via +// `LocalTrackRepublished` because only the full-reconnect path republishes. +// Here the mock transport lets us inject the concurrent request directly and +// observe the escalation as `RoomReconnectingEvent`, which only the full path +// emits. + +@Timeout(Duration(seconds: 10)) +library; + +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/proto/livekit_rtc.pb.dart' as lk_rtc; +import 'package:livekit_client/src/types/internal.dart'; +import '../mock/e2e_container.dart'; +import '../mock/peerconnection_mock.dart'; +import '../mock/websocket_mock.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + late E2EContainer container; + late Room room; + late MockWebSocketConnector ws; + + setUp(() async { + resetMockDataChannels(); + container = E2EContainer(); + room = container.room; + ws = container.wsConnector; + await container.connectRoom(); + }); + + tearDown(() async { + await container.dispose(); + }); + + /// Spin until the SDK opens a new signal socket, returning its URI. + Future awaitNewSocket(Object? previousHandlers) async { + 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: 'SDK never re-opened the signal connection'); + return ws.uri!; + } + + test('a full-reconnect request arriving mid-resume is dispatched after it succeeds', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + + // Cycle 1: a resume, with a full-reconnect request injected while it runs. + final firstHandlers = ws.handlers; + ws.onDispose(); + final resumeUri = await awaitNewSocket(firstHandlers); + expect(resumeUri.queryParameters['reconnect'], '1', reason: 'cycle 1 must be a resume'); + + // What a server `Leave{RECONNECT}` mid-resume does. + container.engine.fullReconnectOnNext = true; + + final resumeHandlers = ws.handlers; + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + + // Cycle 2 must be a full reconnect, dispatched from the request that + // arrived while cycle 1 was in flight — not silently dropped. + final restartUri = await awaitNewSocket(resumeHandlers); + expect(restartUri.queryParameters['reconnect'], isNull, reason: 'cycle 2 must re-join, not resume'); + await cancel(); + + expect( + roomEvents.whereType(), + isNotEmpty, + reason: 'cycle 1 must have been a resume, otherwise the cross-cycle behavior is not under test', + ); + expect( + roomEvents.whereType(), + isNotEmpty, + reason: 'the mid-attempt full-reconnect request must still be honored', + ); + }); + + test('an escalating reason is not lost when a later request replaces it', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + final previousHandlers = ws.handlers; + + // A PeerConnection failure demands a full reconnect. The socket close that + // follows lands a second request whose reason (`signal`) implies only a + // resume, and it replaces the first request's pending timer. + unawaited(container.engine.handleReconnect(ClientDisconnectReason.peerConnectionFailed)); + await container.engine.handleReconnect(ClientDisconnectReason.signal); + + final uri = await awaitNewSocket(previousHandlers); + await cancel(); + + expect(uri.queryParameters['reconnect'], isNull, reason: 'the peer-connection failure must still force a re-join'); + expect(roomEvents.whereType(), isNotEmpty); + expect(roomEvents.whereType(), isEmpty); + }); + + test('a successful resume leaves no full-reconnect state behind', () async { + final previousHandlers = ws.handlers; + ws.onDispose(); + await awaitNewSocket(previousHandlers); + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + + expect(container.engine.fullReconnectOnNext, isFalse); + expect(container.engine.isFullReconnectInProgress, isFalse); + }); +} From d905f517db4e2d4823968b51bbe8227edee04adf Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:05:07 +0800 Subject: [PATCH 4/6] Retry a severed resume and keep requests made during a restart Two ways a reconnect request could still vanish after the consume and dispatch change: restartConnection cleared fullReconnectOnNext after joining. The flag had already been consumed when the attempt started, so a true value there was a new request, typically a RECONNECT leave from the node just joined, and the reset erased it before the finally block could dispatch it. resumeConnection declared success without checking the signal socket. If the socket dropped while the peer connections were being restored, the attempt emitted Resumed with a dead connection and the success path cancelled the retry that the drop had scheduled. Re-check the socket before emitting Resumed and throw a recoverable error so the retry path runs another resume, the same check client-sdk-js and rust-sdks make. Tests cover a peer failure reported through handleReconnect mid-resume, a RECONNECT leave arriving mid-restart, and a socket drop right behind the ReconnectResponse. The last two fail without the engine change. --- .changes/resume-severed-signal-retry | 1 + lib/src/core/engine.dart | 17 +++- .../core/reconnect_request_dispatch_test.dart | 92 +++++++++++++++++++ 3 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 .changes/resume-severed-signal-retry diff --git a/.changes/resume-severed-signal-retry b/.changes/resume-severed-signal-retry new file mode 100644 index 000000000..f32c46c3f --- /dev/null +++ b/.changes/resume-severed-signal-retry @@ -0,0 +1 @@ +patch type="fixed" "A resume whose signal connection drops before it completes is retried instead of being reported as reconnected" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 91d704ab7..02109374c 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -1294,6 +1294,18 @@ class Engine extends Disposable with EventsEmittable { logger.fine('resumeConnection: primary connected'); } + // The socket can drop while the peer connections were being restored. A + // resume that ends with a dead signal connection is a failure, not a + // success: throwing here lets the retry path run another resume instead of + // reporting the room as reconnected and cancelling the pending request. + // Mirrors the re-check in client-sdk-js and rust-sdks. + if (signalClient.connectionState != ConnectionState.connected) { + throw ConnectException( + 'resumeConnection: signal connection severed during resume', + reason: ConnectionErrorReason.InternalError, + ); + } + _isReconnecting = false; events.emit(const EngineResumedEvent()); } @@ -1341,7 +1353,10 @@ class Engine extends Disposable with EventsEmittable { await ensurePublisherConnected(); } - fullReconnectOnNext = false; + // fullReconnectOnNext is not cleared here. attemptReconnect consumed the + // request that started this restart, so a true value at this point is a + // new request (e.g. a RECONNECT leave from the node we just joined) that + // the finally block in attemptReconnect dispatches once we return. _regionUrlProvider?.resetAttempts(); events.emit(const EngineRestartedEvent()); } catch (error) { diff --git a/test/core/reconnect_request_dispatch_test.dart b/test/core/reconnect_request_dispatch_test.dart index 34530250a..9828c24f3 100644 --- a/test/core/reconnect_request_dispatch_test.dart +++ b/test/core/reconnect_request_dispatch_test.dart @@ -32,6 +32,7 @@ import 'dart:async'; 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 'package:livekit_client/src/types/internal.dart'; import '../mock/e2e_container.dart'; @@ -120,6 +121,97 @@ void main() { expect(roomEvents.whereType(), isEmpty); }); + test('a peer failure reported mid-resume is dispatched as a full reconnect afterwards', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + + final firstHandlers = ws.handlers; + ws.onDispose(); + final resumeUri = await awaitNewSocket(firstHandlers); + expect(resumeUri.queryParameters['reconnect'], '1', reason: 'cycle 1 must be a resume'); + + // The real request path, not a direct flag write: a PeerConnection reports + // failed while the resume is in flight. handleReconnect records the + // escalation and schedules a timer that the running attempt swallows. + unawaited(container.engine.handleReconnect(ClientDisconnectReason.peerConnectionFailed)); + + final resumeHandlers = ws.handlers; + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + + final restartUri = await awaitNewSocket(resumeHandlers); + await cancel(); + + expect(restartUri.queryParameters['reconnect'], isNull, reason: 'the peer failure must still force a re-join'); + expect(roomEvents.whereType(), isNotEmpty); + expect(container.engine.fullReconnectOnNext, isFalse, reason: 'the request must be consumed, not left stale'); + }); + + test('a RECONNECT leave arriving mid-restart is not lost', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + + // Cycle 1: a full reconnect, answered with a JoinResponse. + final firstHandlers = ws.handlers; + container.engine.fullReconnectOnNext = true; + ws.onDispose(); + final joinUri = await awaitNewSocket(firstHandlers); + expect(joinUri.queryParameters['reconnect'], isNull, reason: 'cycle 1 must be a re-join'); + + // The node we are joining asks for another full reconnect before the join + // completes. restartConnection used to reset the flag after joining, which + // erased this request. + ws.onData( + lk_rtc.SignalResponse( + leave: lk_rtc.LeaveRequest( + action: lk_rtc.LeaveRequest_Action.RECONNECT, + reason: lk_models.DisconnectReason.STATE_MISMATCH, + ), + ).writeToBuffer(), + ); + + final joinHandlers = ws.handlers; + await container.answerJoin(); + await room.events.waitFor(duration: const Duration(seconds: 5)); + + // Cycle 2: the leave-driven full reconnect must still run. + final secondJoinUri = await awaitNewSocket(joinHandlers); + await cancel(); + + expect(secondJoinUri.queryParameters['reconnect'], isNull, reason: 'cycle 2 must be a re-join too'); + expect(roomEvents.whereType(), hasLength(2)); + }); + + test('a signal drop during the resume is retried instead of reported as success', () async { + final roomEvents = []; + final cancel = room.events.listen(roomEvents.add); + + final firstHandlers = ws.handlers; + ws.onDispose(); + await awaitNewSocket(firstHandlers); + + // The server answers the resume and the socket dies right behind it, before + // the peer connection work finishes. The attempt must not end in + // RoomReconnectedEvent with a dead signal connection. + final resumeHandlers = ws.handlers; + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + ws.onDispose(); + + final retryUri = await awaitNewSocket(resumeHandlers); + expect(retryUri.queryParameters['reconnect'], '1', reason: 'a severed signal is retried as a resume'); + expect( + roomEvents.whereType(), + isEmpty, + reason: 'the attempt with the dead socket must not be reported as a success', + ); + + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await cancel(); + + expect(roomEvents.whereType(), hasLength(1)); + }); + test('a successful resume leaves no full-reconnect state behind', () async { final previousHandlers = ws.handlers; ws.onDispose(); From 467ec5c48349908532a12eac17b60606ea175a7c Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Mon, 14 Sep 2026 23:42:52 +0800 Subject: [PATCH 5/6] Log why a reconnect attempt failed The catch in attemptReconnect decided between retry and give up without recording the error, so a resume that was retried because its signal socket died left no trace of the reason in the log. --- lib/src/core/engine.dart | 1 + 1 file changed, 1 insertion(+) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 02109374c..fab7e18c6 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -1196,6 +1196,7 @@ class Engine extends Disposable with EventsEmittable { succeeded = true; } catch (e) { _reconnectAttempts = _reconnectAttempts + 1; + logger.fine('attemptReconnect: ${fullReconnect ? 'full reconnect' : 'resume'} failed: $e'); bool recoverable = true; if (fullReconnect || e is WebSocketException || e is MediaConnectException) { // a failed full reconnect stays a full reconnect; a resume that failed From 5baf75e9b4d149a647eb8e23bf36a216d1126493 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Tue, 15 Sep 2026 00:05:41 +0800 Subject: [PATCH 6/6] Stop resetting the reconnect attempt counter on signal connect A resume opens its signal socket before the peer connections are restored, and the socket connect reset _reconnectAttempts to zero. Any attempt that failed after that point, a media timeout or the new severed socket check, started counting again from zero, so the retry limit could never be reached and a client stuck in a failing resume loop never emitted a terminal disconnect. client-sdk-js only resets the counter when an attempt has fully succeeded. _clearPendingReconnect already does that here, and cleanUp covers disconnect, so the reset on socket connect was redundant apart from this bug. The new test severs three consecutive resumes and checks the scheduled attempt numbers climb 2, 3, 4 instead of repeating 2. --- lib/src/core/engine.dart | 6 ++- .../core/reconnect_request_dispatch_test.dart | 40 ++++++++++++++++++- 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index fab7e18c6..ba82dfefb 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -1495,7 +1495,11 @@ class Engine extends Disposable with EventsEmittable { }) ..on((event) async { logger.fine('Signal connected'); - _reconnectAttempts = 0; + // The attempt counter is not reset here. A resume opens its socket before + // the peer connections are restored, so a reset on socket connect would + // let an attempt that fails afterwards start again from zero and never + // reach the retry limit. _clearPendingReconnect resets it once an attempt + // has fully succeeded, and cleanUp on disconnect. events.emit(const EngineConnectedEvent()); }) ..on((event) async { diff --git a/test/core/reconnect_request_dispatch_test.dart b/test/core/reconnect_request_dispatch_test.dart index 9828c24f3..8e9e770a0 100644 --- a/test/core/reconnect_request_dispatch_test.dart +++ b/test/core/reconnect_request_dispatch_test.dart @@ -59,8 +59,8 @@ void main() { }); /// Spin until the SDK opens a new signal socket, returning its URI. - Future awaitNewSocket(Object? previousHandlers) async { - for (var i = 0; i < 200 && identical(ws.handlers, previousHandlers); i++) { + Future awaitNewSocket(Object? previousHandlers, {int maxWaitMs = 2000}) async { + for (var i = 0; i < maxWaitMs ~/ 10 && identical(ws.handlers, previousHandlers); i++) { await Future.delayed(const Duration(milliseconds: 10)); } expect(identical(ws.handlers, previousHandlers), isFalse, reason: 'SDK never re-opened the signal connection'); @@ -212,6 +212,42 @@ void main() { expect(roomEvents.whereType(), hasLength(1)); }); + test('repeated severed resumes count towards the retry limit', () async { + final attempts = []; + final cancel = room.events.listen((event) { + if (event is RoomAttemptReconnectEvent) attempts.add(event.attempt); + }); + + // Every resume opens its socket, gets its ReconnectResponse and then loses + // the socket before the attempt completes. The socket connect used to reset + // the attempt counter, so each failure scheduled "attempt 2" again and the + // retry limit was never reached. + var handlers = ws.handlers; + ws.onDispose(); + for (var i = 0; i < 3; i++) { + await awaitNewSocket(handlers, maxWaitMs: 6000); + handlers = ws.handlers; + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + ws.onDispose(); + // let the attempt reach its final check, fail and schedule the retry + await Future.delayed(const Duration(milliseconds: 50)); + } + + // Leave the engine in a clean state: answer the fourth attempt properly. + await awaitNewSocket(handlers, maxWaitMs: 6000); + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await cancel(); + + // Each failure schedules twice (the socket close and the retry), so look at + // the distinct attempt numbers: they must climb, not repeat. + expect(attempts.where((a) => a > 1).toSet().toList(), [ + 2, + 3, + 4, + ], reason: 'every failed attempt must advance the counter'); + }, timeout: const Timeout(Duration(seconds: 30))); + test('a successful resume leaves no full-reconnect state behind', () async { final previousHandlers = ws.handlers; ws.onDispose();