Repair connectivity correctly v1

2026-05-17 15:15:14 +02:00
parent e769a6ee4a
commit 9d0a4478b2
18 changed files with 1125 additions and 25 deletions
--- a/toju-app/src/app/infrastructure/realtime/peer-connection-manager/recovery/peer-recovery.spec.ts
+++ b/toju-app/src/app/infrastructure/realtime/peer-connection-manager/recovery/peer-recovery.spec.ts
@@ -0,0 +1,216 @@
+import { DATA_CHANNEL_RECOVERY_GRACE_MS, DATA_CHANNEL_STATE_OPEN } from '../../realtime.constants';
+import type { PeerData } from '../../realtime.types';
+import {
+  createPeerConnectionManagerState,
+  PeerConnectionManagerContext,
+  RecoveryHandlers
+} from '../shared';
+import { scheduleDataChannelRecovery } from './peer-recovery';
+
+describe('peer recovery', () => {
+  afterEach(() => {
+    vi.clearAllTimers();
+    vi.useRealTimers();
+  });
+
+  it('waits a short grace period before replacing a closed data channel in place', () => {
+    vi.useFakeTimers();
+
+    const channel = createDataChannel('closed');
+    const context = createContext('alice');
+    const handlers = createRecoveryHandlers(context);
+
+    context.state.activePeerConnections.set('bob', createPeerData(channel, 'connected'));
+
+    scheduleDataChannelRecovery(context, 'bob', channel, 'close', handlers);
+
+    vi.advanceTimersByTime(DATA_CHANNEL_RECOVERY_GRACE_MS - 1);
+    expect(handlers.removePeer).not.toHaveBeenCalled();
+    expect(handlers.createPeerConnection).not.toHaveBeenCalled();
+
+    vi.advanceTimersByTime(1);
+    expect(handlers.replaceDataChannel).toHaveBeenCalledWith('bob', channel);
+    expect(handlers.removePeer).not.toHaveBeenCalled();
+    expect(handlers.createPeerConnection).not.toHaveBeenCalled();
+    expect(handlers.createAndSendOffer).not.toHaveBeenCalled();
+  });
+
+  it('falls back to full peer recreation when in-place data channel replacement fails', () => {
+    vi.useFakeTimers();
+
+    const channel = createDataChannel('closed');
+    const context = createContext('alice');
+    const handlers = createRecoveryHandlers(context);
+
+    handlers.replaceDataChannel.mockReturnValueOnce(false);
+    context.state.activePeerConnections.set('bob', createPeerData(channel, 'connected'));
+
+    scheduleDataChannelRecovery(context, 'bob', channel, 'close', handlers);
+    vi.advanceTimersByTime(DATA_CHANNEL_RECOVERY_GRACE_MS);
+
+    expect(handlers.removePeer).toHaveBeenCalledWith('bob', { preserveReconnectState: true });
+    expect(handlers.createPeerConnection).toHaveBeenCalledWith('bob', true);
+    expect(handlers.createAndSendOffer).toHaveBeenCalledWith('bob');
+  });
+
+  it('does not recreate a peer when a replacement data channel is adopted before the grace expires', () => {
+    vi.useFakeTimers();
+
+    const staleChannel = createDataChannel('closed');
+    const replacementChannel = createDataChannel(DATA_CHANNEL_STATE_OPEN);
+    const context = createContext('alice');
+    const handlers = createRecoveryHandlers(context);
+
+    context.state.activePeerConnections.set('bob', createPeerData(staleChannel, 'connected'));
+
+    scheduleDataChannelRecovery(context, 'bob', staleChannel, 'close', handlers);
+    context.state.activePeerConnections.set('bob', createPeerData(replacementChannel, 'connected'));
+
+    vi.advanceTimersByTime(DATA_CHANNEL_RECOVERY_GRACE_MS);
+
+    expect(handlers.removePeer).not.toHaveBeenCalled();
+    expect(handlers.replaceDataChannel).not.toHaveBeenCalled();
+    expect(handlers.createPeerConnection).not.toHaveBeenCalled();
+  });
+
+  it('does not schedule recovery for an open data channel', () => {
+    vi.useFakeTimers();
+
+    const channel = createDataChannel(DATA_CHANNEL_STATE_OPEN);
+    const context = createContext('alice');
+    const handlers = createRecoveryHandlers(context);
+
+    context.state.activePeerConnections.set('bob', createPeerData(channel, 'connected'));
+
+    scheduleDataChannelRecovery(context, 'bob', channel, 'error', handlers);
+    vi.advanceTimersByTime(DATA_CHANNEL_RECOVERY_GRACE_MS);
+
+    expect(handlers.removePeer).not.toHaveBeenCalled();
+    expect(handlers.replaceDataChannel).not.toHaveBeenCalled();
+    expect(handlers.createPeerConnection).not.toHaveBeenCalled();
+  });
+
+  it('preserves a connected non-initiator peer while waiting for the remote initiator to replace the channel', () => {
+    vi.useFakeTimers();
+
+    const channel = createDataChannel('closed');
+    const context = createContext('zoe');
+    const handlers = createRecoveryHandlers(context);
+
+    context.state.activePeerConnections.set('bob', createPeerData(channel, 'connected', false));
+    scheduleDataChannelRecovery(context, 'bob', channel, 'close', handlers);
+    vi.advanceTimersByTime(DATA_CHANNEL_RECOVERY_GRACE_MS);
+
+    expect(handlers.removePeer).not.toHaveBeenCalled();
+    expect(handlers.replaceDataChannel).not.toHaveBeenCalled();
+    expect(handlers.createPeerConnection).not.toHaveBeenCalled();
+    expect(handlers.createAndSendOffer).not.toHaveBeenCalled();
+  });
+
+  it('waits for the remote initiator when a non-connected peer needs full reconnect', () => {
+    vi.useFakeTimers();
+
+    const channel = createDataChannel('closed');
+    const context = createContext('zoe');
+    const handlers = createRecoveryHandlers(context);
+
+    context.state.activePeerConnections.set('bob', createPeerData(channel, 'disconnected'));
+
+    scheduleDataChannelRecovery(context, 'bob', channel, 'close', handlers);
+    vi.advanceTimersByTime(DATA_CHANNEL_RECOVERY_GRACE_MS);
+
+    expect(handlers.createPeerConnection).toHaveBeenCalledWith('bob', false);
+    expect(handlers.createAndSendOffer).not.toHaveBeenCalled();
+  });
+});
+
+function createContext(localOderId: string): PeerConnectionManagerContext {
+  return {
+    logger: {
+      error: vi.fn(),
+      info: vi.fn(),
+      logStream: vi.fn(),
+      traffic: vi.fn(),
+      warn: vi.fn()
+    } as unknown as PeerConnectionManagerContext['logger'],
+    callbacks: {
+      getIceServers: vi.fn(() => []),
+      getIdentifyCredentials: vi.fn(() => ({ oderId: localOderId, displayName: localOderId })),
+      getLocalMediaStream: vi.fn(() => null),
+      getLocalPeerId: vi.fn(() => localOderId),
+      getVoiceStateSnapshot: vi.fn(() => ({
+        isConnected: true,
+        isMuted: false,
+        isDeafened: false,
+        isScreenSharing: false,
+        roomId: 'voice-room',
+        serverId: 'server-1'
+      })),
+      isCameraEnabled: vi.fn(() => false),
+      isScreenSharingActive: vi.fn(() => false),
+      isSignalingConnected: vi.fn(() => true),
+      sendRawMessage: vi.fn()
+    },
+    state: createPeerConnectionManagerState()
+  };
+}
+
+function createRecoveryHandlers(context: PeerConnectionManagerContext): RecoveryHandlers & {
+  createAndSendOffer: ReturnType<typeof vi.fn>;
+  createPeerConnection: ReturnType<typeof vi.fn>;
+  removePeer: ReturnType<typeof vi.fn>;
+  replaceDataChannel: ReturnType<typeof vi.fn>;
+} {
+  return {
+    createAndSendOffer: vi.fn(async () => undefined),
+    createPeerConnection: vi.fn((peerId: string, isInitiator: boolean) => {
+      const peerData = createPeerData(createDataChannel(isInitiator ? 'connecting' : 'closed'), 'new', isInitiator);
+
+      context.state.activePeerConnections.set(peerId, peerData);
+      return peerData;
+    }),
+    removePeer: vi.fn((peerId: string) => {
+      context.state.activePeerConnections.delete(peerId);
+    }),
+    replaceDataChannel: vi.fn((peerId: string, expectedChannel: RTCDataChannel) => {
+      const peerData = context.state.activePeerConnections.get(peerId);
+
+      if (!peerData || peerData.dataChannel !== expectedChannel) {
+        return false;
+      }
+
+      peerData.dataChannel = createDataChannel('connecting');
+      return true;
+    })
+  };
+}
+
+function createPeerData(
+  dataChannel: RTCDataChannel,
+  connectionState: RTCPeerConnectionState,
+  isInitiator = true
+): PeerData {
+  return {
+    audioSender: undefined,
+    connection: {
+      close: vi.fn(),
+      connectionState
+    } as unknown as RTCPeerConnection,
+    createdAt: Date.now(),
+    dataChannel,
+    isInitiator,
+    pendingIceCandidates: [],
+    remoteCameraStreamIds: new Set<string>(),
+    remoteScreenShareStreamIds: new Set<string>(),
+    remoteVoiceStreamIds: new Set<string>(),
+    videoSender: undefined
+  };
+}
+
+function createDataChannel(readyState: RTCDataChannelState): RTCDataChannel {
+  return {
+    bufferedAmount: 0,
+    label: 'chat',
+    readyState
+  } as unknown as RTCDataChannel;
+}
--- a/toju-app/src/app/infrastructure/realtime/peer-connection-manager/recovery/peer-recovery.ts
+++ b/toju-app/src/app/infrastructure/realtime/peer-connection-manager/recovery/peer-recovery.ts
@@ -1,5 +1,6 @@
 import {
  CONNECTION_STATE_CONNECTED,
+  DATA_CHANNEL_RECOVERY_GRACE_MS,
  DATA_CHANNEL_STATE_OPEN,
  P2P_TYPE_VOICE_STATE_REQUEST,
  PEER_DISCONNECT_GRACE_MS,
@@ -27,6 +28,7 @@ export function removePeer(
  const preserveReconnectState = options?.preserveReconnectState === true;

  clearPeerDisconnectGraceTimer(state, peerId);
+  clearDataChannelRecoveryTimer(state, peerId);

  if (!preserveReconnectState) {
    clearPeerReconnectTimer(state, peerId);
@@ -56,6 +58,7 @@ export function removePeer(
 export function closeAllPeers(state: PeerConnectionManagerState): void {
  clearAllPeerReconnectTimers(state);
  clearAllPeerDisconnectGraceTimers(state);
+  clearAllDataChannelRecoveryTimers(state);
  clearAllPingTimers(state);

  state.activePeerConnections.forEach((peerData) => {
@@ -106,6 +109,18 @@ export function clearPeerDisconnectGraceTimer(
  }
 }

+export function clearDataChannelRecoveryTimer(
+  state: PeerConnectionManagerState,
+  peerId: string
+): void {
+  const timer = state.dataChannelRecoveryTimers.get(peerId);
+
+  if (timer) {
+    clearTimeout(timer);
+    state.dataChannelRecoveryTimers.delete(peerId);
+  }
+}
+
 /** Cancel all pending peer reconnect timers and clear the tracker. */
 export function clearAllPeerReconnectTimers(state: PeerConnectionManagerState): void {
  state.peerReconnectTimers.forEach((timer) => clearInterval(timer));
@@ -118,6 +133,85 @@ export function clearAllPeerDisconnectGraceTimers(state: PeerConnectionManagerSt
  state.peerDisconnectGraceTimers.clear();
 }

+export function clearAllDataChannelRecoveryTimers(state: PeerConnectionManagerState): void {
+  state.dataChannelRecoveryTimers.forEach((timer) => clearTimeout(timer));
+  state.dataChannelRecoveryTimers.clear();
+}
+
+export function scheduleDataChannelRecovery(
+  context: PeerConnectionManagerContext,
+  peerId: string,
+  channel: RTCDataChannel,
+  reason: string,
+  handlers: RecoveryHandlers
+): void {
+  const { logger, state } = context;
+  const peerData = state.activePeerConnections.get(peerId);
+
+  if (!peerData || peerData.dataChannel !== channel)
+    return;
+
+  if (channel.readyState === DATA_CHANNEL_STATE_OPEN)
+    return;
+
+  if (state.dataChannelRecoveryTimers.has(peerId))
+    return;
+
+  logger.warn('[data-channel] Control channel unavailable; waiting before reconnect', {
+    channelLabel: channel.label,
+    peerId,
+    readyState: channel.readyState,
+    reason
+  });
+
+  const timer = setTimeout(() => {
+    state.dataChannelRecoveryTimers.delete(peerId);
+
+    const latestPeerData = state.activePeerConnections.get(peerId);
+
+    if (!latestPeerData || latestPeerData.dataChannel !== channel)
+      return;
+
+    if (latestPeerData.dataChannel?.readyState === DATA_CHANNEL_STATE_OPEN)
+      return;
+
+    logger.warn('[data-channel] Control channel did not recover; selecting repair path', {
+      channelLabel: channel.label,
+      connectionState: latestPeerData.connection.connectionState,
+      peerId,
+      readyState: latestPeerData.dataChannel?.readyState ?? null,
+      reason
+    });
+
+    if (latestPeerData.connection.connectionState === CONNECTION_STATE_CONNECTED) {
+      if (latestPeerData.isInitiator && handlers.replaceDataChannel(peerId, channel)) {
+        logger.info('[data-channel] Replaced control channel without recreating media transport', {
+          peerId,
+          reason
+        });
+
+        return;
+      }
+
+      if (!latestPeerData.isInitiator) {
+        logger.info('[data-channel] Waiting for initiator to replace control channel; preserving media transport', {
+          peerId,
+          reason
+        });
+
+        return;
+      }
+    }
+
+    trackDisconnectedPeer(state, peerId);
+    handlers.removePeer(peerId, { preserveReconnectState: true });
+    attemptPeerReconnect(context, peerId, handlers);
+    schedulePeerReconnect(context, peerId, handlers);
+  }, DATA_CHANNEL_RECOVERY_GRACE_MS);
+
+  state.dataChannelRecoveryTimers.set(peerId, timer);
+}
+
 export function schedulePeerDisconnectRecovery(
  context: PeerConnectionManagerContext,
  peerId: string,