Use mutex for disconnect calls (#525)

* use mutex for disconnection

* changeset

* remove buffer drain, as close call will do that already

* simplify closePromise

Author: lukasIO

.changeset/gentle-lizards-smile.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+use mutex to prevent simultaneous calls to disconnect

src/api/SignalClient.ts

@@ -30,7 +30,7 @@ import {
   UpdateTrackSettings,
 } from '../proto/livekit_rtc';
 import { ConnectionError, ConnectionErrorReason } from '../room/errors';
-import { getClientInfo, sleep } from '../room/utils';
+import { getClientInfo, Mutex, sleep } from '../room/utils';
 
 // internal options
 interface ConnectOpts {
@@ -139,12 +139,15 @@ export class SignalClient {
 
   private pingInterval: ReturnType<typeof setInterval> | undefined;
 
+  private closingLock: Mutex;
+
   constructor(useJSON: boolean = false) {
     this.isConnected = false;
     this.isReconnecting = false;
     this.useJSON = useJSON;
     this.requestQueue = new Queue();
     this.queuedRequests = [];
+    this.closingLock = new Mutex();
   }
 
   async join(
@@ -306,35 +309,32 @@ export class SignalClient {
   }
 
   async close() {
-    this.isConnected = false;
-    if (this.ws) {
-      this.ws.onclose = null;
-      this.ws.onmessage = null;
-      this.ws.onopen = null;
-
-      const emptyBufferPromise = new Promise(async (resolve) => {
-        while (this.ws && this.ws.bufferedAmount > 0) {
-          await sleep(50);
-        }
-        resolve(true);
-      });
-      // 250ms grace period for buffer to be cleared
-      await Promise.race([emptyBufferPromise, sleep(250)]);
-
-      let closeResolver: (args: any) => void;
-      const closePromise = new Promise((resolve) => {
-        closeResolver = resolve;
-      });
-
-      // calling `ws.close()` only starts the closing handshake (CLOSING state), prefer to wait until state is actually CLOSED
-      this.ws.onclose = () => closeResolver(true);
+    const unlock = await this.closingLock.lock();
+    try {
+      this.isConnected = false;
+      if (this.ws) {
+        this.ws.onclose = null;
+        this.ws.onmessage = null;
+        this.ws.onopen = null;
+
+        // calling `ws.close()` only starts the closing handshake (CLOSING state), prefer to wait until state is actually CLOSED
+        const closePromise = new Promise((resolve) => {
+          if (this.ws) {
+            this.ws.onclose = resolve;
+          } else {
+            resolve(true);
+          }
+        });
 
-      this.ws.close();
-      // 250ms grace period for ws to close gracefully
-      await Promise.race([closePromise, sleep(250)]);
+        this.ws.close();
+        // 250ms grace period for ws to close gracefully
+        await Promise.race([closePromise, sleep(250)]);
+      }
+      this.ws = undefined;
+      this.clearPingInterval();
+    } finally {
+      unlock();
     }
-    this.ws = undefined;
-    this.clearPingInterval();
   }
 
   // initial offer after joining

src/room/RTCEngine.ts

@@ -38,6 +38,7 @@ import type { TrackPublishOptions, VideoCodec } from './track/options';
 import { Track } from './track/Track';
 import {
   isWeb,
+  Mutex,
   sleep,
   supportsAddTrack,
   supportsSetCodecPreferences,
@@ -130,12 +131,15 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
   /** specifies how often an initial join connection is allowed to retry */
   private maxJoinAttempts: number = 1;
 
+  private closingLock: Mutex;
+
   constructor(private options: InternalRoomOptions) {
     super();
     this.client = new SignalClient();
     this.client.signalLatency = this.options.expSignalLatency;
     this.reconnectPolicy = this.options.reconnectPolicy;
     this.registerOnLineListener();
+    this.closingLock = new Mutex();
   }
 
   async join(
@@ -180,29 +184,34 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
   }
 
   async close() {
-    this._isClosed = true;
-    this.removeAllListeners();
-    this.deregisterOnLineListener();
-    this.clearPendingReconnect();
-    if (this.publisher && this.publisher.pc.signalingState !== 'closed') {
-      this.publisher.pc.getSenders().forEach((sender) => {
-        try {
-          // TODO: react-native-webrtc doesn't have removeTrack yet.
-          if (this.publisher?.pc.removeTrack) {
-            this.publisher?.pc.removeTrack(sender);
+    const unlock = await this.closingLock.lock();
+    try {
+      this._isClosed = true;
+      this.removeAllListeners();
+      this.deregisterOnLineListener();
+      this.clearPendingReconnect();
+      if (this.publisher && this.publisher.pc.signalingState !== 'closed') {
+        this.publisher.pc.getSenders().forEach((sender) => {
+          try {
+            // TODO: react-native-webrtc doesn't have removeTrack yet.
+            if (this.publisher?.pc.removeTrack) {
+              this.publisher?.pc.removeTrack(sender);
+            }
+          } catch (e) {
+            log.warn('could not removeTrack', { error: e });
           }
-        } catch (e) {
-          log.warn('could not removeTrack', { error: e });
-        }
-      });
-      this.publisher.close();
-      this.publisher = undefined;
-    }
-    if (this.subscriber) {
-      this.subscriber.close();
-      this.subscriber = undefined;
+        });
+        this.publisher.close();
+        this.publisher = undefined;
+      }
+      if (this.subscriber) {
+        this.subscriber.close();
+        this.subscriber = undefined;
+      }
+      await this.client.close();
+    } finally {
+      unlock();
     }
-    await this.client.close();
   }
 
   addTrack(req: AddTrackRequest): Promise<TrackInfo> {

src/room/Room.ts

@@ -50,7 +50,7 @@ import { Track } from './track/Track';
 import type { TrackPublication } from './track/TrackPublication';
 import type { AdaptiveStreamSettings } from './track/types';
 import { getNewAudioContext } from './track/utils';
-import { Future, isWeb, supportsSetSinkId, unpackStreamId } from './utils';
+import { Future, isWeb, Mutex, supportsSetSinkId, unpackStreamId } from './utils';
 
 export enum ConnectionState {
   Disconnected = 'disconnected',
@@ -118,6 +118,8 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
   /** future holding client initiated connection attempt */
   private connectFuture?: Future<void>;
 
+  private disconnectLock: Mutex;
+
   /**
    * Creates a new Room, the primary construct for a LiveKit session.
    * @param options
@@ -144,6 +146,8 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
 
     this.maybeCreateEngine();
 
+    this.disconnectLock = new Mutex();
+
     this.localParticipant = new LocalParticipant('', '', this.engine, this.options);
   }
 
@@ -394,26 +398,42 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
    * disconnects the room, emits [[RoomEvent.Disconnected]]
    */
   disconnect = async (stopTracks = true) => {
-    log.info('disconnect from room', { identity: this.localParticipant.identity });
-    if (this.state === ConnectionState.Connecting || this.state === ConnectionState.Reconnecting) {
-      // try aborting pending connection attempt
-      log.warn('abort connection attempt');
-      this.abortController?.abort();
-      // in case the abort controller didn't manage to cancel the connection attempt, reject the connect promise explicitly
-      this.connectFuture?.reject?.(new ConnectionError('Client initiated disconnect'));
-      this.connectFuture = undefined;
-    }
-    // send leave
-    if (this.engine?.client.isConnected) {
-      await this.engine.client.sendLeave();
-    }
-    // close engine (also closes client)
-    if (this.engine) {
-      await this.engine.close();
+    const unlock = await this.disconnectLock.lock();
+    try {
+      if (this.state === ConnectionState.Disconnected) {
+        log.debug('already disconnected');
+        return;
+      }
+      log.info('disconnect from room', { identity: this.localParticipant.identity });
+      if (
+        this.state === ConnectionState.Connecting ||
+        this.state === ConnectionState.Reconnecting
+      ) {
+        // try aborting pending connection attempt
+        log.warn('abort connection attempt');
+        this.abortController?.abort();
+        // in case the abort controller didn't manage to cancel the connection attempt, reject the connect promise explicitly
+        this.connectFuture?.reject?.(new ConnectionError('Client initiated disconnect'));
+        this.connectFuture = undefined;
+      }
+      // send leave
+      if (this.engine?.client.isConnected) {
+        await this.engine.client.sendLeave();
+      }
+      // close engine (also closes client)
+      if (this.engine) {
+        log.info(`closing engine ${this.localParticipant.identity}: ${this.localParticipant.sid}`);
+        await this.engine.close();
+        log.info(
+          `successfully closed engine ${this.localParticipant.identity}: ${this.localParticipant.sid}`,
+        );
+      }
+      this.handleDisconnect(stopTracks, DisconnectReason.CLIENT_INITIATED);
+      /* @ts-ignore */
+      this.engine = undefined;
+    } finally {
+      unlock();
     }
-    this.handleDisconnect(stopTracks, DisconnectReason.CLIENT_INITIATED);
-    /* @ts-ignore */
-    this.engine = undefined;
   };
 
   /**

src/room/utils.ts

@@ -320,3 +320,38 @@ export function createAudioAnalyser(
 
   return { calculateVolume, analyser, cleanup };
 }
+
+export class Mutex {
+  private _locking: Promise<void>;
+
+  private _locks: number;
+
+  constructor() {
+    this._locking = Promise.resolve();
+    this._locks = 0;
+  }
+
+  isLocked() {
+    return this._locks > 0;
+  }
+
+  lock() {
+    this._locks += 1;
+
+    let unlockNext: () => void;
+
+    const willLock = new Promise<void>(
+      (resolve) =>
+        (unlockNext = () => {
+          this._locks -= 1;
+          resolve();
+        }),
+    );
+
+    const willUnlock = this._locking.then(() => unlockNext);
+
+    this._locking = this._locking.then(() => willLock);
+
+    return willUnlock;
+  }
+}