fix: detect and recover from silently dropped websocket connections

apps/client/src/common/utils/__tests__/socket.test.ts

@@ -0,0 +1,162 @@
+// @vitest-environment happy-dom
+import { MessageTag } from 'ontime-types';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { addLog } from '../../stores/logger';
+import { socketConfig } from '../socket';
+
+const { watchdogInterval, silenceTimeout, connectTimeout } = socketConfig;
+
+/** The clock is the server's regular sign of life. */
+const PUBLISH_INTERVAL = 1000;
+
+/** Browser WebSocket stand-in that can stop delivering without closing. */
+class MockWebSocket {
+  static readonly CONNECTING = 0;
+  static readonly OPEN = 1;
+  static readonly CLOSING = 2;
+  static readonly CLOSED = 3;
+
+  static instances: MockWebSocket[] = [];
+
+  readyState = MockWebSocket.CONNECTING;
+  sent: string[] = [];
+
+  onopen: (() => void) | null = null;
+  onclose: (() => void) | null = null;
+  onerror: ((error: unknown) => void) | null = null;
+  onmessage: ((event: { data: string }) => void) | null = null;
+
+  constructor(public readonly url: string) {
+    MockWebSocket.instances.push(this);
+  }
+
+  send(data: string) {
+    this.sent.push(data);
+  }
+
+  close() {
+    this.readyState = MockWebSocket.CLOSED;
+    this.onclose?.();
+  }
+
+  open() {
+    this.readyState = MockWebSocket.OPEN;
+    this.onopen?.();
+  }
+
+  receive(tag: MessageTag, payload: unknown) {
+    this.onmessage?.({ data: JSON.stringify({ tag, payload }) });
+  }
+}
+
+vi.mock('../../api/utils', () => ({ invalidateAllCaches: vi.fn<() => Promise<void>>() }));
+vi.mock('../../stores/logger', () => ({ addLog: vi.fn() }));
+
+describe('socket connection watchdog', () => {
+  let connectSocket: () => void;
+
+  beforeEach(async () => {
+    vi.useFakeTimers();
+    MockWebSocket.instances = [];
+    vi.mocked(addLog).mockClear();
+    vi.stubGlobal('WebSocket', MockWebSocket);
+    // the module keeps connection state in module scope, we need a clean one for each test
+    vi.resetModules();
+    connectSocket = (await import('../socket')).connectSocket;
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.unstubAllGlobals();
+  });
+
+  /** Opens a connection and returns its socket. */
+  function openConnection() {
+    connectSocket();
+    const socket = MockWebSocket.instances[0];
+    socket.open();
+    return socket;
+  }
+
+  it('holds a connection which keeps delivering data', () => {
+    const socket = openConnection();
+
+    // Runtime clock updates keep the connection alive without client polling.
+    for (let elapsed = 0; elapsed < silenceTimeout * 3; elapsed += PUBLISH_INTERVAL) {
+      vi.advanceTimersByTime(PUBLISH_INTERVAL);
+      socket.receive(MessageTag.RuntimeData, { clock: elapsed });
+    }
+
+    expect(socket.sent).toHaveLength(1);
+    expect(MockWebSocket.instances).toHaveLength(1);
+  });
+
+  it('replaces a connection which stops delivering data without closing', () => {
+    const socket = openConnection();
+
+    vi.advanceTimersByTime(PUBLISH_INTERVAL);
+    socket.receive(MessageTag.RuntimeData, { clock: 0 });
+
+    // The browser still considers this socket open.
+    vi.advanceTimersByTime(silenceTimeout);
+    expect(socket.readyState).toBe(MockWebSocket.OPEN);
+
+    vi.advanceTimersByTime(watchdogInterval);
+    expect(socket.readyState).toBe(MockWebSocket.CLOSED);
+    expect(MockWebSocket.instances).toHaveLength(2);
+  });
+
+  it('gives up on a connection attempt which never completes', () => {
+    connectSocket();
+    const socket = MockWebSocket.instances[0];
+
+    // The socket neither opens nor reports an error.
+    vi.advanceTimersByTime(connectTimeout + watchdogInterval);
+
+    expect(socket.readyState).toBe(MockWebSocket.CLOSED);
+    expect(MockWebSocket.instances).toHaveLength(2);
+  });
+
+  it('logs one warning while repeated connection attempts time out', () => {
+    connectSocket();
+
+    vi.advanceTimersByTime(connectTimeout + watchdogInterval + socketConfig.reconnectBaseInterval * 2);
+    vi.advanceTimersByTime(connectTimeout + watchdogInterval);
+
+    expect(addLog).toHaveBeenCalledOnce();
+  });
+
+  it('does not open a second connection while one is alive', () => {
+    openConnection();
+
+    connectSocket();
+
+    expect(MockWebSocket.instances).toHaveLength(1);
+  });
+
+  it('reconnects after the socket closes', () => {
+    const socket = openConnection();
+
+    socket.close();
+    expect(MockWebSocket.instances).toHaveLength(1);
+
+    // Reconnection uses the same backoff policy.
+    vi.advanceTimersByTime(socketConfig.reconnectBaseInterval * 2);
+    expect(MockWebSocket.instances).toHaveLength(2);
+  });
+
+  it('ignores events from a socket which has been replaced', () => {
+    const stale = openConnection();
+
+    vi.advanceTimersByTime(silenceTimeout + watchdogInterval);
+    expect(MockWebSocket.instances).toHaveLength(2);
+    MockWebSocket.instances[1].open();
+
+    // A late close from the old socket must not disturb its replacement.
+    stale.close();
+
+    vi.advanceTimersByTime(watchdogInterval);
+    expect(MockWebSocket.instances).toHaveLength(2);
+  });
+});

