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 65c43e46499c35c3d679c09c054d895ba61d52b9 Mon Sep 17 00:00:00 2001 From: shijing xian Date: Thu, 10 Sep 2026 14:32:34 -0700 Subject: [PATCH 3/4] fix: reconcile the participant roster after a signal resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A resume, unlike a full reconnect, never rebuilds the roster from a JoinResponse, and the DISCONNECTED update for anyone who left during the outage went to a socket we no longer had. Those participants stayed in `room.remoteParticipants` forever. The server answers a resume with the ReconnectResponse followed immediately by a full roster snapshot on the same socket, so the snapshot is authoritative: any participant we still hold that is absent from it left while we were away and its disconnect is synthesized. Mirrors `reconcile_absent_participants` in rust-sdks. The reconciliation is armed off `SignalReconnectResponseEvent` rather than `SignalReconnectedEvent` — the latter is emitted only after the engine's async ReconnectResponse handling (setConfiguration on both transports, reliable-message replay) and can lose the race against the update that follows it on the wire. Arming off the raw signal message keeps the two ordered. If no snapshot ever arrives the reconciliation simply never fires, so a missing snapshot can never be read as "everyone left". Co-Authored-By: Claude Opus 5 (1M context) --- .changes/reconcile-roster-after-resume | 1 + lib/src/core/room.dart | 54 +++++++ test/core/resume_roster_reconcile_test.dart | 169 ++++++++++++++++++++ 3 files changed, 224 insertions(+) create mode 100644 .changes/reconcile-roster-after-resume create mode 100644 test/core/resume_roster_reconcile_test.dart diff --git a/.changes/reconcile-roster-after-resume b/.changes/reconcile-roster-after-resume new file mode 100644 index 000000000..9bea68411 --- /dev/null +++ b/.changes/reconcile-roster-after-resume @@ -0,0 +1 @@ +patch type="fixed" "Reconcile the participant roster after a signal resume: participants who left while the connection was down are now removed instead of lingering as stale entries" \ No newline at end of file diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 13a1079f0..96d76fead 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -131,6 +131,12 @@ class Room extends DisposableChangeNotifier with EventsEmittable { RegionUrlProvider? _regionUrlProvider; String? _regionUrl; + /// Identities seen in participant updates since the signal link came back up + /// during a resume, used to reconcile the roster (see + /// [_reconcileAbsentParticipants]). Non-null only while a resume is waiting + /// for the server's post-`ReconnectResponse` roster snapshot. + Set? _resumeRosterSnapshot; + // Agents final Map _transcriptionReceivedTimes = {}; @@ -402,6 +408,15 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } void _setUpSignalListeners() => _signalListener + ..on((event) { + // The server answers a resume with the `ReconnectResponse` followed + // immediately by a full roster snapshot on the same socket. Arm the + // reconciliation here, off the raw signal message, so it is ordered + // against the update that follows — `SignalReconnectedEvent` is emitted + // only after the engine's async ReconnectResponse handling and can lose + // that race. + _resumeRosterSnapshot = {}; + }) ..on((event) => _onParticipantUpdateEvent(event.participants)) ..on((event) => _onSignalSpeakersChangedEvent(event.speakers)) ..on((event) => _onSignalConnectionQualityUpdateEvent(event.updates)) @@ -601,6 +616,10 @@ class Room extends DisposableChangeNotifier with EventsEmittable { ..on((event) async { events.emit(const RoomReconnectingEvent()); + // a full reconnect rebuilds the roster from the JoinResponse, so any + // armed resume reconciliation is moot + _resumeRosterSnapshot = null; + // reset params _name = null; _metadata = null; @@ -814,6 +833,9 @@ class Room extends DisposableChangeNotifier with EventsEmittable { Future _onParticipantUpdateEvent(List updates) async { // trigger change notifier only if list of participants membership is changed var hasChanged = false; + // Captured before the loop: if a resume armed the reconciliation, this + // batch is the server's post-resume roster snapshot. + final rosterSnapshot = _resumeRosterSnapshot; for (final info in updates) { // The local participant is not ready yet, waiting for the // `RoomConnectedEvent` to create the local participant. @@ -828,6 +850,10 @@ class Room extends DisposableChangeNotifier with EventsEmittable { continue; } + if (info.identity.isNotEmpty) { + _resumeRosterSnapshot?.add(info.identity); + } + final isNew = !_remoteParticipants.containsIdentity(info.identity); if (info.state == lk_models.ParticipantInfo_State.DISCONNECTED) { @@ -862,11 +888,37 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } } + // Disarm before reconciling so a nested update can't reconcile twice. The + // identity check keeps a concurrent re-arm (another resume) intact. + if (rosterSnapshot != null && identical(rosterSnapshot, _resumeRosterSnapshot)) { + _resumeRosterSnapshot = null; + hasChanged = await _reconcileAbsentParticipants(rosterSnapshot) || hasChanged; + } + if (hasChanged) { notifyListeners(); } } + /// Remove participants who left while the signal link was down. + /// + /// A resume, unlike a full reconnect, never rebuilds the roster from a + /// `JoinResponse`, and the `DISCONNECTED` update for anyone who left during + /// the outage was delivered to a socket we no longer had. The server answers + /// a resume with a full roster snapshot, so any participant we still hold + /// that is absent from [presentIdentities] is a ghost and its disconnect has + /// to be synthesized. Mirrors `reconcile_absent_participants` in rust-sdks. + Future _reconcileAbsentParticipants(Set presentIdentities) async { + var hasChanged = false; + for (final participant in _remoteParticipants.toList()) { + if (presentIdentities.contains(participant.identity)) continue; + logger.info('synthesizing disconnect for absent participant: ${participant.identity}'); + final removed = await _handleParticipantDisconnect(participant.identity); + hasChanged = removed || hasChanged; + } + return hasChanged; + } + void _onSignalSpeakersChangedEvent(List speakers) { final lastSpeakers = { for (final p in _activeSpeakers) p.sid: p, @@ -1090,6 +1142,8 @@ extension RoomPrivateMethods on Room { Future _cleanUp({bool disposeLocalParticipant = true}) async { logger.fine('[${objectId}] cleanUp()'); + _resumeRosterSnapshot = null; + // clean up RemoteParticipants final participants = _remoteParticipants.toList(); _remoteParticipants.clear(); diff --git a/test/core/resume_roster_reconcile_test.dart b/test/core/resume_roster_reconcile_test.dart new file mode 100644 index 000000000..78b904b2a --- /dev/null +++ b/test/core/resume_roster_reconcile_test.dart @@ -0,0 +1,169 @@ +// 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. + +// Ported from rust-sdks' `test_resume_synthesizes_disconnect_for_participant_that_left` +// (livekit/tests/reconnection_test.rs). That test needs a live SFU plus a +// fault-injection switch that drops DISCONNECTED updates; here the mock signal +// transport gives the same setup for free — we simply never deliver the +// leaver's disconnect, then answer the resume with a roster snapshot that omits +// them. The observer/leaver/witness shape is kept: the witness proves the +// reconciliation only removes participants that actually left. + +@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(); + await container.simulateRemoteParticipantJoin('leaver'); + await container.simulateRemoteParticipantJoin('witness'); + expect(room.remoteParticipants.keys, containsAll(['leaver', 'witness'])); + }); + + tearDown(() async { + await container.dispose(); + }); + + /// Drop the signal socket, which the engine recovers with a resume, and wait + /// until it has re-opened the connection. + Future resumeSignalConnection() async { + 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(identical(ws.handlers, previousHandlers), isFalse, reason: 'SDK never re-opened the signal connection'); + expect(ws.uri?.queryParameters['reconnect'], '1', reason: 'expected a resume, not a re-join'); + ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); + } + + /// The roster snapshot the server sends right after the `ReconnectResponse`. + void sendRosterSnapshot(List identities) { + ws.onData( + lk_rtc.SignalResponse( + update: lk_rtc.ParticipantUpdate( + participants: identities + .map( + (identity) => lk_models.ParticipantInfo( + sid: '${identity}_sid', + identity: identity, + state: lk_models.ParticipantInfo_State.ACTIVE, + ), + ) + .toList(), + ), + ).writeToBuffer(), + ); + } + + test('resume synthesizes a disconnect for a participant that left', () async { + final disconnected = []; + final cancel = room.events.listen((event) { + if (event is ParticipantDisconnectedEvent) { + disconnected.add(event.participant.identity); + } + }); + + await resumeSignalConnection(); + // The leaver left while we were away; its DISCONNECTED update went to the + // socket we no longer had, so the snapshot is the only evidence. + sendRosterSnapshot(['witness']); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await Future.delayed(const Duration(milliseconds: 50)); + await cancel(); + + expect(disconnected, ['leaver']); + expect(room.remoteParticipants.keys, ['witness']); + }); + + test('resume keeps every participant still present in the snapshot', () async { + final disconnected = []; + final cancel = room.events.listen((event) { + if (event is ParticipantDisconnectedEvent) { + disconnected.add(event.participant.identity); + } + }); + + await resumeSignalConnection(); + sendRosterSnapshot(['leaver', 'witness']); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await Future.delayed(const Duration(milliseconds: 50)); + await cancel(); + + expect(disconnected, isEmpty); + expect(room.remoteParticipants.keys, containsAll(['leaver', 'witness'])); + }); + + test('resume without a roster snapshot leaves the roster untouched', () async { + // Safety property: reconciliation is armed by the signal reconnect but only + // fires on a snapshot. If the server never sends one, we must not conclude + // that everyone left. + final disconnected = []; + final cancel = room.events.listen((event) { + if (event is ParticipantDisconnectedEvent) { + disconnected.add(event.participant.identity); + } + }); + + await resumeSignalConnection(); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await Future.delayed(const Duration(milliseconds: 200)); + await cancel(); + + expect(disconnected, isEmpty); + expect(room.remoteParticipants.keys, containsAll(['leaver', 'witness'])); + }); + + test('a full reconnect does not run the resume reconciliation', () async { + // The full-restart path unwinds the roster itself and rebuilds it from the + // JoinResponse; the armed snapshot must not double-fire on top of that. + final disconnected = []; + final cancel = room.events.listen((event) { + if (event is ParticipantDisconnectedEvent) { + disconnected.add(event.participant.identity); + } + }); + + 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)); + } + await Future.delayed(const Duration(milliseconds: 50)); + await cancel(); + + // exactly one disconnect each, from the restart unwind — not doubled + expect(disconnected..sort(), ['leaver', 'witness']); + }); +} From 655703add30b544ecad16b3795a2e970058f155d Mon Sep 17 00:00:00 2001 From: shijing xian Date: Thu, 10 Sep 2026 15:04:31 -0700 Subject: [PATCH 4/4] fix: bound the resume roster-snapshot window and require the local entry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The arming had no expiry, so if the server's post-resume snapshot never arrived it sat there indefinitely and the next *ordinary* participant update — which lists only what changed — was mistaken for a full roster and evicted everyone else. That is the one failure direction this feature must never have. Two guards: - The arming now expires after 5s. The snapshot follows the ReconnectResponse on the same socket, so the window only has to cover scheduling, never a real wait; if it lapses the reconciliation simply never runs. - An update only counts as the snapshot if it carries the local participant. The server includes it so metadata changes propagate, which is exactly what distinguishes a full roster from a partial update. This also covers the RoomMoved path, which reuses the same handler with `otherParticipants` (no local entry). Both failure modes are now "no reconciliation", never "evict a live participant". Tests cover a partial update arriving while armed, an update after the window lapses, and escalation to a full reconnect while armed. Co-Authored-By: Claude Opus 5 (1M context) --- lib/src/core/room.dart | 53 +++++++++++--- test/core/resume_roster_reconcile_test.dart | 79 ++++++++++++++++++--- 2 files changed, 113 insertions(+), 19 deletions(-) diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index 96d76fead..5848deddd 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -131,12 +131,23 @@ class Room extends DisposableChangeNotifier with EventsEmittable { RegionUrlProvider? _regionUrlProvider; String? _regionUrl; - /// Identities seen in participant updates since the signal link came back up - /// during a resume, used to reconcile the roster (see - /// [_reconcileAbsentParticipants]). Non-null only while a resume is waiting - /// for the server's post-`ReconnectResponse` roster snapshot. + /// Identities seen in the server's post-resume roster snapshot, used to + /// reconcile the roster (see [_reconcileAbsentParticipants]). Non-null only + /// while a resume is waiting for that snapshot. Set? _resumeRosterSnapshot; + /// Disarms [_resumeRosterSnapshot] if the snapshot never arrives. Without + /// this the arming would sit there indefinitely and the next *ordinary* + /// participant update — which lists only what changed — would be mistaken + /// for a full roster and evict everyone else. + Timer? _resumeRosterSnapshotTimeout; + + /// How long after a resume the next qualifying participant update is treated + /// as the roster snapshot. The server sends it immediately after the + /// `ReconnectResponse` on the same socket, so this only has to cover + /// scheduling, never a real wait. + static const _resumeRosterSnapshotWindow = Duration(seconds: 5); + // Agents final Map _transcriptionReceivedTimes = {}; @@ -415,7 +426,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { // against the update that follows — `SignalReconnectedEvent` is emitted // only after the engine's async ReconnectResponse handling and can lose // that race. - _resumeRosterSnapshot = {}; + _armResumeRosterSnapshot(); }) ..on((event) => _onParticipantUpdateEvent(event.participants)) ..on((event) => _onSignalSpeakersChangedEvent(event.speakers)) @@ -618,7 +629,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { // a full reconnect rebuilds the roster from the JoinResponse, so any // armed resume reconciliation is moot - _resumeRosterSnapshot = null; + _disarmResumeRosterSnapshot(); // reset params _name = null; @@ -834,8 +845,12 @@ class Room extends DisposableChangeNotifier with EventsEmittable { // trigger change notifier only if list of participants membership is changed var hasChanged = false; // Captured before the loop: if a resume armed the reconciliation, this - // batch is the server's post-resume roster snapshot. + // batch may be the server's post-resume roster snapshot. The snapshot + // always carries the local participant (the server includes it so metadata + // changes propagate), which is what distinguishes it from an ordinary + // partial update listing only what changed. final rosterSnapshot = _resumeRosterSnapshot; + var sawLocalParticipant = false; for (final info in updates) { // The local participant is not ready yet, waiting for the // `RoomConnectedEvent` to create the local participant. @@ -846,6 +861,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } if (localParticipant?.identity == info.identity) { + sawLocalParticipant = true; await localParticipant?.updateFromInfo(info); continue; } @@ -890,8 +906,8 @@ class Room extends DisposableChangeNotifier with EventsEmittable { // Disarm before reconciling so a nested update can't reconcile twice. The // identity check keeps a concurrent re-arm (another resume) intact. - if (rosterSnapshot != null && identical(rosterSnapshot, _resumeRosterSnapshot)) { - _resumeRosterSnapshot = null; + if (rosterSnapshot != null && sawLocalParticipant && identical(rosterSnapshot, _resumeRosterSnapshot)) { + _disarmResumeRosterSnapshot(); hasChanged = await _reconcileAbsentParticipants(rosterSnapshot) || hasChanged; } @@ -900,6 +916,23 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } } + void _armResumeRosterSnapshot() { + _resumeRosterSnapshotTimeout?.cancel(); + _resumeRosterSnapshot = {}; + _resumeRosterSnapshotTimeout = Timer(_resumeRosterSnapshotWindow, () { + if (_resumeRosterSnapshot != null) { + logger.fine('resume roster snapshot never arrived, skipping reconciliation'); + } + _disarmResumeRosterSnapshot(); + }); + } + + void _disarmResumeRosterSnapshot() { + _resumeRosterSnapshotTimeout?.cancel(); + _resumeRosterSnapshotTimeout = null; + _resumeRosterSnapshot = null; + } + /// Remove participants who left while the signal link was down. /// /// A resume, unlike a full reconnect, never rebuilds the roster from a @@ -1142,7 +1175,7 @@ extension RoomPrivateMethods on Room { Future _cleanUp({bool disposeLocalParticipant = true}) async { logger.fine('[${objectId}] cleanUp()'); - _resumeRosterSnapshot = null; + _disarmResumeRosterSnapshot(); // clean up RemoteParticipants final participants = _remoteParticipants.toList(); diff --git a/test/core/resume_roster_reconcile_test.dart b/test/core/resume_roster_reconcile_test.dart index 78b904b2a..d58b37a4e 100644 --- a/test/core/resume_roster_reconcile_test.dart +++ b/test/core/resume_roster_reconcile_test.dart @@ -30,6 +30,7 @@ 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/test_data.dart'; import '../mock/websocket_mock.dart'; void main() { @@ -67,25 +68,38 @@ void main() { ws.onData(lk_rtc.SignalResponse(reconnect: lk_rtc.ReconnectResponse()).writeToBuffer()); } + lk_models.ParticipantInfo info(String identity) => lk_models.ParticipantInfo( + sid: '${identity}_sid', + identity: identity, + state: lk_models.ParticipantInfo_State.ACTIVE, + ); + /// The roster snapshot the server sends right after the `ReconnectResponse`. + /// It always carries the local participant — the server includes it so + /// metadata changes propagate — which is what marks it as a full roster + /// rather than an ordinary partial update. void sendRosterSnapshot(List identities) { ws.onData( lk_rtc.SignalResponse( update: lk_rtc.ParticipantUpdate( - participants: identities - .map( - (identity) => lk_models.ParticipantInfo( - sid: '${identity}_sid', - identity: identity, - state: lk_models.ParticipantInfo_State.ACTIVE, - ), - ) - .toList(), + participants: [ + localParticipantData, + ...identities.map(info), + ], ), ).writeToBuffer(), ); } + /// An ordinary update: only the participants that changed, no local entry. + void sendPartialUpdate(List identities) { + ws.onData( + lk_rtc.SignalResponse( + update: lk_rtc.ParticipantUpdate(participants: identities.map(info).toList()), + ).writeToBuffer(), + ); + } + test('resume synthesizes a disconnect for a participant that left', () async { final disconnected = []; final cancel = room.events.listen((event) { @@ -144,6 +158,48 @@ void main() { expect(room.remoteParticipants.keys, containsAll(['leaver', 'witness'])); }); + test('a partial update is not mistaken for the roster snapshot', () async { + // An ordinary update lists only the participants that changed. Treating one + // as a full roster would evict everybody else, so it must not disarm or + // trigger the reconciliation. + final disconnected = []; + final cancel = room.events.listen((event) { + if (event is ParticipantDisconnectedEvent) { + disconnected.add(event.participant.identity); + } + }); + + await resumeSignalConnection(); + sendPartialUpdate(['newcomer']); + await room.events.waitFor(duration: const Duration(seconds: 5)); + await Future.delayed(const Duration(milliseconds: 50)); + await cancel(); + + expect(disconnected, isEmpty, reason: 'a partial update must not evict participants'); + expect(room.remoteParticipants.keys, containsAll(['leaver', 'witness', 'newcomer'])); + }); + + test('the arming expires so a much later update cannot trigger it', () async { + final disconnected = []; + final cancel = room.events.listen((event) { + if (event is ParticipantDisconnectedEvent) { + disconnected.add(event.participant.identity); + } + }); + + await resumeSignalConnection(); + await room.events.waitFor(duration: const Duration(seconds: 5)); + // No snapshot ever arrives. Once the window closes, a later snapshot-shaped + // update is just a normal update and must not reconcile against it. + await Future.delayed(const Duration(seconds: 6)); + sendRosterSnapshot(['newcomer']); + await Future.delayed(const Duration(milliseconds: 50)); + await cancel(); + + expect(disconnected, isEmpty); + expect(room.remoteParticipants.keys, containsAll(['leaver', 'witness', 'newcomer'])); + }, timeout: const Timeout(Duration(seconds: 30))); + test('a full reconnect does not run the resume reconciliation', () async { // The full-restart path unwinds the roster itself and rebuilds it from the // JoinResponse; the armed snapshot must not double-fire on top of that. @@ -154,6 +210,11 @@ void main() { } }); + // Arm the reconciliation first, then escalate: a resume that reaches the + // ReconnectResponse and then fails takes exactly this path, and the + // arming must not survive into the restart. + await resumeSignalConnection(); + final previousHandlers = ws.handlers; container.engine.fullReconnectOnNext = true; ws.onDispose();