fix: keep attemptConnection on its own engine when disconnect races the join

.changeset/stale-connect-engine.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+If `room.disconnect()` runs during the signal join, the connection attempt now rejects as cancelled instead of with a `TypeError`. A stale connection attempt no longer closes the engine of a newer connection or disconnects the room.

src/room/CoreRoom.ts

@@ -771,6 +771,14 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
       // at this point the intention to connect has been signalled so we can allow cancelling of the connection via disconnect() again
       unlockDisconnect?.();
 
+      const failAttempt = (error: unknown, reason?: DisconnectReason) => {
+        // once a newer connect() replaced this attempt's controller, the room's state belongs to it
+        if (this.abortController === abortController) {
+          this.handleDisconnect(this.options.stopLocalTrackOnUnpublish, reason);
+        }
+        reject(error);
+      };
+
       try {
         await BackOffStrategy.getInstance().getBackOffPromise(url);
         if (abortController.signal.aborted) {
@@ -788,17 +796,14 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
           let nextUrl: string | null = null;
           try {
             this.log.debug('Fetching next region');
-            nextUrl = await this.regionUrlProvider.getNextBestRegionUrl(
-              this.abortController?.signal,
-            );
+            nextUrl = await this.regionUrlProvider.getNextBestRegionUrl(abortController.signal);
           } catch (regionFetchError) {
             if (
               regionFetchError instanceof ConnectionError &&
               (regionFetchError.status === 401 ||
                 regionFetchError.reason === ConnectionErrorReason.Cancelled)
             ) {
-              this.handleDisconnect(this.options.stopLocalTrackOnUnpublish);
-              reject(regionFetchError);
+              failAttempt(regionFetchError);
               return;
             }
           }
@@ -813,26 +818,21 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
             this.log.debug('Adding failed connection attempt to back off');
             BackOffStrategy.getInstance().addFailedConnectionAttempt(url);
           }
-          if (nextUrl && !this.abortController?.signal.aborted) {
+          if (nextUrl && !abortController.signal.aborted) {
             this.log.info(
               `Initial connection failed with ConnectionError: ${error.message}. Retrying with another region: ${nextUrl}`,
             );
             this.recreateEngine(true);
             await connectFn(resolve, reject, nextUrl);
           } else {
-            this.handleDisconnect(
-              this.options.stopLocalTrackOnUnpublish,
-              getDisconnectReasonFromConnectionError(error),
-            );
-            reject(error);
+            failAttempt(error, getDisconnectReasonFromConnectionError(error));
           }
         } else {
           let disconnectReason = DisconnectReason.UNKNOWN_REASON;
           if (error instanceof ConnectionError) {
             disconnectReason = getDisconnectReasonFromConnectionError(error);
           }
-          this.handleDisconnect(this.options.stopLocalTrackOnUnpublish, disconnectReason);
-          reject(error);
+          failAttempt(error, disconnectReason);
         }
       }
     };
@@ -935,8 +935,10 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
       // create engine if previously disconnected
       this.maybeCreateEngine();
     }
