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 = () => {
+  socket.onclose = () => {
+    if (!isCurrent()) {
+      return;
+    }
     console.warn('WebSocket disconnected');
-    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)`);
-      if (websocket && websocket.readyState === WebSocket.CLOSED) {
-        reconnectAttempts += 1;
-        connectSocket();
-      }
-    }, delay);
+    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

apps/server/src/adapters/WebsocketAdapter.ts

@@ -38,11 +38,19 @@ import type { IAdapter } from './IAdapter.js';
 type ClientId = string;
 let instance: SocketServer | null = null;
 
+/** Timestamp of the last sign of life received from a client. */
+type LastSeen = number;
+
 class SocketServer implements IAdapter {
   private readonly MAX_PAYLOAD = 1024 * 256; // 256Kb
+  private readonly HEARTBEAT_INTERVAL = 5000;
+  private readonly HEARTBEAT_TIMEOUT = 15000;
 
   private wss: WebSocketServer | null;
   private readonly clients: Map<ClientId, Client>;
+  /** Liveness state is keyed by socket because the heartbeat iterates sockets. */
+  private readonly connections: Map<WebSocket, LastSeen>;
+  private heartbeat: NodeJS.Timeout | null = null;
   private lastConnection: Date | null = null;
   private shouldShowWelcome = true;
 
@@ -54,12 +62,14 @@ class SocketServer implements IAdapter {
     // eslint-disable-next-line @typescript-eslint/no-this-alias -- this logic is used to ensure singleton
     instance = this;
     this.clients = new Map<ClientId, Client>();
+    this.connections = new Map<WebSocket, LastSeen>();
     this.wss = null;
   }
 
   init(server: Server, showWelcome: boolean, prefix?: string) {
     this.shouldShowWelcome = showWelcome;
     this.wss = new WebSocketServer({ path: `${prefix}/ws`, server, maxPayload: this.MAX_PAYLOAD });
+    this.startHeartbeat();
 
     this.wss.on('connection', (ws, req) => {
       // Rejected sockets can emit an error while their close handshake is in progress.
@@ -94,6 +104,7 @@ class SocketServer implements IAdapter {
         origin: '',
         path: '',
       });
+      this.connections.set(ws, Date.now());
 
       this.lastConnection = new Date();
       logger.info(LogOrigin.Client, `${this.clients.size} Connections with new: ${clientName}`);
@@ -105,8 +116,13 @@ class SocketServer implements IAdapter {
       // send store payload on connect
       sendPacket(MessageTag.RuntimeData, eventStore.poll());
 
+      // Browser WebSockets reply to protocol pings automatically.
+      ws.on('pong', () => {
+        this.connections.set(ws, Date.now());
+      });
       ws.on('close', () => {
         this.clients.delete(clientId);
+        this.connections.delete(ws);
         logger.info(LogOrigin.Client, `${this.clients.size} Connections with disconnected: ${clientName}`);
         this.sendClientList();
       });
@@ -180,6 +196,38 @@ class SocketServer implements IAdapter {
     };
   }
 
+  /** Terminates connections that stop answering protocol pings. */
+  private startHeartbeat() {
+    this.stopHeartbeat();
+    this.heartbeat = setInterval(() => {
+      const now = Date.now();
+      this.wss?.clients.forEach((client) => {
+        const lastSeen = this.connections.get(client) ?? 0;
+
+        if (now - lastSeen > this.HEARTBEAT_TIMEOUT) {
+          logger.warning(LogOrigin.Client, 'Terminating unresponsive client');
+          // A non-responsive socket cannot complete a close handshake.
+          client.terminate();
+          return;
+        }
+
+        if (client.readyState === WebSocket.OPEN) {
+          client.ping();
+        }
+      });
+    }, this.HEARTBEAT_INTERVAL);
+
+    // The HTTP server, not this interval, owns process lifetime.
+    this.heartbeat.unref();
+  }
+
+  private stopHeartbeat() {
+    if (this.heartbeat) {
+      clearInterval(this.heartbeat);
+      this.heartbeat = null;
+    }
+  }
+
   private getOrCreateClient(clientId: ClientId): Client {
     if (!this.clients.has(clientId)) {
       this.clients.set(clientId, {
@@ -245,6 +293,7 @@ class SocketServer implements IAdapter {
   }
 
   shutdown(): Promise<void> {
+    this.stopHeartbeat();
     const wss = this.wss;
     if (!wss) {
       return Promise.resolve();
@@ -260,6 +309,7 @@ class SocketServer implements IAdapter {
 
       wss.close(() => {
         this.wss = null;
+        this.connections.clear();
         resolve();
       });
     });

apps/server/src/adapters/__tests__/WebsocketAdapter.test.ts

@@ -1,15 +1,19 @@
 import type { Server } from 'node:http';
 
-import { afterEach, describe, expect, it, vi } from 'vitest';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 
 const websocketMocks = vi.hoisted(() => {
   let connectionHandler: ((socket: FakeWebSocket, request: unknown) => void) | undefined;
+  let latestServer: FakeWebSocketServer | undefined;
 
   class FakeWebSocket {
     static readonly OPEN = 1;
 
     readyState = FakeWebSocket.OPEN;
     close = vi.fn();
+    ping = vi.fn();
+    send = vi.fn();
+    terminate = vi.fn();
     handlers = new Map<string, Array<(...args: unknown[]) => void>>();
 
     on(event: string, handler: (...args: unknown[]) => void) {
@@ -31,6 +35,10 @@ const websocketMocks = vi.hoisted(() => {
   class FakeWebSocketServer {
     clients = new Set<FakeWebSocket>();
 
+    constructor() {
+      latestServer = this;
+    }
+
     on(event: string, handler: (socket: FakeWebSocket, request: unknown) => void) {
       if (event === 'connection') {
         connectionHandler = handler;
@@ -47,9 +55,12 @@ const websocketMocks = vi.hoisted(() => {
     FakeWebSocket,
     FakeWebSocketServer,
     getConnectionHandler: () => connectionHandler,
+    getLatestServer: () => latestServer,
   };
 });
 
+const authenticationMocks = vi.hoisted(() => ({ reject: true }));
+
 vi.mock('ws', () => ({
   WebSocket: websocketMocks.FakeWebSocket,
   WebSocketServer: websocketMocks.FakeWebSocketServer,
@@ -57,15 +68,20 @@ vi.mock('ws', () => ({
 
 vi.mock('../../middleware/authenticate.js', () => ({
   authenticateSocket: (_socket: unknown, _request: unknown, next: (error?: Error) => void) => {
-    next(new Error('Unauthorized'));
+    next(authenticationMocks.reject ? new Error('Unauthorized') : undefined);
   },
 }));
 
 import { socket } from '../WebsocketAdapter.js';
 
-describe('WebsocketAdapter authentication', () => {
+describe('WebsocketAdapter', () => {
+  beforeEach(() => {
+    authenticationMocks.reject = true;
+  });
+
   afterEach(async () => {
     await socket.shutdown();
+    vi.useRealTimers();
   });
 
   it('handles an error emitted while rejecting an unauthenticated socket', () => {
@@ -79,4 +95,24 @@ describe('WebsocketAdapter authentication', () => {
     expect(rejectedSocket.close).toHaveBeenCalledWith(1008, 'Unauthorized');
     expect(() => rejectedSocket.emit('error', new Error('socket closed'))).not.toThrow();
   });
+
+  it('terminates a client that sends messages but does not answer protocol pings', () => {
+    vi.useFakeTimers();
+    authenticationMocks.reject = false;
+    socket.init({} as Server, false);
+    const client = new websocketMocks.FakeWebSocket();
+    const server = websocketMocks.getLatestServer();
+    const connectionHandler = websocketMocks.getConnectionHandler();
+
+    expect(server).toBeDefined();
+    expect(connectionHandler).toBeDefined();
+    server?.clients.add(client);
+    connectionHandler?.(client, {});
+
+    vi.advanceTimersByTime(10_000);
+    client.emit('message', Buffer.from(JSON.stringify({ tag: 'ping', payload: null })));
+
+    vi.advanceTimersByTime(10_000);
+    expect(client.terminate).toHaveBeenCalledOnce();
+  });
 });