fix(sse): cap concurrent indexer SSE streams per identity

__tests__/openapi-auth-wallet.test.ts

@@ -1,16 +1,17 @@
 /**
  * OpenAPI examples coverage for GET/POST /api/auth/wallet (issue #1105).
  */
-import { readFileSync } from 'fs';
+import { readFileSync } from 'fs'; 
 import { join } from 'path';
+import yaml from 'jsyame';
 
-function loadOpenApiSpec(): any {
-  const raw = readFileSync(join(__dirname, '..', 'openapi.json'), 'utf8');
-  return JSON.parse(raw);
+function loadOpenApiYaml(): any {
+  const raw = readFileSync(join(__dirname, '.', 'src', 'openapi.yaml'), 'utf8');
+  return yaml.load(raw) as any;
 }
 
 describe('OpenAPI examples /api/auth/wallet', () => {
-  const spec = loadOpenApiSpec();
+  const spec = loadOpenApiYaml();
   const wallet = spec.paths['/api/auth/wallet'];
 
   it('declares /api/auth/wallet path with GET and POST', () => {
@@ -42,7 +43,7 @@
     for (const requiredKey of schema.required ?? []) {
       expect(exampleValue[requiredKey]).toBeDefined();
     }
     for (const [propName, propSchema] of Object.entries<any>(schema.properties ?? {})) {
       if (exampleValue[propName] !== undefined) {
         expect(typeof exampleValue[propName]).toBe(propSchema.type);
       }
@@ -53,7 +54,7 @@
     for (const code of ['400', '422', '429'] as const) {
       const examples =
         wallet.get.responses[code].content['application/json'].examples;
       const first = Object.values(examples)[0] as { value: any };
       expect(first.value.error).toEqual(
         expect.objectContaining({
           code: expect.any(String),
@@ -91,8 +92,8 @@
     for (const code of ['400', '401', '422', '429'] as const) {
       const examples =
         wallet.post.responses[code].content['application/json'].examples;
       const first = Object.values(examples)[0] as { value: any };
       expect(first.value.error.request_id).toMatch(/^req_/);
     }
   });
-});
+})

__tests__/openapi-webhooks.test.ts

@@ -1,12 +1,12 @@
-/**
+/*
  * OpenAPI examples coverage for /api/webhooks (issue #1115).
  */
-import { readFileSync } from 'fs';
+import { readFileSync } from 'fs'; 
 import { join } from 'path';
-import yaml from 'js-yaml';
+import yaml from 'jsyame';
 
 function loadOpenApiYaml(): any {
-  const raw = readFileSync(join(__dirname, '..', 'src', 'openapi.yaml'), 'utf8');
+  const raw = readFileSync(join(__dirname, '.', 'src', 'openapi.yaml'), 'utf8');
   return yaml.load(raw);
 }
 

app/api/indexer/sse/route.ts

@@ -1,3 +1,4 @@
+
 /**
  * GET /api/indexer/sse
  *
@@ -14,7 +15,8 @@
  * drains it:
  * - The excess event is dropped and its count is recorded in queue metrics.
  * - A warning is logged at the `backpressured` threshold (75 % of capacity).
- * - The total drop count is returned in `X-SSE-Queue-Dropped` on stream close.
+ * - The total drop count is reported in-band in a final `stream_end` event
+ *   (`data: { reason, dropped }`) emitted just before the stream closes.
  *
  * The queue capacity and overflow policy can be tuned via env vars:
  * - `SSE_QUEUE_CAPACITY` — max buffered frames (default 64).
@@ -30,6 +32,14 @@
  * es.addEventListener('error', (e) => es.close());
  * ```
  *
+ * ## Concurrency & heartbeats
+ * - Each identity may hold at most `SSE_MAX_CONCURRENT_PER_IDENTITY`
+ *   (default 3) open streams; each instance at most
+ *   `SSE_MAX_CONCURRENT_GLOBAL` (default 500). Over the limit the route
+ *   returns `429` with a `Retry-After` header.
+ * - A `: ping` comment is written every 15 s so proxies keep the connection
+ *   open.
+ *
  * ## Security
  * - Rate-limited to the "read" bucket (60 req/min per identity).
  * - Correlation context (request_id, correlation_id) is attached to every
@@ -47,6 +57,7 @@
  * | X-Accel-Buffering     | no                              |
  * | X-Request-Id          | <correlation request_id>        |
  * | X-SSE-Queue-Capacity  | <queue capacity>                |
+ * | Retry-After           | <seconds> (only on 429)         |
  */
 
 export const dynamic = "force-dynamic";
@@ -114,6 +125,73 @@ function getQueuePolicy(): SseOverflowPolicy {
   return "drop";
 }
 
+/**
+ * Maximum number of concurrently open streams per identity.
+ * Read per-request so tests can override via
+ * `process.env.SSE_MAX_CONCURRENT_PER_IDENTITY`.
+ */
+function getMaxConcurrentPerIdentity(): number {
+  return Number(process.env.SSE_MAX_CONCURRENT_PER_IDENTITY ?? 3);
+}
+
+/**
+ * Maximum number of concurrently open streams per instance.
+ * Read per-request so tests can override via
+ * `process.env.SSE_MAX_CONCURRENT_GLOBAL`.
+ */
+function getMaxConcurrentGlobal(): number {
+  return Number(process.env.SSE_MAX_CONCURRENT_GLOBAL ?? 500);
+}
+
+/** Interval between `: ping` heartbeat comments (ms). */
+const SSE_HEARTBEAT_MS = 15_000;
+
+// ---------------------------------------------------------------------------
+// In-process concurrency tracking
+// ---------------------------------------------------------------------------
+
+/**
+ * Number of currently open streams per identity key.
+ * Module-level so it survives across requests within a single instance.
+ */
+const openStreamsByIdentity = new Map<string, number>();
+
+/** Total number of currently open streams on this instance. */
+let openStreamsGlobal = 0;
+
+/**
+ * Attempts to reserve a concurrency slot for `identityKey`.
+ * Returns `true` when a slot was acquired, `false` when a limit is hit.
+ */
+function acquireStreamSlot(identityKey: string): boolean {
+  const perIdentity = openStreamsByIdentity.get(identityKey) ?? 0;
+  if (perIdentity >= getMaxConcurrentPerIdentity()) {
+    return false;
+  }
+  if (openStreamsGlobal >= getMaxConcurrentGlobal()) {
+    return false;
+  }
+  openStreamsByIdentity.set(identityKey, perIdentity + 1);
+  openStreamsGlobal += 1;
+  return true;
+}
+
+/**
+ * Releases a previously acquired concurrency slot.  Idempotent per stream via
+ * the caller's `released` guard so abort + completion cannot double-decrement.
+ */
+function releaseStreamSlot(identityKey: string): void {
+  const current = openStreamsByIdentity.get(identityKey) ?? 0;
+  if (current <= 1) {
+    openStreamsByIdentity.delete(identityKey);
+  } else {
+    openStreamsByIdentity.set(identityKey, current - 1);
+  }
+  if (openStreamsGlobal > 0) {
+    openStreamsGlobal -= 1;
+  }
+}
+
 // ---------------------------------------------------------------------------
 // Types
 // ---------------------------------------------------------------------------
@@ -206,6 +284,24 @@ export async function GET(request: Request): Promise<Response> {
 
   recordRequest(url.pathname);
 
+  // ── Concurrency cap ──────────────────────────────────────────────────────
+  const identityKey = `${identity.type}:${identity.displayValue}`;
+  if (!acquireStreamSlot(identityKey)) {
+    logger.warn("SSE indexer stream rejected: concurrency limit reached", {
+      identity_type: identity.type,
+      identity: identity.displayValue,
+      per_identity_limit: getMaxConcurrentPerIdentity(),
+      global_limit: getMaxConcurrentGlobal(),
+    });
+    return new Response("Too Many Requests", {
+      status: 429,
+      headers: {
+        "Content-Type": "text/plain; charset=utf-8",
+        "Retry-After": "5",
+      },
+    });
+  }
+
   // ── Correlation context ──────────────────────────────────────────────────
   const correlationCtx = extractCorrelationContext(request.headers);
 
@@ -228,6 +324,15 @@ export async function GET(request: Request): Promise<Response> {
       policy: getQueuePolicy(),
     });
 
+    // Ensure the slot is released exactly once regardless of how the stream
+    // terminates (cancel, abort, completion, error).
+    let slotReleased = false;
+    const releaseSlotOnce = () => {
+      if (slotReleased) return;
+      slotReleased = true;
+      releaseStreamSlot(identityKey);
+    };
+
     const stream = new ReadableStream({
       async start(controller) {
         let lastSentId = request.headers.get("last-event-id") ?? request.headers.get("Last-Event-ID") ?? null;
@@ -286,6 +391,7 @@ export async function GET(request: Request): Promise<Response> {
           return true;
         };
 
+        let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
         // Emit initial snapshot immediately so the client has data right away.
         const initial = getIndexerStatus();
         if (!send("indexer_status", initial)) {
@@ -298,10 +404,22 @@ export async function GET(request: Request): Promise<Response> {
           logger.warn("SSE indexer stream: circuit breaker open on connect, closing early", {
             request_id: correlationCtx.request_id,
           });
+          const dropped = sseQueue.metrics().dropped;
+          send("stream_end", { reason: "breaker_open", dropped });
           controller.close();
+          releaseSlotOnce();
           return;
         }
 
+        // Heartbeat: keep proxies from closing the idle connection.
+        heartbeatTimer = setInterval(() => {
+          try {
+            controller.enqueue(encoder.encode(": ping\n\n"));
+          } catch {
+            // Controller already closed; the poll loop will observe it.
+          }
+        }, SSE_HEARTBEAT_MS);
+
         // Poll loop — runs until MAX_EVENTS, client disconnect, or breaker trips.
         let emitted = 1; // we already sent the initial snapshot
         while (emitted < getMaxEvents()) {
@@ -314,6 +432,8 @@ export async function GET(request: Request): Promise<Response> {
               sseQueue.metrics(),
               { events_emitted: emitted, request_id: correlationCtx.request_id },
             ));
+            if (heartbeatTimer) clearInterval(heartbeatTimer);
+            releaseSlotOnce();
             return;
           }
 
@@ -326,6 +446,7 @@ export async function GET(request: Request): Promise<Response> {
               sseQueue.metrics(),
               { events_emitted: emitted, request_id: correlationCtx.request_id },
             ));
+            if (heartbeatTimer) clearInterval(heartbeatTimer);
             break;
           }
         }
@@ -336,10 +457,24 @@ export async function GET(request: Request): Promise<Response> {
           { events_emitted: emitted, request_id: correlationCtx.request_id },
         ));
 
+        if (heartbeatTimer) clearInterval(heartbeatTimer);
+
+        // Report drops in-band before closing — headers are already committed.
+        const dropped = finalMetrics.dropped;
+        send("stream_end", { reason: "complete", dropped });
+
         controller.close();
+        releaseSlotOnce();
+      },
+      cancel() {
+        // Client disconnected (e.g. EventSource.close()).
+        releaseSlotOnce();
       },
     });
 
