diff --git a/.changes/participant-update-races b/.changes/participant-update-races new file mode 100644 index 000000000..af1f843a6 --- /dev/null +++ b/.changes/participant-update-races @@ -0,0 +1 @@ +patch type="fixed" "Fix remote tracks going missing and duplicate participants when participant updates arrive back to back or out of order" diff --git a/lib/src/core/room.dart b/lib/src/core/room.dart index f10199abb..dceb9fcb0 100644 --- a/lib/src/core/room.dart +++ b/lib/src/core/room.dart @@ -17,6 +17,7 @@ import 'dart:typed_data' show Uint8List; import 'package:collection/collection.dart'; import 'package:meta/meta.dart'; +import 'package:synchronized/synchronized.dart' as sync; import '../audio/audio_manager.dart'; import '../core/signal_client.dart'; @@ -75,6 +76,12 @@ class Room extends DisposableChangeNotifier with EventsEmittable { RoomOptions get roomOptions => engine.roomOptions; final ParticipantCollection _remoteParticipants = ParticipantCollection(); + + // Remote participant creation and updates await between reading and + // writing participant state, so overlapping signal events must not + // interleave. + final _participantUpdateLock = sync.Lock(); + UnmodifiableMapView get remoteParticipants => UnmodifiableMapView(_remoteParticipants.byIdentity); @@ -574,12 +581,19 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } } - for (final info in event.response.otherParticipants) { - logger.fine( - 'Creating RemoteParticipant: sid = ${info.sid}(identity:${info.identity}) ' - 'tracks:${info.tracks.map((e) => e.sid)}', - ); - await _getOrCreateRemoteParticipant(info); + final joinedParticipants = await _participantUpdateLock.synchronized(() async { + final participants = []; + for (final info in event.response.otherParticipants) { + logger.fine( + 'Creating RemoteParticipant: sid = ${info.sid}(identity:${info.identity}) ' + 'tracks:${info.tracks.map((e) => e.sid)}', + ); + participants.add((await _getOrCreateRemoteParticipant(info)).participant); + } + return participants; + }); + for (final participant in joinedParticipants) { + await _flushPendingTracks(participant: participant); } if (e2eeManager != null && event.response.sifTrailer.isNotEmpty) { @@ -819,22 +833,33 @@ class Room extends DisposableChangeNotifier with EventsEmittable { ); _remoteParticipants.set(result.participant); - await _flushPendingTracks(participant: result.participant); return result; } Future _onParticipantUpdateEvent(List updates) async { + // The local participant is not ready yet, waiting for the + // `RoomConnectedEvent` to create the local participant. + if (_localParticipant == null) { + await events.waitFor( + duration: const Duration(seconds: 10), + ); + } + + final updatedParticipants = await _participantUpdateLock.synchronized(() => _applyParticipantUpdates(updates)); + + // A pending track can wait for a later update to publish it, so flush + // only after the lock is released. + for (final participant in updatedParticipants) { + await _flushPendingTracks(participant: participant); + } + } + + /// Returns the remote participants that were created or updated. + Future> _applyParticipantUpdates(List updates) async { + final updatedParticipants = []; // trigger change notifier only if list of participants membership is changed var hasChanged = false; for (final info in updates) { - // The local participant is not ready yet, waiting for the - // `RoomConnectedEvent` to create the local participant. - if (_localParticipant == null) { - await events.waitFor( - duration: const Duration(seconds: 10), - ); - } - if (localParticipant?.identity == info.identity) { await localParticipant?.updateFromInfo(info); continue; @@ -864,12 +889,12 @@ class Room extends DisposableChangeNotifier with EventsEmittable { } } _remoteParticipants.set(result.participant); - await _flushPendingTracks(participant: result.participant); + updatedParticipants.add(result.participant); } else { final wasUpdated = await result.participant.updateFromInfo(info); if (wasUpdated) { _remoteParticipants.set(result.participant); - await _flushPendingTracks(participant: result.participant); + updatedParticipants.add(result.participant); } } } @@ -877,6 +902,7 @@ class Room extends DisposableChangeNotifier with EventsEmittable { if (hasChanged) { notifyListeners(); } + return updatedParticipants; } void _onSignalSpeakersChangedEvent(List speakers) { diff --git a/lib/src/participant/remote.dart b/lib/src/participant/remote.dart index 517d19c80..880fa5b4f 100644 --- a/lib/src/participant/remote.dart +++ b/lib/src/participant/remote.dart @@ -261,7 +261,7 @@ class RemoteParticipant extends Participant { Future updateFromInfo(lk_models.ParticipantInfo info) async { logger.fine('RemoteParticipant.updateFromInfo(info: $info)'); if (!await super.updateFromInfo(info)) { - //return false; + return false; } await _updateTracks(info.tracks); diff --git a/test/core/participant_update_test.dart b/test/core/participant_update_test.dart new file mode 100644 index 000000000..ba834a9ff --- /dev/null +++ b/test/core/participant_update_test.dart @@ -0,0 +1,164 @@ +// 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: 5)) +library; + +import 'package:flutter_test/flutter_test.dart'; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/internal/events.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/mediastream_mock.dart'; +import 'signal_client_test.dart' show exampleUri, joinResponse, offerResponse, token; + +lk_models.ParticipantInfo _info(int version, List trackSids) => lk_models.ParticipantInfo( + sid: 'PA_remote', + identity: 'remote', + version: version, + state: lk_models.ParticipantInfo_State.ACTIVE, + tracks: trackSids.map( + (sid) => lk_models.TrackInfo( + sid: sid, + name: sid, + type: lk_models.TrackType.AUDIO, + source: lk_models.TrackSource.MICROPHONE, + ), + ), +); + +List _update(int version, List trackSids) => lk_rtc.SignalResponse( + update: lk_rtc.ParticipantUpdate(participants: [_info(version, trackSids)]), +).writeToBuffer(); + +/// Connects with [otherParticipants] in the JoinResponse and delivers +/// [updates] right behind it, the way the server flushes the updates it +/// queued while the participant was joining. +Future _connectWithQueuedUpdates( + E2EContainer container, { + List otherParticipants = const [], + required List> updates, +}) async { + final join = joinResponse.deepCopy(); + join.join.otherParticipants + ..clear() + ..addAll(otherParticipants); + + final connectFuture = container.room.connect(exampleUri, token); + await Future.delayed(const Duration(milliseconds: 1)); + container.wsConnector.onData(join.writeToBuffer()); + updates.forEach(container.wsConnector.onData); + container.wsConnector.onData(offerResponse.writeToBuffer()); + await connectFuture; +} + +Future _settle() => Future.delayed(const Duration(milliseconds: 50)); + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + late E2EContainer container; + late Room room; + + setUp(() { + container = E2EContainer(); + room = container.room; + }); + + tearDown(() async { + await container.dispose(); + }); + + test('an older snapshot does not remove tracks from a newer one', () async { + await container.connectRoom(); + final unpublished = []; + room.events.on((event) => unpublished.add(event.publication.sid)); + + container.wsConnector.onData(_update(3, ['TR_A', 'TR_B'])); + await _settle(); + container.wsConnector.onData(_update(2, ['TR_A'])); + await _settle(); + + final participant = room.remoteParticipants['remote']!; + expect(participant.trackPublications.keys, unorderedEquals(['TR_A', 'TR_B'])); + expect(unpublished, isEmpty); + }); + + test('updates queued during join are applied in order', () async { + final events = []; + room.events.on((event) => events.add('published ${event.publication.sid}')); + room.events.on((event) => events.add('unpublished ${event.publication.sid}')); + + await _connectWithQueuedUpdates( + container, + otherParticipants: [_info(2, [])], + updates: [ + _update(3, ['TR_A']), + _update(4, ['TR_A']), + _update(5, ['TR_A', 'TR_B']), + ], + ); + await _settle(); + + final participant = room.remoteParticipants['remote']!; + expect(participant.trackPublications.keys, unorderedEquals(['TR_A', 'TR_B'])); + expect(events, ['published TR_A', 'published TR_B']); + }); + + test('queued updates for a new participant create a single instance', () async { + final connected = []; + room.events.on((event) => connected.add(event.participant)); + + await _connectWithQueuedUpdates( + container, + updates: [ + _update(1, ['TR_A']), + _update(2, ['TR_A', 'TR_B']), + ], + ); + await _settle(); + + final participant = room.remoteParticipants['remote']!; + expect(connected, hasLength(1)); + expect(connected.single, same(participant)); + expect(participant.trackPublications.keys, unorderedEquals(['TR_A', 'TR_B'])); + }); + + test('a pending track waiting for its publication does not block later updates', () async { + await container.connectRoom(); + + // Media for TR_B arrives before the participant is known, so it is queued. + container.engine.events.emit( + EngineTrackAddedEvent( + track: FakeMediaStreamTrack(id: 'TR_B', kind: 'audio'), + stream: FakeMediaStream('PA_remote|stream'), + receiver: null, + ), + ); + final subscribed = room.events.waitFor( + filter: (event) => event.publication.sid == 'TR_B', + duration: const Duration(seconds: 1), + ); + + // The first update creates the participant without TR_B, so flushing the + // queued media waits for a later update to publish it. + container.wsConnector.onData(_update(1, ['TR_A'])); + await _settle(); + container.wsConnector.onData(_update(2, ['TR_A', 'TR_B'])); + + expect((await subscribed).participant.identity, 'remote'); + }); +}