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..5848deddd 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -131,6 +131,23 @@ class Room extends DisposableChangeNotifier with EventsEmittable { RegionUrlProvider? _regionUrlProvider; String? _regionUrl; + /// 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 = {}; @@ -402,6 +419,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. + _armResumeRosterSnapshot(); + }) ..on((event) => _onParticipantUpdateEvent(event.participants)) ..on((event) => _onSignalSpeakersChangedEvent(event.speakers)) ..on((event) => _onSignalConnectionQualityUpdateEvent(event.updates)) @@ -601,6 +627,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 + _disarmResumeRosterSnapshot(); + // reset params _name = null; _metadata = null; @@ -814,6 +844,13 @@ 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 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. @@ -824,10 +861,15 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } if (localParticipant?.identity == info.identity) { + sawLocalParticipant = true; await localParticipant?.updateFromInfo(info); 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 +904,54 @@ 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 && sawLocalParticipant && identical(rosterSnapshot, _resumeRosterSnapshot)) { + _disarmResumeRosterSnapshot(); + hasChanged = await _reconcileAbsentParticipants(rosterSnapshot) || hasChanged; + } + if (hasChanged) { notifyListeners(); } } + 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 + /// `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 +1175,8 @@ extension RoomPrivateMethods on Room { Future _cleanUp({bool disposeLocalParticipant = true}) async { logger.fine('[${objectId}] cleanUp()'); + _disarmResumeRosterSnapshot(); + // 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..d58b37a4e --- /dev/null +++ b/test/core/resume_roster_reconcile_test.dart @@ -0,0 +1,230 @@ +// 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/test_data.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()); + } + + 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: [ + 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) { + 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 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. + final disconnected = []; + final cancel = room.events.listen((event) { + if (event is ParticipantDisconnectedEvent) { + disconnected.add(event.participant.identity); + } + }); + + // 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(); + 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']); + }); +}