From d4d3b816e2e7704ca9feff40144f54f1339634b5 Mon Sep 17 00:00:00 2001 From: Henry Lach Date: Fri, 18 Sep 2026 12:17:56 -0400 Subject: [PATCH] =?UTF-8?q?fix(dashboard):=20SSE=20backpressure=20+=20live?= =?UTF-8?q?ness=20=E2=80=94=20stop=20unbounded=20write=20buffering=20that?= =?UTF-8?q?=20OOM-crashed=20the=20dashboard=20after=20~39h?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stalled SSE client (background-throttled tab, sleeping laptop, half-open TCP) never fires 'close', and res.write() never throws — it buffers. Every 2 s broadcast (~60 KB for a penster batch) accumulated in that buffer until V8's 4 GB heap limit: 140,714,810 ms uptime / 2000 ms ≈ 70k ticks ≈ 60 KB per tick, matching the crash exactly. - writeToSseClients(): skip a client with writableNeedDrain or a write buffer over 2 MB; drop (destroy + remove) after 15 consecutive stalled ticks (~30 s); log drops. Extracted as a pure function for tests. - handleSSE(): ': ping' comment every 15 s and a 90 s socket idle timeout so a dead peer surfaces as error/close; remove the client on close/error from both req and res; clear the ping timer on removal. Test: dashboard-sse-backpressure.test.ts (7) incl. the 39-hour scenario — a permanently stalled client is dropped holding < ~2.1 MB. 4142 pass. --- CHANGELOG.md | 1 + dashboard/server.cjs | 107 ++++++++++++- .../tests/dashboard-sse-backpressure.test.ts | 150 ++++++++++++++++++ 3 files changed, 251 insertions(+), 7 deletions(-) create mode 100644 extensions/tests/dashboard-sse-backpressure.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index c078fe77..4c5d6bed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **Dashboard crashed with `JavaScript heap out of memory` after ~39 hours of a long batch.** `res.write()` on a stalled SSE client never throws — Node buffered every 2-second state broadcast (~60 KB for a large batch) for a browser tab that had stopped reading (background-throttled, laptop asleep, half-open TCP) until V8's 4 GB heap limit. Broadcasts now skip clients with an undrained write buffer, drop them after ~30 s of continuous backpressure, ping every 15 s so dead peers surface, apply a socket idle timeout, and remove clients on `close`/`error` from either side. - **A released hold left the task badged `held` and let its lane successor show `running` with the wrong telemetry** (#651). Three writers: (1) the strict hold-store write that persists a release (or an open) now projects the bound task's record in the same write — open → `held`, released → `running` with the stale hold-timeout reason cleared — instead of waiting for the next unrelated task-transition persist; (2) resume reconciliation decides "session alive" per **task** from the registry manifests' task ids, so a not-yet-started successor sharing the lane's session name stays `pending` (legacy registries without task ids keep the old session semantics); (3) the dashboard attaches a lane's live worker telemetry to the task named by the lane snapshot — including a row still badged `held` — never to a sibling that merely has `status: running`. - **Stale `⏸️ Lane held` alert after a fast ruling.** The escalation reaches the supervisor live, so a ruling is often queued before the worker exits; the hold diff --git a/dashboard/server.cjs b/dashboard/server.cjs index 04152efe..18667d92 100644 --- a/dashboard/server.cjs +++ b/dashboard/server.cjs @@ -1327,6 +1327,71 @@ function serveStatic(req, res) { const sseClients = new Set(); +/** + * SSE backpressure + liveness (heap-OOM fix). + * + * `res.write()` on a stalled client never throws — Node queues the chunk in the + * socket's write buffer and returns false. A browser tab that stopped reading + * (background-throttled, laptop asleep, half-open TCP after a VPN blip) never + * fires `close`, so every broadcast appended another full state payload to a + * buffer nobody drained. A penster batch (~60 KB per tick, every 2 s) reached + * V8's 4 GB heap limit after ~39 hours and the dashboard crashed. + * + * Rules: + * - a client whose write buffer exceeds SSE_MAX_BUFFERED_BYTES is skipped this + * tick (backpressure) and dropped once it has been stalled for + * SSE_MAX_STALLED_TICKS consecutive ticks; + * - a `: ping` comment every SSE_PING_MS and a socket idle timeout make a dead + * peer error out so `close` actually fires; + * - `close`/`error` on BOTH req and res remove the client. + */ +const SSE_MAX_BUFFERED_BYTES = 2 * 1024 * 1024; +const SSE_MAX_STALLED_TICKS = 15; // ≈30 s of consecutive backpressure at POLL_INTERVAL +const SSE_PING_MS = 15_000; +const SSE_SOCKET_TIMEOUT_MS = 90_000; + +/** + * Write one payload to every SSE client with backpressure. Pure over the + * client set: clients expose `write`, `destroy`, `writableLength`, + * `writableNeedDrain` (Node Writable) plus a mutable `_sseStalledTicks`. + * Returns { written, skipped, dropped } for tests/logging. + */ +function writeToSseClients(clients, payload, opts = {}) { + const maxBuffered = opts.maxBufferedBytes ?? SSE_MAX_BUFFERED_BYTES; + const maxStalled = opts.maxStalledTicks ?? SSE_MAX_STALLED_TICKS; + let written = 0; + let skipped = 0; + let dropped = 0; + for (const client of clients) { + const buffered = client.writableLength ?? 0; + const stalled = client.writableNeedDrain === true || buffered > maxBuffered; + if (stalled) { + client._sseStalledTicks = (client._sseStalledTicks ?? 0) + 1; + if (client._sseStalledTicks > maxStalled) { + clients.delete(client); + dropped++; + try { + client.destroy(); + } catch { + /* already gone */ + } + } else { + skipped++; + } + continue; + } + client._sseStalledTicks = 0; + try { + client.write(payload); + written++; + } catch { + clients.delete(client); + dropped++; + } + } + return { written, skipped, dropped }; +} + // ─── Conversation JSONL ───────────────────────────────────────────────── function serveConversation(req, res, prefix) { @@ -1368,19 +1433,47 @@ function handleSSE(req, res) { res.write(`data: ${JSON.stringify(state)}\n\n`); sseClients.add(res); - req.on("close", () => sseClients.delete(res)); + res._sseStalledTicks = 0; + + // Liveness: a dead peer must surface as an error/close, not as an ever-growing + // write buffer. Pings are SSE comments (ignored by EventSource). + const ping = setInterval(() => { + if (!sseClients.has(res)) return; + try { + res.write(": ping\n\n"); + } catch { + remove(); + } + }, SSE_PING_MS); + if (res.socket && typeof res.socket.setTimeout === "function") { + res.socket.setTimeout(SSE_SOCKET_TIMEOUT_MS, () => { + try { + res.destroy(); + } catch { + /* ignore */ + } + }); + } + let removed = false; + const remove = () => { + if (removed) return; + removed = true; + clearInterval(ping); + sseClients.delete(res); + }; + req.on("close", remove); + req.on("error", remove); + res.on("close", remove); + res.on("error", remove); } function broadcastState() { if (sseClients.size === 0) return; const state = buildDashboardState(); const payload = `data: ${JSON.stringify(state)}\n\n`; - for (const client of sseClients) { - try { - client.write(payload); - } catch { - sseClients.delete(client); - } + const r = writeToSseClients(sseClients, payload); + if (r.dropped > 0) { + console.error(`[dashboard] dropped ${r.dropped} stalled SSE client(s) (write buffer never drained)`); } } diff --git a/extensions/tests/dashboard-sse-backpressure.test.ts b/extensions/tests/dashboard-sse-backpressure.test.ts new file mode 100644 index 00000000..e3fcc007 --- /dev/null +++ b/extensions/tests/dashboard-sse-backpressure.test.ts @@ -0,0 +1,150 @@ +/** + * Dashboard SSE backpressure + liveness — heap-OOM fix. + * + * A stalled SSE client used to accumulate every broadcast in its write buffer + * (res.write never throws), reaching V8's heap limit after ~39 h of a penster + * batch. The broadcast path must skip stalled clients, drop them after a + * bounded number of stalled ticks, and the SSE handler must remove clients on + * close/error from either side and ping so dead peers surface. + */ + +import { describe, it } from "node:test"; +import { readFileSync } from "node:fs"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { expect } from "./expect.ts"; + +const HERE = dirname(fileURLToPath(import.meta.url)); +const source = readFileSync(join(HERE, "..", "..", "dashboard", "server.cjs"), "utf-8").replace( + /\r\n/g, + "\n", +); + +function extract(startMarker: string, endMarker: string): string { + const start = source.indexOf(startMarker); + const end = source.indexOf(endMarker, start); + if (start < 0 || end < 0) throw new Error(`cannot extract ${startMarker}`); + return source.slice(start, end); +} + +// Pull the constants + writeToSseClients as a pure module. +const block = extract("const SSE_MAX_BUFFERED_BYTES", "function handleSSE(req, res) {"); +const factory = new Function( + `${block}\nreturn { writeToSseClients, SSE_MAX_BUFFERED_BYTES, SSE_MAX_STALLED_TICKS };`, +); +const { writeToSseClients, SSE_MAX_BUFFERED_BYTES, SSE_MAX_STALLED_TICKS } = factory() as { + writeToSseClients: ( + clients: Set, + payload: string, + opts?: any, + ) => { written: number; skipped: number; dropped: number }; + SSE_MAX_BUFFERED_BYTES: number; + SSE_MAX_STALLED_TICKS: number; +}; + +function fakeClient(over: Record = {}) { + const c: any = { + writes: [] as string[], + destroyed: false, + writableLength: 0, + writableNeedDrain: false, + write(p: string) { + c.writes.push(p); + return true; + }, + destroy() { + c.destroyed = true; + }, + ...over, + }; + return c; +} + +describe("dashboard SSE — writeToSseClients", () => { + it("writes to healthy clients and resets their stall counter", () => { + const a = fakeClient(); + const b = fakeClient({ _sseStalledTicks: 3 }); + const clients = new Set([a, b]); + const r = writeToSseClients(clients, "data: x\n\n"); + expect(r).toEqual({ written: 2, skipped: 0, dropped: 0 }); + expect(a.writes.length).toBe(1); + expect(b._sseStalledTicks).toBe(0); + }); + + it("SKIPS a client whose buffer is over the cap or that needs drain — nothing is appended to its buffer", () => { + const bloated = fakeClient({ writableLength: SSE_MAX_BUFFERED_BYTES + 1 }); + const draining = fakeClient({ writableNeedDrain: true }); + const clients = new Set([bloated, draining]); + const r = writeToSseClients(clients, "data: x\n\n"); + expect(r).toEqual({ written: 0, skipped: 2, dropped: 0 }); + expect(bloated.writes.length).toBe(0); + expect(draining.writes.length).toBe(0); + expect(clients.size).toBe(2); + expect(bloated._sseStalledTicks).toBe(1); + }); + + it("DROPS (destroy + remove) a client stalled for more than SSE_MAX_STALLED_TICKS consecutive ticks; a recovered client survives", () => { + const dead = fakeClient({ writableNeedDrain: true }); + const flaky = fakeClient({ writableNeedDrain: true }); + const clients = new Set([dead, flaky]); + for (let i = 0; i < SSE_MAX_STALLED_TICKS; i++) writeToSseClients(clients, "d"); + expect(clients.size).toBe(2); // at the cap, not yet dropped + flaky.writableNeedDrain = false; // peer caught up + const r = writeToSseClients(clients, "d"); + expect(r).toEqual({ written: 1, skipped: 0, dropped: 1 }); + expect(dead.destroyed).toBe(true); + expect(clients.has(dead)).toBe(false); + expect(clients.has(flaky)).toBe(true); + expect(flaky._sseStalledTicks).toBe(0); + expect(flaky.writes.length).toBe(1); + }); + + it("a write that throws removes the client (existing behaviour preserved)", () => { + const boom = fakeClient({ + write() { + throw new Error("EPIPE"); + }, + }); + const clients = new Set([boom]); + const r = writeToSseClients(clients, "d"); + expect(r.dropped).toBe(1); + expect(clients.size).toBe(0); + }); + + it("39-hour scenario: a permanently stalled client never accumulates more than the cap", () => { + let buffered = 0; + const stalled = fakeClient({ + write(p: string) { + buffered += p.length; + stalled.writableLength = buffered; + return false; + }, + }); + const clients = new Set([stalled]); + const payload = "x".repeat(60_000); + for (let i = 0; i < 70_000 && clients.size > 0; i++) writeToSseClients(clients, payload); + expect(clients.size).toBe(0); + expect(buffered).toBeLessThan(SSE_MAX_BUFFERED_BYTES + 60_000 * 2); + expect(stalled.destroyed).toBe(true); + }); +}); + +describe("dashboard SSE — handler wiring", () => { + it("handleSSE pings, sets a socket timeout, and removes the client on close/error from req AND res", () => { + const fn = extract("function handleSSE(req, res) {", "function broadcastState() {"); + expect(fn).toContain('res.write(": ping\\n\\n");'); + expect(fn).toContain("res.socket.setTimeout(SSE_SOCKET_TIMEOUT_MS"); + expect(fn).toContain('req.on("close", remove);'); + expect(fn).toContain('req.on("error", remove);'); + expect(fn).toContain('res.on("close", remove);'); + expect(fn).toContain('res.on("error", remove);'); + expect(fn).toContain("clearInterval(ping);"); + }); + + it("broadcastState goes through writeToSseClients and logs drops", () => { + const fn = extract("function broadcastState() {", "// ─── Batch History API"); + expect(fn).toContain("const r = writeToSseClients(sseClients, payload);"); + expect(fn).toContain("dropped $" + "{r.dropped} stalled SSE client(s)"); + }); +});