handle late reconnect responses

.changeset/dark-trains-tie.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+fix: add support for late reconnect response

src/api/SignalClient.ts

@@ -241,6 +241,8 @@ export class SignalClient {
 
   onJoined?: (event: JoinResponse) => void;
 
+  onLateReconnectResponse?: (msg: ReconnectResponse) => void;
+
   connectOptions?: ConnectOpts;
 
   ws?: WebSocketStream;
@@ -1077,6 +1079,10 @@ export class SignalClient {
       if (this.onDataTrackSubscriberHandles) {
         this.onDataTrackSubscriberHandles(msg.value);
       }
+    } else if (msg.case === 'reconnect') {
+      if (this.onLateReconnectResponse) {
+        this.onLateReconnectResponse(msg.value);
+      }
     } else {
       this.log.debug('unsupported message', { msgCase: msg.case });
     }

src/room/PCTransportManager.ts

@@ -216,7 +216,9 @@ export class PCTransportManager {
     this.publisher.setConfiguration(config);
     this.subscriber?.setConfiguration(config);
     if (iceRestart) {
-      this.triggerIceRestart();
+      this.triggerIceRestart().catch((error) =>
+        this.iceLog.error('failed to restart ICE', { ...this.logContext, error }),
+      );
     }
   }
 

src/room/RTCEngine.test.ts

@@ -1,10 +1,14 @@
 import {
   DataPacket,
   DataPacket_Kind,
+  ICEServer,
   ConnectionQuality as ProtoConnectionQuality,
+  ReconnectResponse,
+  ServerInfo,
   UserPacket,
 } from '@livekit/protocol';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { SignalConnectionState } from '../api/SignalClient';
 import type { DataPacketBuffer } from '../utils/dataPacketBuffer';
 import { PCTransportState } from './PCTransportManager';
 import RTCEngine, { DataChannelKind } from './RTCEngine';
@@ -919,4 +923,150 @@ describe('RTCEngine', () => {
       expect(handleDisconnect).not.toHaveBeenCalled();
     });
   });
