diff --git a/dashboard/server.cjs b/dashboard/server.cjs index 18667d92..53b60edc 100644 --- a/dashboard/server.cjs +++ b/dashboard/server.cjs @@ -1340,25 +1340,48 @@ 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; @@ -1366,26 +1389,21 @@ function writeToSseClients(clients, payload, opts = {}) { 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); diff --git a/extensions/tests/dashboard-sse-backpressure.test.ts b/extensions/tests/dashboard-sse-backpressure.test.ts index e3fcc007..467bb190 100644 --- a/extensions/tests/dashboard-sse-backpressure.test.ts +++ b/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, - 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, + payload: string, + opts?: any, + ) => { written: number; skipped: number; dropped: number }; + teardownSseClient: (clients: Set, client: any) => void; + SSE_MAX_BUFFERED_BYTES: number; + SSE_MAX_STALLED_MS: number; + }; function fakeClient(over: Record = {}) { const c: any = { @@ -62,57 +64,90 @@ function fakeClient(over: Record = {}) { } 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", () => {