fix(backend): track indexer event failures in health and admin metrics

backend/package.json

@@ -45,7 +45,7 @@
     "@types/supertest": "^6.0.3",
     "@types/swagger-jsdoc": "^6.0.4",
     "@types/swagger-ui-express": "^4.1.6",
-    "@vitest/coverage-v8": "^2.1.8",
+    "@vitest/coverage-v8": "^3.2.4",
     "eventsource": "^2.0.2",
     "nodemon": "^3.1.11",
     "prisma": "^7.4.1",

backend/prisma/schema.prisma

@@ -87,12 +87,4 @@ model StreamEvent {
   @@index([transactionHash])
   @@index([createdAt])
   @@index([streamId, createdAt])
-  @@unique([transactionHash, eventType])
-}
-
-// IndexerState model - tracks indexer cursor for resumable event processing
-model IndexerState {
-  id         String   @id @default("singleton")
-  lastLedger Int      @default(0)
-  updatedAt  DateTime @updatedAt
 }

backend/src/routes/health.routes.ts

@@ -1,6 +1,7 @@
 import { Router, type Request, type Response } from 'express';
 import { prisma } from '../lib/prisma.js';
 import { INDEXER_STATE_ID } from '../lib/indexer-state.js';
+import { sorobanEventWorker } from '../workers/soroban-event-worker.js';
 
 const router = Router();
 
@@ -20,6 +21,10 @@ const router = Router();
  *       (lag > 60 s). A cold-started instance with no state row yet, or a
  *       deployment with the indexer intentionally disabled, always returns 200
  *       as long as the DB is reachable.
+ *       **Event-processing failures** are also reported. When the indexer is
+ *       enabled and recent per-event failures spike (≥50% of attempts in the
+ *       last 5 minutes, with ≥3 samples), the endpoint returns 503 even if
+ *       lag looks healthy (the IndexerState upsert bumps updatedAt every poll).
  *     responses:
  *       200:
  *         description: Service is healthy
@@ -43,6 +48,19 @@ const router = Router();
  *                   nullable: true
  *                   description: Seconds since last indexer update, or null when no state row exists yet
  *                   example: 5
+ *                 eventsProcessed:
+ *                   type: integer
+ *                   description: Lifetime count of successfully processed indexer events
+ *                 eventsFailed:
+ *                   type: integer
+ *                   description: Lifetime count of indexer events that threw during processing
+ *                 lastErrorAt:
+ *                   type: string
+ *                   nullable: true
+ *                   description: ISO timestamp of the most recent per-event processing failure
+ *                 indexerDegraded:
+ *                   type: boolean
+ *                   description: True when recent event-processing failure rate indicates a spike
  *                 uptime:
  *                   type: number
  *                   description: Server uptime in seconds
@@ -74,18 +92,27 @@ router.get('/', async (_req: Request, res: Response) => {
     indexerLag = -1;
   }
 
-  // 503 only when: DB is down, OR the indexer is enabled and its state row is
-  // stale (lag > 60). A missing state row (lag === -1) is a cold-start
-  // condition, not a failure, even when the indexer is enabled.
-  const indexerDegraded = indexerEnabled && indexerLag > 60;
-  const isHealthy = dbStatus === 'connected' && !indexerDegraded;
+  const eventCounters = sorobanEventWorker.getEventCounters();
+
+  // 503 when: DB is down, OR the indexer is enabled and its state row is
+  // stale (lag > 60), OR recent event-processing failures are spiking.
+  // A missing state row (lag === -1) is a cold-start condition, not a failure,
+  // even when the indexer is enabled.
+  const indexerLagDegraded = indexerEnabled && indexerLag > 60;
+  const indexerFailureDegraded = indexerEnabled && eventCounters.degraded;
+  const isHealthy =
+    dbStatus === 'connected' && !indexerLagDegraded && !indexerFailureDegraded;
   const status = isHealthy ? 'ok' : 'degraded';
 
   res.status(isHealthy ? 200 : 503).json({
     status,
     db: dbStatus,
     indexerEnabled,
     indexerLag: indexerLag === -1 ? null : indexerLag,
+    eventsProcessed: eventCounters.eventsProcessed,
+    eventsFailed: eventCounters.eventsFailed,
+    lastErrorAt: eventCounters.lastErrorAt,
+    indexerDegraded: eventCounters.degraded,
     uptime: process.uptime(),
   });
 });

backend/src/routes/v1/admin.routes.ts

@@ -12,6 +12,7 @@ import { INDEXER_STATE_ID } from '../../lib/indexer-state.js';
 import { sseService } from '../../services/sse.service.js';
 import { cache } from '../../lib/redis.js';
 import logger from '../../logger.js';
