fix: report one recovery with the connected region's changes after a region failover

.changeset/restart-region-failover.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Emit `Reconnected` once and keep room metadata and recording changes when a full reconnect fails over to another region

src/room/CoreRoom.ts

@@ -251,6 +251,12 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
 
   private bufferedEvents: Array<any> = [];
 
+  /**
+   * the room as it was before the current full reconnect. Every region the full reconnect tries
+   * starts from it, so the region it connects to reports every change since the outage
+   */
+  private roomInfoBeforeRestart?: RoomModel;
+
   private isResuming: boolean = false;
 
   private pendingTrackAddedCallbacks = new Map<Track.SID, Set<() => void>>();
@@ -1718,6 +1724,11 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
     // in case we went from resuming to full-reconnect, make sure to reflect it on the isResuming flag
     this.isResuming = false;
 
+    // a region that failed after its join leaves its room state and held back events behind
+    this.roomInfoBeforeRestart ??= this.roomInfo;
+    this.roomInfo = this.roomInfoBeforeRestart;
+    this.bufferedEvents = [];
+
     // also unwind existing participants & existing subscriptions
     for (const p of this.remoteParticipants.values()) {
       this.handleParticipantDisconnected(p.identity, p);
@@ -1733,6 +1744,11 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
       region: joinResponse.serverRegion,
     });
     this.bufferedEvents = [];
+    // wait from here, so that this region failing during the republish ends the wait
+    const restarted = this.engine.waitForRestarted().then(
+      () => true,
+      () => false,
+    );
 
     this.applyJoinResponse(joinResponse);
 
@@ -1743,13 +1759,13 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
       this.log.error('error trying to re-publish tracks after reconnection', { error });
     }
 
-    try {
-      await this.engine.waitForRestarted();
-      this.log.debug(`fully reconnected to server`, { region: joinResponse.serverRegion });
-    } catch {
-      // reconnection failed, handleDisconnect is being invoked already, just return here
+    if (!(await restarted)) {
+      // this region failed and the next attempt reports the recovery, or reconnection failed
+      // and handleDisconnect is being invoked already
       return;
     }
+    this.log.debug(`fully reconnected to server`, { region: joinResponse.serverRegion });
+    this.roomInfoBeforeRestart = undefined;
     this.setAndEmitConnectionState(ConnectionState.Connected);
     this.emit(RoomEvent.Reconnected);
     this.registerConnectionReconcile();
@@ -1760,6 +1776,7 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
     this.clearConnectionReconcile();
     this.isResuming = false;
     this.bufferedEvents = [];
+    this.roomInfoBeforeRestart = undefined;
     this.transcriptionReceivedTimes.clear();
     this.extensionResults.forEach((result) => result.onDisconnect?.());
     if (this.state === ConnectionState.Disconnected) {

src/room/RTCEngine.ts

@@ -1667,21 +1667,26 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     }
   }
 
+  /** waits for the current restart attempt's `Restarted`; call it on `SignalRestarted` */
   waitForRestarted = () => {
     return new Promise<void>((resolve, reject) => {
-      if (this.pcState === PCState.Connected) {
-        resolve();
-      }
+      const cleanup = () => {
+        this.off(EngineEvent.Restarted, onRestarted);
+        this.off(EngineEvent.Disconnected, onFailed);
+        this.off(EngineEvent.Restarting, onFailed);
+      };
       const onRestarted = () => {
-        this.off(EngineEvent.Disconnected, onDisconnected);
+        cleanup();
         resolve();
       };
-      const onDisconnected = () => {
-        this.off(EngineEvent.Restarted, onRestarted);
+      // a new Restarting means this attempt's region failed and the next attempt takes over
+      const onFailed = () => {
+        cleanup();
         reject();
       };
       this.once(EngineEvent.Restarted, onRestarted);
-      this.once(EngineEvent.Disconnected, onDisconnected);
+      this.once(EngineEvent.Disconnected, onFailed);
+      this.once(EngineEvent.Restarting, onFailed);
     });
   };
 

src/room/Room.test.ts

@@ -555,3 +555,71 @@ describe('connection attempt racing disconnect (#2150)', () => {
     expect(disconnected).not.toHaveBeenCalled();
   });
 });
+
+describe('full reconnect failing over to a second region', () => {
+  const join = (metadata: string) =>
+    new JoinResponse({
+      room: { sid: 'RM_test', name: 'test', metadata },
+      participant: { sid: 'PA_local', identity: 'local' },
+    });
+  const flush = () => new Promise((resolve) => setTimeout(resolve, 0));
+
+  it('reports the recovery once, with the changes the connected region reports', async () => {
+    const room = new Room();
+    room.state = ConnectionState.Connected;
+    room.engine.emit(EngineEvent.RoomUpdate, join('old').room!);
+    vi.spyOn(room.localParticipant, 'republishAllTracks').mockImplementation(async () => {});
+    const reconnected = vi.fn();
+    const metadataChanged = vi.fn();
+    room.on(RoomEvent.Reconnected, reconnected);
+    room.on(RoomEvent.RoomMetadataChanged, metadataChanged);
+
+    // the first region joins, then its peer connection fails
+    room.engine.emit(EngineEvent.Restarting);
+    room.engine.emit(EngineEvent.SignalRestarted, join('new'));
+    await flush();
+    // the second region joins and connects
+    room.engine.emit(EngineEvent.Restarting);
+    room.engine.emit(EngineEvent.SignalRestarted, join('new'));
+    await flush();
+    room.engine.emit(EngineEvent.Restarted);
+    await flush();
+
+    expect(room.state).toBe(ConnectionState.Connected);
+    expect(reconnected).toHaveBeenCalledTimes(1);
+    expect(metadataChanged).toHaveBeenCalledTimes(1);
+    expect(metadataChanged).toHaveBeenCalledWith('new');
+    await room.disconnect();
+  });
+
+  it('reports the recovery once when the failed region republishes past its failure', async () => {
+    const room = new Room();
+    room.state = ConnectionState.Connected;
+    // the peer connection state from before the outage; only the restart's Restarted counts
+    (room.engine as unknown as { pcState: number }).pcState = 1; // PCState.Connected
+    let finishFirstRepublish!: () => void;
+    vi.spyOn(room.localParticipant, 'republishAllTracks')
+      .mockImplementationOnce(
+        () => new Promise<void>((resolve) => (finishFirstRepublish = resolve)),
+      )
+      .mockImplementation(async () => {});
+    const reconnected = vi.fn();
+    room.on(RoomEvent.Reconnected, reconnected);
+
+    room.engine.emit(EngineEvent.Restarting);
+    room.engine.emit(EngineEvent.SignalRestarted, join('new'));
+    await flush();
+    room.engine.emit(EngineEvent.Restarting);
+    finishFirstRepublish();
+    await flush();
+    room.engine.emit(EngineEvent.SignalRestarted, join('new'));
+    await flush();
+    expect(reconnected).not.toHaveBeenCalled();
+
+    room.engine.emit(EngineEvent.Restarted);
+    await flush();
+
+    expect(reconnected).toHaveBeenCalledTimes(1);
+    await room.disconnect();
+  });
+});