Reconcile the participant roster after a signal resume

.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

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

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']);
+  });
+});

src/room/Room.ts

@@ -225,6 +225,16 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
 
   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<string>;
+
   private pendingTrackAddedCallbacks = new Map<Track.SID, Set<() => void>>();
 
   /**
@@ -634,6 +644,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
       .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<RoomEventCallbacks>)
         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<RoomEventCallbacks>)
     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<RoomEventCallbacks>)
   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<RoomEventCallbacks>)
         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<RoomEventCallbacks>)
     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,