+
+  describe('late ReconnectResponse', () => {
+    // A ReconnectResponse that isn't the first message of a resume (the server can emit it behind
+    // another message on a cross-node migration) used to be dropped as "unsupported", leaving the
+    // client on the previous node's ICE servers with an unflushed reliable buffer.
+    interface LateInternals {
+      _isClosed: boolean;
+      attemptingReconnect: boolean;
+      url: string;
+      token: string;
+      latestJoinResponse: { serverInfo?: ServerInfo };
+      lateReconnectResponse?: ReconnectResponse;
+      client: Record<string, any>;
+      pcManager: unknown;
+      waitForPCReconnected: () => Promise<void>;
+      dataChannelForKind: (kind: DataChannelKind) => RTCDataChannel | undefined;
+      resendReliableMessagesForResume: (seq: number) => Promise<void>;
+      resumeConnection: (reason?: number) => Promise<void>;
+      setupSignalClientCallbacks: () => void;
+    }
+
+    const makeResponse = (lastMessageSeq: number) =>
+      new ReconnectResponse({
+        lastMessageSeq,
+        iceServers: [new ICEServer({ urls: ['turn:new-node'], username: 'u', credential: 'c' })],
+        serverInfo: new ServerInfo({ nodeId: 'new-node', region: 'new-region' }),
+      });
+
+    function primeEngine() {
+      const engine = new RTCEngine(roomOptionDefaults);
+      const internals = engine as unknown as LateInternals;
+      internals._isClosed = false;
+      internals.attemptingReconnect = false;
+      internals.url = 'ws://localhost:7880';
+      internals.token = 'token';
+      internals.latestJoinResponse = { serverInfo: new ServerInfo({ nodeId: 'old-node' }) };
+      const updateConfiguration = vi.fn();
+      internals.pcManager = {
+        currentState: PCTransportState.CONNECTED,
+        updateConfiguration,
+        triggerIceRestart: vi.fn(async () => {}),
+      };
+      internals.waitForPCReconnected = vi.fn(async () => {});
+      internals.dataChannelForKind = vi.fn(() => undefined);
+      const resend = vi.fn(async () => {});
+      internals.resendReliableMessagesForResume = resend;
+      internals.client = {
+        currentState: SignalConnectionState.CONNECTED,
+        setReconnected: vi.fn(),
+        reconnect: vi.fn(async () => undefined),
+      };
+      return { engine, internals, updateConfiguration, resend };
+    }
+
+    it('applies ICE servers, serverInfo and the replay when it arrives outside a resume', () => {
+      const { internals, updateConfiguration, resend } = primeEngine();
+      internals.setupSignalClientCallbacks();
+
+      internals.client.onLateReconnectResponse(makeResponse(7));
+
+      expect(updateConfiguration).toHaveBeenCalledTimes(1);
+      expect(updateConfiguration.mock.calls[0][0].iceServers).toEqual([
+        { urls: ['turn:new-node'], username: 'u', credential: 'c' },
+      ]);
+      expect(internals.latestJoinResponse.serverInfo?.nodeId).toBe('new-node');
+      expect(resend).toHaveBeenCalledWith(7);
+    });
+
+    it('defers the replay to the in-flight resume, which runs it once the transport is back', async () => {
+      const { internals, updateConfiguration, resend } = primeEngine();
+      // no ReconnectResponse as the first message; it lands while the resume is still running
+      internals.client.reconnect = vi.fn(async () => {
+        internals.client.onLateReconnectResponse(makeResponse(11));
+        return undefined;
+      });
+      internals.attemptingReconnect = true;
+
+      await internals.resumeConnection();
+
+      // the node-describing parts are applied immediately, ahead of the ICE restart
+      expect(updateConfiguration).toHaveBeenCalledTimes(1);
+      expect(internals.latestJoinResponse.serverInfo?.nodeId).toBe('new-node');
+      // ...while the replay waits for the resume to reach its usual replay point, and runs once
+      expect(resend).toHaveBeenCalledTimes(1);
+      expect(resend).toHaveBeenCalledWith(11);
+      expect(internals.lateReconnectResponse).toBeUndefined();
+    });
+
+    it('does not replay a response stashed during an earlier attempt', async () => {
+      const { internals, resend } = primeEngine();
+      // left over from an attempt that never reached its replay point
+      internals.lateReconnectResponse = makeResponse(3);
+
+      await internals.resumeConnection();
+
+      expect(resend).not.toHaveBeenCalled();
+      expect(internals.lateReconnectResponse).toBeUndefined();
+    });
+
+    // `setConfiguration` does not rebuild an offer that already went out, so a response landing
+    // once the restart offer is on the wire has to restart ICE again on the new configuration.
+    // Every arrival point qualifies: nothing between `client.reconnect()` resolving and
+    // `triggerIceRestart()` yields to the event loop, so no response can beat the offer.
+    describe('re-restarts ICE, whenever it lands', () => {
+      const arrivals: Array<[string, (i: LateInternals, fire: () => void) => void]> = [
+        [
+          'while triggerIceRestart runs',
+          (i, fire) => ((i.pcManager as any).triggerIceRestart = vi.fn(async () => fire())),
+        ],
+        [
+          'while waitForPCReconnected runs',
+          (i, fire) => (i.waitForPCReconnected = vi.fn(async () => fire())),
+        ],
+      ];
+
+      it.each(arrivals)('%s', async (_name, arrange) => {
+        const { internals, updateConfiguration } = primeEngine();
+        internals.setupSignalClientCallbacks();
+        internals.attemptingReconnect = true;
+        arrange(internals, () => internals.client.onLateReconnectResponse(makeResponse(5)));
+
+        await internals.resumeConnection();
+
+        expect(updateConfiguration).toHaveBeenCalledWith(expect.anything(), true);
+      });
+
+      it('after the resume has completed', async () => {
+        const { internals, updateConfiguration } = primeEngine();
+        internals.setupSignalClientCallbacks();
+
+        await internals.resumeConnection();
+        internals.client.onLateReconnectResponse(makeResponse(5));
+
+        expect(updateConfiguration).toHaveBeenCalledWith(expect.anything(), true);
+      });
+
+      it('but not for an in-order response, whose restart is still to come', async () => {
+        const { internals, updateConfiguration } = primeEngine();
+        internals.client.reconnect = vi.fn(async () => makeResponse(5));
+
+        await internals.resumeConnection();
+
+        expect(updateConfiguration).toHaveBeenCalledWith(expect.anything(), false);
+      });
+    });
+  });
 });

