Serialize browser relay replacements

README.md

@@ -217,12 +217,15 @@ browser storage, log, error, `HELLO`, or `REAUTH` frame.
 
 The control plane returns an up-to-five-minute Miakapp access token atomically
 with its authoritative relay URL. MiakAPI sends only that audience-bound token
-to the returned relay. If a renewal selects a different relay, the client closes
-the old session and opens the replacement with the already-issued credential; it
-does not expose the new token to the old relay or repeat the exchange. Stop and
-discard the client immediately when the Firebase user signs out or the selected
-home changes. Relay routing changes arrive through credentials and do not require
-mutating the client options.
+to the returned relay. If a renewal selects a different relay, the client marks
+the old session stale and closes it. Before any automatic replacement connection,
+including recovery from a transport or protocol failure, the client waits for the
+native transport close event. If closure is not confirmed within ten seconds, the
+client stops fail-closed instead of opening overlapping relay sockets. A routing
+handoff uses the already-issued credential; it does not expose the new token to
+the old relay or repeat the exchange. Stop and discard the client immediately
+when the Firebase user signs out or the selected home changes. Relay routing
+changes arrive through credentials and do not require mutating the client options.
 
 Audience binding limits credential replay; it does not encrypt home traffic from
 the selected relay. Users should still choose an operator they trust with the

src/browser-client.ts

@@ -306,6 +306,10 @@ class BrowserClientImpl implements BrowserClient, UserStateHost, UserCallHost {
               connectionEnd?.resolve({ failure });
             },
           },
+          (createdSession) => {
+            this.#session = createdSession;
+            this.#sessionRelayUrl = credential.relayUrl;
+          },
         );
         if (this.#loopController.signal.aborted) {
           session.terminate();
@@ -314,12 +318,9 @@ class BrowserClientImpl implements BrowserClient, UserStateHost, UserCallHost {
         }
         if (credential.expiresAtMs <= this.#runtime.now()) {
           session.terminate();
-          session.detach();
           throw browserUnavailable('Browser relay credential expired during authentication');
         }
         this.#reconnectAttempt = 0;
-        this.#session = session;
-        this.#sessionRelayUrl = credential.relayUrl;
         this.#setHomeStatus(Object.freeze({
           enrolled: session.welcome.readySession.enrolled,
           coordinators: session.welcome.readySession.coordinators,
@@ -366,12 +367,26 @@ class BrowserClientImpl implements BrowserClient, UserStateHost, UserCallHost {
             ? error
             : browserUnavailable('Browser relay connection attempt failed') };
       }
