From 55e8b26ace442fbed669ed2c5d9f43cb45a8da8d Mon Sep 17 00:00:00 2001 From: adevwithpurpose Date: Sat, 15 Aug 2026 03:03:22 -0300 Subject: [PATCH] fix(dashboard): send periodic WS heartbeat pings to stop live-dashboard reconnect churn The live-dashboard WS client (src/hooks/useLiveDashboard.ts) only sent a subscribe frame on open and never emitted the protocol's { type: "ping" } heartbeat. The server (src/server/ws/liveServer.ts) refreshes client liveness only from inbound messages and terminates any client idle past HEARTBEAT_TIMEOUT_MS (35s), so a healthy, connected-but-idle dashboard client was force-terminated roughly every 35-45s, causing constant reconnect churn (#10319). Fix (both directions, per the analyzed plan): - Client: start a 15s ping interval on open, cleared on close/unmount/ reconnect, so the connection stays inside the server's liveness window. - Server (defense in depth): the outbound heartbeat pong now also bumps client.lastActivity, so even a third-party client that never pings is not dropped for being idle. Regression coverage: - tests/unit/useLiveDashboard-heartbeat.test.tsx: fast fake-timer check that the hook emits periodic ping frames and cleans up the interval on close/unmount (no leaked timers). - tests/integration/live-ws-heartbeat-keepalive.test.ts: real WS-server integration test asserting a silent-but-subscribed client stays connected past the 35s heartbeat timeout (~50s window), converted from the plan file's TDD RED repro. Closes #10319 --- .../fixes/10319-live-ws-heartbeat-ping.md | 1 + src/hooks/useLiveDashboard.ts | 37 +++- src/server/ws/liveServer.ts | 6 +- .../live-ws-heartbeat-keepalive.test.ts | 167 ++++++++++++++++ .../unit/useLiveDashboard-heartbeat.test.tsx | 183 ++++++++++++++++++ 5 files changed, 391 insertions(+), 3 deletions(-) create mode 100644 changelog.d/fixes/10319-live-ws-heartbeat-ping.md create mode 100644 tests/integration/live-ws-heartbeat-keepalive.test.ts create mode 100644 tests/unit/useLiveDashboard-heartbeat.test.tsx diff --git a/changelog.d/fixes/10319-live-ws-heartbeat-ping.md b/changelog.d/fixes/10319-live-ws-heartbeat-ping.md new file mode 100644 index 0000000000..91ed7b00c5 --- /dev/null +++ b/changelog.d/fixes/10319-live-ws-heartbeat-ping.md @@ -0,0 +1 @@ +- fix(dashboard): send periodic WS heartbeat pings so live dashboard connections stop dropping every ~35s (#10319) diff --git a/src/hooks/useLiveDashboard.ts b/src/hooks/useLiveDashboard.ts index 05c991a31f..d257605237 100644 --- a/src/hooks/useLiveDashboard.ts +++ b/src/hooks/useLiveDashboard.ts @@ -19,6 +19,11 @@ import { deriveLiveWsPath } from "@/shared/utils/wsPath"; const WS_RECONNECT_DELAYS = [1000, 2000, 4000, 8000, 16000, 30000]; +// Must stay <= the server's HEARTBEAT_TIMEOUT_MS (35s in src/server/ws/liveServer.ts) so a +// healthy, connected-but-idle client is never force-terminated by the server's heartbeat sweep. +// Matches the server's own HEARTBEAT_INTERVAL_MS (15s). +const CLIENT_PING_INTERVAL_MS = 15_000; + /** Only accept ws:// or wss:// URLs (mirrors the guard in src/app/api/v1/ws/route.ts). */ function sanitizeWsPublicUrl(url: unknown): string | null { if (typeof url !== "string" || url.length === 0) return null; @@ -152,7 +157,15 @@ export function useLiveDashboard({ const [events, setEvents] = useState([]); const wsRef = useRef(null); const reconnectTimeoutRef = useRef | null>(null); + const pingIntervalRef = useRef | null>(null); const mountedRef = useRef(true); + + const stopPingHeartbeat = useCallback(() => { + if (pingIntervalRef.current) { + clearInterval(pingIntervalRef.current); + pingIntervalRef.current = null; + } + }, []); const maxEvents = 500; const onEventRef = useRef(onEvent); @@ -189,6 +202,16 @@ export function useLiveDashboard({ // Subscribe to channels ws.send(JSON.stringify({ type: "subscribe", channels })); + + // Heartbeat: send a periodic ping so the server (which only refreshes + // liveness from inbound messages) never terminates a healthy, idle + // connection for exceeding its inactivity timeout (#10319). + stopPingHeartbeat(); + pingIntervalRef.current = setInterval(() => { + if (ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ type: "ping" })); + } + }, CLIENT_PING_INTERVAL_MS); }; ws.onmessage = (event) => { @@ -236,6 +259,7 @@ export function useLiveDashboard({ }; ws.onclose = () => { + stopPingHeartbeat(); if (!mountedRef.current) return; wsRef.current = null; setConnection((prev) => ({ @@ -271,7 +295,14 @@ export function useLiveDashboard({ error: err instanceof Error ? err.message : "Connection failed", })); } - }, [effectiveWsUrl, apiKey, channels.join(","), autoReconnect, connection.reconnectAttempt]); + }, [ + effectiveWsUrl, + apiKey, + channels.join(","), + autoReconnect, + connection.reconnectAttempt, + stopPingHeartbeat, + ]); // Connect on mount and on reconnect trigger useEffect(() => { @@ -281,6 +312,7 @@ export function useLiveDashboard({ clearTimeout(reconnectTimeoutRef.current); reconnectTimeoutRef.current = null; } + stopPingHeartbeat(); wsRef.current?.close(); wsRef.current = null; setConnection({ @@ -302,9 +334,10 @@ export function useLiveDashboard({ if (reconnectTimeoutRef.current) { clearTimeout(reconnectTimeoutRef.current); } + stopPingHeartbeat(); wsRef.current?.close(); }; - }, [connect, enabled, wsUrlResolved]); + }, [connect, enabled, wsUrlResolved, stopPingHeartbeat]); // Connect (for manual retry) const reconnect = useCallback(() => { diff --git a/src/server/ws/liveServer.ts b/src/server/ws/liveServer.ts index 3c5f91c4e1..5922f84a83 100644 --- a/src/server/ws/liveServer.ts +++ b/src/server/ws/liveServer.ts @@ -428,7 +428,11 @@ function startHeartbeat(server: WebSocketServer): void { clients.delete(clientId); continue; } - // Send ping + // Send ping. Also refresh liveness on our own outbound heartbeat (defense in + // depth, #10319): lastActivity was previously inbound-only, so a client that + // never sends { type: "ping" } (e.g. a third-party/bespoke client) would still + // get force-terminated even though the connection is healthy. + client.lastActivity = now; sendTo(client.ws, { type: "pong" } as WsServerMessage); } }, HEARTBEAT_INTERVAL_MS); diff --git a/tests/integration/live-ws-heartbeat-keepalive.test.ts b/tests/integration/live-ws-heartbeat-keepalive.test.ts new file mode 100644 index 0000000000..2caa96b716 --- /dev/null +++ b/tests/integration/live-ws-heartbeat-keepalive.test.ts @@ -0,0 +1,167 @@ +// Integration test for #10319: a healthy, subscribed-but-otherwise-silent LiveWS +// client must never be terminated by the server's heartbeat sweep. Uses the same +// spawn-the-real-server harness pattern as tests/integration/live-ws-startup.test.ts +// (serial, --test-concurrency=1 integration runner — this test needs a ~50s window +// to cross the server's HEARTBEAT_TIMEOUT_MS, which is intentionally NOT inflated +// here; do not shrink this test's window to "make it fast" — that would stop +// exercising the real timeout). +import assert from "node:assert/strict"; +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import net from "node:net"; +import test from "node:test"; +import WebSocket from "ws"; + +function getFreePort(): Promise { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + server.close(() => { + if (address && typeof address === "object") resolve(address.port); + else reject(new Error("Failed to allocate a local port")); + }); + }); + }); +} + +function terminateTree(child: ChildProcessWithoutNullStreams): void { + if (!child.pid) return; + try { + process.kill(-child.pid, "SIGTERM"); + } catch { + child.kill("SIGTERM"); + } +} + +function waitForStartup( + child: ChildProcessWithoutNullStreams, + getOutput: () => string +): Promise { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error(`LiveWS startup timed out. Output:\n${getOutput()}`)); + }, 30_000); + + const onData = () => { + const output = getOutput(); + if (output.includes("Dashboard WebSocket server listening")) { + cleanup(); + resolve(); + } + }; + + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + cleanup(); + reject( + new Error(`LiveWS exited before listening: code=${code} signal=${signal}\n${getOutput()}`) + ); + }; + + const cleanup = () => { + clearTimeout(timeout); + child.stdout.off("data", onData); + child.stderr.off("data", onData); + child.off("exit", onExit); + }; + + child.stdout.on("data", onData); + child.stderr.on("data", onData); + child.once("exit", onExit); + onData(); + }); +} + +test( + "LiveWS keeps a subscribed-but-silent client connected past the heartbeat timeout (#10319)", + { timeout: 65_000 }, + async () => { + const port = await getFreePort(); + const apiKey = "test-live-ws-heartbeat-key"; + const jwtSecret = "test-live-ws-heartbeat-jwt-secret"; + const origin = "http://localhost"; + let output = ""; + + const child = spawn(process.execPath, ["scripts/start-ws-server.mjs"], { + cwd: process.cwd(), + detached: process.platform !== "win32", + env: { + ...process.env, + NODE_ENV: "test", + OMNIROUTE_API_KEY: apiKey, + JWT_SECRET: jwtSecret, + LIVE_WS_HOST: "127.0.0.1", + LIVE_WS_PORT: String(port), + LIVE_WS_ALLOWED_ORIGINS: origin, + }, + }); + + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + output += chunk; + }); + child.stderr.on("data", (chunk) => { + output += chunk; + }); + + try { + await waitForStartup(child, () => output); + + let closed = false; + let closeCode: number | undefined; + + const welcomeReceived = new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error(`Timed out waiting for welcome. Output:\n${output}`)); + }, 5_000); + + const ws = new WebSocket(`ws://127.0.0.1:${port}/live-ws`, { + headers: { Authorization: `Bearer ${apiKey}`, Origin: origin }, + }); + + ws.once("open", () => { + ws.send(JSON.stringify({ type: "subscribe", channels: ["requests"] })); + }); + + ws.on("message", (data) => { + const msg = JSON.parse(data.toString()); + if (msg.type === "welcome") { + clearTimeout(timeout); + resolve(); + } + }); + + ws.once("close", (code) => { + closed = true; + closeCode = code; + }); + + ws.once("error", (error) => { + clearTimeout(timeout); + reject(new Error(`LiveWS client failed: ${error.message}. Output:\n${output}`)); + }); + + // Deliberately stay silent after subscribing — this models the buggy + // client (never pings). The FIX under test lives server-side: the + // server's own outbound heartbeat pong now refreshes lastActivity, so + // even a silent client must not be terminated. + }); + + await welcomeReceived; + + // Wait past HEARTBEAT_TIMEOUT_MS (35s) + a full HEARTBEAT_INTERVAL_MS (15s) + // margin so at least one heartbeat sweep has had the chance to (wrongly) + // terminate an idle-but-healthy connection. + await new Promise((resolve) => setTimeout(resolve, 50_000)); + + assert.equal( + closed, + false, + `Silent-but-subscribed client was terminated (closeCode=${closeCode}) — #10319 regressed. Output:\n${output}` + ); + } finally { + terminateTree(child); + } + } +); diff --git a/tests/unit/useLiveDashboard-heartbeat.test.tsx b/tests/unit/useLiveDashboard-heartbeat.test.tsx new file mode 100644 index 0000000000..1514065647 --- /dev/null +++ b/tests/unit/useLiveDashboard-heartbeat.test.tsx @@ -0,0 +1,183 @@ +// @vitest-environment jsdom +// +// Fast regression guard for #10319 (live dashboard WS client never sends a +// heartbeat ping, so the server's 35s inactivity sweep terminates a healthy, +// idle-but-subscribed client). This is the cheap unit-ish check; the slow +// end-to-end confirmation (real server, ~50s window) lives in +// tests/integration/live-ws-heartbeat-keepalive.test.ts. +import React, { act } from "react"; +import { createRoot } from "react-dom/client"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { useLiveDashboard } from "../../src/hooks/useLiveDashboard"; + +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; + onmessage: ((event: { data: string }) => void) | null = null; + onclose: (() => void) | null = null; + onerror: (() => void) | null = null; + + constructor(public url: string) { + MockWebSocket.instances.push(this); + } + + send(data: string): void { + this.sent.push(data); + } + + close(): void { + if (this.readyState === MockWebSocket.CLOSED) return; + this.readyState = MockWebSocket.CLOSED; + this.onclose?.(); + } + + triggerOpen(): void { + this.readyState = MockWebSocket.OPEN; + this.onopen?.(); + } +} + +const cleanupCallbacks: Array<() => void> = []; + +function makeContainer(): HTMLElement { + const container = document.createElement("div"); + document.body.appendChild(container); + cleanupCallbacks.push(() => container.remove()); + return container; +} + +describe("useLiveDashboard heartbeat ping (#10319)", () => { + beforeEach(() => { + ( + globalThis as typeof globalThis & { IS_REACT_ACT_ENVIRONMENT?: boolean } + ).IS_REACT_ACT_ENVIRONMENT = true; + MockWebSocket.instances = []; + vi.stubGlobal("WebSocket", MockWebSocket); + vi.useFakeTimers(); + }); + + afterEach(() => { + while (cleanupCallbacks.length > 0) { + cleanupCallbacks.pop()?.(); + } + document.body.innerHTML = ""; + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + it("sends a periodic { type: 'ping' } frame after the connection opens", () => { + const container = makeContainer(); + const root = createRoot(container); + + function C() { + useLiveDashboard({ wsUrl: "ws://localhost:20132/live-ws", channels: ["requests"] }); + return null; + } + + act(() => { + root.render(); + }); + + const instance = MockWebSocket.instances[0]; + expect(instance).toBeDefined(); + + act(() => { + instance.triggerOpen(); + }); + + // On open, only the subscribe frame has been sent — no ping yet. + expect(instance.sent.some((m) => JSON.parse(m).type === "ping")).toBe(false); + expect(instance.sent.some((m) => JSON.parse(m).type === "subscribe")).toBe(true); + + // Advance past the client's heartbeat interval (15s) — must now have pinged. + act(() => { + vi.advanceTimersByTime(15_000); + }); + + const pings = instance.sent.filter((m) => JSON.parse(m).type === "ping"); + expect(pings.length).toBeGreaterThanOrEqual(1); + + // And it keeps pinging periodically (regression guard against a one-shot timer). + act(() => { + vi.advanceTimersByTime(30_000); + }); + const pingsAfterMore = instance.sent.filter((m) => JSON.parse(m).type === "ping"); + expect(pingsAfterMore.length).toBeGreaterThan(pings.length); + }); + + it("stops pinging after the socket closes (no leaked interval)", () => { + const container = makeContainer(); + const root = createRoot(container); + + function C() { + useLiveDashboard({ + wsUrl: "ws://localhost:20132/live-ws", + channels: ["requests"], + autoReconnect: false, + }); + return null; + } + + act(() => { + root.render(); + }); + + const instance = MockWebSocket.instances[0]; + act(() => { + instance.triggerOpen(); + }); + act(() => { + vi.advanceTimersByTime(15_000); + }); + const pingsBeforeClose = instance.sent.filter((m) => JSON.parse(m).type === "ping").length; + expect(pingsBeforeClose).toBeGreaterThanOrEqual(1); + + act(() => { + instance.close(); + }); + + act(() => { + vi.advanceTimersByTime(60_000); + }); + const pingsAfterClose = instance.sent.filter((m) => JSON.parse(m).type === "ping").length; + expect(pingsAfterClose).toBe(pingsBeforeClose); + }); + + it("clears the ping interval on unmount", () => { + const container = makeContainer(); + const root = createRoot(container); + + function C() { + useLiveDashboard({ wsUrl: "ws://localhost:20132/live-ws", channels: ["requests"] }); + return null; + } + + act(() => { + root.render(); + }); + const instance = MockWebSocket.instances[0]; + act(() => { + instance.triggerOpen(); + }); + act(() => { + vi.advanceTimersByTime(15_000); + }); + const pingsBeforeUnmount = instance.sent.filter((m) => JSON.parse(m).type === "ping").length; + + act(() => { + root.unmount(); + }); + act(() => { + vi.advanceTimersByTime(60_000); + }); + const pingsAfterUnmount = instance.sent.filter((m) => JSON.parse(m).type === "ping").length; + expect(pingsAfterUnmount).toBe(pingsBeforeUnmount); + }); +});