fix(client): harden SFU reconnection and ICE-restart recovery (#2285)

### 💡 Overview

Four reconnection-robustness changes for the core client.

**1. Signal-close revival hardening (`StreamSfuClient`, `Call`).** A
wedged signal WebSocket can be detected as dead by two independent
sources (the health watchdog and the transport `close` event), which on
a stuck socket can arrive seconds apart. Revival is now driven by a
one-shot latch so it fires at most once per client, triggers immediately
on an unhealthy close instead of waiting for the (possibly long-delayed)
`close` event, and ignores closes from a superseded client so a stale
socket cannot drive a reconnect after a new client has swapped in.

**2. Subscriber negotiation-failure recovery (`Subscriber`).**
`Subscriber.negotiate` reset `isIceRestarting` only as the last
statement of its happy path, so a negotiation failure left the flag
stuck at `true` and disabled all future automatic ICE-restart /
reconnect handling on the subscriber. Negotiation is now wrapped in
try/catch/finally so the flag always clears; on failure the remote
description is rolled back when the peer connection is still awaiting an
answer (`have-remote-offer`), the rollback is itself guarded so it can
never mask the original error, the failure is traced, and the subscriber
escalates via `tryRestartIce` so a failed restart falls through to a
reconnect instead of being silently dropped.

**3. Revert the stability-gated publisher fast-reconnect skip (`Call`,
`BasePeerConnection`).** `main` skips the publisher's ICE restart on a
fast reconnect when the publisher PC still looks stable (`isStable()`),
to avoid SDP/ICE churn on signal-WS-only drops. In practice that skip
can strand a cratered send-side bandwidth estimate: a "stable" publisher
PC never restarts ICE, so the network-route-change BWE reset never fires
and the encoder stays bitrate-starved after a handover. This restores
the earlier behavior (on a fast reconnect the publisher's ICE is
restarted unconditionally whenever it is publishing) and removes the
now-unused `isStable()` helper.

**4. Generation-aware ICE trickle buffer (`IceTrickleBuffer`,
`BasePeerConnection`, `Subscriber`).** After an ICE restart the SFU
rotates its ICE credentials and trickles candidates for the new
generation, but `IceTrickleBuffer` (an unbounded `ReplaySubject`)
replayed every buffered candidate on each renegotiation, re-adding
superseded-generation candidates against the current generation and
contributing to stuck (DTLS-`connecting`) subscribers. The buffer is now
generation-aware: each peer connection declares the active generation
via `updateActiveGeneration` (derived from the applied remote
description's `ice-ufrag`), and the candidate streams emit only the
current generation, hold not-yet-applied (future) generations until they
become active, and drop superseded ones; candidates with no detectable
generation fail open. On a failed subscriber negotiation the buffer is
restored to the rolled-back generation.

### 📝 Implementation notes

- `StreamSfuClient`: `handleWebSocketClose` becomes `notifySignalClose`,
guarded by a `signalClosed` one-shot latch; `close()` notifies on a
non-clean close without waiting for the transport event.
- `Call.handleSfuSignalClose` ignores closes from any client that is no
longer the active `sfuClient`.
- `Call` fast reconnect: back to `restoreICE(sfuClient, {
includeSubscriber: false })` (publisher restart governed by
`includePublisher`'s `true` default); the `isStable()`-gated skip is
gone.
- `Subscriber.negotiate` wrapped in try/catch/finally; guarded rollback
on `have-remote-offer`; `subscriber.negotiationFailed` trace;
`tryRestartIce` escalation on failure.
- `BasePeerConnection.isStable()` removed (no remaining callers).
- `IceTrickleBuffer`: per-peer `ReplaySubject`s replaced by per-peer
`CandidateGenerationBuffer`s (mutable store + live `Subject`, exposed as
`subscriber` / `publisher`); `updateActiveGeneration(peerType, sdp)`
sets the active generation from the remote `ice-ufrag` and evicts
superseded candidates.
- `BasePeerConnection.addTrickledIceCandidates` declares the active
generation from `pc.remoteDescription`, then subscribes; the
candidate-string / SDP ufrag helpers (`getCandidateUfrag`,
`parseIceUfrag`, `toJSON`) are consolidated in
`rtc/helpers/iceCandiates.ts`.
- `Subscriber.negotiate` snapshots the committed generation and restores
it in the buffer when a negotiation rolls back.
- Tests: `StreamSfuClient` revival, `Call` superseded-client guard,
`Subscriber` success / failure / rollback / rollback-failure branches;
removed the `isStable()` and FAST-skip tests. Added `IceTrickleBuffer`
generation emit/hold/drop/fail-open, `iceCandiates` helper, and
subscriber trickle wiring + rollback-restore coverage.

🎫 Ticket:
https://linear.app/stream/issue/REACT-1016/improve-bandwidth-estimation-ramp-up-and-harden-ice-restarts

Author: oliverlaz

packages/client/src/Call.ts

@@ -1333,23 +1333,9 @@ export class Call {
     // when performing fast reconnect, or when we reuse the same SFU client,
     // (ws remained healthy), we just need to restore the ICE connection
     if (performingFastReconnect) {
-      // The SFU automatically issues an ICE restart on the subscriber,
-      // so we only need to decide about the publisher. If the publisher's
-      // peer connection is still stable (ICE still connected end-to-end),
-      // the signal WebSocket drop was the only problem — the new WS alone
-      // is enough, and restarting ICE would add unnecessary SDP/ICE churn.
-      const publisherIsStable = this.publisher?.isStable() ?? true;
-      const includePublisher =
-        !!this.publisher?.isPublishing() && !publisherIsStable;
-      if (!includePublisher && this.publisher?.isPublishing()) {
-        this.logger.info(
-          '[Reconnect] FAST: skipping publisher ICE restart, publisher PC is stable',
-        );
-      }
-      await this.restoreICE(sfuClient, {
-        includeSubscriber: false,
-        includePublisher,
-      });
+      // the SFU automatically issues an ICE restart on the subscriber
+      // we don't have to do it ourselves
+      await this.restoreICE(sfuClient, { includeSubscriber: false });
     } else {
       const connectionConfig = toRtcConfiguration(this.credentials.ice_servers);
       await this.initPublisherAndSubscriber({
@@ -1688,6 +1674,10 @@ export class Call {
     reason: string,
   ) => {
     this.logger.debug('[Reconnect] SFU signal connection closed');
+    // Ignore a close from a superseded client (e.g. an old socket's delayed
+    // `onclose` arriving after a reconnection already swapped in a new client).
+    // Only the currently active client's death may drive a reconnection.
+    if (sfuClient !== this.sfuClient) return;
     const { callingState } = this.state;
     if (
       // SFU WS closed before we finished current join,
@@ -1728,11 +1718,12 @@ export class Call {
     strategy: WebsocketReconnectStrategy,
     reason: ReconnectReason,
   ): Promise<void> => {
+    const { callingState } = this.state;
     if (
-      this.state.callingState === CallingState.JOINING ||
-      this.state.callingState === CallingState.RECONNECTING ||
-      this.state.callingState === CallingState.MIGRATING ||
-      this.state.callingState === CallingState.RECONNECTING_FAILED
+      callingState === CallingState.JOINING ||
+      callingState === CallingState.RECONNECTING ||
+      callingState === CallingState.MIGRATING ||
+      callingState === CallingState.RECONNECTING_FAILED
     )
       return;
 

packages/client/src/StreamSfuClient.ts

@@ -164,13 +164,24 @@ export class StreamSfuClient {
    */
   isClosingClean = false;
 
+  /**
+   * One-shot latch guarding `onSignalClose`. The signal connection can be
+   * detected as dead by more than one source (the health watchdog and the
+   * WebSocket `close` event, which on a wedged socket can arrive seconds
+   * apart). This ensures revival is triggered at most once per client.
+   */
+  private signalClosed = false;
+
   private readonly rpc: SignalServerClient;
   private keepAliveInterval?: number;
-  private connectionCheckTimeout?: NodeJS.Timeout;
+  private connectionCheckInterval?: number;
   private migrateAwayTimeout?: NodeJS.Timeout;
   private readonly pingIntervalInMs = 5 * 1000;
-  private readonly unhealthyTimeoutInMs = 15 * 1000;
-  private lastMessageTimestamp?: Date;
+  private readonly unhealthyTimeoutInMs = this.pingIntervalInMs * 2 + 2 * 1000;
+  private readonly connectionCheckIntervalInMs = Math.round(
+    this.unhealthyTimeoutInMs / 3,
+  );
+  private lastMessageTimestamp?: number;
   private readonly tracer?: Tracer;
   private readonly unsubscribeIceTrickle: () => void;
   private readonly unsubscribeNetworkChanged: () => void;
@@ -209,7 +220,7 @@ export class StreamSfuClient {
   /**
    * The error code used when the SFU connection is unhealthy.
    * Usually, this means that no message has been received from the SFU for
-   * a certain amount of time (`connectionCheckTimeout`).
+   * a certain amount of time (`unhealthyTimeoutInMs`).
    */
   static ERROR_CONNECTION_UNHEALTHY = 4001;
   /**
@@ -311,7 +322,7 @@ export class StreamSfuClient {
       endpoint: `${this.credentials.server.ws_endpoint}?${new URLSearchParams(params).toString()}`,
       tracer: this.tracer,
       onMessage: (message) => {
-        this.lastMessageTimestamp = new Date();
+        this.lastMessageTimestamp = Date.now();
         this.scheduleConnectionCheck();
         const eventKind = message.eventPayload.oneofKind;
         if (eventsToTrace[eventKind]) {
@@ -336,7 +347,7 @@ export class StreamSfuClient {
           this.signalWs.addEventListener('open', onOpen);
 
           this.signalWs.addEventListener('close', (e) => {
-            this.handleWebSocketClose(e);
+            this.notifySignalClose(`${e.code} ${e.reason ?? ''}`);
             // Normally, this shouldn't have any effect, because WS should never emit 'close'
             // before emitting 'open'. However, stranger things have happened, and we don't
             // want to leave signalReady in a pending state.
@@ -371,11 +382,13 @@ export class StreamSfuClient {
     return this.joinResponseTask.promise;
   }
 
-  private handleWebSocketClose = (e: CloseEvent) => {
-    this.signalWs.removeEventListener('close', this.handleWebSocketClose);
-    getTimers().clearInterval(this.keepAliveInterval);
-    clearTimeout(this.connectionCheckTimeout);
-    this.onSignalClose?.(`${e.code} ${e.reason}`);
+  private notifySignalClose = (reason: string) => {
+    if (this.signalClosed) return;
+    this.signalClosed = true;
+    const timers = getTimers();
+    timers.clearInterval(this.keepAliveInterval);
+    timers.clearInterval(this.connectionCheckInterval);
+    this.onSignalClose?.(reason.trim());
   };
 
   close = (code: number = StreamSfuClient.NORMAL_CLOSURE, reason?: string) => {
@@ -392,7 +405,9 @@ export class StreamSfuClient {
     ) {
       this.logger.debug(`Closing SFU WS connection: ${code} - ${reason}`);
       ws.close(code, `js-client: ${reason}`);
-      ws.removeEventListener('close', this.handleWebSocketClose);
+    }
+    if (!this.isClosingClean) {
+      this.notifySignalClose(`${code} ${reason ?? ''}`);
     }
     this.dispose(reason);
   };
@@ -401,8 +416,9 @@ export class StreamSfuClient {
     this.logger.debug('Disposing SFU client');
     this.unsubscribeIceTrickle();
     this.unsubscribeNetworkChanged();
-    clearInterval(this.keepAliveInterval);
-    clearTimeout(this.connectionCheckTimeout);
+    const timers = getTimers();
+    timers.clearInterval(this.keepAliveInterval);
+    timers.clearInterval(this.connectionCheckInterval);
     clearTimeout(this.migrateAwayTimeout);
     this.abortController.abort();
     this.migrationTask?.resolve();
@@ -711,19 +727,18 @@ export class StreamSfuClient {
   };
 
   private scheduleConnectionCheck = () => {
-    clearTimeout(this.connectionCheckTimeout);
-    this.connectionCheckTimeout = setTimeout(() => {
+    const timers = getTimers();
+    timers.clearInterval(this.connectionCheckInterval);
+    this.connectionCheckInterval = timers.setInterval(() => {
       if (this.lastMessageTimestamp) {
-        const timeSinceLastMessage =
-          new Date().getTime() - this.lastMessageTimestamp.getTime();
-
+        const timeSinceLastMessage = Date.now() - this.lastMessageTimestamp;
         if (timeSinceLastMessage > this.unhealthyTimeoutInMs) {
           this.close(
             StreamSfuClient.ERROR_CONNECTION_UNHEALTHY,
             `SFU connection unhealthy. Didn't receive any message for ${this.unhealthyTimeoutInMs}ms`,
           );
         }
       }
-    }, this.unhealthyTimeoutInMs);
+    }, this.connectionCheckIntervalInMs);
   };
 }

packages/client/src/__tests__/StreamSfuClient.test.ts

@@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { StreamSfuClient } from '../StreamSfuClient';
 import { Dispatcher } from '../rtc';
 import { StreamClient } from '../coordinator/connection/client';
+import { getTimers } from '../timers';
 
 /**
  * Minimal `WebSocket` stub used to drive `StreamSfuClient.close()` while the
@@ -38,9 +39,13 @@ class CapturingWebSocket {
     this.closeArgs = { code, reason };
     this.readyState = CapturingWebSocket.CLOSED;
   }
+  /** Test helper: synchronously fire a registered event (e.g. `close`). */
+  emit(event: string, payload: unknown) {
+    this.listeners.get(event)?.forEach((listener) => listener(payload));
+  }
 }
 
-const buildSfuClient = () => {
+const buildSfuClient = (onSignalClose?: (reason: string) => void) => {
   const dispatcher = new Dispatcher();
   const streamClient = new StreamClient('test-key');
   return new StreamSfuClient({
@@ -59,9 +64,109 @@ const buildSfuClient = () => {
     },
     tag: 'test',
     enableTracing: false,
+    onSignalClose,
   });
 };
 
+describe('StreamSfuClient unhealthy watchdog timer source', () => {
+  beforeEach(() => {
+    CapturingWebSocket.instances = [];
+    vi.stubGlobal('WebSocket', CapturingWebSocket);
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+    vi.restoreAllMocks();
+  });
+
+  it('arms the unhealthy watchdog on the worker timer, not the main-thread setInterval', () => {
+    const sfuClient = buildSfuClient();
+    const workerSetInterval = vi
+      .spyOn(getTimers(), 'setInterval')
+      .mockReturnValue(1 as unknown as number);
+    const mainSetInterval = vi.spyOn(globalThis, 'setInterval');
+
+    (
+      sfuClient as unknown as { scheduleConnectionCheck: () => void }
+    ).scheduleConnectionCheck();
+
+    expect(workerSetInterval).toHaveBeenCalledTimes(1);
+    expect(mainSetInterval).not.toHaveBeenCalled();
+
+    sfuClient.close(1000, 'test cleanup');
+  });
+});
+
+describe('StreamSfuClient unhealthy watchdog resilience', () => {
+  beforeEach(() => {
+    CapturingWebSocket.instances = [];
+    vi.stubGlobal('WebSocket', CapturingWebSocket);
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.unstubAllGlobals();
+    vi.restoreAllMocks();
+  });
+
+  it('re-arms the unhealthy watchdog after a check passes (not single-shot)', () => {
+    vi.useFakeTimers();
+    const sfuClient = buildSfuClient();
+    const closeSpy = vi.spyOn(sfuClient, 'close').mockImplementation(() => {});
+    const c = sfuClient as unknown as {
+      lastMessageTimestamp?: number;
+      unhealthyTimeoutInMs: number;
+      scheduleConnectionCheck: () => void;
+    };
+    const window = c.unhealthyTimeoutInMs;
+
+    c.lastMessageTimestamp = Date.now();
+    c.scheduleConnectionCheck();
+
+    // At exactly the threshold the connection is still healthy (strict `>`),
+    // so no poll within the first window closes it. A single-shot watchdog
+    // armed for the threshold would now be dead.
+    vi.advanceTimersByTime(window);
+    expect(closeSpy).not.toHaveBeenCalled();
+
+    // No further messages arrive; the self-rescheduling watchdog keeps polling
+    // and detects the connection as unhealthy on a later tick.
+    vi.advanceTimersByTime(window);
+    expect(closeSpy).toHaveBeenCalledWith(
+      StreamSfuClient.ERROR_CONNECTION_UNHEALTHY,
+      expect.stringContaining('unhealthy'),
+    );
+  });
+
+  it('detects an unhealthy connection shortly after the threshold, not a full window later', () => {
+    vi.useFakeTimers();
+    const sfuClient = buildSfuClient();
+    const closeSpy = vi.spyOn(sfuClient, 'close').mockImplementation(() => {});
+    const c = sfuClient as unknown as {
+      lastMessageTimestamp?: number;
+      unhealthyTimeoutInMs: number;
+      scheduleConnectionCheck: () => void;
+    };
+    const window = c.unhealthyTimeoutInMs;
+
+    c.lastMessageTimestamp = Date.now();
+    c.scheduleConnectionCheck();
+
+    // healthy up to (and exactly at) the threshold
+    vi.advanceTimersByTime(window);
+    expect(closeSpy).not.toHaveBeenCalled();
+
+    // the watchdog polls finer than the window, so silence is caught well
+    // before a second full window elapses (the old period == window design
+    // could take up to 2x the window to notice).
+    vi.advanceTimersByTime(window / 2);
+    expect(closeSpy).toHaveBeenCalledWith(
+      StreamSfuClient.ERROR_CONNECTION_UNHEALTHY,
+      expect.stringContaining('unhealthy'),
+    );
+  });
+});
+
 describe('StreamSfuClient.close()', () => {
   beforeEach(() => {
     CapturingWebSocket.instances = [];
@@ -164,6 +269,59 @@ describe('StreamSfuClient.close()', () => {
   });
 });
 
+describe('StreamSfuClient signal-close revival', () => {
+  beforeEach(() => {
+    CapturingWebSocket.instances = [];
+    vi.stubGlobal('WebSocket', CapturingWebSocket);
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+    vi.clearAllMocks();
+  });
+
+  it('drives revival immediately on an unhealthy close, without waiting for the onclose event', () => {
+    const onSignalClose = vi.fn();
+    const sfuClient = buildSfuClient(onSignalClose);
+
+    // A wedged socket may fire `onclose` only after the OS TCP timeout. The
+    // health watchdog closes with ERROR_CONNECTION_UNHEALTHY; revival must
+    // start now, not when (or if) the transport `close` event arrives.
+    sfuClient.close(
+      StreamSfuClient.ERROR_CONNECTION_UNHEALTHY,
+      'SFU connection unhealthy',
+    );
+
+    expect(onSignalClose).toHaveBeenCalledTimes(1);
+  });
+
+  it('notifies revival only once when the late onclose event follows an unhealthy close', () => {
+    const onSignalClose = vi.fn();
+    const sfuClient = buildSfuClient(onSignalClose);
+    const ws = CapturingWebSocket.instances.at(-1)!;
+
+    // watchdog closes the dead socket (revival triggered proactively)...
+    sfuClient.close(
+      StreamSfuClient.ERROR_CONNECTION_UNHEALTHY,
+      'SFU connection unhealthy',
+    );
+    // ...then the OS finally surfaces the wedged socket's `close` event.
+    ws.emit('close', { code: 1006, reason: '' });
+
+    expect(onSignalClose).toHaveBeenCalledTimes(1);
+  });
+
+  it('notifies revival when only the onclose event fires (server-initiated close)', () => {
+    const onSignalClose = vi.fn();
+    buildSfuClient(onSignalClose);
+    const ws = CapturingWebSocket.instances.at(-1)!;
+
+    ws.emit('close', { code: 1006, reason: '' });
+
+    expect(onSignalClose).toHaveBeenCalledTimes(1);
+  });
+});
+
 describe('StreamSfuClient.leaveAndClose()', () => {
   beforeEach(() => {
     CapturingWebSocket.instances = [];