From daeca0050ddf433fd2452a1edce63521773dede7 Mon Sep 17 00:00:00 2001 From: Carlos Valente Date: Tue, 15 Sep 2026 18:20:36 +0200 Subject: [PATCH] fix: detect and recover from silently dropped websocket connections --- .../src/common/utils/__tests__/socket.test.ts | 162 +++++++++++++ apps/client/src/common/utils/socket.ts | 220 +++++++++++++++--- apps/server/src/adapters/WebsocketAdapter.ts | 50 ++++ .../__tests__/WebsocketAdapter.test.ts | 42 +++- 4 files changed, 437 insertions(+), 37 deletions(-) create mode 100644 apps/client/src/common/utils/__tests__/socket.test.ts diff --git a/apps/client/src/common/utils/__tests__/socket.test.ts b/apps/client/src/common/utils/__tests__/socket.test.ts new file mode 100644 index 000000000..21a96191c --- /dev/null +++ b/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>() })); +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); + }); +}); diff --git a/apps/client/src/common/utils/socket.ts b/apps/client/src/common/utils/socket.ts index a9366cc36..f843227e3 100644 --- a/apps/client/src/common/utils/socket.ts +++ b/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 diff --git a/apps/server/src/adapters/WebsocketAdapter.ts b/apps/server/src/adapters/WebsocketAdapter.ts index 008266c44..617e97969 100644 --- a/apps/server/src/adapters/WebsocketAdapter.ts +++ b/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; + /** Liveness state is keyed by socket because the heartbeat iterates sockets. */ + private readonly connections: Map; + 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(); + this.connections = new Map(); 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 { + 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(); }); }); diff --git a/apps/server/src/adapters/__tests__/WebsocketAdapter.test.ts b/apps/server/src/adapters/__tests__/WebsocketAdapter.test.ts index 0b0d080d1..38ac09e15 100644 --- a/apps/server/src/adapters/__tests__/WebsocketAdapter.test.ts +++ b/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 void>>(); on(event: string, handler: (...args: unknown[]) => void) { @@ -31,6 +35,10 @@ const websocketMocks = vi.hoisted(() => { class FakeWebSocketServer { clients = new Set(); + 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(); + }); });