diff --git a/.changeset/resume-roster-reconciliation.md b/.changeset/resume-roster-reconciliation.md new file mode 100644 index 0000000000..4ae9e69379 --- /dev/null +++ b/.changeset/resume-roster-reconciliation.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +Reconcile the participant roster after a signal resume, removing remote participants that left while the signal connection was down diff --git a/src/e2ee/subscriberBlackScreen.test.ts b/src/e2ee/subscriberBlackScreen.test.ts index 6dbc392091..4c90fcb047 100644 --- a/src/e2ee/subscriberBlackScreen.test.ts +++ b/src/e2ee/subscriberBlackScreen.test.ts @@ -103,13 +103,7 @@ function setupRoomWithE2EE() { } ).getOrCreateParticipant('jake', new ParticipantInfo({ sid: 'PA_jake', identity: 'jake' })); - const trackInfo = new TrackInfo({ - sid: 'TR_video', - type: TrackType.VIDEO, - name: 'camera', - mimeType: 'video/h264', - encryption: Encryption_Type.GCM, - }); + const trackInfo = jakeTrackInfo(); const publication = new RemoteTrackPublication(Track.Kind.Video, trackInfo, true); // addTrackPublication is what wires TrackEvent.Subscribed -> ParticipantEvent.TrackSubscribed ( @@ -126,6 +120,27 @@ function setupRoomWithE2EE() { return { room, worker, publication, receiver, subscribe, unsubscribe }; } +function jakeTrackInfo() { + return new TrackInfo({ + sid: 'TR_video', + type: TrackType.VIDEO, + name: 'camera', + mimeType: 'video/h264', + encryption: Encryption_Type.GCM, + }); +} + +/** + * The server replays the full participant roster after the ReconnectResponse; Room reconciles + * against it to drop participants that left while the link was down, so a resume simulation has + * to include it or jake gets (correctly) evicted. + */ +function replayRosterDuringResume(room: Room) { + room.engine.emit(EngineEvent.ParticipantUpdate, [ + new ParticipantInfo({ sid: 'PA_jake', identity: 'jake', tracks: [jakeTrackInfo()] }), + ]); +} + describe('subscriber black screen', () => { beforeEach(() => { encryptionEnabledMap.clear(); @@ -164,6 +179,7 @@ describe('subscriber black screen', () => { // signal comes back: Room.ts:642 throws the buffered events away... room.engine.emit(EngineEvent.SignalResumed); + replayRosterDuringResume(room); // ...so the flush on Resumed has nothing left to flush. room.engine.emit(EngineEvent.Resumed); @@ -188,6 +204,7 @@ describe('subscriber black screen', () => { subscribe(); unsubscribe(); room.engine.emit(EngineEvent.SignalResumed); + replayRosterDuringResume(room); room.engine.emit(EngineEvent.Resumed); // CURRENTLY FAILS with ['unsubscribed']: subscribe is buffered then diff --git a/src/room/Room.test.ts b/src/room/Room.test.ts index 08fc3ffa2d..20eb8f1c79 100644 --- a/src/room/Room.test.ts +++ b/src/room/Room.test.ts @@ -1,6 +1,8 @@ import { ClientInfo_Capability, JoinResponse, + ParticipantInfo, + ParticipantInfo_State, StreamState as ProtoStreamState, StreamStateUpdate, SubscriptionError, @@ -321,3 +323,114 @@ describe('stream state updates', () => { expect(participantEvents).not.toHaveBeenCalled(); }); }); + +describe('participant roster reconciliation after resume', () => { + const rooms: Room[] = []; + + afterEach(() => { + while (rooms.length > 0) { + const room = rooms.pop()!; + // the Resumed handler starts a reconcile interval; don't leak it across tests + (room as unknown as { clearConnectionReconcile(): void }).clearConnectionReconcile(); + } + }); + + function info(identity: string) { + return new ParticipantInfo({ + sid: `PA_${identity}`, + identity, + state: ParticipantInfo_State.ACTIVE, + }); + } + + function pushUpdate(room: Room, ...infos: ParticipantInfo[]) { + room.engine.emit(EngineEvent.ParticipantUpdate, infos); + } + + /** A connected room with the given remote participants already in the roster. */ + function setupConnectedRoom(...identities: string[]) { + const room = new Room(); + rooms.push(room); + room.state = ConnectionState.Connected; + pushUpdate(room, ...identities.map(info)); + expect([...room.remoteParticipants.keys()]).toEqual(identities); + return room; + } + + it('removes participants that left while the signal connection was down', () => { + const room = setupConnectedRoom('alice', 'bob'); + + const disconnected = vi.fn(); + room.on(RoomEvent.ParticipantDisconnected, disconnected); + + room.engine.emit(EngineEvent.Resuming); + // server replays the full roster after the ReconnectResponse — bob left while we were down + pushUpdate(room, info('alice')); + room.engine.emit(EngineEvent.Resumed); + + expect([...room.remoteParticipants.keys()]).toEqual(['alice']); + expect(disconnected).toHaveBeenCalledTimes(1); + expect(disconnected.mock.calls[0][0].identity).toBe('bob'); + }); + + it('accumulates identities across updates interleaved during the resume', () => { + const room = setupConnectedRoom('alice', 'bob', 'carol'); + + room.engine.emit(EngineEvent.Resuming); + // the roster snapshot can arrive split across several batched updates + pushUpdate(room, info('alice')); + pushUpdate(room, info('bob')); + room.engine.emit(EngineEvent.Resumed); + + expect([...room.remoteParticipants.keys()]).toEqual(['alice', 'bob']); + }); + + it('keeps participants that joined during the resume', () => { + const room = setupConnectedRoom('alice'); + + room.engine.emit(EngineEvent.Resuming); + pushUpdate(room, info('alice'), info('dave')); + room.engine.emit(EngineEvent.Resumed); + + expect([...room.remoteParticipants.keys()]).toEqual(['alice', 'dave']); + }); + + it('emits ParticipantDisconnected before Reconnected', () => { + const room = setupConnectedRoom('alice', 'bob'); + + const order: string[] = []; + room.on(RoomEvent.ParticipantDisconnected, (p) => order.push(`disconnected:${p.identity}`)); + room.on(RoomEvent.Reconnected, () => order.push('reconnected')); + + room.engine.emit(EngineEvent.Resuming); + pushUpdate(room, info('alice')); + room.engine.emit(EngineEvent.Resumed); + + expect(order).toEqual(['disconnected:bob', 'reconnected']); + }); + + it('does not reconcile against updates received outside of a resume', () => { + const room = setupConnectedRoom('alice', 'bob'); + + // a routine partial update (e.g. alice changed metadata) must not evict bob + pushUpdate(room, info('alice')); + + expect([...room.remoteParticipants.keys()]).toEqual(['alice', 'bob']); + }); + + it('does not evict participants when the resume escalates to a full reconnect', () => { + const room = setupConnectedRoom('alice', 'bob'); + + room.engine.emit(EngineEvent.Resuming); + pushUpdate(room, info('alice')); + // resume failed; the engine falls back to a full reconnect, which rebuilds from the + // JoinResponse instead — the abandoned resume's roster must not be applied later + room.engine.emit(EngineEvent.Restarting); + expect(room.remoteParticipants.size).toBe(0); + + pushUpdate(room, info('alice'), info('bob')); + room.engine.emit(EngineEvent.Resumed); + + expect([...room.remoteParticipants.keys()]).toEqual(['alice', 'bob']); + }); +}); diff --git a/src/room/Room.ts b/src/room/Room.ts index d18ef0f30e..45c35241e9 100644 --- a/src/room/Room.ts +++ b/src/room/Room.ts @@ -225,6 +225,16 @@ class Room extends (EventEmitter as new () => TypedEmitter) private isResuming: boolean = false; + /** + * Identities seen in participant updates since the current resume started, or `undefined` when + * no resume is in progress. A resume — unlike a full reconnect — never rebuilds the roster from + * a `JoinResponse`, and `DISCONNECTED` updates for participants who left while the signal link + * was down went to a socket we no longer had. The server replays the roster after the + * `ReconnectResponse`, but it can interleave batched updates around it, so no single update is + * identifiable as the snapshot — we accumulate the union and reconcile once the resume settles. + */ + private resumeSeenIdentities?: Set; + private pendingTrackAddedCallbacks = new Map void>>(); /** @@ -634,6 +644,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) .on(EngineEvent.Resuming, () => { this.clearConnectionReconcile(); this.isResuming = true; + this.resumeSeenIdentities = new Set(); this.log.debug('Resuming signal connection'); if (this.setAndEmitConnectionState(ConnectionState.SignalReconnecting)) { this.emit(RoomEvent.SignalReconnecting); @@ -643,6 +654,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) this.registerConnectionReconcile(); this.isResuming = false; this.log.debug('Resumed signal connection'); + this.reconcileParticipantsAfterResume(); this.updateSubscriptions(); if (this.setAndEmitConnectionState(ConnectionState.Connected)) { this.emit(RoomEvent.Reconnected); @@ -1785,6 +1797,9 @@ class Room extends (EventEmitter as new () => TypedEmitter) this.clearConnectionReconcile(); // in case we went from resuming to full-reconnect, make sure to reflect it on the isResuming flag this.isResuming = false; + // a full reconnect rebuilds the roster from the JoinResponse, so the abandoned resume's + // partial view of it must not be applied afterwards + this.resumeSeenIdentities = undefined; // also unwind existing participants & existing subscriptions for (const p of this.remoteParticipants.values()) { @@ -1832,6 +1847,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) private handleDisconnect(shouldStopTracks = true, reason?: DisconnectReason) { this.clearConnectionReconcile(); this.isResuming = false; + this.resumeSeenIdentities = undefined; this.bufferedEvents = []; this.transcriptionReceivedTimes.clear(); this.incomingDataStreamManager.clearControllers(); @@ -1921,6 +1937,8 @@ class Room extends (EventEmitter as new () => TypedEmitter) info.identity = this.sidToIdentity.get(info.sid) ?? ''; } + this.resumeSeenIdentities?.add(info.identity); + let remoteParticipant = this.remoteParticipants.get(info.identity); // when it's disconnected, send updates @@ -1949,6 +1967,30 @@ class Room extends (EventEmitter as new () => TypedEmitter) this.incomingDataTrackManager.receiveSfuPublicationUpdates(mapped); }; + /** + * Synthesizes disconnects for remote participants that the server didn't mention in any + * participant update during a resume — they left while the signal connection was down, so their + * `DISCONNECTED` update never reached us. Called before `RoomEvent.Reconnected` is emitted so + * consumers never observe a reconnected room with a stale roster. + */ + private reconcileParticipantsAfterResume() { + const seenIdentities = this.resumeSeenIdentities; + this.resumeSeenIdentities = undefined; + if (!seenIdentities) { + return; + } + + for (const [identity, participant] of [...this.remoteParticipants]) { + if (!seenIdentities.has(identity)) { + this.log.debug( + `removing participant ${identity} absent from the roster replayed after resume`, + this.logContext, + ); + this.handleParticipantDisconnected(identity, participant); + } + } + } + private handleParticipantDisconnected( identity: string, participant?: RemoteParticipant,