From 7b84455416484b8e1c4fa03512c79dbb2db93ea5 Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Fri, 23 Oct 2026 04:14:50 +0800 Subject: [PATCH 1/3] Ignore stale ParticipantInfo in RemoteParticipant.updateFromInfo Participant.updateFromInfo returns false when the incoming info has an older version for the same participant sid, but the remote override had its early return commented out. The stale info then went on to reconcile tracks and removed every publication that only the newer info contained. Return early like LocalParticipant and the JS SDK already do. The server only increases the version for a given sid and a rejoin gets a new sid, so this never drops a legitimate update. --- .changes/ignore-stale-participant-updates | 1 + lib/src/participant/remote.dart | 2 +- test/core/participant_update_test.dart | 75 +++++++++++++++++++++++ 3 files changed, 77 insertions(+), 1 deletion(-) create mode 100644 .changes/ignore-stale-participant-updates create mode 100644 test/core/participant_update_test.dart diff --git a/.changes/ignore-stale-participant-updates b/.changes/ignore-stale-participant-updates new file mode 100644 index 000000000..75fcea2ae --- /dev/null +++ b/.changes/ignore-stale-participant-updates @@ -0,0 +1 @@ +patch type="fixed" "A remote participant update with an older version than the one already applied is now ignored, so it no longer removes tracks published in the newer update" 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..d43f40d7c --- /dev/null +++ b/test/core/participant_update_test.dart @@ -0,0 +1,75 @@ +// 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/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'; + +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(); + +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); + }); +} From 132b8dc3fb85c4b1de680980aa785994888ddd6c Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Fri, 23 Oct 2026 04:16:35 +0800 Subject: [PATCH 2/3] Serialize remote participant updates in Room The room signal listener is not synchronized, so each ParticipantUpdate runs _onParticipantUpdateEvent concurrently with the ones before it. Right after joining this is the common case: the server sends the JoinResponse and then flushes the updates it queued, and every one of them parks on the RoomConnectedEvent wait and resumes at once. Their track reconciliation then interleaves across awaits, and an older update removes the publication a newer one added. The same window lets two updates for a new participant both miss the lookup in _getOrCreateRemoteParticipant and create two instances, each with its own ParticipantConnectedEvent. Run JoinResponse participant creation and ParticipantUpdate handling under one lock. Pending track flushing moves out of the locked section because addSubscribedMediaTrack can wait for a later update to publish the track, which would otherwise block on the lock until the publish timeout. --- .changes/ignore-stale-participant-updates | 1 - lib/src/core/room.dart | 60 ++++++++++----- test/core/participant_update_test.dart | 89 +++++++++++++++++++++++ 3 files changed, 132 insertions(+), 18 deletions(-) delete mode 100644 .changes/ignore-stale-participant-updates diff --git a/.changes/ignore-stale-participant-updates b/.changes/ignore-stale-participant-updates deleted file mode 100644 index 75fcea2ae..000000000 --- a/.changes/ignore-stale-participant-updates +++ /dev/null @@ -1 +0,0 @@ -patch type="fixed" "A remote participant update with an older version than the one already applied is now ignored, so it no longer removes tracks published in the newer update" 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/test/core/participant_update_test.dart b/test/core/participant_update_test.dart index d43f40d7c..ba834a9ff 100644 --- a/test/core/participant_update_test.dart +++ b/test/core/participant_update_test.dart @@ -18,9 +18,12 @@ 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', @@ -41,6 +44,27 @@ 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() { @@ -72,4 +96,69 @@ void main() { 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'); + }); } From b3e69f00a6105701f812d487fa220098909680bc Mon Sep 17 00:00:00 2001 From: Hiroshi Horie <548776+hiroshihorie@users.noreply.github.com> Date: Fri, 23 Oct 2026 04:46:28 +0800 Subject: [PATCH 3/3] Add changeset --- .changes/participant-update-races | 1 + 1 file changed, 1 insertion(+) create mode 100644 .changes/participant-update-races 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"