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

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

@@ -14,6 +14,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();
 
@@ -105,6 +106,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
@@ -139,20 +142,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/workers/soroban-event-worker.ts

@@ -67,6 +67,23 @@ export function decodeMap(val: xdr.ScVal): Record<string, xdr.ScVal> {
   return result;
 }
 
+// ─── Event-processing counters / degraded signal ─────────────────────────────
+
+/** Sliding window used to decide whether recent failures count as a "spike". */
+const FAILURE_WINDOW_MS = 5 * 60 * 1000;
+/** Need at least this many attempts in the window before marking degraded. */
+const MIN_SAMPLES_FOR_DEGRADED = 3;
+/** Degraded when recent failure rate is at or above this threshold. */
+const FAILURE_RATE_THRESHOLD = 0.5;
+
+export interface IndexerEventCounters {
+  eventsProcessed: number;
+  eventsFailed: number;
+  lastErrorAt: string | null;
+  /** True when recent failure rate indicates a spike (not just lifetime totals). */
+  degraded: boolean;
+}
+
 // ─── Worker Class ─────────────────────────────────────────────────────────────
 
 export class SorobanEventWorker {
@@ -85,6 +102,15 @@ export class SorobanEventWorker {
   /** Promise chain that serializes all `fetchAndProcessEvents` invocations. */
   private batchMutex: Promise<void> = Promise.resolve();
 
+  /** Lifetime count of events that processed without throwing. */
+  private eventsProcessed = 0;
+  /** Lifetime count of events that threw during processing. */
+  private eventsFailed = 0;
+  /** Timestamp of the most recent per-event processing failure. */
+  private lastErrorAt: Date | null = null;
+  /** Recent attempt outcomes for sliding-window spike detection. */
+  private recentOutcomes: { ok: boolean; at: number }[] = [];
+
   constructor() {
     const rpcUrl =
       process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org";
@@ -97,6 +123,44 @@ export class SorobanEventWorker {
     this.server = new rpc.Server(rpcUrl, { allowHttp: true });
   }
 
+  /**
+   * Snapshot of event-processing counters for /health and admin metrics.
+   * `degraded` is true when ≥50% of attempts in the last 5 minutes failed
+   * (with at least 3 samples), so a broken indexer fails the health check
+   * even when lag stays low because `updatedAt` is bumped every poll.
+   */
+  getEventCounters(): IndexerEventCounters {
+    return {
+      eventsProcessed: this.eventsProcessed,
+      eventsFailed: this.eventsFailed,
+      lastErrorAt: this.lastErrorAt?.toISOString() ?? null,
+      degraded: this.isFailureSpike(),
+    };
+  }
+
+  /** @internal Reset counters — used by unit tests. */
+  resetEventCounters(): void {
+    this.eventsProcessed = 0;
+    this.eventsFailed = 0;
+    this.lastErrorAt = null;
+    this.recentOutcomes = [];
+  }
+
+  private recordOutcome(ok: boolean): void {
+    const now = Date.now();
+    this.recentOutcomes.push({ ok, at: now });
+    const cutoff = now - FAILURE_WINDOW_MS;
+    this.recentOutcomes = this.recentOutcomes.filter((o) => o.at >= cutoff);
+  }
+
+  private isFailureSpike(): boolean {
+    const cutoff = Date.now() - FAILURE_WINDOW_MS;
+    const recent = this.recentOutcomes.filter((o) => o.at >= cutoff);
+    if (recent.length < MIN_SAMPLES_FOR_DEGRADED) return false;
+    const failed = recent.filter((o) => !o.ok).length;
+    return failed / recent.length >= FAILURE_RATE_THRESHOLD;
+  }
+
   /**
    * Start the polling worker. If `STREAM_CONTRACT_ID` is not configured the
    * worker logs a warning and exits gracefully instead of throwing.
@@ -266,10 +330,15 @@ export class SorobanEventWorker {
 
       try {
         await this.processEvent(event);
+        this.eventsProcessed += 1;
+        this.recordOutcome(true);
         // Use the event ID as the cursor if pagingToken is not available
         lastCursor = event.id;
         lastLedger = event.ledger;
       } catch (err) {
+        this.eventsFailed += 1;
+        this.lastErrorAt = new Date();
+        this.recordOutcome(false);
         logger.error(
           `[SorobanWorker] Failed to process event ${event.id}:`,
           err,

backend/tests/health.test.ts

@@ -17,7 +17,20 @@ vi.mock('../src/lib/prisma.js', () => ({
   default: prismaMock,
 }));
 
+vi.mock('../src/workers/soroban-event-worker.js', () => ({
+  sorobanEventWorker: {
+    getEventCounters: vi.fn().mockReturnValue({
+      eventsProcessed: 0,
+      eventsFailed: 0,
+      lastErrorAt: null,
+      degraded: false,
+    }),
+  },
+  SorobanEventWorker: vi.fn(),
+}));
+
 import app from '../src/app.js';
+import { sorobanEventWorker } from '../src/workers/soroban-event-worker.js';
 
 function makeState(lagSeconds: number) {
   const updatedAt = new Date(Date.now() - lagSeconds * 1000);
@@ -29,6 +42,12 @@ describe('GET /health', () => {
     vi.unstubAllEnvs();
     prismaMock.$queryRaw.mockResolvedValue([{ '?column?': 1n }]);
     prismaMock.indexerState.findUnique.mockResolvedValue(null);
+    vi.mocked(sorobanEventWorker.getEventCounters).mockReturnValue({
+      eventsProcessed: 0,
+      eventsFailed: 0,
+      lastErrorAt: null,
+      degraded: false,
+    });
   });
 
   afterEach(() => {
@@ -44,6 +63,10 @@ describe('GET /health', () => {
     expect(res.body.db).toBe('connected');
     expect(res.body.indexerEnabled).toBe(false);
     expect(res.body.indexerLag).toBeNull();
+    expect(res.body.eventsProcessed).toBe(0);
+    expect(res.body.eventsFailed).toBe(0);
+    expect(res.body.lastErrorAt).toBeNull();
+    expect(res.body.indexerDegraded).toBe(false);
   });
 
   it('returns 200 when DB is up and indexer is enabled but has no state row yet (cold start)', async () => {
@@ -97,4 +120,24 @@ describe('GET /health', () => {
     expect(typeof res.body.indexerLag).toBe('number');
     expect(typeof res.body.uptime).toBe('number');
   });
+
+  it('returns 503 when indexer is enabled and event-processing failures spike (#844)', async () => {
+    vi.stubEnv('STREAM_CONTRACT_ID', 'CSOME_CONTRACT_ADDRESS');
+    prismaMock.indexerState.findUnique.mockResolvedValue(makeState(5));
+    vi.mocked(sorobanEventWorker.getEventCounters).mockReturnValue({
+      eventsProcessed: 1,
+      eventsFailed: 10,
+      lastErrorAt: '2026-07-27T08:00:00.000Z',
+      degraded: true,
+    });
+
+    const res = await request(app).get('/health');
+    expect(res.status).toBe(503);
+    expect(res.body.status).toBe('degraded');
+    expect(res.body.indexerLag).toBeLessThanOrEqual(60);
+    expect(res.body.eventsProcessed).toBe(1);
+    expect(res.body.eventsFailed).toBe(10);
+    expect(res.body.lastErrorAt).toBe('2026-07-27T08:00:00.000Z');
+    expect(res.body.indexerDegraded).toBe(true);
+  });
 });

backend/tests/integration/admin-metrics.test.ts

@@ -90,9 +90,22 @@
   replayFromLedger: vi.fn().mockResolvedValue(undefined),
 }));
 
+vi.mock('../../src/workers/soroban-event-worker.js', () => ({
+  sorobanEventWorker: {
+    getEventCounters: vi.fn().mockReturnValue({
+      eventsProcessed: 0,
+      eventsFailed: 0,
+      lastErrorAt: null,
+      degraded: false,
+    }),
+  },
+  SorobanEventWorker: vi.fn(),
+}));
+
 // ─── Import app after mocks are registered ────────────────────────────────────
 
 import app from '../../src/app.js';
+import { sorobanEventWorker } from '../../src/workers/soroban-event-worker.js';
 
 // ─── Helpers ──────────────────────────────────────────────────────────────────
 
@@ -134,7 +147,7 @@
 
     const res = await request(app).get('/v1/admin/metrics');
 
     expect(res.status).toBe(200);
     expect(res.body).toMatchObject({
       total_streams: 12,
       active_streams: 7,
@@ -156,7 +169,7 @@
 
     const res = await request(app).get('/v1/admin/metrics');
 
     expect(res.status).toBe(200);
     expect(res.body.total_volume_streamed).toBe('18014398509481986');
   });
 
@@ -164,7 +177,7 @@
     setupCounts({ total: 4, active: 4 });
 
     const first = await request(app).get('/v1/admin/metrics');
     expect(first.status).toBe(200);
     expect(first.headers['x-cache']).toBe('MISS');
 
     expect(mocks.cache.set).toHaveBeenCalledTimes(1);
@@ -183,6 +196,15 @@
       completed_streams: 30,
       cancelled_streams: 14,
       total_volume_streamed: '123456789',
+      indexer: {
+        lastLedger: 10,
+        lagSeconds: 1,
+        lastUpdated: null,
+        eventsProcessed: 0,
+        eventsFailed: 0,
+        lastErrorAt: null,
+        degraded: false,
+      },
     };
     mocks.cache.get.mockReturnValueOnce(cachedPayload);
 
@@ -194,4 +216,57 @@
     expect(mocks.prisma.stream.count).not.toHaveBeenCalled();
     expect(mocks.prisma.stream.findMany).not.toHaveBeenCalled();
   });
+
+  it('exposes indexer event-processing counters and degraded signal (#844)', async () => {
+    setupCounts();
+    vi.mocked(sorobanEventWorker.getEventCounters).mockReturnValueOnce({
+      eventsProcessed: 40,
+      eventsFailed: 12,
+      lastErrorAt: '2026-07-27T08:00:00.000Z',
+      degraded: true,
+    });
+
+    const res = await request(app).get('/v1/admin/metrics');
+
+    expect(res.status).toBe(200);
+    expect(res.body.indexer).toMatchObject({
+      eventsProcessed: 40,
+      eventsFailed: 12,
+      lastErrorAt: '2026-07-27T08:00:00.000Z',
+      degraded: true,
+    });
+  });
+
+  it('merges live indexer counters into a cached metrics response', async () => {
+    mocks.cache.get.mockReturnValueOnce({
+      total_streams: 1,
+      indexer: {
+        lastLedger: 5,
+        lagSeconds: 2,
+        lastUpdated: null,
+        eventsProcessed: 0,
+        eventsFailed: 0,
+        lastErrorAt: null,
+        degraded: false,
+      },
+    });
+    vi.mocked(sorobanEventWorker.getEventCounters).mockReturnValueOnce({
+      eventsProcessed: 9,
+      eventsFailed: 3,
+      lastErrorAt: '2026-07-27T09:00:00.000Z',
+      degraded: true,
+    });
+
+    const res = await request(app).get('/v1/admin/metrics');
+
+    expect(res.status).toBe(200);
+    expect(res.headers['x-cache']).toBe('HIT');
+    expect(res.body.indexer).toMatchObject({
+      lastLedger: 5,
+      eventsProcessed: 9,
+      eventsFailed: 3,
+      lastErrorAt: '2026-07-27T09:00:00.000Z',
+      degraded: true,
+    });
+  });
 });