src/room/RTCEngine.ts

@@ -219,6 +219,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
   private attemptingReconnect: boolean = false;
 
+  /** Late-arriving ReconnectResponse, replayed by the in-flight resume. */
+  private lateReconnectResponse?: ReconnectResponse;
+
   private reconnectPolicy: ReconnectPolicy;
 
   private reconnectTimeout?: ReturnType<typeof setTimeout>;
@@ -741,6 +744,26 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       this.emit(EngineEvent.DataTrackSubscriberHandles, event);
     };
 
+    this.client.onLateReconnectResponse = (res: ReconnectResponse) => {
+      this.log.warn('received reconnect response out of order, applying it late', {
+        ...this.logContext,
+      });
+      // The new node's ICE servers can't wait for the resume to finish: without them we'd keep
+      // the previous node's TURN credentials, which can fail the ICE restart outright. Nothing
+      // between `client.reconnect` resolving and `triggerIceRestart` yields to the event loop, so
+      // by the time a response can reach us the restart offer is already out — and
+      // `setConfiguration` alone does not rebuild it. Restart again on the new configuration.
+      this.applyReconnectResponse(res, true);
+
+      if (this.attemptingReconnect) {
+        // A resume is still running — let it replay once the peer connection is back, rather than
+        // pushing the backlog into a data channel that is about to be torn down.
+        this.lateReconnectResponse = res;
+      } else {
+        this.replayReliableMessages(res);
+      }
+    };
+
     this.client.onClose = () => {
       this.handleDisconnect('signal', ReconnectReason.RR_SIGNAL_DISCONNECTED);
     };
@@ -1440,6 +1463,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
 
     this.log.info(`resuming signal connection, attempt ${this.reconnectAttempts}`);
     this.emit(EngineEvent.Resuming);
+    // Anything stashed by a previous attempt belongs to a session we've since left.
+    this.lateReconnectResponse = undefined;
     let res: ReconnectResponse | undefined;
     try {
       this.setupSignalClientCallbacks();
@@ -1461,12 +1486,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
     this.emit(EngineEvent.SignalResumed);
 
     if (res) {
-      const rtcConfig = this.makeRTCConfiguration(res);
-      this.pcManager.updateConfiguration(rtcConfig);
-      if (this.latestJoinResponse) {
-        this.latestJoinResponse.serverInfo = res.serverInfo;
-      }
+      this.applyReconnectResponse(res);
     } else {
+      // Not necessarily fatal: the response may still be in flight behind another message, in
+      // which case `onLateReconnectResponse` applies it as soon as it lands.
       this.log.warn('Did not receive reconnect response');
     }
 
@@ -1493,19 +1516,36 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       this.createDataChannels();
     }
 
-    if (res?.lastMessageSeq) {
-      this.resendReliableMessagesForResume(res.lastMessageSeq).catch((error) => {
-        this.log.warn('failed to resend reliable messages after resume', {
-          ...this.logContext,
-          error,
-        });
-      });
-    }
+    // A response that arrived out of order gets replayed here too — now that the transport is
+    // back, this is the same point in the resume the in-order one would have been replayed at.
+    const reconnectResponse = res ?? this.lateReconnectResponse;
+    this.lateReconnectResponse = undefined;
+    this.replayReliableMessages(reconnectResponse);
 
     // resume success
     this.emit(EngineEvent.Resumed);
   }
 
+  /** Applies the node-describing half of a ReconnectResponse. Idempotent. */
+  private applyReconnectResponse(res: ReconnectResponse, iceRestart = false) {
+    this.pcManager?.updateConfiguration(this.makeRTCConfiguration(res), iceRestart);
+    if (this.latestJoinResponse) {
+      this.latestJoinResponse.serverInfo = res.serverInfo;
+    }
+  }
+
+  private replayReliableMessages(res: ReconnectResponse | undefined) {
+    if (res?.lastMessageSeq === undefined) {
+      return;
+    }
+    this.resendReliableMessagesForResume(res.lastMessageSeq).catch((error) =>
+      this.log.warn('failed to resend reliable messages after resume', {
+        ...this.logContext,
+        error,
+      }),
+    );
+  }
+
   async waitForPCInitialConnection(timeout?: number, abortController?: AbortController) {
     if (!this.pcManager) {
       throw new UnexpectedConnectionState('PC manager is closed');