-      this.#disconnectSession();
+      const endedSession = this.#session;
+      this.#deactivateSession();
       if (this.#loopController.signal.aborted) break;
       if (end.failure !== undefined) this.#emitFailure(end.failure);
       if (this.#loopController.signal.aborted) break;
       this.#transition('reconnecting', undefined, end.failure);
       if (this.#loopController.signal.aborted) break;
+      if (endedSession !== undefined
+        && !endedSession.transportClosed
+        && !await this.#waitForTransportClose(endedSession)) {
+        if (!this.#loopController.signal.aborted) {
+          const failure = browserUnavailable('Browser relay transport close timed out');
+          this.#emitFailure(failure);
+          this.#startDeferred?.reject(failure);
+          void this.stop();
+        }
+        break;
+      }
+      this.#disconnectSession();
+      if (this.#loopController.signal.aborted) break;
       if (end.handoffCredential !== undefined) {
         pendingCredential = end.handoffCredential;
         reason = 'reconnect';
@@ -478,8 +493,8 @@ class BrowserClientImpl implements BrowserClient, UserStateHost, UserCallHost {
         if (this.#sessionEnd === undefined || this.#sessionEnd.settled) {
           throw browserUnavailable('Browser relay handoff state is unavailable');
         }
-        this.#sessionEnd.resolve({ handoffCredential: credential });
         session.terminate();
+        this.#sessionEnd.resolve({ handoffCredential: credential });
         return;
       }
       const remaining = Math.min(currentExpiresAtMs, credential.expiresAtMs)
@@ -513,6 +528,27 @@ class BrowserClientImpl implements BrowserClient, UserStateHost, UserCallHost {
       && this.#status !== 'draining';
   }
 
+  async #waitForTransportClose(session: UserRelaySession): Promise<boolean> {
+    const interrupted = createDeferred<boolean>();
+    const abort = () => interrupted.resolve(false);
+    if (this.#loopController.signal.aborted) return false;
+    this.#loopController.signal.addEventListener('abort', abort, { once: true });
+    if (this.#loopController.signal.aborted) abort();
+    const timeout = this.#runtime.setTimer(
+      () => interrupted.resolve(false),
+      SESSION_PHASE_TIMEOUT_MS,
+    );
+    try {
+      return await Promise.race([
+        session.waitForTransportClose().then(() => true),
+        interrupted.promise,
+      ]);
+    } finally {
+      timeout.cancel();
+      this.#loopController.signal.removeEventListener('abort', abort);
+    }
+  }
+
   #clearReauthentication(): void {
     this.#reauthTimer?.cancel();
     this.#reauthTimer = undefined;
@@ -625,13 +661,7 @@ class BrowserClientImpl implements BrowserClient, UserStateHost, UserCallHost {
   }
 
   #disconnectSession(): void {
-    this.#bootstrapTimer?.cancel();
-    this.#bootstrapTimer = undefined;
-    this.#clearReauthentication();
-    this.#abortCredentialRequest();
-    this.state.disconnected();
-    this.calls.disconnected();
-    this.#markHomeStale();
+    this.#deactivateSession();
     this.#session?.detach();
     this.#session = undefined;
     this.#sessionRelayUrl = undefined;
@@ -643,6 +673,16 @@ class BrowserClientImpl implements BrowserClient, UserStateHost, UserCallHost {
     this.#topicReady = false;
   }
 
+  #deactivateSession(): void {
+    this.#bootstrapTimer?.cancel();
+    this.#bootstrapTimer = undefined;
+    this.#clearReauthentication();
+    this.#abortCredentialRequest();
+    this.state.disconnected();
+    this.calls.disconnected();
+    this.#markHomeStale();
+  }
+
   #emitFailure(failure: BrowserClientFailure): void {
     this.#errorListeners.emit(failure, () => {
       safeBrowserLog(this.#options.logger, { level: 'error', event: 'error_listener_failed' });

src/internal/browser-socket.ts

@@ -213,6 +213,7 @@ export class BrowserSocketFactory implements SocketFactory {
     url: string,
     handlers: SocketHandlers,
     signal: AbortSignal,
+    onSocket?: (socket: ManagedSocket) => void,
   ): Promise<ManagedSocket> {
     if (signal.aborted) throw signal.reason;
     const ManagedWebSocket = nativeConstructor();
@@ -222,12 +223,13 @@ export class BrowserSocketFactory implements SocketFactory {
       signal,
       this.#now,
     );
+    onSocket?.(socket);
     try {
       await socket.ready();
       return socket;
     } catch (error) {
-      socket.detach();
       socket.terminate();
+      if (onSocket === undefined) socket.detach();
       throw error;
     }
   }

src/internal/runtime.ts

@@ -15,7 +15,13 @@ export interface ManagedSocket {
 }
 
 export interface SocketFactory {
-  connect(url: string, handlers: SocketHandlers, signal: AbortSignal): Promise<ManagedSocket>;
+  /** Transfers a created transport to its owner before asynchronous readiness work begins. */
+  connect(
+    url: string,
+    handlers: SocketHandlers,
+    signal: AbortSignal,
+    onSocket?: (socket: ManagedSocket) => void,
+  ): Promise<ManagedSocket>;
 }
 
 export interface RuntimeTimer {

src/internal/socket.ts

@@ -150,6 +150,7 @@ export class WsSocketFactory implements SocketFactory {
     url: string,
     handlers: SocketHandlers,
     signal: AbortSignal,
+    onSocket?: (socket: ManagedSocket) => void,
   ): Promise<ManagedSocket> {
     if (signal.aborted) throw signal.reason;
     const options: BoundedClientOptions = {
@@ -165,8 +166,15 @@ export class WsSocketFactory implements SocketFactory {
       handlers,
       signal,
     );
-    await managed.ready();
-    return managed;
+    onSocket?.(managed);
+    try {
+      await managed.ready();
+      return managed;
+    } catch (error) {
+      managed.terminate();
+      if (onSocket === undefined) managed.detach();
+      throw error;
+    }
   }
 }
 

src/internal/user-session.ts

@@ -130,6 +130,7 @@ export class UserRelaySession {
   readonly #callbacks: UserRelaySessionCallbacks;
   readonly #now: () => number;
   readonly #protocol = new UserProtocolSession();
+  readonly #transportClosed = createDeferred<void>();
   readonly #welcome = createDeferred<UserRelayWelcome>();
   readonly #queuedFrames: Frame[] = [];
   #queuedFrameBytes = 0;
@@ -152,9 +153,13 @@ export class UserRelaySession {
     token: string,
     signal: AbortSignal,
     callbacks: UserRelaySessionCallbacks,
+    onCreated: (session: UserRelaySession) => void,
   ): Promise<UserRelaySession> {
     const session = new UserRelaySession(callbacks, () => runtime.now());
+    onCreated(session);
     const handshake = childAbortController(signal);
+    const abortWelcome = () => session.#welcome.reject(handshake.controller.signal.reason);
+    handshake.controller.signal.addEventListener('abort', abortWelcome, { once: true });
     const timeout = runtime.setTimer(() => {
       handshake.controller.abort(new Error('Browser relay handshake timed out'));
     }, HANDSHAKE_TIMEOUT_MS);
@@ -168,6 +173,10 @@ export class UserRelaySession {
         relayUrl,
         handlers,
         handshake.controller.signal,
+        (socket) => {
+          session.#socket = socket;
+          if (session.#closed) socket.terminate();
+        },
       );
       session.#connectedAtMs = runtime.now();
       await session.#socket.write(session.#protocol.encode({
@@ -178,10 +187,10 @@ export class UserRelaySession {
       return session;
     } catch (error) {
       session.terminate();
-      session.detach();
       throw error;
     } finally {
       timeout.cancel();
+      handshake.controller.signal.removeEventListener('abort', abortWelcome);
       handshake.dispose();
     }
   }
@@ -232,6 +241,16 @@ export class UserRelaySession {
     this.#socket?.terminate();
   }
 
+  waitForTransportClose(): Promise<void> {
+    return this.#socket === undefined
+      ? Promise.resolve()
+      : this.#transportClosed.promise;
+  }
+
+  get transportClosed(): boolean {
+    return this.#socket === undefined || this.#transportClosed.settled;
+  }
+
   detach(): void {
     this.#socket?.detach();
   }
@@ -270,6 +289,7 @@ export class UserRelaySession {
     const wasClosed = this.#closed;
     this.#closed = true;
     this.#protocol.close();
+    this.#transportClosed.resolve(undefined);
     this.#welcome.reject(browserProtocolFailure('Relay closed before WELCOME'));
     if (!wasClosed) this.#callbacks.closed(code, reason);
   }