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 08c1f8b307..bcdb2384ac 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,167 @@ 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(); + } + }); + it('does not validate the url when aborted while the websocket is opening', async () => { + const abortController = new AbortController(); + const internals = signalClient as unknown as { connectionLock: Mutex }; + const opened = new Future(); + opened.promise.catch(() => {}); + mockWebSocketStream({ opened: opened.promise, readyState: 0 }); + vi.mocked(global.fetch).mockClear(); + + const joining = signalClient.join( + 'wss://test.livekit.io', + 'test-token', + defaultOptions, + abortController.signal, + ); + const cancelled = expect(joining).rejects.toMatchObject({ + reason: ConnectionErrorReason.Cancelled, + }); + await vi.waitFor(() => expect(WebSocketStream).toHaveBeenCalled()); + abortController.abort(); + opened.reject?.(new Error('closed by abort')); + await cancelled; + // let the attempt's continuation run to completion + const unlock = await internals.connectionLock.lock(); + unlock(); + + try { + expect(global.fetch).not.toHaveBeenCalled(); + } finally { + await signalClient.close(); + vi.mocked(WebSocketStream).mockReset(); + } + }); + + it('does not report a join that arrives after the attempt was aborted', async () => { + const abortController = new AbortController(); + const internals = signalClient as unknown as { connectionLock: Mutex }; + let enqueue: ((chunk: ArrayBuffer) => void) | undefined; + const readable = new ReadableStream({ + start(controller) { + enqueue = (chunk) => controller.enqueue(chunk); + }, + }); + const writerReady = new Future(); + const writable = new WritableStream(); + const getWriter = writable.getWriter.bind(writable); + writable.getWriter = () => { + const writer = getWriter(); + writerReady.resolve?.(); + return writer; + }; + mockWebSocketStream({ + connection: { readable, writable, protocol: '', extensions: '' }, + }); + const onJoined = vi.fn(); + signalClient.onJoined = onJoined; + + const joining = signalClient.join( + 'wss://test.livekit.io', + 'test-token', + defaultOptions, + abortController.signal, + ); + const cancelled = expect(joining).rejects.toMatchObject({ + reason: ConnectionErrorReason.Cancelled, + }); + await writerReady.promise; + abortController.abort(); + enqueue?.( + createSignalResponse('join', createJoinResponse()).toBinary().buffer as ArrayBuffer, + ); + await cancelled; + const unlock = await internals.connectionLock.lock(); + unlock(); + + try { + expect(onJoined).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(); @@ -926,7 +1089,8 @@ describe('SignalClient.connect', () => { it('does not let a timed-out attempt block the next one', async () => { // A validate request on a dead pooled connection only settles once its abort signal fires. - // Without a deadline it held the connection lock, so the next attempt never started. + // Without a deadline it held the connection lock, so the next attempt never started. A + // timed-out attempt now skips validation entirely, since its caller has already been rejected. vi.useFakeTimers(); try { const hangingFetch = (_url: unknown, init?: RequestInit) => @@ -953,8 +1117,12 @@ describe('SignalClient.connect', () => { await vi.waitFor(() => expect(WebSocketStream).toHaveBeenCalledTimes(2)); await vi.advanceTimersByTimeAsync(defaultOptions.websocketTimeout); await second; + + expect(fetch).not.toHaveBeenCalled(); } finally { vi.useRealTimers(); + // drop the unused hanging fetches so they don't leak into later tests + vi.mocked(fetch).mockReset(); } }); }); diff --git a/src/api/SignalClient.ts b/src/api/SignalClient.ts index f210733856..905bd0f1c1 100644 --- a/src/api/SignalClient.ts +++ b/src/api/SignalClient.ts @@ -453,6 +453,12 @@ export class SignalClient { throw e; } + if (abortSignal?.aborted) { + unlock(); + this.log.debug('signal connect abandoned before opening a transport'); + throw ConnectionError.cancelled('Connection aborted'); + } + return new Promise(async (resolve, reject) => { try { let alreadyAborted = false; @@ -507,6 +513,10 @@ 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) { + this.log.debug('signal connect abandoned after tearing down the previous transport'); + return; + } // the transport created below belongs to this attempt; events arriving from it after a // newer attempt has started are dropped by the machine @@ -571,6 +581,12 @@ export class SignalClient { } }); const connection = await this.ws.opened.catch(async (reason: unknown) => { + if (alreadyAborted) { + // the abort handler has already rejected this attempt; don't probe the validate + // endpoint for a failure nobody is waiting on + this.log.debug('signal connect abandoned while the websocket was opening'); + return; + } if (this.lifecycleState !== 'connected') { // claimed synchronously, before the await below: the socket's close event is already // on its way and would otherwise reject first with a less useful error @@ -647,6 +663,13 @@ export class SignalClient { return; } signalReader.releaseLock(); + if (alreadyAborted) { + // The abort handler has already rejected this attempt and is closing the transport. + // Don't act on a join that raced it: `onJoined` would publish an abandoned session's + // state to the room. + this.log.debug('signal connect abandoned while waiting for the first message'); + return; + } // Validate the first message const validation = this.validateFirstMessage( diff --git a/src/room/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index fd0bbc5a58..825ad40ec5 100644 --- a/src/room/RTCEngine.test.ts +++ b/src/room/RTCEngine.test.ts @@ -1,19 +1,22 @@ 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 DefaultReconnectPolicy from './DefaultReconnectPolicy'; import { PCTransportState } from './PCTransportManager'; import RTCEngine, { DataChannelKind } from './RTCEngine'; import type { ReconnectContext, ReconnectPolicy } from './ReconnectPolicy'; import { roomOptionDefaults } from './defaults'; -import { PublishDataError, UnexpectedConnectionState } from './errors'; +import { ConnectionError, PublishDataError, UnexpectedConnectionState } from './errors'; import { EngineEvent } from './events'; import CriticalTimers from './timers'; +import { Future } from './utils'; describe('RTCEngine', () => { const originalRTCRtpSender = window.RTCRtpSender; @@ -741,6 +744,123 @@ describe('RTCEngine', () => { }); }); + describe('closing during a full reconnect', () => { + interface ReconnectInternals { + _isClosed: boolean; + url: string; + token: string; + signalOpts: SignalOptions; + attemptReconnect: () => Promise; + configure: () => Promise; + waitForPCReconnected: () => Promise; + reconnectAbortController?: AbortController; + } + + 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; + const controller = internals.reconnectAbortController; + expect(controller).toBeDefined(); + await engine.close(); + expect(controller?.signal.aborted).toBe(true); + expect(internals.reconnectAbortController).toBeUndefined(); + finishCleanup.resolve?.(); + await reconnect; + + expect(internals.reconnectAbortController).toBeUndefined(); + 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 @@ -751,7 +871,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 d2424b664e..e71dab0ee4 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; @@ -401,6 +403,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit useV0Path, offerProto, ); + if (abortSignal?.aborted) { + this.log.info('join cancelled after the signal join response'); + throw ConnectionError.cancelled('Connection aborted'); + } this._isClosed = false; this.latestJoinResponse = joinResponse; this.participantSid = joinResponse.participant?.sid; @@ -414,6 +420,10 @@ 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) { + this.log.info('join cancelled after configuring peer connections'); + 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 @@ -478,6 +488,8 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit * having given up on reconnecting. */ async close(reason?: string) { + this.reconnectAbortController?.abort(); + this.reconnectAbortController = undefined; const unlock = await this.closingLock.lock(); if (this.isClosed) { unlock(); @@ -1377,16 +1389,26 @@ 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) { + this.log.info('reconnect attempt cancelled, not marking it as succeeded'); + return; + } this.clearPendingReconnect(); succeeded = true; } catch (e) { + if (abortController.signal.aborted) { + this.log.info('reconnect attempt cancelled, not retrying', { error: e }); + return; + } this.reconnectAttempts += 1; let recoverable = true; if (e instanceof UnexpectedConnectionState) { @@ -1415,6 +1437,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit ); } } finally { + // close() may already have cleared this (or a later attempt replaced it); only release our own + if (this.reconnectAbortController === abortController) { + this.reconnectAbortController = undefined; + } this.attemptingReconnect = false; // A full reconnect requested while this attempt was running (e.g. a `RECONNECT` leave @@ -1438,8 +1464,12 @@ 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) { + this.log.info('full reconnect cancelled before starting', { regionUrl }); + return; + } if (!this.url || !this.token) { // permanent failure, don't attempt reconnection throw new UnexpectedConnectionState('could not reconnect, url or token not saved'); @@ -1453,6 +1483,12 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } await this.cleanupPeerConnections(); await this.cleanupClient(); + if (abortSignal.aborted) { + this.log.info('full reconnect cancelled after cleaning up the previous session', { + regionUrl, + }); + return; + } let joinResponse: JoinResponse; try { @@ -1466,7 +1502,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit regionUrl ?? this.url, this.token, this.signalOpts, - undefined, + abortSignal, !this.options.singlePeerConnection, ) ).joinResponse; @@ -1476,6 +1512,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } throw new SignalReconnectError(); } + if (abortSignal.aborted) { + this.log.info('full reconnect cancelled after the signal join response', { regionUrl }); + return; + } if (this.shouldFailNext) { this.shouldFailNext = false; @@ -1486,6 +1526,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.emit(EngineEvent.SignalRestarted, joinResponse); await this.waitForPCReconnected(); + if (abortSignal.aborted) { + this.log.info('full reconnect cancelled after peer connections reconnected', { regionUrl }); + return; + } // re-check signal connection state before setting engine as resumed if (this.client.currentState !== SignalConnectionState.CONNECTED) { @@ -1496,9 +1540,17 @@ 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) { + this.log.info('full reconnect cancelled after a failed attempt', { regionUrl }); + return; + } + const nextRegionUrl = await this.regionStrategy?.getNextUrl(abortSignal); + if (abortSignal.aborted) { + this.log.info('full reconnect cancelled during region selection', { regionUrl }); + 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)