diff --git a/.changeset/cancel-pending-reconnect.md b/.changeset/cancel-pending-reconnect.md new file mode 100644 index 0000000000..7e9014c6fa --- /dev/null +++ b/.changeset/cancel-pending-reconnect.md @@ -0,0 +1,5 @@ +--- +'livekit-client': patch +--- + +Cancel in-flight full reconnects when the engine closes, preventing delayed cleanup, region failover, or join responses from reopening a disconnected session. Honor cancellation before opening a signal transport. diff --git a/AGENTS.md b/AGENTS.md index 9dc3928dd9..d0ef5e9949 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -19,6 +19,11 @@ the primary way to react to state changes — tracks published, participants joi Important: event handlers should be registered **before** calling `room.connect()` because some events (like `DataTrackPublished`) fire during the connection handshake. +When changing full reconnects, carry the attempt's `AbortSignal` through joins and region +selection. Cancellation must cover continuations after awaited cleanup and late join responses: +engine close removes room listeners, so room-event guards cannot catch those continuations. +Regression tests live in `RTCEngine.test.ts` and `SignalClient.test.ts`. + ### Data transport APIs LiveKit has four data transport mechanisms, each suited to different patterns: diff --git a/src/api/SignalClient.test.ts b/src/api/SignalClient.test.ts index 662898f18c..72794d69ee 100644 --- a/src/api/SignalClient.test.ts +++ b/src/api/SignalClient.test.ts @@ -11,8 +11,10 @@ import { WrappedJoinRequest_Compression, } from '@livekit/protocol'; import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { Mutex } from '@livekit/mutex'; import { ConnectionError, ConnectionErrorReason } from '../room/errors'; import CriticalTimers from '../room/timers'; +import { Future } from '../room/utils'; import { SignalClient, SignalConnectionState } from './SignalClient'; import type { WebSocketCloseInfo, WebSocketConnection } from './WebSocketStream'; import { WebSocketStream } from './WebSocketStream'; @@ -147,6 +149,86 @@ describe('SignalClient.connect', () => { return JoinRequest.fromBinary(bytes); } + describe('Cancelled connection attempts', () => { + it.each(['before join', 'while waiting for the connection lock'])( + 'does not open a transport when aborted %s', + async (when) => { + const abortController = new AbortController(); + const internals = signalClient as unknown as { connectionLock: Mutex }; + const unlock = await internals.connectionLock.lock(); + mockWebSocketStream({ + connection: createMockConnection( + createMockReadableStream([createSignalResponse('join', createJoinResponse())]), + ), + }); + if (when === 'before join') abortController.abort(); + const joining = signalClient.join( + 'wss://test.livekit.io', + 'test-token', + defaultOptions, + abortController.signal, + ); + abortController.abort(); + unlock(); + + try { + await expect(joining).rejects.toMatchObject({ reason: ConnectionErrorReason.Cancelled }); + expect(WebSocketStream).not.toHaveBeenCalled(); + } finally { + await signalClient.close(); + vi.mocked(WebSocketStream).mockReset(); + } + }, + ); + + it('does not replace a transport when aborted during its teardown', async () => { + const abortController = new AbortController(); + const internals = signalClient as unknown as { + connectionLock: Mutex; + teardownTransport: (reason: string) => Promise; + }; + const teardownStarted = new Future(); + const finishTeardown = new Future(); + vi.spyOn(internals, 'teardownTransport').mockImplementationOnce(async () => { + teardownStarted.resolve?.(); + await finishTeardown.promise; + }); + signalClient.ws = { + close: () => {}, + closed: Promise.resolve({ closeCode: 1000, reason: '' }), + } as WebSocketStream; + mockWebSocketStream({ + connection: createMockConnection( + createMockReadableStream([createSignalResponse('join', createJoinResponse())]), + ), + }); + + const joining = signalClient.join( + 'wss://test.livekit.io', + 'test-token', + defaultOptions, + abortController.signal, + ); + const cancelled = expect(joining).rejects.toMatchObject({ + reason: ConnectionErrorReason.Cancelled, + }); + await teardownStarted.promise; + abortController.abort(); + finishTeardown.resolve?.(); + await cancelled; + // The rejection can settle before the transport replacement resumes. + const unlock = await internals.connectionLock.lock(); + unlock(); + + try { + expect(WebSocketStream).not.toHaveBeenCalled(); + } finally { + await signalClient.close(); + vi.mocked(WebSocketStream).mockReset(); + } + }); + }); + describe('Happy Path - Initial Join', () => { it('should successfully connect and receive join response', async () => { const joinResponse = createJoinResponse(); diff --git a/src/api/SignalClient.ts b/src/api/SignalClient.ts index b143e07545..535437449a 100644 --- a/src/api/SignalClient.ts +++ b/src/api/SignalClient.ts @@ -443,6 +443,11 @@ export class SignalClient { const rtcUrl = createRtcUrl(url, params, useV0Path).toString(); const validateUrl = createValidateUrl(rtcUrl).toString(); + if (abortSignal?.aborted) { + unlock(); + throw ConnectionError.cancelled('Connection aborted'); + } + return new Promise(async (resolve, reject) => { try { let alreadyAborted = false; @@ -497,6 +502,7 @@ export class SignalClient { await this.teardownTransport('replaced by a new connection attempt'); this.log.debug(`closed previous ws connection in ${performance.now() - startClose}ms`); } + if (alreadyAborted) return; // the transport created below belongs to this attempt; events arriving from it after a // newer attempt has started are dropped by the machine diff --git a/src/room/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index e8be48da8f..be805362fa 100644 --- a/src/room/RTCEngine.test.ts +++ b/src/room/RTCEngine.test.ts @@ -1,16 +1,19 @@ import { DataPacket, DataPacket_Kind, + JoinResponse, ConnectionQuality as ProtoConnectionQuality, UserPacket, } from '@livekit/protocol'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { SignalConnectionState, type SignalOptions } from '../api/SignalClient'; import type { DataPacketBuffer } from '../utils/dataPacketBuffer'; import { PCTransportState } from './PCTransportManager'; import RTCEngine, { DataChannelKind } from './RTCEngine'; import { roomOptionDefaults } from './defaults'; -import { PublishDataError, UnexpectedConnectionState } from './errors'; +import { ConnectionError, PublishDataError, UnexpectedConnectionState } from './errors'; import { EngineEvent } from './events'; +import { Future } from './utils'; describe('RTCEngine', () => { const originalRTCRtpSender = window.RTCRtpSender; @@ -701,6 +704,117 @@ describe('RTCEngine', () => { }); }); + describe('closing during a full reconnect', () => { + interface ReconnectInternals { + _isClosed: boolean; + url: string; + token: string; + signalOpts: SignalOptions; + attemptReconnect: () => Promise; + configure: () => Promise; + waitForPCReconnected: () => Promise; + } + + function primeEngine() { + const engine = new RTCEngine({ ...roomOptionDefaults, singlePeerConnection: false }); + const internals = engine as unknown as ReconnectInternals; + internals._isClosed = false; + internals.url = 'wss://test.livekit.io'; + internals.token = 'test-token'; + internals.signalOpts = { autoSubscribe: true, maxRetries: 1, websocketTimeout: 15_000 }; + engine.fullReconnectOnNext = true; + const configure = vi.spyOn(internals, 'configure').mockResolvedValue(); + vi.spyOn(internals, 'waitForPCReconnected').mockResolvedValue(); + const join = vi + .spyOn(engine.client, 'join') + .mockResolvedValue(new JoinResponse({ subscriberPrimary: true })); + return { engine, internals, configure, join }; + } + + it('does not join after close completes while reconnect cleanup is pending', async () => { + const { engine, internals, join } = primeEngine(); + const cleanupStarted = new Future(); + const finishCleanup = new Future(); + vi.spyOn(engine, 'cleanupPeerConnections').mockImplementationOnce(async () => { + cleanupStarted.resolve?.(); + await finishCleanup.promise; + }); + + const reconnect = internals.attemptReconnect(); + await cleanupStarted.promise; + await engine.close(); + finishCleanup.resolve?.(); + await reconnect; + + expect(join).not.toHaveBeenCalled(); + expect(engine.isClosed).toBe(true); + }); + + it('does not join another region after close completes during region selection', async () => { + const { engine, internals, join } = primeEngine(); + join.mockRejectedValue(ConnectionError.internal('signal connection failed')); + const selectingRegion = new Future(); + const nextRegion = new Future(); + const getNextUrl = vi.fn(async () => { + selectingRegion.resolve?.(); + return nextRegion.promise; + }); + engine.setRegionStrategy({ getNextUrl, resetAttempts: () => {} }); + + const reconnect = internals.attemptReconnect(); + await selectingRegion.promise; + await engine.close(); + nextRegion.resolve?.('wss://another-region.livekit.io'); + // A second selection must terminate even on the unfixed implementation. + getNextUrl.mockResolvedValue(null); + await reconnect; + + expect(join).toHaveBeenCalledTimes(1); + expect(getNextUrl).toHaveBeenCalledTimes(1); + expect(engine.isClosed).toBe(true); + }); + + it('does not reopen the engine when a signal join resolves after close', async () => { + const { engine, internals, configure, join } = primeEngine(); + const joining = new Future(); + const joinResponse = new Future(); + join.mockImplementationOnce(async () => { + joining.resolve?.(); + return joinResponse.promise; + }); + + const reconnect = internals.attemptReconnect(); + await joining.promise; + await engine.close(); + joinResponse.resolve?.(new JoinResponse({ subscriberPrimary: true })); + await reconnect; + + expect(engine.isClosed).toBe(true); + expect(configure).not.toHaveBeenCalled(); + expect(join.mock.calls[0][3]?.aborted).toBe(true); + }); + + it('still completes a full reconnect while the engine is active', async () => { + const { engine, internals, join } = primeEngine(); + vi.spyOn(engine.client, 'sendLeave').mockResolvedValue(); + vi.spyOn(engine.client, 'currentState', 'get').mockReturnValue( + SignalConnectionState.CONNECTED, + ); + const restarted = vi.fn(); + engine.on(EngineEvent.Restarted, restarted); + + try { + await internals.attemptReconnect(); + + expect(join).toHaveBeenCalledTimes(1); + expect(restarted).toHaveBeenCalledTimes(1); + expect(engine.isClosed).toBe(false); + } finally { + await engine.close(); + } + }); + }); + describe('reconnect requested mid-attempt', () => { // A full reconnect requested while a resume is already in flight (e.g. a server // RECONNECT leave racing the resume) sets `fullReconnectOnNext` mid-attempt. A @@ -711,7 +825,7 @@ describe('RTCEngine', () => { clientConfiguration: unknown; pcManager: unknown; resumeConnection: (reason?: number) => Promise; - restartConnection: (regionUrl?: string) => Promise; + restartConnection: (abortSignal: AbortSignal, regionUrl?: string) => Promise; clearPendingReconnect: () => void; handleDisconnect: (connection: string, reason?: number) => void; attemptReconnect: (reason?: number) => Promise; diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index fd0fc1fd61..41046a778a 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -227,6 +227,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit private attemptingReconnect: boolean = false; + private reconnectAbortController?: AbortController; + private reconnectPolicy: ReconnectPolicy; private reconnectTimeout?: ReturnType; @@ -392,6 +394,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit useV0Path, offerProto, ); + if (abortSignal?.aborted) { + throw ConnectionError.cancelled('Connection aborted'); + } this._isClosed = false; this.latestJoinResponse = joinResponse; this.participantSid = joinResponse.participant?.sid; @@ -405,6 +410,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit // the server's ICE servers and topology, then negotiate separately rather than bundling // the offer with the join. await this.configure(joinResponse, !useV0Path); + if (abortSignal?.aborted) { + throw ConnectionError.cancelled('Connection aborted'); + } if (!useV0Path) { // The V1 first offer must carry the media layout so Firefox binds receive decoders for // subscribed tracks — without it, subscribed audio/video arrive as RTP but @@ -463,6 +471,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit * having given up on reconnecting. */ async close(reason?: string) { + this.reconnectAbortController?.abort(); const unlock = await this.closingLock.lock(); if (this.isClosed) { unlock(); @@ -1335,16 +1344,20 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.fullReconnectOnNext = false; let succeeded = false; + const abortController = new AbortController(); + this.reconnectAbortController = abortController; try { this.attemptingReconnect = true; if (fullReconnect) { - await this.restartConnection(); + await this.restartConnection(abortController.signal); } else { await this.resumeConnection(reason); } + if (abortController.signal.aborted) return; this.clearPendingReconnect(); succeeded = true; } catch (e) { + if (abortController.signal.aborted) return; this.reconnectAttempts += 1; let recoverable = true; if (e instanceof UnexpectedConnectionState) { @@ -1373,6 +1386,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit ); } } finally { + this.reconnectAbortController = undefined; this.attemptingReconnect = false; // A full reconnect requested while this attempt was running (e.g. a `RECONNECT` leave @@ -1396,8 +1410,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit return null; } - private async restartConnection(regionUrl?: string) { + private async restartConnection(abortSignal: AbortSignal, regionUrl?: string) { try { + if (abortSignal.aborted) return; if (!this.url || !this.token) { // permanent failure, don't attempt reconnection throw new UnexpectedConnectionState('could not reconnect, url or token not saved'); @@ -1411,6 +1426,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } await this.cleanupPeerConnections(); await this.cleanupClient(); + if (abortSignal.aborted) return; let joinResponse: JoinResponse; try { @@ -1424,7 +1440,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit regionUrl ?? this.url, this.token, this.signalOpts, - undefined, + abortSignal, !this.options.singlePeerConnection, ) ).joinResponse; @@ -1434,6 +1450,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } throw new SignalReconnectError(); } + if (abortSignal.aborted) return; if (this.shouldFailNext) { this.shouldFailNext = false; @@ -1444,6 +1461,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.emit(EngineEvent.SignalRestarted, joinResponse); await this.waitForPCReconnected(); + if (abortSignal.aborted) return; // re-check signal connection state before setting engine as resumed if (this.client.currentState !== SignalConnectionState.CONNECTED) { @@ -1454,9 +1472,11 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit // reconnect success this.emit(EngineEvent.Restarted); } catch (error) { - const nextRegionUrl = await this.regionStrategy?.getNextUrl(); + if (abortSignal.aborted) return; + const nextRegionUrl = await this.regionStrategy?.getNextUrl(abortSignal); + if (abortSignal.aborted) return; if (nextRegionUrl) { - await this.restartConnection(nextRegionUrl); + await this.restartConnection(abortSignal, nextRegionUrl); return; } else { // no more regions to try (or we're not on cloud)