fix(dashboard): SSE stall eviction is elapsed-time based; one idempotent teardown for every exit path (Sage review of #664)

dashboard/server.cjs

@@ -1340,52 +1340,70 @@ const sseClients = new Set();
  * 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;
+ *    SSE_MAX_STALLED_MS of continuous backpressure (elapsed time, not 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
+// Elapsed-time grace, not a tick count: fs.watch debounce bursts during a busy
+// batch would otherwise burn a tick budget in seconds (Sage review).
+const SSE_MAX_STALLED_MS = 30_000;
 const SSE_PING_MS = 15_000;
 const SSE_SOCKET_TIMEOUT_MS = 90_000;
 
+/**
+ * One idempotent teardown for every SSE exit path (client close/error, stall
+ * eviction, synchronous write failure): unregister, clear the ping timer,
+ * destroy the connection. Nothing may retain a response after this runs.
+ */
+function teardownSseClient(clients, client) {
+  if (client._sseTornDown) return;
+  client._sseTornDown = true;
+  clients.delete(client);
+  if (client._ssePing) {
+    clearInterval(client._ssePing);
+    client._ssePing = null;
+  }
+  try {
+    client.destroy();
+  } catch {
+    /* already gone */
+  }
+}
+
 /**
  * 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`.
+ * `writableNeedDrain` (Node Writable) plus mutable `_sseStalledSince` / `_sseTornDown` / `_ssePing`.
  * 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;
+	const maxStalledMs = opts.maxStalledMs ?? SSE_MAX_STALLED_MS;
+	const now = opts.now ?? Date.now();
 	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);
+			if (client._sseStalledSince == null) client._sseStalledSince = now;
+			if (now - client._sseStalledSince > maxStalledMs) {
+				teardownSseClient(clients, client);
 				dropped++;
-				try {
-					client.destroy();
-				} catch {
-					/* already gone */
-				}
 			} else {
 				skipped++;
 			}
 			continue;
 		}
-		client._sseStalledTicks = 0;
+		client._sseStalledSince = null;
 		try {
 			client.write(payload);
 			written++;
 		} catch {
-			clients.delete(client);
+			teardownSseClient(clients, client);
 			dropped++;
 		}
 	}
