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,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<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();
+      }
+    });
+    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<WebSocketConnection, Error>();
+      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<ArrayBuffer>({
+        start(controller) {
+          enqueue = (chunk) => controller.enqueue(chunk);
+        },
+      });
+      const writerReady = new Future<void, Error>();
+      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();
       }
     });
   });

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