diff --git a/.changeset/dark-trains-tie.md b/.changeset/dark-trains-tie.md new file mode 100644 index 0000000000..479737c82b --- /dev/null +++ b/.changeset/dark-trains-tie.md @@ -0,0 +1,5 @@ +--- +"livekit-client": patch +--- + +fix: add support for late reconnect response diff --git a/src/api/SignalClient.ts b/src/api/SignalClient.ts index 72f58d684c..9fb301f9f6 100644 --- a/src/api/SignalClient.ts +++ b/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 }); } diff --git a/src/room/PCTransportManager.ts b/src/room/PCTransportManager.ts index 43f3d2af34..fb6c900c39 100644 --- a/src/room/PCTransportManager.ts +++ b/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 }), + ); } } diff --git a/src/room/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index 1dc28a2cf6..bfc0765f64 100644 --- a/src/room/RTCEngine.test.ts +++ b/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; + pcManager: unknown; + waitForPCReconnected: () => Promise; + dataChannelForKind: (kind: DataChannelKind) => RTCDataChannel | undefined; + resendReliableMessagesForResume: (seq: number) => Promise; + resumeConnection: (reason?: number) => Promise; + 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); + }); + }); + }); }); diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index e8bd78f39e..4f6f3ec9e0 100644 --- a/src/room/RTCEngine.ts +++ b/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; @@ -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');