fix(dashboard): close SSE connections on outbound backpressure
The global SSE broadcast called res.write() without checking the return value, so a paused or backgrounded client would silently accumulate every store event for every entity (tasks, missions, plugins, agents, chat, ...) into res.outputData until the dashboard process OOMed. Add a 4 MB writableLength threshold; when exceeded, tear down the connection so the OS releases the buffer. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -144,4 +144,41 @@ describe("createSSE client cleanup", () => {
|
|||||||
expect(connection.res.end).toHaveBeenCalledTimes(1);
|
expect(connection.res.end).toHaveBeenCalledTimes(1);
|
||||||
expect(getActiveSSEConnections()).toBe(baseline);
|
expect(getActiveSSEConnections()).toBe(baseline);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("closes the connection when the outbound buffer exceeds the backpressure threshold", () => {
|
||||||
|
// Capture the task:created listener so we can fire a send after the
|
||||||
|
// socket buffer has been bloated past the threshold.
|
||||||
|
let onCreated: ((task: unknown) => void) | undefined;
|
||||||
|
const store = {
|
||||||
|
on: vi.fn((event: string, handler: (task: unknown) => void) => {
|
||||||
|
if (event === "task:created") onCreated = handler;
|
||||||
|
}),
|
||||||
|
off: vi.fn(),
|
||||||
|
} as unknown as TaskStore;
|
||||||
|
|
||||||
|
const baseline = getActiveSSEConnections();
|
||||||
|
const socket = new MockSocket();
|
||||||
|
const req = new EventEmitter() as Request & { query: Record<string, string>; socket: MockSocket };
|
||||||
|
req.query = { clientId: "backpressure-client" };
|
||||||
|
req.socket = socket;
|
||||||
|
const res = new MockResponse(socket) as MockResponse & { writableLength: number };
|
||||||
|
res.writableLength = 0;
|
||||||
|
|
||||||
|
createSSE(store)(req, res as unknown as Response);
|
||||||
|
expect(getActiveSSEConnections()).toBe(baseline + 1);
|
||||||
|
expect(typeof onCreated).toBe("function");
|
||||||
|
|
||||||
|
// Simulate a stuck client: kernel + Node buffers full beyond 4 MB.
|
||||||
|
res.writableLength = 5 * 1024 * 1024;
|
||||||
|
const writeCountBefore = res.write.mock.calls.length;
|
||||||
|
|
||||||
|
onCreated?.({ id: "task-1" });
|
||||||
|
|
||||||
|
// Backpressure was detected before the write, so res.write must NOT have
|
||||||
|
// been called for this event, and the connection should be torn down.
|
||||||
|
expect(res.write.mock.calls.length).toBe(writeCountBefore);
|
||||||
|
expect(res.end).toHaveBeenCalledTimes(1);
|
||||||
|
expect(socket.destroy).toHaveBeenCalledTimes(1);
|
||||||
|
expect(getActiveSSEConnections()).toBe(baseline);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -19,8 +19,14 @@ let nextConnectionId = 1;
|
|||||||
|
|
||||||
const SSE_CLIENT_ID_MAX_LENGTH = 128;
|
const SSE_CLIENT_ID_MAX_LENGTH = 128;
|
||||||
const SSE_CLIENT_STALE_MS = 5_000;
|
const SSE_CLIENT_STALE_MS = 5_000;
|
||||||
|
// If a client's outbound buffer exceeds this, treat the connection as stuck
|
||||||
|
// and close it. Without this, res.write() silently queues into res.outputData
|
||||||
|
// for a paused/backgrounded client, and every store event for every entity
|
||||||
|
// accumulates there until the process OOMs.
|
||||||
|
const SSE_MAX_BUFFERED_BYTES = 4 * 1024 * 1024;
|
||||||
|
|
||||||
type SSECloseReason =
|
type SSECloseReason =
|
||||||
|
| "backpressure"
|
||||||
| "client-disconnect"
|
| "client-disconnect"
|
||||||
| "close"
|
| "close"
|
||||||
| "error"
|
| "error"
|
||||||
@@ -107,16 +113,23 @@ export function getSSEHighWaterMark(): number {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Safely write to an SSE response stream.
|
* Safely write to an SSE response stream.
|
||||||
* Returns `true` if the write succeeded, `false` if the connection is dead.
|
* Returns "ok" on success, "dead" if the socket is gone, or "backpressure" if
|
||||||
* On failure the caller should clean up event listeners.
|
* the outbound buffer has grown past SSE_MAX_BUFFERED_BYTES (caller should
|
||||||
|
* tear down — Node will otherwise queue indefinitely into res.outputData).
|
||||||
*/
|
*/
|
||||||
function safeWrite(res: Response, data: string): boolean {
|
type SafeWriteResult = "ok" | "dead" | "backpressure";
|
||||||
|
|
||||||
|
function safeWrite(res: Response, data: string): SafeWriteResult {
|
||||||
try {
|
try {
|
||||||
if (res.writableEnded || res.destroyed) return false;
|
if (res.writableEnded || res.destroyed) return "dead";
|
||||||
|
// Pre-check: if the buffer is already full, refuse the write.
|
||||||
|
if (typeof res.writableLength === "number" && res.writableLength > SSE_MAX_BUFFERED_BYTES) {
|
||||||
|
return "backpressure";
|
||||||
|
}
|
||||||
res.write(data);
|
res.write(data);
|
||||||
return true;
|
return "ok";
|
||||||
} catch {
|
} catch {
|
||||||
return false;
|
return "dead";
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -299,9 +312,20 @@ export function createSSE(
|
|||||||
// Send initial heartbeat
|
// Send initial heartbeat
|
||||||
res.write(": connected\n\n");
|
res.write(": connected\n\n");
|
||||||
|
|
||||||
/** Write an SSE message; clean up on failure. */
|
/** Write an SSE message; tear down on failure or backpressure. */
|
||||||
const send = (data: string) => {
|
const send = (data: string) => {
|
||||||
if (!safeWrite(res, data)) cleanup("send-failed");
|
const result = safeWrite(res, data);
|
||||||
|
if (result === "ok") return;
|
||||||
|
if (result === "backpressure") {
|
||||||
|
console.warn(
|
||||||
|
`[sse] connection ${connectionId} backpressure exceeded ` +
|
||||||
|
`(buffered=${res.writableLength}B, threshold=${SSE_MAX_BUFFERED_BYTES}B); closing`,
|
||||||
|
);
|
||||||
|
closeConnection("backpressure");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// "dead" — socket already gone; cleanup is enough.
|
||||||
|
cleanup("send-failed");
|
||||||
};
|
};
|
||||||
|
|
||||||
// --- Event handler definitions ---
|
// --- Event handler definitions ---
|
||||||
|
|||||||
Reference in New Issue
Block a user