fix(dashboard): SSE backpressure + liveness — stop unbounded write buffering that OOM-crashed the dashboard after ~39h

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

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)`);
   }
 }
 

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<any>,
+		payload: string,
+		opts?: any,
+	) => { written: number; skipped: number; dropped: number };
+	SSE_MAX_BUFFERED_BYTES: number;
+	SSE_MAX_STALLED_TICKS: number;
+};
+
+function fakeClient(over: Record<string, unknown> = {}) {
+	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)");
+	});
+});