apps/client/src/common/utils/socket.ts

@@ -1,6 +1,8 @@
 import {
   ApiActionTag,
   Log,
+  LogLevel,
+  LogOrigin,
   MaybeNumber,
   MessageTag,
   RefetchKey,
@@ -8,6 +10,7 @@ import {
   WsPacketToClient,
   WsPacketToServer,
 } from 'ontime-types';
+import { generateId, millisToString } from 'ontime-utils';
 
 import { isProduction, websocketUrl } from '../../externals';
 import {
@@ -39,28 +42,53 @@ import {
 import { addDialog } from '../stores/dialogStore';
 import { addLog } from '../stores/logger';
 import { patchRuntime, patchRuntimeProperty } from '../stores/runtime';
+import { nowInMillis } from './time';
 
 let websocket: WebSocket | null = null;
 let reconnectTimeout: NodeJS.Timeout | null = null;
-const socketConfig = {
-  reconnectBaseInterval: 1000, // 1 second
-  reconnectMaxInterval: 30000, // 30 seconds
-  reconnectMinInterval: 500, // 0.5 seconds
+let watchdogInterval: NodeJS.Timeout | null = null;
+export const socketConfig = {
+  reconnectBaseInterval: 1000,
+  reconnectMaxInterval: 30000,
+  reconnectMinInterval: 500,
   reconnectJitter: 0.25,
-  offlineAttemptsThreshold: 2, // when we consider the client disconnected
+  offlineAttemptsThreshold: 2,
+  watchdogInterval: 2000,
+  silenceTimeout: 10000,
+  connectTimeout: 10000,
 } as const;
 
 export const getConnectionState = () => hasConnected;
 export const getReconnectAttempts = () => reconnectAttempts;
 let hasConnected = false;
 let reconnectAttempts = 0;
+let lastContact = 0;
+let hasLoggedConnectionIssue = false;
 
 export const connectSocket = () => {
-  websocket = new WebSocket(websocketUrl);
+  if (websocket && (websocket.readyState === WebSocket.CONNECTING || websocket.readyState === WebSocket.OPEN)) {
+    return;
+  }
+
+  if (reconnectTimeout) {
+    clearTimeout(reconnectTimeout);
+    reconnectTimeout = null;
+  }
+
+  const socket = new WebSocket(websocketUrl);
+  websocket = socket;
+  registerConnectionAttempt();
+  startWatchdog();
 
   const preferredClientName = getClientName();
 
-  websocket.onopen = () => {
+  // Replaced sockets must not schedule reconnects for their replacement.
+  const isCurrent = () => websocket === socket;
+
+  socket.onopen = () => {
+    if (!isCurrent()) {
+      return;
+    }
     const isReconnect = hasConnected;
     if (reconnectTimeout) {
       clearTimeout(reconnectTimeout);
@@ -68,6 +96,7 @@ export const connectSocket = () => {
     }
     hasConnected = true;
     reconnectAttempts = 0;
+    registerContact();
 
     sendSocket(MessageTag.ClientSet, {
       type: 'ontime',
@@ -82,38 +111,26 @@ export const connectSocket = () => {
     setOnlineStatus(true);
   };
 
-  websocket.onclose = () => {
-    console.warn('WebSocket disconnected');
-    if (reconnectTimeout) {
-      clearTimeout(reconnectTimeout);
-      reconnectTimeout = null;
+  socket.onclose = () => {
+    if (!isCurrent()) {
+      return;
     }
-
-    const exponentialDelay = Math.min(
-      socketConfig.reconnectBaseInterval * 2 ** reconnectAttempts,
-      socketConfig.reconnectMaxInterval,
-    );
-    const jitterOffset = exponentialDelay * socketConfig.reconnectJitter * (Math.random() * 2 - 1);
-    const delay = Math.max(socketConfig.reconnectMinInterval, Math.round(exponentialDelay + jitterOffset));
-
-    reconnectTimeout = setTimeout(() => {
-      reconnectTimeout = null;
-      if (reconnectAttempts > socketConfig.offlineAttemptsThreshold) {
-        setOnlineStatus(false);
-      }
-      console.warn(`WebSocket: reconnecting now (#${reconnectAttempts + 1}, waited ${delay}ms)`);
-      if (websocket && websocket.readyState === WebSocket.CLOSED) {
-        reconnectAttempts += 1;
-        connectSocket();
-      }
-    }, delay);
+    console.warn('WebSocket disconnected');
+    scheduleReconnect();
   };
 
-  websocket.onerror = (error) => {
+  socket.onerror = (error) => {
     console.error('WebSocket error:', error);
   };
 
-  websocket.onmessage = async (event) => {
+  socket.onmessage = async (event) => {
+    if (!isCurrent()) {
+      return;
+    }
+
+    // Any server message proves the connection is still delivering.
+    registerContact();
+
     try {
       const data = JSON.parse(event.data) as WsPacketToClient;
 
@@ -125,7 +142,8 @@ export const connectSocket = () => {
 
       switch (tag) {
         case MessageTag.Pong: {
-          const offset = (new Date().getTime() - new Date(payload).getTime()) * 0.5;
+          // a round trip can be faster than the clock resolution, we keep the value positive since a ping <= 0 means offline
+          const offset = Math.max(1, (new Date().getTime() - new Date(payload).getTime()) * 0.5);
           patchRuntimeProperty('ping', offset);
           updateDevTools({ ping: offset });
           break;
@@ -234,6 +252,140 @@ export const connectSocket = () => {
   };
 };
 
+function scheduleReconnect() {
+  if (reconnectTimeout) {
+    clearTimeout(reconnectTimeout);
+    reconnectTimeout = null;
+  }
+
+  const exponentialDelay = Math.min(
+    socketConfig.reconnectBaseInterval * 2 ** reconnectAttempts,
+    socketConfig.reconnectMaxInterval,
+  );
+  const jitterOffset = exponentialDelay * socketConfig.reconnectJitter * (Math.random() * 2 - 1);
+  const delay = Math.max(socketConfig.reconnectMinInterval, Math.round(exponentialDelay + jitterOffset));
+
+  reconnectTimeout = setTimeout(() => {
+    reconnectTimeout = null;
+    if (reconnectAttempts > socketConfig.offlineAttemptsThreshold) {
+      setOnlineStatus(false);
+    }
+    console.warn(`WebSocket: reconnecting now (#${reconnectAttempts + 1}, waited ${delay}ms)`);
+    reconnectAttempts += 1;
+    connectSocket();
+  }, delay);
+}
+
+/**
+ * Drops the current connection and immediately opens a new one.
+ * Used when we have reason to believe the socket is no longer delivering data.
+ */
+function reconnectNow(reason: string) {
+  logConnectionIssue(reason);
+  detachSocket();
+  connectSocket();
+}
+
+/** Detaches a socket before closing it so late events cannot affect its replacement. */
+function detachSocket() {
+  const previous = websocket;
+  websocket = null;
+
+  if (!previous) {
+    return;
+  }
+
+  previous.onopen = null;
+  previous.onclose = null;
+  previous.onerror = null;
+  previous.onmessage = null;
+  try {
+    previous.close();
+  } catch (_) {
+    // The socket is unusable either way.
+  }
+}
+
+function registerContact() {
+  lastContact = Date.now();
+  hasLoggedConnectionIssue = false;
+}
+
+function registerConnectionAttempt() {
+  lastContact = Date.now();
+}
+
+/**
+ * Replaces silent connections. The server publishes a clock update every second, even
+ * while paused, so extended silence is evidence of a dropped connection.
+ */
+function checkConnection() {
+  if (!websocket) {
+    return;
+  }
+
+  const silentFor = Date.now() - lastContact;
+
+  if (websocket.readyState === WebSocket.CONNECTING) {
+    // A connection attempt can otherwise hang indefinitely.
+    if (silentFor > socketConfig.connectTimeout) {
+      reconnectNow('WebSocket: connection attempt timed out');
+    }
+    return;
+  }
+
+  if (websocket.readyState === WebSocket.OPEN && silentFor > socketConfig.silenceTimeout) {
+    reconnectNow('WebSocket: no data from server, reconnecting');
+  }
+}
+
+function startWatchdog() {
+  if (watchdogInterval) {
+    return;
+  }
+  watchdogInterval = setInterval(checkConnection, socketConfig.watchdogInterval);
+}
+
+/** Records connection failures locally because server logs may be unreachable. */
+function logConnectionIssue(text: string) {
+  if (hasLoggedConnectionIssue) {
+    return;
+  }
+
+  hasLoggedConnectionIssue = true;
+  console.warn(text);
+  addLog({
+    id: generateId(),
+    origin: LogOrigin.Client,
+    time: millisToString(nowInMillis()),
+    level: LogLevel.Warn,
+    text,
+  });
+}
+
+/** Rechecks a socket when a resumed browser may have lost it while timers were suspended. */
+function handleEnvironmentChange() {
+  if (!websocket || websocket.readyState === WebSocket.CLOSED || websocket.readyState === WebSocket.CLOSING) {
+    // A pending reconnect may be delayed by backoff, so retry immediately.
+    reconnectAttempts = 0;
+    connectSocket();
+    return;
+  }
+
+  checkConnection();
+}
+
+if (typeof window !== 'undefined') {
+  window.addEventListener('online', handleEnvironmentChange);
+  window.addEventListener('pageshow', handleEnvironmentChange);
+  window.addEventListener('focus', handleEnvironmentChange);
+  document.addEventListener('visibilitychange', () => {
+    if (document.visibilityState === 'visible') {
+      handleEnvironmentChange();
+    }
+  });
+}
+
 export function maybeInvalidateRundownCache(revision: MaybeNumber, rundownId?: string) {
   if (!rundownId) {
     // we omit rundownId to signify invalidate all rundowns