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..b9d0240f3 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -1064,6 +1064,22 @@ class Engine extends Disposable with EventsEmittable { logger.info('onDisconnected state:${connectionState} reason:${reason.name}'); + // Decide the escalation now rather than when the retry timer fires. A later + // request replaces the pending timer together with its reason, so a + // Leave{RESUME} that lands right after a peer connection failure would + // otherwise downgrade that failure into a resume. + // + // `leaveReconnect` is intentionally not escalated: 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. + if ([ + ClientDisconnectReason.negotiationFailed, + ClientDisconnectReason.peerConnectionFailed, + ].contains(reason)) { + fullReconnectOnNext = true; + } + _isReconnecting = true; if (_reconnectAttempts == 0) { @@ -1126,12 +1142,9 @@ class Engine extends Disposable with EventsEmittable { return; } - if (_clientConfiguration?.resumeConnection == lk_models.ClientConfigSetting.DISABLED || - [ - ClientDisconnectReason.leaveReconnect, - ClientDisconnectReason.negotiationFailed, - ClientDisconnectReason.peerConnectionFailed, - ].contains(reason)) { + // Reason based escalation is decided in handleReconnect. The server side + // switch is checked here so the latest ClientConfiguration wins. + if (_clientConfiguration?.resumeConnection == lk_models.ClientConfigSetting.DISABLED) { fullReconnectOnNext = true; } @@ -1526,7 +1539,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..8a148021f --- /dev/null +++ b/test/core/leave_action_test.dart @@ -0,0 +1,225 @@ +// 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 'package:livekit_client/src/support/websocket.dart'; +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(() { + resetMockDataChannels(); + container = E2EContainer(); + room = container.room; + ws = container.wsConnector; + }); + + tearDown(() async { + await container.dispose(); + }); + + /// Connect and inject one remote participant so the tests can observe + /// whether the roster survives the reconnect. + Future connectWithRemoteParticipant({lk_models.ClientConfiguration? clientConfiguration}) async { + await container.connectRoom(clientConfiguration: clientConfiguration); + await container.simulateRemoteParticipantJoin('bob'); + expect(room.remoteParticipants, hasLength(1)); + } + + /// 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). + Future waitForNewSignalConnection(WebSocketEventHandlers? previous) async { + for (var i = 0; i < 200 && identical(ws.handlers, previous); i++) { + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(identical(ws.handlers, previous), isFalse, reason: 'SDK never re-opened the signal connection'); + } + + /// Answer a resume attempt the way the receiving node would. + Future answerResume(WebSocketEventHandlers? previous) async { + await waitForNewSignalConnection(previous); + expect( + ws.uri?.queryParameters['reconnect'], + '1', + reason: 'a resume must re-open the signal connection with reconnect=1', + ); + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + } + + /// Answer a full reconnect attempt: the SDK re-joins, so it gets a JoinResponse. + Future answerFullReconnect(WebSocketEventHandlers? previous) async { + await waitForNewSignalConnection(previous); + expect( + ws.uri?.queryParameters.containsKey('reconnect'), + isFalse, + reason: 'a full reconnect must re-join without reconnect=1', + ); + await container.answerJoin(); + } + + void expectFullReconnect(List roomEvents) { + expect( + roomEvents.whereType(), + isNotEmpty, + reason: 'a full reconnect must emit RoomReconnectingEvent', + ); + expect(roomEvents.whereType(), isEmpty); + expect(roomEvents.whereType(), hasLength(1)); + expect(roomEvents.whereType(), hasLength(1)); + expect(room.remoteParticipants, isEmpty); + expect(container.engine.fullReconnectOnNext, isFalse); + } + + test('Leave{RESUME} (node migration) resumes and keeps remote participants', () async { + await connectWithRemoteParticipant(); + + final roomEvents = []; + 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 answerResume(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); + + // The ICE servers from the ReconnectResponse must reach both transports. + final publisher = container.engine.publisher?.pc as MockPeerConnection?; + final subscriber = container.engine.subscriber?.pc as MockPeerConnection?; + expect(publisher?.appliedConfiguration, isNotNull); + expect(subscriber?.appliedConfiguration, isNotNull); + }); + + test('Leave{RECONNECT} performs a full reconnect', () async { + await connectWithRemoteParticipant(); + + final roomEvents = []; + final sub = room.events.listen(roomEvents.add); + final previousHandlers = ws.handlers; + + sendLeave(lk_rtc.LeaveRequest_Action.RECONNECT, lk_models.DisconnectReason.SERVER_SHUTDOWN); + + await answerFullReconnect(previousHandlers); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await sub(); + + expectFullReconnect(roomEvents); + }); + + test('Leave{RESUME} does not downgrade a pending full reconnect', () async { + await connectWithRemoteParticipant(); + + final roomEvents = []; + final sub = room.events.listen(roomEvents.add); + final previousHandlers = ws.handlers; + + // An earlier failure already decided the next attempt must be a full + // reconnect. The server asking for a resume must not undo that. + container.engine.fullReconnectOnNext = true; + sendLeave(lk_rtc.LeaveRequest_Action.RESUME, lk_models.DisconnectReason.MIGRATION); + + await answerFullReconnect(previousHandlers); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await sub(); + + expectFullReconnect(roomEvents); + }); + + test('Leave{RESUME} performs a full reconnect when the server disabled resume', () async { + await connectWithRemoteParticipant( + clientConfiguration: lk_models.ClientConfiguration( + resumeConnection: lk_models.ClientConfigSetting.DISABLED, + ), + ); + + final roomEvents = []; + final sub = room.events.listen(roomEvents.add); + final previousHandlers = ws.handlers; + + sendLeave(lk_rtc.LeaveRequest_Action.RESUME, lk_models.DisconnectReason.MIGRATION); + + await answerFullReconnect(previousHandlers); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await sub(); + + expectFullReconnect(roomEvents); + }); + + test('a Leave{RESUME} arriving before a peer failure retry keeps the escalation', () async { + await connectWithRemoteParticipant(); + + final roomEvents = []; + final sub = room.events.listen(roomEvents.add); + final previousHandlers = ws.handlers; + + // The peer connection failure schedules a retry that must be a full + // reconnect. The Leave replaces that pending retry with its own reason; + // the escalation decided for the failure must survive the swap. + final failure = container.engine.handleReconnect(ClientDisconnectReason.peerConnectionFailed); + sendLeave(lk_rtc.LeaveRequest_Action.RESUME, lk_models.DisconnectReason.MIGRATION); + await failure; + + await answerFullReconnect(previousHandlers); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await sub(); + + expectFullReconnect(roomEvents); + }); +} diff --git a/test/mock/e2e_container.dart b/test/mock/e2e_container.dart index fb79fd904..3c9527213 100644 --- a/test/mock/e2e_container.dart +++ b/test/mock/e2e_container.dart @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +import 'dart:async'; import 'dart:typed_data'; import 'package:fixnum/fixnum.dart'; @@ -63,6 +64,7 @@ class E2EContainer { bool captureOutbound = false, ConnectOptions? connectOptions, @Deprecated('mirrors the deprecated Room.connect parameter') RoomOptions? roomOptions, + lk_models.ClientConfiguration? clientConfiguration, }) async { final connectFuture = room.connect( exampleUri, @@ -71,11 +73,7 @@ class E2EContainer { // ignore: deprecated_member_use_from_same_package roomOptions: roomOptions, ); - Future.delayed(const Duration(milliseconds: 1), () { - final resp = _buildJoinResponse(localClientProtocol); - wsConnector.onData(resp.writeToBuffer()); - wsConnector.onData(offerResponse.writeToBuffer()); - }); + unawaited(answerJoin(localClientProtocol: localClientProtocol, clientConfiguration: clientConfiguration)); await connectFuture; @@ -111,20 +109,42 @@ class E2EContainer { } } - lk_rtc.SignalResponse _buildJoinResponse(int? localClientProtocol) { - if (localClientProtocol == null) { + /// Answer the signal connection the SDK just opened the way the server does + /// for a (re)join: a `JoinResponse` followed by the subscriber offer. Used by + /// [connectRoom] and by tests that drive a full reconnect. + Future answerJoin({ + int? localClientProtocol, + lk_models.ClientConfiguration? clientConfiguration, + }) async { + // Give the SDK a tick to start waiting for the join response. + await Future.delayed(const Duration(milliseconds: 1)); + final resp = _buildJoinResponse(localClientProtocol, clientConfiguration); + wsConnector.onData(resp.writeToBuffer()); + wsConnector.onData(offerResponse.writeToBuffer()); + } + + lk_rtc.SignalResponse _buildJoinResponse( + int? localClientProtocol, + lk_models.ClientConfiguration? clientConfiguration, + ) { + if (localClientProtocol == null && clientConfiguration == null) { return joinResponse; } - final localInfo = localParticipantData.deepCopy()..clientProtocol = localClientProtocol; - return lk_rtc.SignalResponse( - join: lk_rtc.JoinResponse( - room: lk_models.Room(name: 'room_name', sid: 'room_sid'), - participant: localInfo, - subscriberPrimary: true, - serverVersion: '99.999', - serverInfo: lk_models.ServerInfo(version: '1.8.0'), - ), + final localInfo = localParticipantData.deepCopy(); + if (localClientProtocol != null) { + localInfo.clientProtocol = localClientProtocol; + } + final join = lk_rtc.JoinResponse( + room: lk_models.Room(name: 'room_name', sid: 'room_sid'), + participant: localInfo, + subscriberPrimary: true, + serverVersion: '99.999', + serverInfo: lk_models.ServerInfo(version: '1.8.0'), ); + if (clientConfiguration != null) { + join.clientConfiguration = clientConfiguration; + } + return lk_rtc.SignalResponse(join: join); } void _installOutboundCapture() { 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(