Typesafe errors in connection path

src/api/SignalClient.ts

@@ -274,7 +274,7 @@
     const self = this;
 
     return withMutex(
       safeTry<U, ConnectionError>(async function* () {
         self.connectOptions = opts;
 
         const clientInfo = getClientInfo();
@@ -883,7 +883,7 @@
     connection: WebSocketConnection,
   ): ResultAsync<SignalResponse, ConnectionError> {
     // TODO: If inferring from the return type this could be more granular here than ConnectionError
     return safeTry<SignalResponse, ConnectionError>(async function* () {
       const signalReader = connection.readable.getReader();
 
       const firstMessage = await signalReader.read().finally(() => signalReader.releaseLock());
@@ -986,9 +986,6 @@
         );
       }
     } catch (e) {
-      if (!(e instanceof ConnectionError)) {
-        console.warn('received unexpected error', e);
-      }
       return err(
         e instanceof ConnectionError
           ? e

src/room/PCTransportManager.ts

@@ -1,5 +1,6 @@
 import { Mutex } from '@livekit/mutex';
 import { SignalTarget } from '@livekit/protocol';
+import { ResultAsync, ok, okAsync } from 'neverthrow';
 import log, { LoggerNames, getLogger } from '../logger';
 import PCTransport, { PCEvents } from './PCTransport';
 import { roomConnectOptionDefaults } from './defaults';
@@ -198,7 +199,10 @@ export class PCTransportManager {
     }
   }
 
-  async ensurePCTransportConnection(abortController?: AbortController, timeout?: number) {
+  async ensurePCTransportConnection(
+    abortController?: AbortController,
+    timeout?: number,
+  ): Promise<ResultAsync<void, ConnectionError>> {
     const unlock = await this.connectionLock.lock();
     try {
       if (
@@ -209,11 +213,11 @@ export class PCTransportManager {
         this.log.debug('negotiation required, start negotiating', this.logContext);
         this.publisher.negotiate();
       }
-      await Promise.all(
+      return await ResultAsync.combine(
         this.requiredTransports?.map((transport) =>
           this.ensureTransportConnected(transport, abortController, timeout),
         ),
-      );
+      ).andThen(() => ok());
     } finally {
       unlock();
     }
@@ -330,43 +334,46 @@ export class PCTransportManager {
     }
   };
 
-  private async ensureTransportConnected(
+  private ensureTransportConnected(
     pcTransport: PCTransport,
     abortController?: AbortController,
     timeout: number = this.peerConnectionTimeout,
-  ) {
+  ): ResultAsync<void, ConnectionError> {
     const connectionState = pcTransport.getConnectionState();
     if (connectionState === 'connected') {
-      return;
+      return okAsync();
     }
 
-    return new Promise<void>(async (resolve, reject) => {
-      const abortHandler = () => {
-        this.log.warn('abort transport connection', this.logContext);
-        CriticalTimers.clearTimeout(connectTimeout);
-
-        reject(ConnectionError.cancelled('room connection has been cancelled'));
-      };
-      if (abortController?.signal.aborted) {
-        abortHandler();
-      }
-      abortController?.signal.addEventListener('abort', abortHandler);
-
-      const connectTimeout = CriticalTimers.setTimeout(() => {
-        abortController?.signal.removeEventListener('abort', abortHandler);
-        reject(ConnectionError.internal('could not establish pc connection'));
-      }, timeout);
+    return ResultAsync.fromPromise(
+      new Promise<void>(async (resolve, reject) => {
+        const abortHandler = () => {
+          this.log.warn('abort transport connection', this.logContext);
+          CriticalTimers.clearTimeout(connectTimeout);
 
-      while (this.state !== PCTransportState.CONNECTED) {
-        await sleep(50); // FIXME we shouldn't rely on `sleep` in the connection paths, as it invokes `setTimeout` which can be drastically throttled by browser implementations
-        if (abortController?.signal.aborted) {
           reject(ConnectionError.cancelled('room connection has been cancelled'));
-          return;
+        };
+        if (abortController?.signal.aborted) {
+          abortHandler();
         }
-      }
-      CriticalTimers.clearTimeout(connectTimeout);
-      abortController?.signal.removeEventListener('abort', abortHandler);
-      resolve();
-    });
+        abortController?.signal.addEventListener('abort', abortHandler);
+
+        const connectTimeout = CriticalTimers.setTimeout(() => {
+          abortController?.signal.removeEventListener('abort', abortHandler);
+          reject(ConnectionError.internal('could not establish pc connection'));
+        }, timeout);
+
+        while (this.state !== PCTransportState.CONNECTED) {
+          await sleep(50); // FIXME we shouldn't rely on `sleep` in the connection paths, as it invokes `setTimeout` which can be drastically throttled by browser implementations
+          if (abortController?.signal.aborted) {
+            reject(ConnectionError.cancelled('room connection has been cancelled'));
+            return;
+          }
+        }
+        CriticalTimers.clearTimeout(connectTimeout);
+        abortController?.signal.removeEventListener('abort', abortHandler);
+        resolve();
+      }),
+      (e: unknown) => e as ConnectionError,
+    );
   }
 }

src/room/RTCEngine.ts

@@ -1079,7 +1079,7 @@
     regionUrl?: string,
   ): Promise<Result<void, UnexpectedConnectionState | SignalReconnectError>> {
     const self = this;
     const restartResultAsync = safeTry(async function* () {
       if (!self.url || !self.token) {
         // permanent failure, don't attempt reconnection
         return err(new UnexpectedConnectionState('could not reconnect, url or token not saved'));
@@ -1119,7 +1119,7 @@
       self.client.setReconnected();
       self.emit(EngineEvent.SignalRestarted, joinResult.value);
 
-      await self.waitForPCReconnected();
+      yield* await self.waitForPCReconnected();
 
       // re-check signal connection state before setting engine as resumed
       if (self.client.currentState !== SignalConnectionState.CONNECTED) {
@@ -1150,7 +1150,7 @@
 
   private async resumeConnection(
     reason?: ReconnectReason,
-  ): Promise<Result<void, SignalReconnectError>> {
+  ): Promise<Result<void, SignalReconnectError | UnexpectedConnectionState>> {
     if (!this.url || !this.token) {
       // permanent failure, don't attempt reconnection
       return errAsync(new UnexpectedConnectionState('could not reconnect, url or token not saved'));
@@ -1172,7 +1172,11 @@
     );
 
     if (reconnectResult.isErr()) {
-      return err(reconnectResult.error);
+      return err(
+        new SignalReconnectError(
+          `${reconnectResult.error.reasonName}: ${reconnectResult.error.message}`,
+        ),
+      );
     }
 
     this.emit(EngineEvent.SignalResumed);
@@ -1224,24 +1228,30 @@
     if (!this.pcManager) {
       throw new UnexpectedConnectionState('PC manager is closed');
     }
-    await this.pcManager.ensurePCTransportConnection(abortController, timeout);
+    return this.pcManager.ensurePCTransportConnection(abortController, timeout);
   }
 
-  private async waitForPCReconnected() {
+  private async waitForPCReconnected(): Promise<
+    Result<void, UnexpectedConnectionState | ConnectionError>
+  > {
     this.pcState = PCState.Reconnecting;
 
     this.log.debug('waiting for peer connection to reconnect', this.logContext);
     try {
       await sleep(minReconnectWait); // FIXME setTimeout again not ideal for a connection critical path
       if (!this.pcManager) {
-        throw new UnexpectedConnectionState('PC manager is closed');
+        return err(new UnexpectedConnectionState('PC manager is closed'));
       }
-      await this.pcManager.ensurePCTransportConnection(undefined, this.peerConnectionTimeout);
+      const res = await this.pcManager.ensurePCTransportConnection(
+        undefined,
+        this.peerConnectionTimeout,
+      );
       this.pcState = PCState.Connected;
+      return res;
     } catch (e: any) {
       // TODO do we need a `failed` state here for the PC?
       this.pcState = PCState.Disconnected;
-      throw ConnectionError.internal(`could not establish PC connection: ${e.message}`);
+      return err(ConnectionError.internal(`could not establish PC connection: ${e.message}`));
     }
   }
 

src/room/Room.ts

@@ -925,10 +925,15 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     }
 
     try {
-      await this.engine.waitForPCInitialConnection(
+      const result = await this.engine.waitForPCInitialConnection(
         this.connOptions.peerConnectionTimeout,
         abortController,
       );
+      if (result.isErr()) {
+        await this.engine.close();
+        this.recreateEngine();
+        throw result.error;
+      }
     } catch (e) {
       await this.engine.close();
       this.recreateEngine();