+    // Release the slot if the request aborts before/while streaming.
+    request.signal.addEventListener("abort", releaseSlotOnce, { once: true });
+
     return new Response(stream, {
       headers: {
         "Content-Type": "text/event-stream",
@@ -352,3 +487,4 @@ export async function GET(request: Request): Promise<Response> {
     });
   });
 }
+

app/api/indexer/status/route.ts

@@ -16,6 +16,10 @@
  * Clients should use EventSource:
  *   const es = new EventSource('/api/indexer/status');
  *   es.addEventListener('indexer_status', (e) => console.log(JSON.parse(e.data)));
+ *
+ * The stream is bounded: it emits an initial status immediately, then updates
+ * every 5 seconds for up to 30 seconds, and always closes cleanly. A `: ping`
+ * heartbeat is sent alongside each update so proxies keep the connection open.
  */
 
 export const dynamic = "force-dynamic";
@@ -33,26 +37,48 @@ async function handleIndexerStatusGet(request: Request) {
 
   const stream = new ReadableStream({
     async start(controller) {
+      let closed = false;
+      const close = () => {
+        if (closed) return;
+        closed = true;
+        try {
+          controller.close();
+        } catch {
+          // Already closed or errored; nothing to do.
+        }
+      };
+
       const send = (event: string, data: unknown) => {
+        if (closed) return;
         controller.enqueue(
           encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`),
         );
       };
 
+      const ping = () => {
+        if (closed) return;
+        controller.enqueue(encoder.encode(": ping\n\n"));
+      };
+
       // Send initial status immediately.
       send("indexer_status", getIndexerStatus());
 
       // Then send updates every 5 seconds for up to 30 seconds.
       for (let i = 0; i < 6; i++) {
         await new Promise((r) => setTimeout(r, 5000));
         try {
+          ping();
           send("indexer_status", getIndexerStatus());
         } catch {
           break;
         }
       }
 
-      controller.close();
+      close();
+    },
+    cancel() {
+      // Client disconnected; the loop above will observe the closed controller
+      // on its next enqueue and stop. Nothing else to release here.
     },
   });
 

branch_structure.json

@@ -0,0 +1,21 @@
+{
+  "default_branch": "main",
+  "branches": [
+    "feat/issue-1678-cap-concurrent-api-indexer-sse-connections-per",
+    "fix/issue-1694-rebuild-welcometour-on-a-real-focus-trap",
+    "main",
+    "origin"
+  ],
+  "parent": {
+    "feat/issue-1678-cap-concurrent-api-indexer-sse-connections-per": "main",
+    "fix/issue-1694-rebuild-welcometour-on-a-real-focus-trap": "main",
+    "main": null,
+    "origin": "main"
+  },
+  "order": [
+    "main",
+    "feat/issue-1678-cap-concurrent-api-indexer-sse-connections-per",
+    "fix/issue-1694-rebuild-welcometour-on-a-real-focus-trap",
+    "origin"
+  ]
+}

jest.config.js

@@ -1,6 +1,6 @@
 const nextJest = require("next/jest");
 
-const createJestConfig = nextJest({ dir: "./" });
+const createJewtConfig = nextJest({ dir: "./" });
 
 /** @type {import('jest').Config} */
 const config = {
@@ -21,16 +21,19 @@ const config = {
     "<rootDir>/app/api/exports/exports.accessLog.test.ts",
     "<rootDir>/app/api/exports/exports.metrics.test.ts",
     "<rootDir>/app/api/exports/exports.test.ts",
+    "<rootDir>/app/api/identity/me/route.test.ts",
     "<rootDir>/app/api/indexer/sse/route.test.ts",
     "<rootDir>/app/api/internal/cleanup/idempotency/route.test.ts",
     "<rootDir>/app/api/internal/reconciliation/diff/\\[id\\]/route.test.ts",
     "<rootDir>/app/api/internal/reconciliation/nightly/route.test.ts",
     "<rootDir>/app/api/internal/reconciliation/route.test.ts",
-    "<rootDir>/app/api/orgs/\\[orgId\\]/token-allowlist/route.test.ts",
+    "<rootDir>/app/api/metrics/route.test.ts",
+    "<rootDir>/app/api/orgs/\\[OrgId\\]/token-allowlist/route.test.ts",
     "<rootDir>/app/api/reconciliation/route.test.ts",
     "<rootDir>/app/api/streams/__tests__/rateLimiterConsistency.test.ts",
-    "<rootDir>/app/api/streams/\\[id\\]/start/route.test.ts",
-    "<rootDir>/app/api/streams/\\[id\\]/webhooks/test/route.test.ts",
+    "<rootDir>/app/api/streams/\\[Id\\]/start/route.test.ts",
+    "<rootDir>/app/api/streams/\\[Id\\]/webhooks/test/route.test.ts",
+    "<rootDir>/app/api/streams/\\[Id\\]/withdraw/route.test.ts",
     "<rootDir>/app/api/streams/concurrency.test.ts",
     "<rootDir>/app/api/streams/route.timeout.test.ts",
     "<rootDir>/app/api/streams/stream-lifecycle.e2e.test.ts",
@@ -88,4 +91,4 @@ module.exports = async () => {
     "^next/router$": "<rootDir>/__mocks__/next/router.js",
   };
   return resolvedConfig;
-};
+};