fix(realtime): survive signal outages and per-server identities

Peer recovery burned its whole retry budget while signaling was down, then dropped the tracker with no re-arm, so a peer stayed dead until an unrelated roster event healed it. Recovery now waits for a usable transport before spending an attempt. Initiator election also compared a home actor id against foreign roster ids, which is not antisymmetric across signal servers - both sides offered, or neither did. Election moves into `peer-role.rules` and compares ids only within one signal server's identity space.
2026-08-14 03:19:29 +02:00
parent f9e8538c80
commit a83f5aa750
14 changed files with 1176 additions and 64 deletions
@@ -7,6 +7,7 @@ import {
  PEER_RECONNECT_INTERVAL_MS,
  PEER_RECONNECT_MAX_ATTEMPTS
 } from '../../realtime.constants';
+import { shouldInitiatePeerConnection } from '../../peer-role.rules';
 import {
  PeerConnectionManagerContext,
  PeerConnectionManagerState,
@@ -205,6 +206,15 @@ export function scheduleDataChannelRecovery(
  state.dataChannelRecoveryTimers.set(peerId, timer);
 }

+/**
+ * Repair a failed control channel with the least destructive option available.
+ *
+ * While the `RTCPeerConnection` is still connected the channel is replaced on that same
+ * connection, so voice, camera, and screen share keep flowing. Only the deterministically
+ * elected initiator creates the replacement; the other side adopts the incoming channel.
+ * A full peer rebuild is the fallback when the connection itself is gone, when the
+ * replacement cannot be created, or when the replacement never opens.
+ */
 function repairUnavailableDataChannel(
  context: PeerConnectionManagerContext,
  peerId: string,
@@ -212,7 +222,7 @@ function repairUnavailableDataChannel(
  reason: string,
  handlers: RecoveryHandlers
 ): void {
-  const { logger, state } = context;
+  const { callbacks, logger, state } = context;
  const peerData = state.activePeerConnections.get(peerId);

  if (!peerData || peerData.dataChannel !== channel)
@@ -221,11 +231,103 @@ function repairUnavailableDataChannel(
  if (peerData.dataChannel?.readyState === DATA_CHANNEL_STATE_OPEN)
    return;

-  logger.warn('[data-channel] Recreating peer transport after control channel failure', {
+  if (peerData.connection.connectionState !== CONNECTION_STATE_CONNECTED) {
+    rebuildPeerAfterDataChannelFailure(context, peerId, reason, handlers);
+    return;
+  }
+
+  const localOderId = callbacks.getIdentifyCredentialsForPeer(peerId)?.oderId ?? null;
+
+  if (!localOderId) {
+    logger.warn('[data-channel] Logical identity unknown; rebuilding peer instead of replacing channel', {
+      peerId,
+      reason
+    });
+
+    rebuildPeerAfterDataChannelFailure(context, peerId, reason, handlers);
+    return;
+  }
+
+  if (shouldInitiatePeerConnection(localOderId, peerId)) {
+    logger.warn('[data-channel] Replacing control channel on the live connection', {
+      channelLabel: channel.label,
+      peerId,
+      reason
+    });
+
+    if (!handlers.replaceDataChannel(peerId, channel)) {
+      rebuildPeerAfterDataChannelFailure(context, peerId, reason, handlers);
+      return;
+    }
+
+    watchDataChannelReplacement(context, peerId, reason, handlers);
+    return;
+  }
+
+  logger.info('[data-channel] Waiting for the remote replacement control channel', {
    channelLabel: channel.label,
-    connectionState: peerData.connection.connectionState,
+    localOderId,
    peerId,
-    readyState: peerData.dataChannel?.readyState ?? null,
+    reason
+  });
+
+  watchDataChannelReplacement(context, peerId, reason, handlers);
+}
+
+/** Rebuild the peer if the replacement control channel does not open within the grace period. */
+function watchDataChannelReplacement(
+  context: PeerConnectionManagerContext,
+  peerId: string,
+  reason: string,
+  handlers: RecoveryHandlers
+): void {
+  const { logger, state } = context;
+
+  clearDataChannelRecoveryTimer(state, peerId);
+
+  const timer = setTimeout(() => {
+    state.dataChannelRecoveryTimers.delete(peerId);
+
+    const latestPeerData = state.activePeerConnections.get(peerId);
+
+    if (!latestPeerData)
+      return;
+
+    if (latestPeerData.dataChannel?.readyState === DATA_CHANNEL_STATE_OPEN) {
+      logger.info('[data-channel] Replacement control channel is open; peer kept alive', {
+        peerId,
+        reason
+      });
+
+      return;
+    }
+
+    logger.warn('[data-channel] Replacement control channel never opened; rebuilding peer', {
+      connectionState: latestPeerData.connection.connectionState,
+      peerId,
+      readyState: latestPeerData.dataChannel?.readyState ?? null,
+      reason
+    });
+
+    rebuildPeerAfterDataChannelFailure(context, peerId, reason, handlers);
+  }, DATA_CHANNEL_RECOVERY_GRACE_MS);
+
+  state.dataChannelRecoveryTimers.set(peerId, timer);
+}
+
+function rebuildPeerAfterDataChannelFailure(
+  context: PeerConnectionManagerContext,
+  peerId: string,
+  reason: string,
+  handlers: RecoveryHandlers
+): void {
+  const { logger, state } = context;
+  const peerData = state.activePeerConnections.get(peerId);
+
+  logger.warn('[data-channel] Recreating peer transport after control channel failure', {
+    connectionState: peerData?.connection.connectionState ?? null,
+    peerId,
+    readyState: peerData?.dataChannel?.readyState ?? null,
    reason
  });

@@ -299,30 +401,84 @@ export function schedulePeerReconnect(
      return;
    }

+    // An attempt costs nothing while the socket is down and there is no way to send an
+    // offer, so the budget must only be spent on attempts that can actually reach the peer.
+    if (!callbacks.isSignalingConnected()) {
+      logger.info('Deferring P2P reconnect - no signaling connection', {
+        peerId,
+        attemptsSpent: info.reconnectAttempts
+      });
+
+      return;
+    }
+
+    if (info.reconnectAttempts >= PEER_RECONNECT_MAX_ATTEMPTS) {
+      failPeerRecovery(context, peerId, info.reconnectAttempts);
+      return;
+    }
+
    info.reconnectAttempts++;
    logger.info('P2P reconnect attempt', {
      peerId,
      attempt: info.reconnectAttempts
    });

-    if (info.reconnectAttempts >= PEER_RECONNECT_MAX_ATTEMPTS) {
-      logger.info('P2P reconnect max attempts reached', { peerId });
-      clearPeerReconnectTimer(state, peerId);
-      state.disconnectedPeerTracker.delete(peerId);
-      return;
-    }
-
-    if (!callbacks.isSignalingConnected()) {
-      logger.info('Skipping P2P reconnect - no signaling connection', { peerId });
-      return;
-    }
-
    attemptPeerReconnect(context, peerId, handlers);
  }, PEER_RECONNECT_INTERVAL_MS);

  state.peerReconnectTimers.set(peerId, timer);
 }

+/**
+ * Stop retrying a peer and say so. The tracker entry is kept (marked failed) so
+ * `resumeStalledPeerRecovery` can re-arm it once signaling is usable again.
+ */
+function failPeerRecovery(
+  context: PeerConnectionManagerContext,
+  peerId: string,
+  attempts: number
+): void {
+  const { logger, state } = context;
+  const info = state.disconnectedPeerTracker.get(peerId);
+
+  logger.warn('P2P reconnect gave up', { attempts, peerId });
+  clearPeerReconnectTimer(state, peerId);
+
+  if (info) {
+    info.recoveryFailed = true;
+  }
+
+  state.peerRecoveryStatus$.next({ attempts, peerId, status: 'failed' });
+}
+
+/**
+ * Re-arm every peer whose recovery gave up. Called when signaling reconnects, so a peer
+ * that ran out of attempts during an outage is retried instead of staying dead silently.
+ */
+export function resumeStalledPeerRecovery(
+  context: PeerConnectionManagerContext,
+  handlers: RecoveryHandlers
+): void {
+  const { callbacks, logger, state } = context;
+
+  if (!callbacks.isSignalingConnected())
+    return;
+
+  state.disconnectedPeerTracker.forEach((info, peerId) => {
+    if (!info.recoveryFailed)
+      return;
+
+    logger.info('Re-arming P2P recovery after signaling returned', { peerId });
+
+    info.recoveryFailed = false;
+    info.reconnectAttempts = 0;
+    state.peerRecoveryStatus$.next({ attempts: 0, peerId, status: 'retrying' });
+
+    attemptPeerReconnect(context, peerId, handlers);
+    schedulePeerReconnect(context, peerId, handlers);
+  });
+}
+
 export function attemptPeerReconnect(
  context: PeerConnectionManagerContext,
  peerId: string,
@@ -334,7 +490,7 @@ export function attemptPeerReconnect(
    handlers.removePeer(peerId, { preserveReconnectState: true });
  }

-  const localOderId = callbacks.getIdentifyCredentials()?.oderId ?? null;
+  const localOderId = callbacks.getIdentifyCredentialsForPeer(peerId)?.oderId ?? null;

  if (!localOderId) {
    logger.info('Skipping reconnect offer until logical identity is ready', { peerId });
@@ -342,7 +498,7 @@ export function attemptPeerReconnect(
    return;
  }

-  const shouldInitiate = peerId !== localOderId && localOderId < peerId;
+  const shouldInitiate = shouldInitiatePeerConnection(localOderId, peerId);

  handlers.createPeerConnection(peerId, shouldInitiate);