From 856c2925284bb6ad471ee07a1a28738fb499c29d Mon Sep 17 00:00:00 2001 From: Max Date: Wed, 19 Aug 2026 21:46:05 +0200 Subject: [PATCH] fix(sse): add a protocol-level ping/pong heartbeat to LiveWS startHeartbeat() only refreshed a client's liveness on inbound application messages (#10452), evicting a read-only subscriber that never sends anything. Send a protocol-level ping (RFC 6455 5.5.2) on each tick; conformant clients answer with a pong control frame automatically, refreshed in a new ws.on("pong") handler. A socket that stopped reading still goes unanswered and is reaped as before, so #10452's protection is unchanged. The application-level pong is kept for clients that still rely on it. Extended the integration test with a third connection mode covering both properties. --- .../fixes/10782-ws-heartbeat-ping-pong.md | 1 + src/server/ws/liveServer.ts | 24 ++++-- .../live-ws-heartbeat-keepalive.test.ts | 77 +++++++++++++------ 3 files changed, 71 insertions(+), 31 deletions(-) create mode 100644 changelog.d/fixes/10782-ws-heartbeat-ping-pong.md diff --git a/changelog.d/fixes/10782-ws-heartbeat-ping-pong.md b/changelog.d/fixes/10782-ws-heartbeat-ping-pong.md new file mode 100644 index 000000000000..23aeaf3d3a38 --- /dev/null +++ b/changelog.d/fixes/10782-ws-heartbeat-ping-pong.md @@ -0,0 +1 @@ +- fix(sse): replace LiveWS's application-only liveness check with a protocol-level `ws.ping()`/`pong` heartbeat (RFC 6455 §5.5.2) alongside the existing one, so a read-only dashboard subscriber that never sends anything survives the connection timeout — a socket that stops reading frames entirely is still reaped exactly as before (#10782) diff --git a/src/server/ws/liveServer.ts b/src/server/ws/liveServer.ts index fc640415eb90..4c2287fa20ab 100644 --- a/src/server/ws/liveServer.ts +++ b/src/server/ws/liveServer.ts @@ -11,6 +11,10 @@ * Server → Client: { type: "pong" } * Server → Client: { type: "welcome", version, sessionId, channels, backlog } * Server → Client: { type: "error", code, message } + * + * Liveness: besides the application ping/pong above, the server sends a protocol-level + * ping (RFC 6455 §5.5.2) each HEARTBEAT_INTERVAL_MS. Conformant clients answer it with a + * pong control frame automatically, so a quiet-but-alive subscriber survives. */ import { WebSocketServer, WebSocket } from "ws"; @@ -428,10 +432,13 @@ function startHeartbeat(server: WebSocketServer): void { clients.delete(clientId); continue; } - // Send the application-level heartbeat response. Only inbound client - // messages (including { type: "ping" }) refresh lastActivity; renewing it - // here would keep a half-open socket alive indefinitely (#10452). + // Send the application-level heartbeat response for clients that still rely on it. sendTo(client.ws, { type: "pong" } as WsServerMessage); + // Protocol-level ping: the client's automatic pong reply is what keeps a + // silent-but-alive subscriber alive, while a half-open socket stays silent and is + // still reaped. Nothing here refreshes lastActivity — only a received pong does, + // so #10452 holds. + client.ws.ping(); } }, HEARTBEAT_INTERVAL_MS); // Don't keep the process alive solely for the heartbeat (it is also cleared on close). @@ -567,6 +574,13 @@ export async function startLiveDashboardServer( console.error("[LiveWS] Client error %s: %s", clientId, err.message); clients.delete(clientId); }); + + // A control-frame pong (RFC 6455 §5.5.3) from the client is the only signal that + // proves the socket is not half-open (see startHeartbeat). + ws.on("pong", () => { + const current = clients.get(clientId); + if (current) current.lastActivity = Date.now(); + }); }); // Heartbeat @@ -610,9 +624,7 @@ export async function startLiveDashboardServer( // Build/test environments never auto-start regardless of the flag. function isBuildOrTest(): boolean { - return ( - isBuildProcess() || isAutomatedTestProcess() - ); + return isBuildProcess() || isAutomatedTestProcess(); } export function isLiveWsEnabled(): boolean { diff --git a/tests/integration/live-ws-heartbeat-keepalive.test.ts b/tests/integration/live-ws-heartbeat-keepalive.test.ts index b3ba3f882c73..1d04edcec5f9 100644 --- a/tests/integration/live-ws-heartbeat-keepalive.test.ts +++ b/tests/integration/live-ws-heartbeat-keepalive.test.ts @@ -1,8 +1,13 @@ -// Integration test for #10452: a server-emitted application-level pong must not -// keep a half-open client alive. Uses the real server harness from -// 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). +// Integration test for #10452 and its follow-up: a server-emitted application pong must +// not keep a half-open client alive (#10452), and a subscriber that never sends anything +// at the application level must not be evicted either. The server now also sends a +// protocol-level ping (RFC 6455 §5.5.2) that conformant clients auto-answer, which +// separates a client that stopped reading frames (simulated below by pausing the +// underlying socket) from one that is alive but silent at the application level. +// +// Uses the real server harness from tests/integration/live-ws-startup.test.ts (serial, +// --test-concurrency=1 integration runner — needs a ~50s window to cross the server's +// HEARTBEAT_TIMEOUT_MS, intentionally NOT inflated here). import assert from "node:assert/strict"; import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; import net from "node:net"; @@ -71,7 +76,7 @@ function waitForStartup( } test( - "LiveWS removes a silent socket but keeps one answering protocol heartbeats (#10452)", + "LiveWS reaps a socket that stops reading frames, but keeps a quiet-but-alive subscriber", // A fresh DATA_DIR can spend up to the server-startup allowance running the // complete migration set before the 50s heartbeat observation window. { timeout: 95_000 }, @@ -108,19 +113,22 @@ test( try { await waitForStartup(child, () => output); - const connect = (answerHeartbeat: boolean) => { + type ConnectMode = "appHeartbeat" | "quiet" | "deadSocket"; + + const connect = (mode: ConnectMode) => { const ws = new WebSocket(`ws://127.0.0.1:${port}/live-ws`, { headers: { Authorization: `Bearer ${apiKey}`, Origin: origin }, }); let heartbeat: NodeJS.Timeout | undefined; - const welcome = new Promise((resolve, reject) => { + let sessionId: string | undefined; + const welcome = new Promise((resolve, reject) => { const timeout = setTimeout(() => { reject(new Error(`Timed out waiting for welcome. Output:\n${output}`)); }, 5_000); ws.once("open", () => { ws.send(JSON.stringify({ type: "subscribe", channels: ["requests"] })); - if (answerHeartbeat) { + if (mode === "appHeartbeat") { heartbeat = setInterval(() => { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: "ping" })); @@ -130,9 +138,17 @@ test( }); ws.on("message", (data) => { - if (JSON.parse(data.toString()).type === "welcome") { + const message = JSON.parse(data.toString()); + if (message.type === "welcome") { clearTimeout(timeout); - resolve(); + sessionId = message.sessionId; + // "deadSocket": pause the underlying TCP socket so the `ws` receiver never + // sees (and auto-answers) the server's ping. It also cannot observe its own + // closure, so the assertion checks the server's disconnect log instead. + if (mode === "deadSocket") { + (ws as unknown as { _socket: import("net").Socket })._socket.pause(); + } + resolve(sessionId as string); } }); @@ -145,30 +161,41 @@ test( return { ws, welcome, stop: () => clearInterval(heartbeat) }; }; - const silent = connect(false); - const responsive = connect(true); - await Promise.all([silent.welcome, responsive.welcome]); + const deadSocket = connect("deadSocket"); + const quiet = connect("quiet"); + const appHeartbeat = connect("appHeartbeat"); + const [deadSocketId] = await Promise.all([ + deadSocket.welcome, + quiet.welcome, + appHeartbeat.welcome, + ]); // Wait past HEARTBEAT_TIMEOUT_MS (35s) plus one heartbeat interval (15s). - // The server emits application-level pong frames during this period, but - // only the responsive client sends the inbound { type: "ping" } signal. await new Promise((resolve) => setTimeout(resolve, 50_000)); - assert.equal( - silent.ws.readyState, + assert.ok( + output.includes(`Client disconnected: ${deadSocketId}`), + `Server never disconnected the socket that stopped reading frames — protocol ` + + `ping/pong must not mask a truly half-open connection (#10452). Output:\n${output}` + ); + assert.notEqual( + quiet.ws.readyState, WebSocket.CLOSED, - `Silent socket remained alive after timeout — server pong renewed lastActivity. Output:\n${output}` + `A real subscriber that sent no application-level messages was evicted — the protocol ` + + `ping/pong (auto-answered by every conformant client) must keep it alive. Output:\n${output}` ); assert.notEqual( - responsive.ws.readyState, + appHeartbeat.ws.readyState, WebSocket.CLOSED, - `Protocol-heartbeat client was terminated. Output:\n${output}` + `Application-heartbeat client was terminated. Output:\n${output}` ); - silent.stop(); - responsive.stop(); - silent.ws.close(); - responsive.ws.close(); + deadSocket.stop(); + quiet.stop(); + appHeartbeat.stop(); + quiet.ws.close(); + appHeartbeat.ws.close(); + deadSocket.ws.terminate(); } finally { terminateTree(child); }