fix: cancel in-flight full reconnects when the engine closes

.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.

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:

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<void>;
+      };
+      const teardownStarted = new Future<void, Error>();
+      const finishTeardown = new Future<void, Error>();
+      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();

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<JoinResponse | ReconnectResponse | undefined>(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

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<void>;
+      configure: () => Promise<void>;
+      waitForPCReconnected: () => Promise<void>;
+    }
+
+    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<void, Error>();
+      const finishCleanup = new Future<void, Error>();
+      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<void, Error>();
+      const nextRegion = new Future<string | null, Error>();
+      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<void, Error>();
+      const joinResponse = new Future<JoinResponse, Error>();
+      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<void>;
-      restartConnection: (regionUrl?: string) => Promise<void>;
+      restartConnection: (abortSignal: AbortSignal, regionUrl?: string) => Promise<void>;
       clearPendingReconnect: () => void;
       handleDisconnect: (connection: string, reason?: number) => void;
       attemptReconnect: (reason?: number) => Promise<void>;