From 86ebcb446c9349f36884e2558eabfaaf6b684d8e Mon Sep 17 00:00:00 2001 From: shijing xian Date: Thu, 10 Sep 2026 13:46:34 -0700 Subject: [PATCH 1/4] 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/4] 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 f16c2f2a81df09e3be6072ba17dd6d9e64cf4b04 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Mon, 14 Sep 2026 14:39:21 +0800 Subject: [PATCH 3/4] Decide reconnect escalation when the request is made handleReconnect replaces the pending retry timer together with its reason, and the reason based escalation ran only when that timer fired. A Leave{RESUME} arriving right after a peer connection failure therefore swapped the reason to leaveReconnect and the failed connection was resumed instead of restarted. Now that leaveReconnect no longer escalates on its own, decide the escalation in handleReconnect so a later request cannot drop it. The server side resumeConnection switch stays in attemptReconnect so the latest ClientConfiguration wins. --- lib/src/core/engine.dart | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index 8ab16fbc9..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,15 +1142,9 @@ 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 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; } From 1471d115badc7a9ebf513f9d612564881c3c82d6 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Mon, 14 Sep 2026 14:39:21 +0800 Subject: [PATCH 4/4] Cover the full reconnect leave paths end to end The RECONNECT test now answers the re-join and waits for RoomReconnectedEvent instead of tearing down mid restart, and asserts the signal URL carries no reconnect flag. Add cases for a stale fullReconnectOnNext, for resumeConnection DISABLED from the server, and for a Leave{RESUME} racing a pending peer failure retry. The RESUME test also checks the ReconnectResponse configuration reached both transports. E2EContainer gains answerJoin() and a clientConfiguration option so tests can drive a full reconnect and shape the join response. --- test/core/leave_action_test.dart | 147 +++++++++++++++++++++++++------ test/mock/e2e_container.dart | 52 +++++++---- 2 files changed, 155 insertions(+), 44 deletions(-) diff --git a/test/core/leave_action_test.dart b/test/core/leave_action_test.dart index a68a7875f..8a148021f 100644 --- a/test/core/leave_action_test.dart +++ b/test/core/leave_action_test.dart @@ -20,6 +20,8 @@ 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'; @@ -31,18 +33,25 @@ void main() { late Room room; late MockWebSocketConnector ws; - setUp(() async { + setUp(() { resetMockDataChannels(); container = E2EContainer(); room = container.room; ws = container.wsConnector; - await container.connectRoom(); }); 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( @@ -52,19 +61,51 @@ void main() { ); } - /// 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++) { + /// 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, previousHandlers), isFalse, reason: 'SDK never re-opened the signal connection'); + 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 container.simulateRemoteParticipantJoin('bob'); - expect(room.remoteParticipants, hasLength(1)); + await connectWithRemoteParticipant(); final roomEvents = []; final sub = room.events.listen(roomEvents.add); @@ -77,15 +118,10 @@ void main() { // 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 answerResume(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, @@ -103,11 +139,16 @@ void main() { ); 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 container.simulateRemoteParticipantJoin('bob'); - expect(room.remoteParticipants, hasLength(1)); + await connectWithRemoteParticipant(); final roomEvents = []; final sub = room.events.listen(roomEvents.add); @@ -115,20 +156,70 @@ void main() { 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 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(); - expect( - roomEvents.whereType(), - isNotEmpty, - reason: 'a RECONNECT leave must trigger a full reconnect', + 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, + ), ); - expect(roomEvents.whereType(), isEmpty); - expect(roomEvents.whereType(), 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 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() {