From 9f798801d406301835d1a5c68b440d07500f33c4 Mon Sep 17 00:00:00 2001 From: Zan Horvat Date: Fri, 11 Sep 2026 16:07:00 +0200 Subject: [PATCH 1/6] fix: cancel in-flight full reconnects when the engine closes --- .changeset/cancel-pending-reconnect.md | 5 ++ AGENTS.md | 5 ++ src/api/SignalClient.test.ts | 82 +++++++++++++++++ src/api/SignalClient.ts | 6 ++ src/room/RTCEngine.test.ts | 118 ++++++++++++++++++++++++- src/room/RTCEngine.ts | 30 +++++-- 6 files changed, 239 insertions(+), 7 deletions(-) create mode 100644 .changeset/cancel-pending-reconnect.md 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..a48350eb5a 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 f210733856..cb1a79d993 100644 --- a/src/api/SignalClient.ts +++ b/src/api/SignalClient.ts @@ -453,6 +453,11 @@ export class SignalClient { throw e; } + if (abortSignal?.aborted) { + unlock(); + throw ConnectionError.cancelled('Connection aborted'); + } + return new Promise(async (resolve, reject) => { try { let alreadyAborted = false; @@ -507,6 +512,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 fd0bbc5a58..3cc575ded8 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,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 @@ -751,7 +865,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..e85c739570 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,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; @@ -414,6 +419,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 @@ -478,6 +486,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(); @@ -1377,16 +1386,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) { @@ -1415,6 +1428,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 @@ -1438,8 +1452,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'); @@ -1453,6 +1468,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } await this.cleanupPeerConnections(); await this.cleanupClient(); + if (abortSignal.aborted) return; let joinResponse: JoinResponse; try { @@ -1466,7 +1482,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 +1492,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } throw new SignalReconnectError(); } + if (abortSignal.aborted) return; if (this.shouldFailNext) { this.shouldFailNext = false; @@ -1486,6 +1503,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) { @@ -1496,9 +1514,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) From ae6efa1d4be2110d4c4a91a3b1560eed9e5c902a Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 15 Oct 2026 11:58:18 +1300 Subject: [PATCH 2/6] fix: release the reconnect abort controller when the engine closes close() now clears `reconnectAbortController` after aborting it, so the field's presence reliably means a full reconnect is in flight. The attempt's own finally only clears the field when it still holds that attempt's controller, so it can't clobber a newer attempt's. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/room/RTCEngine.test.ts | 6 ++++++ src/room/RTCEngine.ts | 6 +++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/src/room/RTCEngine.test.ts b/src/room/RTCEngine.test.ts index 3cc575ded8..825ad40ec5 100644 --- a/src/room/RTCEngine.test.ts +++ b/src/room/RTCEngine.test.ts @@ -753,6 +753,7 @@ describe('RTCEngine', () => { attemptReconnect: () => Promise; configure: () => Promise; waitForPCReconnected: () => Promise; + reconnectAbortController?: AbortController; } function primeEngine() { @@ -782,10 +783,15 @@ describe('RTCEngine', () => { 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); }); diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index e85c739570..45c8af4c56 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -487,6 +487,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit */ async close(reason?: string) { this.reconnectAbortController?.abort(); + this.reconnectAbortController = undefined; const unlock = await this.closingLock.lock(); if (this.isClosed) { unlock(); @@ -1428,7 +1429,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit ); } } finally { - this.reconnectAbortController = undefined; + // 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 From 3f34cd28fc2f5a78eded604d7644f1e5b9d51601 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 15 Oct 2026 11:59:15 +1300 Subject: [PATCH 3/6] fix: stop an aborted signal connect from running on after its awaits The abort listener rejects the join and closes the transport, but the connect continuation kept running once its pending await settled: a failed upgrade still probed the validate endpoint, and a join that raced the abort still fired `onJoined`, publishing the abandoned session's participants to the room. Bail out after each of those awaits, and log each early exit so a cancelled connect is visible when debugging. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/api/SignalClient.test.ts | 81 ++++++++++++++++++++++++++++++++++++ src/api/SignalClient.ts | 19 ++++++++- 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/src/api/SignalClient.test.ts b/src/api/SignalClient.test.ts index a48350eb5a..b826b9aba1 100644 --- a/src/api/SignalClient.test.ts +++ b/src/api/SignalClient.test.ts @@ -227,6 +227,87 @@ describe('SignalClient.connect', () => { 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', () => { diff --git a/src/api/SignalClient.ts b/src/api/SignalClient.ts index cb1a79d993..905bd0f1c1 100644 --- a/src/api/SignalClient.ts +++ b/src/api/SignalClient.ts @@ -455,6 +455,7 @@ export class SignalClient { if (abortSignal?.aborted) { unlock(); + this.log.debug('signal connect abandoned before opening a transport'); throw ConnectionError.cancelled('Connection aborted'); } @@ -512,7 +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) return; + 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 @@ -577,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 @@ -653,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( From f25cbba30affbf5414e9870b4f1126d5e4674912 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 15 Oct 2026 12:00:05 +1300 Subject: [PATCH 4/6] chore: log each cancelled reconnect and join bail-out Every early exit taken because the engine closed mid-attempt now logs which stage it bailed out at, so a connection failure can be traced to a deliberate cancellation rather than looking like a silent stop. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/room/RTCEngine.ts | 32 ++++++++++++++++++++++++-------- 1 file changed, 24 insertions(+), 8 deletions(-) diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index 45c8af4c56..31d4e93498 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -404,6 +404,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit offerProto, ); if (abortSignal?.aborted) { + this.log.info('join cancelled after the signal join response'); throw ConnectionError.cancelled('Connection aborted'); } this._isClosed = false; @@ -420,6 +421,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit // 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) { @@ -1396,11 +1398,17 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } else { await this.resumeConnection(reason); } - if (abortController.signal.aborted) return; + 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) return; + 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) { @@ -1457,8 +1465,16 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } private async restartConnection(abortSignal: AbortSignal, regionUrl?: string) { + // Each bail-out names its stage so a cancelled reconnect can be traced in the logs. + const cancelled = (stage: string) => { + if (!abortSignal.aborted) { + return false; + } + this.log.info(`full reconnect cancelled ${stage}`, { regionUrl }); + return true; + }; try { - if (abortSignal.aborted) return; + if (cancelled('before starting')) return; if (!this.url || !this.token) { // permanent failure, don't attempt reconnection throw new UnexpectedConnectionState('could not reconnect, url or token not saved'); @@ -1472,7 +1488,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } await this.cleanupPeerConnections(); await this.cleanupClient(); - if (abortSignal.aborted) return; + if (cancelled('after cleaning up the previous session')) return; let joinResponse: JoinResponse; try { @@ -1496,7 +1512,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } throw new SignalReconnectError(); } - if (abortSignal.aborted) return; + if (cancelled('after the signal join response')) return; if (this.shouldFailNext) { this.shouldFailNext = false; @@ -1507,7 +1523,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.emit(EngineEvent.SignalRestarted, joinResponse); await this.waitForPCReconnected(); - if (abortSignal.aborted) return; + if (cancelled('after peer connections reconnected')) return; // re-check signal connection state before setting engine as resumed if (this.client.currentState !== SignalConnectionState.CONNECTED) { @@ -1518,9 +1534,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit // reconnect success this.emit(EngineEvent.Restarted); } catch (error) { - if (abortSignal.aborted) return; + if (cancelled('after a failed attempt')) return; const nextRegionUrl = await this.regionStrategy?.getNextUrl(abortSignal); - if (abortSignal.aborted) return; + if (cancelled('during region selection')) return; if (nextRegionUrl) { await this.restartConnection(abortSignal, nextRegionUrl); return; From 609885a0b3ea91db321e517324412281ce47f110 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 15 Oct 2026 12:08:38 +1300 Subject: [PATCH 5/6] refactor: inline the reconnect cancellation checks Co-Authored-By: Claude Opus 5.5 (1M context) --- src/room/RTCEngine.ts | 40 ++++++++++++++++++++++++++-------------- 1 file changed, 26 insertions(+), 14 deletions(-) diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index 31d4e93498..e71dab0ee4 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -1465,16 +1465,11 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } private async restartConnection(abortSignal: AbortSignal, regionUrl?: string) { - // Each bail-out names its stage so a cancelled reconnect can be traced in the logs. - const cancelled = (stage: string) => { - if (!abortSignal.aborted) { - return false; - } - this.log.info(`full reconnect cancelled ${stage}`, { regionUrl }); - return true; - }; try { - if (cancelled('before starting')) return; + 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'); @@ -1488,7 +1483,12 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } await this.cleanupPeerConnections(); await this.cleanupClient(); - if (cancelled('after cleaning up the previous session')) return; + if (abortSignal.aborted) { + this.log.info('full reconnect cancelled after cleaning up the previous session', { + regionUrl, + }); + return; + } let joinResponse: JoinResponse; try { @@ -1512,7 +1512,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } throw new SignalReconnectError(); } - if (cancelled('after the signal join response')) return; + if (abortSignal.aborted) { + this.log.info('full reconnect cancelled after the signal join response', { regionUrl }); + return; + } if (this.shouldFailNext) { this.shouldFailNext = false; @@ -1523,7 +1526,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.emit(EngineEvent.SignalRestarted, joinResponse); await this.waitForPCReconnected(); - if (cancelled('after peer connections reconnected')) return; + 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) { @@ -1534,9 +1540,15 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit // reconnect success this.emit(EngineEvent.Restarted); } catch (error) { - if (cancelled('after a failed attempt')) return; + if (abortSignal.aborted) { + this.log.info('full reconnect cancelled after a failed attempt', { regionUrl }); + return; + } const nextRegionUrl = await this.regionStrategy?.getNextUrl(abortSignal); - if (cancelled('during region selection')) return; + if (abortSignal.aborted) { + this.log.info('full reconnect cancelled during region selection', { regionUrl }); + return; + } if (nextRegionUrl) { await this.restartConnection(abortSignal, nextRegionUrl); return; From 4caf3a344d6ae14a13865aeeebcd22893fd3a585 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 15 Oct 2026 12:11:56 +1300 Subject: [PATCH 6/6] test: a timed-out signal connect no longer validates the url Since an abandoned connect skips the validate request, the hanging- validate regression test from #2148 no longer consumes its queued fetch mocks, which leaked into the handleConnectionError tests. Assert that no validation happens and reset the mocks afterwards. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/api/SignalClient.test.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/api/SignalClient.test.ts b/src/api/SignalClient.test.ts index b826b9aba1..bcdb2384ac 100644 --- a/src/api/SignalClient.test.ts +++ b/src/api/SignalClient.test.ts @@ -1089,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) => @@ -1116,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(); } }); });