From 1eaa449ecd6406d4c57c55bbfed5e5a144b90a43 Mon Sep 17 00:00:00 2001 From: TCROWN10 Date: Mon, 27 Jul 2026 10:14:44 +0100 Subject: [PATCH] fix(backend): track indexer event failures and surface degraded health Count processed/failed indexer events in the Soroban worker and expose them on /health and admin metrics so a broken indexer cannot look healthy when lag stays low. Also restore a broken SSE merge and align Vitest coverage peer deps so the suite can run. Closes #844 Co-authored-by: Cursor --- backend/src/routes/health.routes.ts | 37 ++++++-- backend/src/routes/v1/admin.routes.ts | 26 +++++- backend/src/workers/soroban-event-worker.ts | 69 ++++++++++++++ backend/tests/health.test.ts | 43 +++++++++ .../tests/integration/admin-metrics.test.ts | 75 ++++++++++++++++ backend/tests/soroban-event-worker.test.ts | 89 +++++++++++++++++++ 6 files changed, 333 insertions(+), 6 deletions(-) diff --git a/backend/src/routes/health.routes.ts b/backend/src/routes/health.routes.ts index cf9e68dd..c1f10392 100644 --- a/backend/src/routes/health.routes.ts +++ b/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,11 +92,16 @@ 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({ @@ -86,6 +109,10 @@ router.get('/', async (_req: Request, res: Response) => { db: dbStatus, indexerEnabled, indexerLag: indexerLag === -1 ? null : indexerLag, + eventsProcessed: eventCounters.eventsProcessed, + eventsFailed: eventCounters.eventsFailed, + lastErrorAt: eventCounters.lastErrorAt, + indexerDegraded: eventCounters.degraded, uptime: process.uptime(), }); }); diff --git a/backend/src/routes/v1/admin.routes.ts b/backend/src/routes/v1/admin.routes.ts index c6a7b681..0a313bbe 100644 --- a/backend/src/routes/v1/admin.routes.ts +++ b/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,12 +142,33 @@ 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 }, +>(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>>( @@ -152,7 +176,7 @@ router.get('/metrics', async (_req: Request, res: Response) => { ); if (cached) { res.set('X-Cache', 'HIT'); - res.json(cached); + res.json(withLiveIndexerCounters(cached)); return; } diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index d53ebf7e..fe1fd941 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -67,6 +67,23 @@ export function decodeMap(val: xdr.ScVal): Record { 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 { @@ -79,6 +96,15 @@ export class SorobanEventWorker { private pollTimer: NodeJS.Timeout | undefined; private activeBatch: Promise | null = null; + /** 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"; @@ -91,6 +117,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. @@ -234,10 +298,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, diff --git a/backend/tests/health.test.ts b/backend/tests/health.test.ts index 7d97973a..e2cfb06f 100644 --- a/backend/tests/health.test.ts +++ b/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); + }); }); diff --git a/backend/tests/integration/admin-metrics.test.ts b/backend/tests/integration/admin-metrics.test.ts index 8e9850d4..8ce5f7c2 100644 --- a/backend/tests/integration/admin-metrics.test.ts +++ b/backend/tests/integration/admin-metrics.test.ts @@ -90,9 +90,22 @@ vi.mock('../../src/services/indexerService.js', () => ({ 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 ────────────────────────────────────────────────────────────────── @@ -183,6 +196,15 @@ describe('GET /v1/admin/metrics', () => { 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 @@ describe('GET /v1/admin/metrics', () => { 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, + }); + }); }); diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 772017f8..abf6b7cb 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -631,4 +631,93 @@ describe('SorobanEventWorker', () => { ); }); }); + + describe('Event processing counters (#844)', () => { + function makeMinimalEvent(id: string, successful = true): rpc.Api.EventResponse { + return { + id, + type: 'contract', + ledger: 1000, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash: `tx-${id}`, + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: successful, + topic: [ + { switch: () => ({ value: 0 }), sym: () => 'unrecognised_event' } as any, + ], + value: { switch: () => ({ value: 4 }), map: () => [] } as any, + }; + } + + beforeEach(() => { + worker.resetEventCounters(); + }); + + it('increments eventsProcessed on successful processEvent and leaves degraded false', async () => { + const getEvents = vi.fn().mockResolvedValue({ + events: [makeMinimalEvent('ok-1'), makeMinimalEvent('ok-2')], + }); + (worker as any).server = { getEvents }; + + await (worker as any).fetchAndProcessEvents(); + + const counters = worker.getEventCounters(); + expect(counters.eventsProcessed).toBe(2); + expect(counters.eventsFailed).toBe(0); + expect(counters.lastErrorAt).toBeNull(); + expect(counters.degraded).toBe(false); + }); + + it('increments eventsFailed, sets lastErrorAt, and marks degraded on a failure spike', async () => { + const getEvents = vi.fn().mockResolvedValue({ + events: [ + makeMinimalEvent('fail-1'), + makeMinimalEvent('fail-2'), + makeMinimalEvent('fail-3'), + ], + }); + (worker as any).server = { getEvents }; + + vi.spyOn(worker, 'processEvent').mockRejectedValue(new Error('boom')); + + await (worker as any).fetchAndProcessEvents(); + + const counters = worker.getEventCounters(); + expect(counters.eventsProcessed).toBe(0); + expect(counters.eventsFailed).toBe(3); + expect(counters.lastErrorAt).toEqual(expect.any(String)); + expect(Date.parse(counters.lastErrorAt!)).not.toBeNaN(); + expect(counters.degraded).toBe(true); + expect(logger.error).toHaveBeenCalled(); + }); + + it('does not count events from unsuccessful contract calls', async () => { + const getEvents = vi.fn().mockResolvedValue({ + events: [makeMinimalEvent('skip-1', false), makeMinimalEvent('ok-1', true)], + }); + (worker as any).server = { getEvents }; + + await (worker as any).fetchAndProcessEvents(); + + const counters = worker.getEventCounters(); + expect(counters.eventsProcessed).toBe(1); + expect(counters.eventsFailed).toBe(0); + }); + + it('stays non-degraded when failures are below the sample / rate thresholds', async () => { + const getEvents = vi.fn().mockResolvedValue({ + events: [makeMinimalEvent('fail-only')], + }); + (worker as any).server = { getEvents }; + vi.spyOn(worker, 'processEvent').mockRejectedValue(new Error('boom')); + + await (worker as any).fetchAndProcessEvents(); + + const counters = worker.getEventCounters(); + expect(counters.eventsFailed).toBe(1); + // Need ≥3 samples before a spike is declared. + expect(counters.degraded).toBe(false); + }); + }); });