+    // a concurrent disconnect() or connect() can replace or clear `this.engine` while we await below
+    const engine = this.engine;
     if (this.regionUrlProvider?.isCloud()) {
-      this.engine.setRegionStrategy(this.createRegionStrategy());
+      engine.setRegionStrategy(this.createRegionStrategy());
     }
 
     this.acquireAudioContext(true).catch((error) =>
@@ -946,17 +948,17 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
     this.connOptions = { ...roomConnectOptionDefaults, ...opts } as InternalRoomConnectOptions;
 
     if (this.connOptions.rtcConfig) {
-      this.engine.rtcConfig = this.connOptions.rtcConfig;
+      engine.rtcConfig = this.connOptions.rtcConfig;
     }
     if (this.connOptions.peerConnectionTimeout) {
-      this.engine.peerConnectionTimeout = this.connOptions.peerConnectionTimeout;
+      engine.peerConnectionTimeout = this.connOptions.peerConnectionTimeout;
     }
 
     try {
       const joinResponse = await this.connectSignal(
         url,
         token,
-        this.engine,
+        engine,
         this.connOptions,
         this.options,
         abortController,
@@ -967,8 +969,8 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
       this.setupLocalParticipantEvents();
       this.emit(RoomEvent.SignalConnected);
     } catch (err) {
-      await this.engine.close();
-      this.recreateEngine();
+      await engine.close();
+      this.recreateEngine(false, engine);
 
       const resultingError = abortController.signal.aborted
         ? ConnectionError.cancelled('Signal connection aborted')
@@ -988,19 +990,19 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
     }
 
     if (abortController.signal.aborted) {
-      await this.engine.close();
-      this.recreateEngine();
+      await engine.close();
+      this.recreateEngine(false, engine);
       throw ConnectionError.cancelled(`Connection attempt aborted`);
     }
 
     try {
-      await this.engine.waitForPCInitialConnection(
+      await engine.waitForPCInitialConnection(
         this.connOptions.peerConnectionTimeout,
         abortController,
       );
     } catch (e) {
-      await this.engine.close();
-      this.recreateEngine();
+      await engine.close();
+      this.recreateEngine(false, engine);
       throw e;
     }
 
@@ -1501,14 +1503,16 @@ export class CoreRoom extends (EventEmitter as new () => TypedEmitter<RoomEventC
       );
   }
 
-  private recreateEngine(sendLeave?: boolean) {
-    const oldEngine = this.engine;
-
+  private recreateEngine(sendLeave?: boolean, oldEngine: RTCEngine | undefined = this.engine) {
     if (sendLeave && oldEngine && !oldEngine.client.isDisconnected) {
       oldEngine.client.sendLeave().finally(() => oldEngine.close());
     } else {
       oldEngine?.close();
     }
+    // the room already moved on to a newer engine (or none); leave that one alone
+    if (oldEngine !== this.engine) {
+      return;
+    }
     /* @ts-ignore */
     this.engine = undefined;
     this.isResuming = false;

src/room/Room.test.ts

@@ -11,8 +11,10 @@ import {
 import { afterEach, describe, expect, it, vi } from 'vitest';
 import MockMediaStreamTrack from '../test/MockMediaStreamTrack';
 import { stubMissingDeflateRawSupport } from '../test/compressionStreamStubs';
+import RTCEngine from './RTCEngine';
 import { ConnectionState, Room } from './Room';
 import { roomConnectOptionDefaults, roomOptionDefaults } from './defaults';
+import { ConnectionError, ConnectionErrorReason } from './errors';
 import { EngineEvent, ParticipantEvent, RoomEvent } from './events';
 import RemoteParticipant from './participant/RemoteParticipant';
 import RemoteTrackPublication from './track/RemoteTrackPublication';
@@ -463,3 +465,93 @@ describe('audio context ownership', () => {
     expect(created[0].closed).toBe(true);
   });
 });
+
+describe('connection attempt racing disconnect (#2150)', () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.unstubAllGlobals();
+  });
+
+  type AttemptConnection = (
+    url: string,
+    token: string,
+    opts: undefined,
+    abortController: AbortController,
+  ) => Promise<void>;
+
+  /** Starts a connection attempt whose signal join stays pending until `failJoin` is called. */
+  async function startAttempt(room: Room) {
+    let failJoin!: () => void;
+    vi.spyOn(RTCEngine.prototype, 'join').mockImplementationOnce(
+      () =>
+        new Promise((_, reject) => {
+          failJoin = () => reject(ConnectionError.cancelled('Connection aborted'));
+        }),
+    );
+    const engine = room.engine;
+    const abortController = new AbortController();
+    (room as unknown as { abortController: AbortController }).abortController = abortController;
+    room.state = ConnectionState.Connecting;
+    const attempt = (room as unknown as { attemptConnection: AttemptConnection }).attemptConnection(
+      'ws://localhost:7880',
+      'token',
+      undefined,
+      abortController,
+    );
+    await vi.waitFor(() => expect(failJoin).toBeDefined());
+    return { attempt, engine, failJoin };
+  }
+
+  it('rejects as cancelled when disconnect() clears the engine mid-join', async () => {
+    const room = new Room();
+    const { attempt, engine, failJoin } = await startAttempt(room);
+    const close = vi.spyOn(engine, 'close');
+
+    await room.disconnect();
+    expect(room.engine).toBeUndefined();
+    failJoin();
+
+    await expect(attempt).rejects.toMatchObject({ reason: ConnectionErrorReason.Cancelled });
+    expect(close).toHaveBeenCalled();
+  });
+
+  it('leaves a newer connect() alone when a stale join fails', async () => {
+    // connect() bails out unless the runtime looks like it supports WebRTC
+    vi.stubGlobal(
+      'RTCPeerConnection',
+      class {
+        addTrack() {}
+      },
+    );
+    let failStaleJoin!: () => void;
+    vi.spyOn(RTCEngine.prototype, 'join')
+      .mockImplementationOnce(
+        () =>
+          new Promise((_, reject) => {
+            failStaleJoin = () => reject(ConnectionError.cancelled('Connection aborted'));
+          }),
+      )
+      .mockImplementationOnce(() => new Promise(() => {}));
+    const room = new Room();
+
+    const stale = room.connect('ws://localhost:7880', 'token');
+    await vi.waitFor(() => expect(failStaleJoin).toBeDefined());
+    await room.disconnect();
+    await expect(stale).rejects.toMatchObject({ reason: ConnectionErrorReason.Cancelled });
+
+    room.connect('ws://localhost:7880', 'token').catch(() => {});
+    await vi.waitFor(() => expect(RTCEngine.prototype.join).toHaveBeenCalledTimes(2));
+    const newerEngine = room.engine;
+    const close = vi.spyOn(newerEngine, 'close');
+    const disconnected = vi.fn();
+    room.on(RoomEvent.Disconnected, disconnected);
+
+    failStaleJoin();
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    expect(room.state).toBe(ConnectionState.Connecting);
+    expect(room.engine).toBe(newerEngine);
+    expect(close).not.toHaveBeenCalled();
+    expect(disconnected).not.toHaveBeenCalled();
+  });
+});