From e94c73edb87fa1df1525655581d0a3dafa0fdd78 Mon Sep 17 00:00:00 2001 From: vjuliaife Date: Sat, 17 Oct 2026 04:10:43 -0700 Subject: [PATCH] feat(health): add livez, dependency-aware readyz and admin status --- .env.example | 11 ++ docs/api.md | 10 +- docs/architecture.md | 4 + scripts/smoke.sh | 3 + src/app.ts | 21 ++- src/config/index.ts | 18 +++ src/controllers/admin.controller.ts | 44 ++++++ src/cron/ingestion.cron.ts | 4 + src/cron/monitor.cron.ts | 10 ++ src/cron/registry.ts | 26 ++++ src/lib/gemini.ts | 7 + src/routes/admin.routes.ts | 3 +- src/routes/health.routes.ts | 41 ++++++ src/server.ts | 16 ++- src/services/health.service.ts | 188 +++++++++++++++++++++++++ src/services/indexer.service.ts | 19 ++- tests/health.test.ts | 204 ++++++++++++++++++++++++++++ 17 files changed, 606 insertions(+), 23 deletions(-) create mode 100644 src/cron/registry.ts create mode 100644 src/routes/health.routes.ts create mode 100644 src/services/health.service.ts create mode 100644 tests/health.test.ts diff --git a/.env.example b/.env.example index 0b2c5a4..2e52df0 100644 --- a/.env.example +++ b/.env.example @@ -78,6 +78,17 @@ RESOLUTION_DISPUTE_WINDOW_SECS=86400 INDEXER_ENABLED=true INDEXER_POLL_MS=5000 +# Health probes (#111): /livez (process only) and /readyz (dependencies). +# HEALTH_CRITICAL: comma list of database,redis,rpc,indexer that fail readiness. +HEALTH_CRITICAL=database,redis +HEALTH_PROBE_TIMEOUT_MS=1500 +HEALTH_CACHE_TTL_MS=2000 +INDEXER_MAX_LAG_LEDGERS=120 +INDEXER_MAX_STALENESS_MS=60000 +LIVENESS_MAX_LOOP_DELAY_MS=2000 +# Drain + close must fit the 15 s forced-exit window. +SHUTDOWN_DRAIN_MS=5000 + # ── LLM providers (market and trace generation) ──────────────────────────── ANTHROPIC_API_KEY= GEMINI_API_KEY= diff --git a/docs/api.md b/docs/api.md index d9e7313..db26657 100644 --- a/docs/api.md +++ b/docs/api.md @@ -4,7 +4,7 @@ This is the contract between `OracleDesk-Frontend` and this backend. The Soroban contracts in `OracleDesk-SmartContract/oracledesk-stellar` are the source of truth for anything on-chain; this API never overrides them. -Base URL: `/api/v1` (liveness is `/health`, readiness is `/readyz`). +Base URL: `/api/v1` (liveness is `/livez`, readiness is `/readyz`). ## Conventions @@ -127,9 +127,13 @@ useMutation({ ### Health -`GET /health` → `{ status: "healthy", uptime: number }` +`GET /livez` → `{ status: "alive" }`. Process health only (event-loop delay, fatal flag); never touches a dependency, so an outage cannot restart every instance. `503 NOT_ALIVE` when the loop is stalled. -`GET /readyz` → `200` when the database is reachable, `503 NOT_READY` otherwise. +`GET /readyz` → `200 { status: "ready", checks }` when every critical dependency is up, `503 NOT_READY` with `status: "not_ready"` otherwise, or `status: "draining"` once shutdown has begun. Each check is `{ name, ok, critical, latencyMs, error? }` for `database`, `redis`, `rpc` and `indexer`; `error` is one of `timeout`, `connection_refused`, `rpc_unhealthy`, `lagging`, `unavailable`. Probes time out after `HEALTH_PROBE_TIMEOUT_MS`, results are cached for `HEALTH_CACHE_TTL_MS` (failures for 500 ms) and concurrent calls share one run. `HEALTH_CRITICAL` (default `database,redis`) picks which dependencies fail readiness. Both endpoints bypass the rate limiter. + +`GET /health` → `{ status: "healthy", uptime: number }`. Deprecated: it sends `Deprecation: true` and `Link: ; rel="successor-version"`. + +`GET /api/v1/admin/status` (admin JWT) → probe details with latencies, indexer cursor vs RPC latest ledger and lag, last success/failure per cron job (in memory, reset on restart), chain execution mode, agent public key, LLM provider configuration and Gemini breaker state, and process info. ### Auth (signed challenge, SEP-10 style) diff --git a/docs/architecture.md b/docs/architecture.md index fc3989f..dfa4bf2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -73,6 +73,10 @@ Outcomes are decided on-chain by the resolver contract. The backend only reads ` - The ledger cursor lives in `indexer_cursors`. If it is older than RPC retention (about 7 days on testnet), the indexer warns and resumes from the oldest retained ledger; anything in between is lost to it. - New `Trade` and `TracePublished` events are pushed to clients over socket.io as `TRADE_EXECUTED` and `REASONING_PUBLISHED`. +## Health and degraded mode (#111) + +`/livez` checks only the process; `/readyz` probes Postgres, Redis, RPC and the indexer lag. `HEALTH_CRITICAL` decides which failures return 503. By default Redis and Postgres are critical (login and every data endpoint need them) and RPC is not: an instance without RPC stays ready and keeps serving database reads, while on-chain state, resolution, trace and payment endpoints return 502. The indexer fails when it is more than `INDEXER_MAX_LAG_LEDGERS` behind or its cursor is older than `INDEXER_MAX_STALENESS_MS`; with no cursor yet it reports `initialising`. On SIGTERM `/readyz` returns 503 `draining` for `SHUTDOWN_DRAIN_MS` before the server stops accepting connections. + ## Generated code `src/generated/` is copied from the pinned `contracts` submodule by `scripts/sync-contracts.mjs`, with a do-not-edit header and `// @ts-nocheck`. CI fails if it's stale. diff --git a/scripts/smoke.sh b/scripts/smoke.sh index ce5a44c..2e4fb87 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -17,6 +17,9 @@ need() { node -e "const j=JSON.parse(require('fs').readFileSync(0));if(!j.ok){co step "GET /health" curl -fsS "$API/health" | need 'console.log(j.data.status)' +step "GET /readyz" +curl -fsS "$API/readyz" | need 'console.log(j.data.status)' + step "throwaway keypair" KEYS=$(node -e 'const {Keypair}=require("@stellar/stellar-sdk");const k=Keypair.random();console.log(k.publicKey()+" "+k.secret())') ADDRESS=${KEYS% *}; SECRET=${KEYS#* } diff --git a/src/app.ts b/src/app.ts index 3793adc..9c11b6d 100644 --- a/src/app.ts +++ b/src/app.ts @@ -6,18 +6,23 @@ import morgan from 'morgan'; import hpp from 'hpp'; import { config } from './config'; -import { prisma } from './lib/prisma'; import { logger } from './lib/logger'; import { errorMiddleware } from './middlewares/error.middleware'; import { globalLimiter } from './middlewares/rate-limit.middleware'; import { abortMiddleware } from './middlewares/abort.middleware'; import router from './routes'; +import { createHealthRouter } from './routes/health.routes'; +import { health } from './services/health.service'; const app = express(); // BigInt (u64 market ids, i128 amounts) serialises as a decimal string. app.set('json replacer', (_key: string, value: unknown) => (typeof value === 'bigint' ? value.toString() : value)); +// ─── Liveness / Readiness (#111) ─── +// Before the rate limiter, compression and CORS: orchestrators poll often from one IP. +app.use(createHealthRouter(health)); + // ─── Security ─── app.use(helmet()); app.use(hpp()); @@ -58,21 +63,13 @@ app.use( app.use(abortMiddleware); // ─── Health Check ─── +// Deprecated in favour of /livez and /readyz; response shape is unchanged. app.get('/health', (_req, res) => { + res.setHeader('Deprecation', 'true'); + res.setHeader('Link', '; rel="successor-version"'); res.json({ ok: true, data: { status: 'healthy', uptime: process.uptime() }, error: null }); }); -// Readiness for rolling deploys (#144): the process is only ready when it can -// reach the database. Keep this cheap — it runs every few seconds. -app.get('/readyz', async (_req, res) => { - try { - await prisma.$queryRaw`SELECT 1`; - res.json({ ok: true, data: { status: 'ready' }, error: null }); - } catch { - res.status(503).json({ ok: false, data: null, error: { code: 'NOT_READY', message: 'Database is not reachable' } }); - } -}); - // ─── API Routes ─── app.use('/api/v1', router); diff --git a/src/config/index.ts b/src/config/index.ts index 1864342..42d96cb 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -25,6 +25,12 @@ const optionalAddress = z 'must be a G… account or C… contract address', ); +export const PROBE_NAMES = ['database', 'redis', 'rpc', 'indexer'] as const; + +const probeName = z + .string() + .refine((v) => (PROBE_NAMES as readonly string[]).includes(v), `must be one of ${PROBE_NAMES.join(', ')}`); + const commaList = (item: z.ZodType, fallback = "") => z .string() @@ -101,6 +107,18 @@ const envSchema = z INDEXER_ENABLED: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'), INDEXER_POLL_MS: z.string().default('5000').transform(Number), + // ── Health probes (#111) ────────────────────────────────────────────── + // Dependencies that make /readyz fail. RPC is non-critical by default so + // cached DB reads can still be served while chain reads return 502. + HEALTH_CRITICAL: commaList(probeName, 'database,redis'), + HEALTH_PROBE_TIMEOUT_MS: z.string().default('1500').transform(Number), + HEALTH_CACHE_TTL_MS: z.string().default('2000').transform(Number), + INDEXER_MAX_LAG_LEDGERS: z.string().default('120').transform(Number), + INDEXER_MAX_STALENESS_MS: z.string().default('60000').transform(Number), + LIVENESS_MAX_LOOP_DELAY_MS: z.string().default('2000').transform(Number), + // Time /readyz reports "draining" before the HTTP server stops accepting. + SHUTDOWN_DRAIN_MS: z.string().default('5000').transform(Number), + // ── LLM providers (optional; generation fails at runtime without one) ─ ANTHROPIC_API_KEY: optionalString, GEMINI_API_KEY: optionalString, diff --git a/src/controllers/admin.controller.ts b/src/controllers/admin.controller.ts index e7a8132..d3d2ceb 100644 --- a/src/controllers/admin.controller.ts +++ b/src/controllers/admin.controller.ts @@ -5,6 +5,13 @@ import { TraceAuditStatus } from '@prisma/client'; import { runTraceAudit } from '../services/trace-audit.service'; import { sendSuccess, sendError, parsePagination, buildPaginationMeta } from '../utils/response.util'; import { logger } from '../lib/logger'; +import { config } from '../config'; +import { secret } from '../config/secrets'; +import { health } from '../services/health.service'; +import { indexerStatus } from '../services/indexer.service'; +import { agentSigner } from '../services/stellar/clients'; +import { geminiBreakerState } from '../lib/gemini'; +import { getCronStatus } from '../cron/registry'; /** * GET /admin/trace-audit @@ -83,3 +90,40 @@ export async function triggerTraceAuditRun(req: Request, res: Response): Promise sendSuccess(res, { runId, message: 'Trace audit run scheduled' }, 202); } + +/** + * GET /admin/status (#111) + * Probe details, indexer lag, cron runs, chain mode and LLM state. Unlike + * /readyz this includes ledger numbers, so it is admin-only. + */ +export async function getSystemStatus(_req: Request, res: Response): Promise { + const { ready, checks } = await health.readiness(); + const latestLedger = checks.find((c) => c.name === 'rpc')?.detail?.latestLedger; + const indexer = indexerStatus(); + sendSuccess(res, { + ready, + draining: health.isDraining(), + checks, + indexer: { + ...indexer, + rpcLatestLedger: typeof latestLedger === 'number' ? latestLedger : null, + lagLedgers: typeof latestLedger === 'number' && indexer.lastLedger !== null ? latestLedger - indexer.lastLedger : null, + }, + crons: getCronStatus(), + chain: { + mode: config.CHAIN_EXECUTION_MODE, + agentPublicKey: agentSigner().publicKey ?? null, + preflight: 'not_implemented', + }, + llm: { + anthropicConfigured: secret('ANTHROPIC_API_KEY') !== '', + geminiConfigured: secret('GEMINI_API_KEY') !== '', + geminiBreaker: geminiBreakerState(), + }, + process: { + uptimeSeconds: process.uptime(), + memory: process.memoryUsage(), + version: process.env.npm_package_version ?? null, + }, + }); +} diff --git a/src/cron/ingestion.cron.ts b/src/cron/ingestion.cron.ts index 6f8ccc2..9d2e0cd 100644 --- a/src/cron/ingestion.cron.ts +++ b/src/cron/ingestion.cron.ts @@ -1,6 +1,7 @@ import cron from 'node-cron'; import { runMarketMakerCycle } from '../agents/market-maker.agent'; import { logger } from '../lib/logger'; +import { recordCronRun } from './registry'; let isRunning = false; @@ -22,9 +23,12 @@ export function startIngestionCron(): void { } isRunning = true; + const startedAt = new Date(); try { await runMarketMakerCycle(); + recordCronRun('ingestion', 'success', startedAt, new Date()); } catch (err) { + recordCronRun('ingestion', 'failure', startedAt, new Date(), err); logger.error({ err }, 'Ingestion cron: cycle failed'); } finally { isRunning = false; diff --git a/src/cron/monitor.cron.ts b/src/cron/monitor.cron.ts index 9a2bb49..fad4c31 100644 --- a/src/cron/monitor.cron.ts +++ b/src/cron/monitor.cron.ts @@ -3,6 +3,7 @@ import { runTraderCycle } from '../agents/trader.agent'; import { expireStaleSubscriptions } from '../services/subscription.service'; import { logger } from '../lib/logger'; import { purgeExpiredIdempotencyKeys } from '../middlewares/idempotency.middleware'; +import { recordCronRun } from './registry'; let traderRunning = false; @@ -24,9 +25,12 @@ export function startMonitorCron(): void { } traderRunning = true; + const startedAt = new Date(); try { await runTraderCycle(); + recordCronRun('trader', 'success', startedAt, new Date()); } catch (err) { + recordCronRun('trader', 'failure', startedAt, new Date(), err); logger.error({ err }, 'Monitor cron: trader cycle failed'); } finally { traderRunning = false; @@ -35,19 +39,25 @@ export function startMonitorCron(): void { // Subscription expiry — every hour cron.schedule('0 * * * *', async () => { + const startedAt = new Date(); try { await expireStaleSubscriptions(); + recordCronRun('subscription-expiry', 'success', startedAt, new Date()); } catch (err) { + recordCronRun('subscription-expiry', 'failure', startedAt, new Date(), err); logger.error({ err }, 'Monitor cron: subscription expiry failed'); } }); // Expired Idempotency-Key rows (24h TTL, #123) — hourly at :10, in batches of 5 000 cron.schedule('10 * * * *', async () => { + const startedAt = new Date(); try { const deleted = await purgeExpiredIdempotencyKeys(); + recordCronRun('idempotency-purge', 'success', startedAt, new Date()); if (deleted > 0) logger.info({ deleted }, 'Monitor cron: purged expired idempotency keys'); } catch (err) { + recordCronRun('idempotency-purge', 'failure', startedAt, new Date(), err); logger.error({ err }, 'Monitor cron: idempotency key purge failed'); } }); diff --git a/src/cron/registry.ts b/src/cron/registry.ts new file mode 100644 index 0000000..3c6c6a4 --- /dev/null +++ b/src/cron/registry.ts @@ -0,0 +1,26 @@ +export type CronOutcome = 'success' | 'failure'; + +interface CronRun { + startedAt: string; + finishedAt: string; + error?: string; +} + +const lastRuns = new Map(); + +/** In-memory last success/failure per cron job (#111); not persisted. */ +export function recordCronRun(job: string, outcome: CronOutcome, startedAt: Date, finishedAt: Date, error?: unknown): void { + const entry = lastRuns.get(job) ?? {}; + const run: CronRun = { startedAt: startedAt.toISOString(), finishedAt: finishedAt.toISOString() }; + if (outcome === 'success') { + entry.lastSuccess = run; + } else { + run.error = error instanceof Error ? error.message : String(error); + entry.lastFailure = run; + } + lastRuns.set(job, entry); +} + +export function getCronStatus(): Record { + return Object.fromEntries(lastRuns); +} diff --git a/src/lib/gemini.ts b/src/lib/gemini.ts index d71b2af..1f0ac92 100644 --- a/src/lib/gemini.ts +++ b/src/lib/gemini.ts @@ -56,6 +56,13 @@ function isQuotaExhausted(): boolean { return Date.now() < _quotaExhaustedUntil; } +/** Read-only breaker state for the admin status endpoint. */ +export function geminiBreakerState(): { open: boolean; until: string | null } { + return isQuotaExhausted() + ? { open: true, until: new Date(_quotaExhaustedUntil).toISOString() } + : { open: false, until: null }; +} + function markQuotaExhausted(): void { _quotaExhaustedUntil = Date.now() + QUOTA_COOLDOWN_MS; logger.error( diff --git a/src/routes/admin.routes.ts b/src/routes/admin.routes.ts index a50318c..ee993eb 100644 --- a/src/routes/admin.routes.ts +++ b/src/routes/admin.routes.ts @@ -1,11 +1,12 @@ import { Router } from 'express'; import { requireAuth, requireAdmin } from '../middlewares/auth.middleware'; -import { getTraceAuditSummary, triggerTraceAuditRun } from '../controllers/admin.controller'; +import { getTraceAuditSummary, triggerTraceAuditRun, getSystemStatus } from '../controllers/admin.controller'; const router = Router(); router.use(requireAuth, requireAdmin); +router.get('/status', getSystemStatus); router.get('/trace-audit', getTraceAuditSummary); router.post('/trace-audit/run', triggerTraceAuditRun); diff --git a/src/routes/health.routes.ts b/src/routes/health.routes.ts new file mode 100644 index 0000000..df5dfd0 --- /dev/null +++ b/src/routes/health.routes.ts @@ -0,0 +1,41 @@ +import { Router } from 'express'; +import type { HealthService } from '../services/health.service'; + +/** + * /livez and /readyz (#111). Mounted before the global rate limiter so + * orchestrators polling from one IP are never throttled. Responses carry only + * name, ok, critical, latencyMs and a sanitised error class; probe detail + * (ledger numbers, lag) is admin-only. + */ +export function createHealthRouter(health: HealthService): Router { + const router = Router(); + + router.get('/livez', (_req, res) => { + const live = health.liveness(); + if (live.alive) { + res.json({ ok: true, data: { status: 'alive' }, error: null }); + return; + } + res.status(503).json({ ok: false, data: { status: 'unhealthy' }, error: { code: 'NOT_ALIVE', message: 'Process is not healthy' } }); + }); + + router.get('/readyz', async (_req, res) => { + if (health.isDraining()) { + res.status(503).json({ ok: false, data: { status: 'draining', checks: [] }, error: { code: 'NOT_READY', message: 'Server is shutting down' } }); + return; + } + const { ready, checks } = await health.readiness(); + const publicChecks = checks.map(({ name, ok, critical, latencyMs, error }) => ({ name, ok, critical, latencyMs, ...(error ? { error } : {}) })); + if (ready) { + res.json({ ok: true, data: { status: 'ready', checks: publicChecks }, error: null }); + return; + } + res.status(503).json({ + ok: false, + data: { status: 'not_ready', checks: publicChecks }, + error: { code: 'NOT_READY', message: 'A critical dependency is unavailable' }, + }); + }); + + return router; +} diff --git a/src/server.ts b/src/server.ts index 5f78ba5..24ecfaf 100644 --- a/src/server.ts +++ b/src/server.ts @@ -11,6 +11,7 @@ import { startTraceAuditCron } from './cron/trace-audit.cron'; import { startTracePublicationCron } from './cron/trace-publication.cron'; import { startEventListener } from './services/indexer.service'; import { closeRedis } from './lib/redis'; +import { health } from './services/health.service'; let stopEventListener: (() => void) | null = null; let stopSecretRefresh: (() => void) | null = null; @@ -81,6 +82,14 @@ async function bootstrap(): Promise { // ─── Graceful Shutdown ─── const shutdown = async (signal: string) => { logger.info({ signal }, 'Shutting down gracefully...'); + // Force exit if shutdown (drain + close) takes too long + setTimeout(() => { + logger.error('Forced shutdown after timeout'); + process.exit(1); + }, 15_000); + // Report 503 from /readyz first so load balancers stop routing before the socket closes. + health.startDraining(); + await new Promise((resolve) => setTimeout(resolve, config.SHUTDOWN_DRAIN_MS)); const { triggerShutdown } = await import('./lib/shutdown'); triggerShutdown(); @@ -93,12 +102,6 @@ async function bootstrap(): Promise { logger.info('Database disconnected'); process.exit(0); }); - - // Force exit if shutdown takes too long - setTimeout(() => { - logger.error('Forced shutdown after timeout'); - process.exit(1); - }, 15_000); }; process.on('SIGTERM', () => shutdown('SIGTERM')); @@ -110,6 +113,7 @@ async function bootstrap(): Promise { }); process.on('uncaughtException', (err) => { + health.markFatal(); logger.fatal({ err }, 'Uncaught exception — shutting down'); process.exit(1); }); diff --git a/src/services/health.service.ts b/src/services/health.service.ts new file mode 100644 index 0000000..8274cdf --- /dev/null +++ b/src/services/health.service.ts @@ -0,0 +1,188 @@ +/** + * Liveness and readiness (#111). + * + * Liveness is process-level only (event-loop delay, fatal flag) and never + * touches a dependency, so a database outage cannot restart every instance. + * Readiness runs one probe per dependency with a timeout, caches results for a + * short TTL and shares one in-flight run between concurrent callers. + */ +import { monitorEventLoopDelay } from 'perf_hooks'; +import { config } from '../config'; +import { prisma } from '../lib/prisma'; +import { logger } from '../lib/logger'; +import { getRedis } from '../lib/redis'; +import { rpcServer } from './stellar/clients'; +import { CURSOR_ID } from './indexer.service'; + +export type ProbeName = 'database' | 'redis' | 'rpc' | 'indexer'; + +export interface ProbeResult { + name: ProbeName; + ok: boolean; + critical: boolean; + latencyMs: number; + checkedAt: string; + detail?: Record; + error?: string; +} + +export interface ProbeContext { + /** Cached or in-flight result of another probe (single run, never a second call). */ + result(name: ProbeName): Promise; +} + +export interface Probe { + name: ProbeName; + run(signal: AbortSignal, ctx: ProbeContext): Promise>; +} + +export interface HealthServiceDeps { + probes: Probe[]; + ttlMs: number; + timeoutMs: number; + criticality: Record; + maxLoopDelayMs?: number; + now?: () => number; +} + +const FAILURE_TTL_MS = 500; +const LOG_INTERVAL_MS = 60_000; + +/** Only these classes reach responses; the full error is logged instead. */ +function sanitise(err: unknown): string { + const text = err instanceof Error ? `${err.message} ${(err as { code?: string }).code ?? ''}` : String(err); + if (/^timeout\b/.test(text)) return 'timeout'; + if (/ECONNREFUSED|ENOTFOUND|EAI_AGAIN|Can't reach database/i.test(text)) return 'connection_refused'; + return 'unavailable'; +} + +export function createHealthService(deps: HealthServiceDeps) { + const now = deps.now ?? Date.now; + const probes = new Map(deps.probes.map((p) => [p.name, p])); + const cache = new Map(); + const inflight = new Map>(); + const lastLogged = new Map(); + const ctx: ProbeContext = { result: (name) => get(name) }; + let draining = false; + let fatal = false; + const loop = monitorEventLoopDelay({ resolution: 10 }); + loop.enable(); + + async function execute(probe: Probe): Promise { + const started = now(); + const controller = new AbortController(); + let timer: NodeJS.Timeout | undefined; + const base = { name: probe.name, critical: deps.criticality[probe.name] }; + try { + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => { + controller.abort(); + reject(new Error('timeout')); + }, deps.timeoutMs); + timer.unref(); + }); + const outcome = await Promise.race([probe.run(controller.signal, ctx), timeout]); + return { ...base, ...outcome, latencyMs: now() - started, checkedAt: new Date(now()).toISOString() }; + } catch (err) { + if (now() - (lastLogged.get(probe.name) ?? -Infinity) >= LOG_INTERVAL_MS) { + lastLogged.set(probe.name, now()); + logger.warn({ probe: probe.name, err: err instanceof Error ? err.message : err }, 'Health probe failed'); + } + return { ...base, ok: false, error: sanitise(err), latencyMs: now() - started, checkedAt: new Date(now()).toISOString() }; + } finally { + clearTimeout(timer); + } + } + + function get(name: ProbeName): Promise { + const hit = cache.get(name); + if (hit && hit.expiresAt > now()) return Promise.resolve(hit.result); + const running = inflight.get(name); + if (running) return running; + const run = execute(probes.get(name)!) + .then((result) => { + cache.set(name, { result, expiresAt: now() + (result.ok ? deps.ttlMs : Math.min(FAILURE_TTL_MS, deps.ttlMs)) }); + return result; + }) + .finally(() => inflight.delete(name)); + inflight.set(name, run); + return run; + } + + return { + async readiness(): Promise<{ ready: boolean; checks: ProbeResult[] }> { + const checks = await Promise.all(deps.probes.map((p) => get(p.name))); + return { ready: checks.every((c) => c.ok || !c.critical), checks }; + }, + /** Process-level only: never calls a dependency probe. */ + liveness(): { alive: boolean; loopDelayMs: number; fatal: boolean } { + const p99 = loop.percentile(99) / 1e6; + loop.reset(); + const loopDelayMs = Number.isFinite(p99) ? p99 : 0; + return { alive: !fatal && loopDelayMs <= (deps.maxLoopDelayMs ?? 2000), loopDelayMs, fatal }; + }, + markFatal(): void { fatal = true; }, + isDraining: () => draining, + startDraining(): void { draining = true; }, + }; +} + +export type HealthService = ReturnType; + +export const defaultProbes: Probe[] = [ + { + name: 'database', + async run() { + await prisma.$queryRaw`SELECT 1`; + return { ok: true }; + }, + }, + { + name: 'redis', + async run() { + const redis = getRedis(); + // ioredis queues commands while reconnecting; fail fast instead of waiting on retries. + if (redis.status !== 'ready') return { ok: false, error: 'connection_refused' }; + await redis.ping(); + return { ok: true }; + }, + }, + { + name: 'rpc', + async run() { + const health = await rpcServer.getHealth(); + const detail = { latestLedger: health.latestLedger, oldestLedger: health.oldestLedger }; + return health.status === 'healthy' ? { ok: true, detail } : { ok: false, detail, error: 'rpc_unhealthy' }; + }, + }, + { + name: 'indexer', + async run(_signal, ctx) { + if (!config.INDEXER_ENABLED) return { ok: true, detail: { enabled: false } }; + const row = await prisma.indexerCursor.findUnique({ where: { id: CURSOR_ID } }); + // A fresh instance has no cursor yet; that is not a reason to fail readiness. + if (!row) return { ok: true, detail: { enabled: true, state: 'initialising' } }; + const rpc = await ctx.result('rpc'); + const latest = rpc.detail?.latestLedger; + const lagLedgers = typeof latest === 'number' ? latest - row.lastLedger : null; + const staleMs = Date.now() - row.updatedAt.getTime(); + const detail = { enabled: true, lastLedger: row.lastLedger, lagLedgers, staleMs }; + const lagging = + (lagLedgers !== null && lagLedgers > config.INDEXER_MAX_LAG_LEDGERS) || staleMs > config.INDEXER_MAX_STALENESS_MS; + return lagging ? { ok: false, detail, error: 'lagging' } : { ok: true, detail }; + }, + }, +]; + +export const health = createHealthService({ + probes: defaultProbes, + ttlMs: config.HEALTH_CACHE_TTL_MS, + timeoutMs: config.HEALTH_PROBE_TIMEOUT_MS, + criticality: { + database: config.HEALTH_CRITICAL.includes('database'), + redis: config.HEALTH_CRITICAL.includes('redis'), + rpc: config.HEALTH_CRITICAL.includes('rpc'), + indexer: config.HEALTH_CRITICAL.includes('indexer'), + }, + maxLoopDelayMs: config.LIVENESS_MAX_LOOP_DELAY_MS, +}); diff --git a/src/services/indexer.service.ts b/src/services/indexer.service.ts index 6a4ec0f..9834ee7 100644 --- a/src/services/indexer.service.ts +++ b/src/services/indexer.service.ts @@ -320,6 +320,18 @@ export function decodedFromStored(row: { }; } +// Poll state for health reporting (#111); module-level because one indexer runs per process. +const pollState: { lastLedger: number | null; updatedAt: string | null; lastPollAt: string | null; lastPollError: string | null } = { + lastLedger: null, + updatedAt: null, + lastPollAt: null, + lastPollError: null, +}; + +export function indexerStatus() { + return { enabled: config.INDEXER_ENABLED, ...pollState }; +} + /** Polls every INDEXER_POLL_MS. Returns a stop function. */ export function startEventListener(emit: IndexerDeps['emit']): () => void { if (!config.INDEXER_ENABLED) { @@ -330,10 +342,15 @@ export function startEventListener(emit: IndexerDeps['emit']): () => void { let timer: NodeJS.Timeout | null = null; const tick = async () => { try { - await backfillEvents(undefined, { source: rpcServer, emit }); + const result = await backfillEvents(undefined, { source: rpcServer, emit }); + pollState.lastLedger = result.lastLedger; + pollState.updatedAt = new Date().toISOString(); + pollState.lastPollError = null; } catch (err) { + pollState.lastPollError = err instanceof Error ? err.message : String(err); logger.warn({ err: err instanceof Error ? err.message : err }, 'Indexer poll failed; will retry'); } + pollState.lastPollAt = new Date().toISOString(); if (!stopped) timer = setTimeout(tick, config.INDEXER_POLL_MS); }; void tick(); diff --git a/tests/health.test.ts b/tests/health.test.ts new file mode 100644 index 0000000..2961b86 --- /dev/null +++ b/tests/health.test.ts @@ -0,0 +1,204 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import type { AddressInfo } from 'node:net'; +import express from 'express'; +import rateLimit from 'express-rate-limit'; +import { Keypair } from '@stellar/stellar-sdk'; +import { setupTestEnv } from './helpers/env'; + +setupTestEnv(); + +type Mod = typeof import('../src/services/health.service'); +type ProbeName = import('../src/services/health.service').ProbeName; + +const allCritical: Record = { database: true, redis: true, rpc: true, indexer: true }; + +async function service( + probes: import('../src/services/health.service').Probe[], + over: Partial[0]> = {}, +) { + const { createHealthService } = await import('../src/services/health.service'); + return createHealthService({ probes, ttlMs: 2000, timeoutMs: 1500, criticality: allCritical, ...over }); +} + +const probe = (name: ProbeName, run: () => Promise<{ ok: boolean; error?: string }>) => ({ name, run }); +const up = (name: ProbeName) => probe(name, async () => ({ ok: true })); + +async function withRouter(health: Awaited>, fn: (base: string) => Promise) { + const { createHealthRouter } = await import('../src/routes/health.routes'); + const app = express(); + // A limiter of 1 request proves the probe routes sit in front of it. + app.use(createHealthRouter(health)); + app.use(rateLimit({ windowMs: 60_000, limit: 1 })); + app.get('/limited', (_req, res) => void res.json({})); + const server = app.listen(0); + try { + await fn(`http://127.0.0.1:${(server.address() as AddressInfo).port}`); + } finally { + server.close(); + } +} + +test('livez returns 200 without touching any dependency', async () => { + const boom = (name: ProbeName) => probe(name, async () => { throw new Error('probe must not run'); }); + const health = await service([boom('database'), boom('redis')]); + await withRouter(health, async (base) => { + const res = await fetch(`${base}/livez`); + assert.equal(res.status, 200); + assert.equal((await res.json()).data.status, 'alive'); + }); +}); + +test('readyz returns 200 when all critical probes pass and a non-critical one fails', async () => { + const health = await service( + [up('database'), probe('rpc', async () => ({ ok: false, error: 'rpc_unhealthy' }))], + { criticality: { ...allCritical, rpc: false } }, + ); + await withRouter(health, async (base) => { + const res = await fetch(`${base}/readyz`); + assert.equal(res.status, 200); + const body = await res.json(); + assert.equal(body.data.status, 'ready'); + assert.equal(body.data.checks.find((c: { name: string }) => c.name === 'rpc').ok, false); + }); +}); + +test('readyz returns 503 when a critical probe fails, with sanitised error', async () => { + const health = await service([probe('database', async () => { throw new Error('connect ECONNREFUSED 10.0.0.5:5432'); })]); + await withRouter(health, async (base) => { + const res = await fetch(`${base}/readyz`); + assert.equal(res.status, 503); + const body = await res.json(); + assert.equal(body.error.code, 'NOT_READY'); + assert.equal(body.data.checks[0].error, 'connection_refused'); + assert.ok(!JSON.stringify(body).includes('10.0.0.5')); + }); +}); + +test('a hanging probe times out within HEALTH_PROBE_TIMEOUT_MS', async () => { + const hang = probe('redis', () => new Promise((resolve) => setTimeout(() => resolve({ ok: true }), 500))); + const health = await service([hang], { timeoutMs: 30 }); + const started = Date.now(); + const { ready, checks } = await health.readiness(); + assert.ok(Date.now() - started < 300); + assert.equal(ready, false); + assert.equal(checks[0].error, 'timeout'); +}); + +test('concurrent readyz calls share one in-flight probe run', async () => { + let calls = 0; + const slow = probe('database', async () => { + calls++; + await new Promise((resolve) => setTimeout(resolve, 20)); + return { ok: true }; + }); + const health = await service([slow]); + await Promise.all([health.readiness(), health.readiness(), health.readiness()]); + assert.equal(calls, 1); +}); + +test('results are cached for ttl and failures for the shorter ttl', async () => { + let t = 1_000_000; + let calls = 0; + let ok = true; + const flaky = probe('database', async () => { calls++; return { ok }; }); + const health = await service([flaky], { now: () => t }); + await health.readiness(); + await health.readiness(); + assert.equal(calls, 1); + t += 2001; + ok = false; + await health.readiness(); + assert.equal(calls, 2); + t += 300; // inside the 500 ms failure ttl + await health.readiness(); + assert.equal(calls, 2); + t += 300; // past it + ok = true; + const { ready } = await health.readiness(); + assert.equal(calls, 3); + assert.equal(ready, true); +}); + +test('indexer probe fails when lag exceeds INDEXER_MAX_LAG_LEDGERS and passes when disabled', async () => { + const { defaultProbes } = await import('../src/services/health.service'); + const { config } = await import('../src/config'); + const { prisma } = await import('../src/lib/prisma'); + const indexer = defaultProbes.find((p) => p.name === 'indexer')!; + const signal = new AbortController().signal; + const ctx = (latestLedger: number) => ({ + result: async () => ({ name: 'rpc' as const, ok: true, critical: false, latencyMs: 1, checkedAt: '', detail: { latestLedger } }), + }); + const cursors = prisma.indexerCursor as unknown as { findUnique: unknown }; + const original = cursors.findUnique; + const enabled = config.INDEXER_ENABLED; + try { + config.INDEXER_ENABLED = false; + assert.equal((await indexer.run(signal, ctx(1000))).ok, true); + + config.INDEXER_ENABLED = true; + cursors.findUnique = async () => ({ lastLedger: 1000 - config.INDEXER_MAX_LAG_LEDGERS - 1, updatedAt: new Date() }); + const lagging = await indexer.run(signal, ctx(1000)); + assert.equal(lagging.ok, false); + assert.equal(lagging.error, 'lagging'); + + cursors.findUnique = async () => ({ lastLedger: 990, updatedAt: new Date() }); + assert.equal((await indexer.run(signal, ctx(1000))).ok, true); + + cursors.findUnique = async () => null; + assert.equal((await indexer.run(signal, ctx(1000))).ok, true); + } finally { + cursors.findUnique = original; + config.INDEXER_ENABLED = enabled; + } +}); + +test('readyz returns 503 draining after startDraining()', async () => { + const health = await service([up('database')]); + health.startDraining(); + await withRouter(health, async (base) => { + const res = await fetch(`${base}/readyz`); + assert.equal(res.status, 503); + assert.equal((await res.json()).data.status, 'draining'); + }); +}); + +test('readyz and livez are not rate limited', async () => { + const health = await service([up('database')]); + await withRouter(health, async (base) => { + for (let i = 0; i < 5; i++) { + assert.equal((await fetch(`${base}/livez`)).status, 200); + assert.equal((await fetch(`${base}/readyz`)).status, 200); + } + assert.equal((await fetch(`${base}/limited`)).status, 200); + assert.equal((await fetch(`${base}/limited`)).status, 429); + }); +}); + +test('/health keeps its shape and is marked deprecated', async () => { + const { default: app } = await import('../src/app'); + const server = app.listen(0); + try { + const res = await fetch(`http://127.0.0.1:${(server.address() as AddressInfo).port}/health`); + assert.equal(res.status, 200); + assert.equal(res.headers.get('deprecation'), 'true'); + assert.equal((await res.json()).data.status, 'healthy'); + } finally { + server.close(); + } +}); + +test('admin status requires an admin JWT', async () => { + const { default: app } = await import('../src/app'); + const { signToken } = await import('../src/middlewares/auth.middleware'); + const server = app.listen(0); + try { + const base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + assert.equal((await fetch(`${base}/api/v1/admin/status`)).status, 401); + const token = signToken('u1', Keypair.random().publicKey()); + const nonAdmin = await fetch(`${base}/api/v1/admin/status`, { headers: { authorization: `Bearer ${token}` } }); + assert.equal(nonAdmin.status, 403); + } finally { + server.close(); + } +});