From 8d823814a9612e00ecc32d3c4d5959c8990e04d0 Mon Sep 17 00:00:00 2001 From: lukasIO Date: Tue, 1 Sep 2026 09:35:42 +0200 Subject: [PATCH 1/5] add support for late reconnect response --- src/api/SignalClient.ts | 6 +++ src/room/RTCEngine.test.ts | 104 +++++++++++++++++++++++++++++++++++++ src/room/RTCEngine.ts | 71 ++++++++++++++++++++----- 3 files changed, 168 insertions(+), 13 deletions(-) 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/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index 1dc28a2cf6..df41fd0a82 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,104 @@ 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 { engine, 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); + expect(engine).toBeDefined(); + }); + + 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(); + }); + }); }); diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index e8bd78f39e..6f417ddefa 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -219,6 +219,13 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit private attemptingReconnect: boolean = false; + /** + * A `ReconnectResponse` that arrived *after* the resume had already been declared connected + * (see {@link SignalClient.onLateReconnectResponse}). The in-flight resume picks it up once the + * peer connection is back, so the reliable replay still lands on a usable transport. + */ + private lateReconnectResponse?: ReconnectResponse; + private reconnectPolicy: ReconnectPolicy; private reconnectTimeout?: ReturnType; @@ -741,6 +748,23 @@ 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. + this.applyReconnectResponse(res); + + 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 +1464,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 +1487,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 +1517,40 @@ 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 parts of a `ReconnectResponse` that describe the node we've resumed onto. + * Idempotent, so a duplicate or late response is + * harmless. + */ + private applyReconnectResponse(res: ReconnectResponse) { + this.pcManager?.updateConfiguration(this.makeRTCConfiguration(res)); + if (this.latestJoinResponse) { + this.latestJoinResponse.serverInfo = res.serverInfo; + } + } + + private replayReliableMessages(res: ReconnectResponse | undefined) { + if (!res?.lastMessageSeq) { + 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'); From 041b59ab9a4d70f251197883eff3d4f4458c1de5 Mon Sep 17 00:00:00 2001 From: lukasIO Date: Tue, 1 Sep 2026 17:13:26 +0200 Subject: [PATCH 2/5] cleanup --- src/room/RTCEngine.ts | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index 6f417ddefa..3944daa66d 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -219,11 +219,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit private attemptingReconnect: boolean = false; - /** - * A `ReconnectResponse` that arrived *after* the resume had already been declared connected - * (see {@link SignalClient.onLateReconnectResponse}). The in-flight resume picks it up once the - * peer connection is back, so the reliable replay still lands on a usable transport. - */ + /** Late-arriving ReconnectResponse, replayed by the in-flight resume. */ private lateReconnectResponse?: ReconnectResponse; private reconnectPolicy: ReconnectPolicy; @@ -1527,11 +1523,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.emit(EngineEvent.Resumed); } - /** - * Applies the parts of a `ReconnectResponse` that describe the node we've resumed onto. - * Idempotent, so a duplicate or late response is - * harmless. - */ + /** Applies the node-describing half of a ReconnectResponse. Idempotent. */ private applyReconnectResponse(res: ReconnectResponse) { this.pcManager?.updateConfiguration(this.makeRTCConfiguration(res)); if (this.latestJoinResponse) { @@ -1543,12 +1535,12 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit if (!res?.lastMessageSeq) { return; } - this.resendReliableMessagesForResume(res.lastMessageSeq).catch((error) => { + 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) { From b9c0fe46cd43e2f830cac1a71b1560deef4aaf04 Mon Sep 17 00:00:00 2001 From: lukasIO Date: Tue, 1 Sep 2026 17:43:43 +0200 Subject: [PATCH 3/5] Fix late reconnect response support --- .changeset/dark-trains-tie.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/dark-trains-tie.md 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 From 31f3f3bd6eca2e0b240990315f3d42b8f26280a7 Mon Sep 17 00:00:00 2001 From: lukasIO Date: Tue, 1 Sep 2026 18:17:35 +0200 Subject: [PATCH 4/5] address comment --- src/room/PCTransportManager.ts | 4 ++- src/room/RTCEngine.test.ts | 50 ++++++++++++++++++++++++++++++++-- src/room/RTCEngine.ts | 11 +++++--- 3 files changed, 58 insertions(+), 7 deletions(-) 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 df41fd0a82..bfc0765f64 100644 --- a/src/room/RTCEngine.test.ts +++ b/src/room/RTCEngine.test.ts @@ -978,7 +978,7 @@ describe('RTCEngine', () => { } it('applies ICE servers, serverInfo and the replay when it arrives outside a resume', () => { - const { engine, internals, updateConfiguration, resend } = primeEngine(); + const { internals, updateConfiguration, resend } = primeEngine(); internals.setupSignalClientCallbacks(); internals.client.onLateReconnectResponse(makeResponse(7)); @@ -989,7 +989,6 @@ describe('RTCEngine', () => { ]); expect(internals.latestJoinResponse.serverInfo?.nodeId).toBe('new-node'); expect(resend).toHaveBeenCalledWith(7); - expect(engine).toBeDefined(); }); it('defers the replay to the in-flight resume, which runs it once the transport is back', async () => { @@ -1022,5 +1021,52 @@ describe('RTCEngine', () => { 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 3944daa66d..f6d7d4badd 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -749,8 +749,11 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit ...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. - this.applyReconnectResponse(res); + // 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 @@ -1524,8 +1527,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } /** Applies the node-describing half of a ReconnectResponse. Idempotent. */ - private applyReconnectResponse(res: ReconnectResponse) { - this.pcManager?.updateConfiguration(this.makeRTCConfiguration(res)); + private applyReconnectResponse(res: ReconnectResponse, iceRestart = false) { + this.pcManager?.updateConfiguration(this.makeRTCConfiguration(res), iceRestart); if (this.latestJoinResponse) { this.latestJoinResponse.serverInfo = res.serverInfo; } From 6746683e3082c7957c2ca93c14c50f1982e0ea33 Mon Sep 17 00:00:00 2001 From: lukasIO Date: Tue, 1 Sep 2026 18:18:38 +0200 Subject: [PATCH 5/5] address comment --- src/room/RTCEngine.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index f6d7d4badd..4f6f3ec9e0 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -1535,7 +1535,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } private replayReliableMessages(res: ReconnectResponse | undefined) { - if (!res?.lastMessageSeq) { + if (res?.lastMessageSeq === undefined) { return; } this.resendReliableMessagesForResume(res.lastMessageSeq).catch((error) =>