@@ -1433,34 +1451,24 @@ function handleSSE(req, res) {
   res.write(`data: ${JSON.stringify(state)}\n\n`);
 
   sseClients.add(res);
-  res._sseStalledTicks = 0;
+  res._sseStalledSince = null;
+  res._sseTornDown = false;
 
   // 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;
+  // write buffer. Pings are SSE comments (ignored by EventSource). The timer is
+  // owned by the client record so every teardown path can clear it.
+  res._ssePing = setInterval(() => {
+    if (res._sseTornDown) return;
     try {
       res.write(": ping\n\n");
     } catch {
-      remove();
+      teardownSseClient(sseClients, res);
     }
   }, SSE_PING_MS);
   if (res.socket && typeof res.socket.setTimeout === "function") {
-    res.socket.setTimeout(SSE_SOCKET_TIMEOUT_MS, () => {
-      try {
-        res.destroy();
-      } catch {
-        /* ignore */
-      }
-    });
+    res.socket.setTimeout(SSE_SOCKET_TIMEOUT_MS, () => teardownSseClient(sseClients, res));
   }
-  let removed = false;
-  const remove = () => {
-    if (removed) return;
-    removed = true;
-    clearInterval(ping);
-    sseClients.delete(res);
-  };
+  const remove = () => teardownSseClient(sseClients, res);
   req.on("close", remove);
   req.on("error", remove);
   res.on("close", remove);

extensions/tests/dashboard-sse-backpressure.test.ts

@@ -4,7 +4,7 @@
  * 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
+ * bounded stall duration (elapsed time, not ticks), and the SSE handler must remove clients on
  * close/error from either side and ping so dead peers surface.
  */
 
@@ -31,17 +31,19 @@ function extract(startMarker: string, endMarker: string): string {
 // 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 };`,
+	`${block}\nreturn { writeToSseClients, teardownSseClient, SSE_MAX_BUFFERED_BYTES, SSE_MAX_STALLED_MS };`,
 );
-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;
-};
+const { writeToSseClients, teardownSseClient, SSE_MAX_BUFFERED_BYTES, SSE_MAX_STALLED_MS } =
+	factory() as {
+		writeToSseClients: (
+			clients: Set<any>,
+			payload: string,
+			opts?: any,
+		) => { written: number; skipped: number; dropped: number };
+		teardownSseClient: (clients: Set<any>, client: any) => void;
+		SSE_MAX_BUFFERED_BYTES: number;
+		SSE_MAX_STALLED_MS: number;
+	};
 
 function fakeClient(over: Record<string, unknown> = {}) {
 	const c: any = {
@@ -62,57 +64,90 @@ function fakeClient(over: Record<string, unknown> = {}) {
 }
 
 describe("dashboard SSE — writeToSseClients", () => {
-	it("writes to healthy clients and resets their stall counter", () => {
+	const T0 = 1_000_000;
+
+	it("writes to healthy clients and clears their stall mark", () => {
 		const a = fakeClient();
-		const b = fakeClient({ _sseStalledTicks: 3 });
+		const b = fakeClient({ _sseStalledSince: T0 - 5000 });
 		const clients = new Set([a, b]);
-		const r = writeToSseClients(clients, "data: x\n\n");
+		const r = writeToSseClients(clients, "data: x\n\n", { now: T0 });
 		expect(r).toEqual({ written: 2, skipped: 0, dropped: 0 });
 		expect(a.writes.length).toBe(1);
-		expect(b._sseStalledTicks).toBe(0);
+		expect(b._sseStalledSince).toBe(null);
 	});
 
-	it("SKIPS a client whose buffer is over the cap or that needs drain — nothing is appended to its buffer", () => {
+	it("SKIPS a client whose buffer is over the cap or that needs drain — nothing is appended; stall clock starts", () => {
 		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");
+		const r = writeToSseClients(clients, "data: x\n\n", { now: T0 });
 		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);
+		expect(bloated._sseStalledSince).toBe(T0);
 	});
 
-	it("DROPS (destroy + remove) a client stalled for more than SSE_MAX_STALLED_TICKS consecutive ticks; a recovered client survives", () => {
+	it("eviction is ELAPSED-TIME based (Sage): 100 bursty broadcasts inside 3 s do not evict; > SSE_MAX_STALLED_MS does; 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");
+		// fs.watch debounce storm: 100 broadcasts in 3 seconds
+		for (let i = 0; i < 100; i++) writeToSseClients(clients, "d", { now: T0 + i * 30 });
+		expect(clients.size).toBe(2);
+		// at the boundary: still kept
+		writeToSseClients(clients, "d", { now: T0 + SSE_MAX_STALLED_MS });
+		expect(clients.size).toBe(2);
+		flaky.writableNeedDrain = false; // peer caught up just in time
+		const r = writeToSseClients(clients, "d", { now: T0 + SSE_MAX_STALLED_MS + 1 });
 		expect(r).toEqual({ written: 1, skipped: 0, dropped: 1 });
 		expect(dead.destroyed).toBe(true);
+		expect(dead._sseTornDown).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);
+		expect(flaky._sseStalledSince).toBe(null);
 	});
 
-	it("a write that throws removes the client (existing behaviour preserved)", () => {
+	it("a write that throws goes through the SAME teardown: removed, destroyed, ping timer cleared (Sage)", () => {
+		let cleared = false;
+		const realClear = globalThis.clearInterval;
 		const boom = fakeClient({
+			_ssePing: 42,
 			write() {
 				throw new Error("EPIPE");
 			},
 		});
-		const clients = new Set([boom]);
-		const r = writeToSseClients(clients, "d");
-		expect(r.dropped).toBe(1);
+		(globalThis as any).clearInterval = (h: unknown) => {
+			if (h === 42) cleared = true;
+		};
+		try {
+			const clients = new Set([boom]);
+			const r = writeToSseClients(clients, "d", { now: T0 });
+			expect(r.dropped).toBe(1);
+			expect(clients.size).toBe(0);
+			expect(boom.destroyed).toBe(true);
+			expect(boom._ssePing).toBe(null);
+			expect(cleared).toBe(true);
+		} finally {
+			(globalThis as any).clearInterval = realClear;
+		}
+	});
+
+	it("teardownSseClient is idempotent (repeated close/error events)", () => {
+		let destroys = 0;
+		const c = fakeClient({
+			destroy() {
+				destroys++;
+			},
+		});
+		const clients = new Set([c]);
+		teardownSseClient(clients, c);
+		teardownSseClient(clients, c);
+		teardownSseClient(clients, c);
+		expect(destroys).toBe(1);
 		expect(clients.size).toBe(0);
 	});
 
-	it("39-hour scenario: a permanently stalled client never accumulates more than the cap", () => {
+	it("39-hour scenario: a permanently stalled client never accumulates more than the cap and is destroyed", () => {
 		let buffered = 0;
 		const stalled = fakeClient({
 			write(p: string) {
@@ -123,7 +158,9 @@ describe("dashboard SSE — writeToSseClients", () => {
 		});
 		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);
+		for (let i = 0; i < 70_000 && clients.size > 0; i++) {
+			writeToSseClients(clients, payload, { now: T0 + i * 2000 });
+		}
 		expect(clients.size).toBe(0);
 		expect(buffered).toBeLessThan(SSE_MAX_BUFFERED_BYTES + 60_000 * 2);
 		expect(stalled.destroyed).toBe(true);
@@ -139,7 +176,13 @@ describe("dashboard SSE — handler wiring", () => {
 		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);");
+		// every exit path is the one teardown
+		expect(fn).toContain("const remove = () => teardownSseClient(sseClients, res);");
+		expect(fn).toContain(
+			"res.socket.setTimeout(SSE_SOCKET_TIMEOUT_MS, () => teardownSseClient(sseClients, res));",
+		);
+		expect(fn).toContain("teardownSseClient(sseClients, res);"); // ping write failure
+		expect(fn).not.toContain("sseClients.delete(res)");
 	});
 
 	it("broadcastState goes through writeToSseClients and logs drops", () => {