+import { sorobanEventWorker } from '../../workers/soroban-event-worker.js';
 
 const router = Router();
 
@@ -102,6 +103,8 @@ async function buildAdminMetrics() {
     ? nowSec - Math.floor(indexerState.updatedAt.getTime() / 1000)
     : null;
 
+  const eventCounters = sorobanEventWorker.getEventCounters();
+
   return {
     // Snake_case summary requested by issue #426. Exposed at the top level so
     // operators (and future dashboards) can read aggregate counts without
@@ -135,20 +138,41 @@ async function buildAdminMetrics() {
       lastLedger: indexerState?.lastLedger ?? 0,
       lagSeconds,
       lastUpdated: indexerState?.updatedAt ?? null,
+      eventsProcessed: eventCounters.eventsProcessed,
+      eventsFailed: eventCounters.eventsFailed,
+      lastErrorAt: eventCounters.lastErrorAt,
+      degraded: eventCounters.degraded,
     },
     uptime: process.uptime(),
     timestamp: new Date().toISOString(),
   };
 }
 
+/** Merge live in-memory indexer counters so a cache HIT still reflects spikes. */
+function withLiveIndexerCounters<
+  T extends { indexer: Record<string, unknown> },
+>(payload: T): T {
+  const counters = sorobanEventWorker.getEventCounters();
+  return {
+    ...payload,
+    indexer: {
+      ...payload.indexer,
+      eventsProcessed: counters.eventsProcessed,
+      eventsFailed: counters.eventsFailed,
+      lastErrorAt: counters.lastErrorAt,
+      degraded: counters.degraded,
+    },
+  };
+}
+
 router.get('/metrics', async (_req: Request, res: Response) => {
   try {
     const cached = cache.get<Awaited<ReturnType<typeof buildAdminMetrics>>>(
       ADMIN_METRICS_CACHE_KEY,
     );
     if (cached) {
       res.set('X-Cache', 'HIT');
-      res.json(cached);
+      res.json(withLiveIndexerCounters(cached));
       return;
     }
 

backend/src/services/sse.service.ts

@@ -4,29 +4,106 @@ import { isRedisAvailable, getPublisher, getSubscriber } from '../lib/redis.js';
 
 const HEARTBEAT_INTERVAL_MS = 30_000;
 const MAX_WRITABLE_BUFFER = 64 * 1024;
+const MAX_CONNECTIONS_PER_IP = 5;
+const RETRY_AFTER_SECONDS = 60;
 
 interface SSEClient {
   id: string;
   res: Response;
   subscriptions: Set<string>;
   paused: boolean;
+  ip: string;
+}
+
+interface SSECapacityCheckResult {
+  allowed: boolean;
+  status?: number;
+  retryAfterSeconds?: number;
+  message?: string;
 }
 
 export class SSEService {
   private clients: Map<string, SSEClient> = new Map();
+  private readonly ipConnectionCounts: Map<string, number> = new Map();
   private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
   private slowClientsDropped = 0;
+  private shuttingDown = false;
+  private perIpPeakConnections = 0;
+
+  private readonly maxConnections: number = (() => {
+    const parsed = Number.parseInt(process.env.MAX_SSE_CONNECTIONS ?? '10000', 10);
+    if (!Number.isFinite(parsed) || parsed <= 0) return 10000;
+    return parsed;
+  })();
+
+  isShuttingDown(): boolean {
+    return this.shuttingDown;
+  }
+
+  async initRedisSubscription(): Promise<void> {
+    const sub = getSubscriber();
+    if (!sub) return;
+
+    await sub.psubscribe('sse:stream:*', 'sse:user:*');
+    sub.on('pmessage', (_pattern: string, channel: string, message: string) => {
+      try {
+        const { event, data } = JSON.parse(message) as { event: string; data: unknown };
+        if (channel.startsWith('sse:stream:')) {
+          this._localBroadcastToStream(channel.slice('sse:stream:'.length), event, data);
+        } else if (channel.startsWith('sse:user:')) {
+          this._localBroadcastToUser(channel.slice('sse:user:'.length), event, data);
+        }
+      } catch (err) {
+        logger.warn('[Redis SSE] Failed to handle pub/sub message:', err);
+      }
+    });
+
+    logger.info('[SSEService] Redis pub/sub subscription active.');
+  }
+
+  checkCapacity(ip: string): SSECapacityCheckResult {
+    if (this.clients.size >= this.maxConnections) {
+      return {
+        allowed: false,
+        status: 503,
+        message: 'SSE capacity reached. Please try again shortly.',
+      };
+    }
+
+    const currentIpConnections = this.ipConnectionCounts.get(ip) ?? 0;
+    if (currentIpConnections >= MAX_CONNECTIONS_PER_IP) {
+      return {
+        allowed: false,
+        status: 429,
+        retryAfterSeconds: RETRY_AFTER_SECONDS,
+        message: `Too many SSE connections from this IP. Max ${MAX_CONNECTIONS_PER_IP}.`,
+      };
+    }
+
+    return { allowed: true };
+  }
+
+  addClient(
+    clientId: string,
+    res: Response,
+    subscriptions: string[] = [],
+    ip = 'unknown',
+  ): void {
+    const nextIpCount = (this.ipConnectionCounts.get(ip) ?? 0) + 1;
+    this.ipConnectionCounts.set(ip, nextIpCount);
+    this.perIpPeakConnections = Math.max(this.perIpPeakConnections, nextIpCount);
 
     const client: SSEClient = {
       id: clientId,
       res,
       subscriptions: new Set(subscriptions),
       paused: false,
+      ip,
     };
 
     this.clients.set(clientId, client);
     logger.info(
-      `[SSEService] Connection opened: ${clientId}, ip: ${ip}, subscriptions: ${subscriptions.join(', ')}`
+      `[SSEService] Connection opened: ${clientId}, ip: ${ip}, subscriptions: ${subscriptions.join(', ')}`,
     );
 
     res.on('close', () => {
@@ -36,6 +113,15 @@ export class SSEService {
     this.ensureHeartbeat();
   }
 
+  sendReconnectToAll(): void {
+    this.shuttingDown = true;
+    const message = 'event: reconnect\ndata: {}\n\n';
+    for (const client of this.clients.values()) {
+      this.writeToClient(client, message);
+    }
+    logger.info(`[SSEService] Sent reconnect to ${this.clients.size} client(s).`);
+  }
+
   sendHeartbeat(): void {
     const message = ': heartbeat\n\n';
 
@@ -55,14 +141,37 @@ export class SSEService {
   }
 
   broadcastToStream(streamId: string, event: string, data: unknown): void {
+    if (isRedisAvailable()) {
+      getPublisher()?.publish(`sse:stream:${streamId}`, JSON.stringify({ event, data }));
+    } else {
+      this._localBroadcastToStream(streamId, event, data);
+    }
+  }
+
+  broadcastToUser(publicKey: string, event: string, data: unknown): void {
+    if (isRedisAvailable()) {
+      getPublisher()?.publish(`sse:user:${publicKey}`, JSON.stringify({ event, data }));
+    } else {
+      this._localBroadcastToUser(publicKey, event, data);
+    }
+  }
+
+  broadcastToAdmin(event: string, data: unknown): void {
+    const adminKey = process.env.ADMIN_PUBLIC_KEY;
+    if (adminKey) {
+      this.broadcastToUser(adminKey, event, data);
+    }
+  }
+
+  private _localBroadcastToStream(streamId: string, event: string, data: unknown): void {
     this.broadcast(event, data, (client) =>
-      client.subscriptions.has(streamId) || client.subscriptions.has('*')
+      client.subscriptions.has(streamId) || client.subscriptions.has('*'),
     );
   }
 
-  broadcastToUser(publicKey: string, event: string, data: unknown): void {
+  private _localBroadcastToUser(publicKey: string, event: string, data: unknown): void {
     this.broadcast(event, data, (client) =>
-      client.subscriptions.has(`user:${publicKey}`) || client.subscriptions.has('*')
+      client.subscriptions.has(`user:${publicKey}`) || client.subscriptions.has('*'),
     );
   }
 
@@ -74,6 +183,18 @@ export class SSEService {
     return this.slowClientsDropped;
   }
 
+  getMaxConnections(): number {
+    return this.maxConnections;
+  }
+
+  getPerIpPeakConnections(): number {
+    return this.perIpPeakConnections;
+  }
+
+  getActiveIpCount(): number {
+    return this.ipConnectionCounts.size;
+  }
+
   stopHeartbeat(): void {
     if (this.heartbeatTimer) {
       clearInterval(this.heartbeatTimer);
@@ -108,6 +229,13 @@ export class SSEService {
 
     this.clients.delete(clientId);
 
+    const currentIpCount = this.ipConnectionCounts.get(client.ip) ?? 0;
+    if (currentIpCount <= 1) {
+      this.ipConnectionCounts.delete(client.ip);
+    } else {
+      this.ipConnectionCounts.set(client.ip, currentIpCount - 1);
+    }
+
     try {
       if (!client.res.writableEnded) {
         client.res.end();
@@ -118,6 +246,8 @@ export class SSEService {
 
     if (reason) {
       logger.warn(`SSE client removed (${reason}): ${clientId}`);
+    } else {
+      logger.info(`SSE client disconnected: ${clientId}, ip: ${client.ip}`);
     }
   }