diff --git a/.changeset/reliability-resilience-cluster.md b/.changeset/reliability-resilience-cluster.md new file mode 100644 index 00000000..fcd34f03 --- /dev/null +++ b/.changeset/reliability-resilience-cluster.md @@ -0,0 +1,15 @@ +--- +'lens': minor +--- + +Comprehensive Process Resiliency, Reliability, Health Probes & Telemetry Cluster: +- #348: Handle Horizon 429 rate limiting with Retry-After backoff, non-2xx pre-parse detection, and remaining quota gauge metrics +- #345: Active retention path mode telemetry, Redis periodic reconnection recovery, and advisory-locked safe concurrent pruning +- #344: Unhandled rejection and uncaught exception guard logging full stack trace and exiting non-zero cleanly +- #342: Webhook dead-letter store recording exhausted deliveries without persisting secrets, with failure categorisation +- #341: Full jitter on exponential webhook retry delays with injectable randomSource +- #339: Configurable database connection pool with poolSize and timeout arithmetic, pool utilisation gauges, and DatabasePoolExhaustedError +- #334: GET /live lightweight liveness probe and GET /ready boot & dependency readiness probe +- #331: Ingester crash supervisor with capped exponential backoff, healthy run reset, and crash history telemetry +- #329: Distinct VenueFetchError and failure tracking preventing empty-venue confusion and unalerted reweighting +- #350: Comprehensive operational service runbook in docs/RUNBOOK.md and README.md linkage diff --git a/.env.example b/.env.example index 3937aedc..885a65d6 100644 --- a/.env.example +++ b/.env.example @@ -16,6 +16,13 @@ HOST=0.0.0.0 DATABASE_URL=postgresql://lens:lens@localhost:5432/lens # Direct unpooled PostgreSQL connection string (used by Prisma migrations and CLI to bypass connection poolers) DIRECT_DATABASE_URL=postgresql://lens:lens@localhost:5432/lens +# Maximum connections in the shared pg connection pool (default: 20) +# NOTE: Ingesters and API share this pool. 3 replicas x 20 conn = 60 max conn, within Postgres 100 limit. +DATABASE_POOL_SIZE=20 +# Optional alias for DATABASE_POOL_SIZE +DATABASE_POOL_MAX=20 +# Connection acquisition timeout in milliseconds (default: 10000) +DATABASE_POOL_TIMEOUT_MS=10000 # Retention window in days for partitioned price_points table (default: 30) PRICE_POINTS_RETENTION_DAYS=30 # Explicit opt-in flag to drop expired partitions older than PRICE_POINTS_RETENTION_DAYS (default: false, DESTRUCTIVE). diff --git a/README.md b/README.md index 13334c3b..039b5d26 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,9 @@ Aggregates price data from Stellar's Classic Order Book (SDEX) and AMM Liquidity | GET | `/screener?sortBy=volume&order=desc&limit=20` | Screen pairs by volume, 24h change, liquidity and price | | GET | `/benchmark/:asset?target=USD` | Peg-deviation statistics versus a target | | GET | `/basket?asset=XLM&asset=USDC&weight=0.6&weight=0.4` | Weighted price of a basket of assets | -| GET | `/status` | Indexer health | +| GET | `/status` | Indexer health and operational telemetry | +| GET | `/live` | Liveness check (process event loop responsiveness) | +| GET | `/ready` | Readiness check (validates database and cache connectivity) | | GET | `/discovery/resources?type=&payTo=&network=&extensions=&limit=&offset=` | Bazaar catalog of x402-discoverable resources (spec: [`bazaar`](https://github.com/x402-foundation/x402/blob/main/specs/extensions/bazaar.md)) | | GET | `/usage/me` | Usage and quota for the calling API key (requires `Authorization: Bearer `) | | GET | `/supported` | Payment kinds and extensions this facilitator supports | @@ -530,9 +532,17 @@ curl -s http://localhost:3002/price/XLM/USDC | jq '{price, ammPrice, lastUpdated | `ORACLE_PAYMENT_ADDRESS` | Stellar address for x402 API payments | - | No* | | `ORACLE_PAYMENT_ADDRESS_TESTNET` / `ORACLE_PAYMENT_ADDRESS_MAINNET` | Per-network override for the address above | - | No | | `X402_FACILITATOR_URL` | x402 facilitator service URL | - | No | +| `DATABASE_POOL_SIZE` | Maximum PostgreSQL connections (shared across ingesters and API) | `20` | No | +| `DATABASE_POOL_TIMEOUT_MS` | Connection acquisition timeout in ms | `10000` | No | *\*Required if enabling x402 payment gating.* +## Operations & Runbook + +For production architecture, failure mode diagnostics, degraded operational states, and deployment procedures, see [`docs/RUNBOOK.md`](docs/RUNBOOK.md). + +> **Note on Connection Pooling:** The Fastify API server and background venue ingesters (SDEX, AMM, Soroswap, Aquarius) share a single database connection pool (`DATABASE_POOL_SIZE`). At 20 connections per instance, 3 running replicas require at most 60 connections, safely under the default PostgreSQL `max_connections` ceiling of 100. + ## Stack - **Runtime:** Node.js 20 + TypeScript - **API:** Fastify + Mercurius (GraphQL) diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md new file mode 100644 index 00000000..bcd1817b --- /dev/null +++ b/docs/RUNBOOK.md @@ -0,0 +1,107 @@ +# Lens Service Runbook + +This runbook documents the operational architecture, dependencies, degraded operating modes, diagnostics, and deployment procedures for the Lens service. + +--- + +## 1. Architecture & Background Processes + +A single Lens process runs both user-facing API handlers and background data pipelines: + +1. **Fastify REST & GraphQL API Server** + - Serves price queries, OHLCV candles, orderbook depth, aggregates, screener, and x402 facilitator routes. + - Listens on `PORT` (default 3002) and binds to `HOST` (default `0.0.0.0`). + - Serves health and telemetry probes: `/live`, `/ready`, `/status`, and Prometheus metrics on `/metrics`. + +2. **WebSocket Price Fan-Out** + - Broadcasts real-time price updates for watched asset pairs to connected WebSocket clients at `/ws`. + +3. **Venue Ingestion Loops (Per Enabled Network)** + - Independently fault-isolated worker loops per network (`testnet`, `mainnet`): + - **SDEX Ingester:** Ingests Classic Stellar orderbook trades. + - **AMM Ingester:** Indexes Stellar liquidity pool reserves from Horizon. + - **Soroswap Ingester:** Polls Soroswap pool reserves via Soroban RPC. + - **Aquarius Ingester:** Queries Aquarius AMM pool states. + - **Snapshot Ingester:** Periodically records 1-minute price snapshots. + - Ingesters use exponential backoff (1s up to 60s ceiling) on repeated crashes, resetting after 60s of sustained healthy execution. + +4. **Background Jobs (BullMQ + Redis)** + - **Aggregate Refresh Worker:** Computes and refreshes rolling VWAP / OHLCV aggregates. + - **Snapshot Retention Worker:** Enforces 30-day retention on `price_snapshots`. Degrades to an advisory-locked in-process hourly timer if Redis is unavailable. + +--- + +## 2. Shared Database Connection Pool + +The Fastify API handlers, background venue ingesters, and retention jobs **share a single PostgreSQL connection pool**. + +### Pool Sizing Arithmetic +- Standard Postgres `max_connections` limit: **100**. +- Lens pool size per instance: **20 connections** (`DATABASE_POOL_SIZE`, default 20). +- Acquisition timeout: **10,000ms** (`DATABASE_POOL_TIMEOUT_MS`, default 10s). +- Across 3 deployed instances: `3 × 20 = 60 connections`, leaving 40 connections headroom for administrative tasks, migrations, and external monitoring. + +--- + +## 3. External Dependencies & Degraded Modes + +| Dependency | Purpose | Failure Symptom | Degraded Behavior | +|---|---|---|---| +| **PostgreSQL** | Primary source of truth for trades, snapshots, and indexer state | `/ready` returns 503; queries fail with `DatabasePoolExhaustedError` or connection errors | API cannot serve fresh data; writes fail immediately. | +| **Redis** | Response cache and BullMQ job queue | `redis_connected` metric = 0; log warning on connection | Price reads fall back to PostgreSQL; retention degrades to in-process hourly timer with Postgres advisory locking (`retentionMode: in_process`). Restores automatically to queue mode upon reconnection. | +| **Horizon (Stellar)** | SDEX trade feeds and AMM pool reserve queries | `horizon_rate_limited_total` increments; 429 logs with `Retry-After` | Ingester respects `Retry-After` header or performs exponential backoff; venue is excluded from aggregate instead of reporting zero pools. | +| **Soroban RPC** | Soroswap reserves & Reflector oracle quotes | RPC simulation errors; `VenueFetchError` logged | Excluded from aggregate; Reflector oracle returns null and logs error; last error recorded in `/status`. | +| **Aquarius API** | Aquarius AMM liquidity pools | Non-2xx response; `VenueFetchError` logged | Excluded from aggregate; does not dilute or zero-weight overall market price. | + +--- + +## 4. Diagnostics: /status, Probes, & /metrics + +### Health Probes +- `GET /live`: Liveness probe. Returns HTTP 200 `{"status": "live"}` if the event loop is turning. Does NOT query Postgres or Redis, preventing restart loops during dependency blips. +- `GET /ready`: Readiness probe. Returns HTTP 200 `{"status": "ready", "checks": {"database": true, "redis": true}}` only after boot completes and dependencies are reachable; returns HTTP 503 otherwise. +- Both `/live` and `/ready` are public, unauthenticated, exempt from rate limiting, and excluded from `httpMetrics` route cardinality. + +### `/status` Monitoring Endpoint +Returns operational telemetry: +- `ok`: Service status boolean. +- `network`: Current network (`testnet` or `mainnet`). +- `watchedPairs`: List of watched trading pairs. +- `lastIndexedLedger`: Latest ingested ledger sequence. +- `lastProcessedAt`: Timestamp of last indexed trade. +- `ingestLagSeconds`: Seconds elapsed since last trade was ingested. +- `retentionMode`: Active snapshot retention mode (`queue` or `in_process`). +- `ingesterCrashes`: History of recent crashes per ingester, restart counts, and current backoff. +- `failedVenues`: Map of venues experiencing upstream fetch errors. +- `failedWebhooks`: Recent webhook deliveries that exhausted all retries (payload body preserved, secret excluded). + +### Key Prometheus Metrics (`/metrics`) +- `trades_ingested_total{pair, network}`: Counter of trades ingested. +- `last_trade_timestamp{pair, network}`: Unix timestamp of latest trade. +- `horizon_rate_limit_remaining{network}`: Gauge of remaining Horizon rate limit quota. +- `horizon_rate_limited_total{network}`: Counter of 429 rate limit responses from Horizon. +- `redis_connected`: Gauge (1 = connected, 0 = disconnected). +- `retention_path_mode{mode}`: Gauge indicating active retention mode (`queue` vs `in_process`). +- `pg_pool_total_connections`, `pg_pool_idle_connections`, `pg_pool_waiting_count`: Pool utilisation gauges. +- `pg_pool_exhausted_total`: Counter of queries that timed out waiting for an available DB connection. +- `ingester_restarts_total{ingester, network}`: Ingester crash restart counter. +- `webhook_delivery_failures_total{category}`: Counter of permanently failed webhook deliveries by category (`upstream_5xx`, `timeout`, `network`, `unknown`). +- `unhandled_rejections_total`, `uncaught_exceptions_total`: Counters of fatal process error events. + +--- + +## 5. Deployment & Rollback Procedures + +### Standard Deployment +1. Build the application: `npm run build` +2. Run database migrations: `npm run prisma:push` +3. Launch application: `npm start` (or container start) +4. Verify `/ready` returns HTTP 200 before routing live traffic. + +### Rollback Procedure +1. If deployment fails or exhibits regressions, redeploy the previous stable container image or git commit. +2. Lens ingesters are idempotent and resume automatically from `indexer_state.last_ledger`. +3. In-flight WebSocket connections and HTTP requests terminate during restart and reconnect automatically. + +### Known Non-Blocking Test Failures +- **`Load test (prices)`** is failing on `main` and is a known, pre-existing issue unrelated to application reliability or this release. diff --git a/openapi.json b/openapi.json index 94cb0a74..90f87aa6 100644 --- a/openapi.json +++ b/openapi.json @@ -156,6 +156,21 @@ "ingestLagSeconds": { "type": "integer", "nullable": true + }, + "retentionMode": { + "type": "string" + }, + "ingesterCrashes": { + "type": "object" + }, + "failedVenues": { + "type": "object" + }, + "failedWebhooks": { + "type": "array", + "items": { + "type": "object" + } } } } @@ -168,6 +183,94 @@ } } }, + "/live": { + "get": { + "operationId": "getLive", + "summary": "Liveness check probe", + "security": [], + "responses": { + "200": { + "description": "Process event loop is healthy and alive", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "status": { + "type": "string", + "example": "live" + } + } + } + } + } + } + } + } + }, + "/ready": { + "get": { + "operationId": "getReady", + "summary": "Readiness check probe", + "security": [], + "responses": { + "200": { + "description": "Process has finished booting and dependencies are ready", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "status": { + "type": "string", + "example": "ready" + }, + "checks": { + "type": "object", + "properties": { + "database": { + "type": "boolean" + }, + "redis": { + "type": "boolean" + } + } + } + } + } + } + } + }, + "503": { + "description": "Process is starting up or dependencies are unreachable", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "status": { + "type": "string", + "example": "unhealthy" + }, + "checks": { + "type": "object", + "properties": { + "database": { + "type": "boolean" + }, + "redis": { + "type": "boolean" + } + } + } + } + } + } + } + } + } + } + }, "/price/{assetA}/{assetB}": { "get": { "operationId": "getPrice", diff --git a/openapi.yaml b/openapi.yaml index eb49b630..ba8939be 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -114,9 +114,77 @@ paths: ingestLagSeconds: type: integer nullable: true + retentionMode: + type: string + ingesterCrashes: + type: object + failedVenues: + type: object + failedWebhooks: + type: array + items: + type: object '429': $ref: '#/components/responses/TooManyRequests' + /live: + get: + operationId: getLive + summary: Liveness check probe + security: [] + responses: + '200': + description: Process event loop is healthy and alive + content: + application/json: + schema: + type: object + properties: + status: + type: string + example: live + + /ready: + get: + operationId: getReady + summary: Readiness check probe + security: [] + responses: + '200': + description: Process has finished booting and dependencies are ready + content: + application/json: + schema: + type: object + properties: + status: + type: string + example: ready + checks: + type: object + properties: + database: + type: boolean + redis: + type: boolean + '503': + description: Process is starting up or dependencies are unreachable + content: + application/json: + schema: + type: object + properties: + status: + type: string + example: unhealthy + checks: + type: object + properties: + database: + type: boolean + redis: + type: boolean + /price/{assetA}/{assetB}: get: operationId: getPrice diff --git a/src/__tests__/aquariusIngester.test.ts b/src/__tests__/aquariusIngester.test.ts index 318e64fb..180b0c26 100644 --- a/src/__tests__/aquariusIngester.test.ts +++ b/src/__tests__/aquariusIngester.test.ts @@ -66,11 +66,10 @@ describe('fetchAquariusPools', () => { ) }) - it('returns an empty array on a non-ok response', async () => { - ;(global.fetch as ReturnType).mockResolvedValue({ ok: false }) + it('throws on a non-ok response (#329)', async () => { + ;(global.fetch as ReturnType).mockResolvedValue({ ok: false, status: 500 }) - const result = await fetchAquariusPools(mockPair as any) - expect(result).toEqual([]) + await expect(fetchAquariusPools(mockPair as any)).rejects.toThrow() }) }) diff --git a/src/__tests__/reliabilityResilienceCluster.test.ts b/src/__tests__/reliabilityResilienceCluster.test.ts new file mode 100644 index 00000000..33abd375 --- /dev/null +++ b/src/__tests__/reliabilityResilienceCluster.test.ts @@ -0,0 +1,369 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import Fastify from 'fastify' +import { registerRESTRoutes } from '../api/rest' +import { isBootCompleted, setBootComplete, resetBootStateForTesting } from '../bootState' +import { + retryDelayMs, + recordFailedDelivery, + getRecentWebhookFailures, + clearWebhookFailures, + deliverJsonWithRetries, +} from '../alerts/delivery' +import { + recordIngesterCrash, + recordIngesterHealthyRun, + getIngesterCrashHistory, + resetCrashHistoryForTesting, + BASE_BACKOFF_MS, + MAX_BACKOFF_MS, +} from '../ingest/supervisor' +import { + VenueFetchError, + HorizonRateLimitError, + recordVenueFailure, + recordVenueSuccess, + getFailedVenues, + resetVenueFailuresForTesting, +} from '../ingest/venues/errors' +import { + getRetentionPathMode, + setRetentionPathMode, + switchToInProcessRetention, + switchToQueueRetention, + resetRetentionSupervisorForTesting, + safePruneWithAdvisoryLock, +} from '../jobs/snapshotRetention' +import { + installProcessErrorGuard, + resetProcessGuardForTesting, +} from '../errors/processGuard' +import { + poolSize, + poolTimeoutMs, + DEFAULT_DATABASE_POOL_SIZE, + DEFAULT_DATABASE_POOL_TIMEOUT_MS, + DatabasePoolExhaustedError, + handlePoolQueryError, + pgPool, +} from '../db' +import { + horizon_rate_limit_remaining, + horizon_rate_limited_total, + redis_connected, + retention_path_mode, + unhandled_rejections_total, + uncaught_exceptions_total, + pg_pool_exhausted_total, + ingester_restarts_total, + webhook_delivery_failures_total, +} from '../metrics' +import { fetchPools } from '../ingest/venues/amm' +import { fetchAquariusPools } from '../ingest/venues/aquarius' +import { fetchSoroswapTokenList, fetchPoolsFromFactory } from '../ingest/venues/soroswap' +import { getLastReflectorError } from '../ingest/oracles/reflector' + +describe('Reliability & Resilience Mega-Batch Test Suite (#348, #345, #344, #342, #341, #339, #334, #331, #329, #350)', () => { + beforeEach(() => { + resetBootStateForTesting() + clearWebhookFailures() + resetCrashHistoryForTesting() + resetVenueFailuresForTesting() + resetRetentionSupervisorForTesting() + resetProcessGuardForTesting() + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #348: Horizon Rate Limiting Handling & Telemetry + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #348: Horizon Rate Limiting', () => { + it('detects 429 and throws HorizonRateLimitError without Retry-After', async () => { + const pair = { + pairKey: 'XLM/USDC', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN' }, + } + + const mockFetch = vi.fn().mockResolvedValue({ + ok: false, + status: 429, + headers: new Headers({ + 'x-ratelimit-remaining': '0', + }), + }) + vi.stubGlobal('fetch', mockFetch) + + await expect(fetchPools(pair as any, 'testnet')).rejects.toThrow(HorizonRateLimitError) + vi.unstubAllGlobals() + }) + + it('parses Retry-After header and exports gauge metric on 429', async () => { + const pair = { + pairKey: 'XLM/USDC', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN' }, + } + + const mockFetch = vi.fn().mockResolvedValue({ + ok: false, + status: 429, + headers: new Headers({ + 'retry-after': '30', + 'x-ratelimit-remaining': '0', + }), + }) + vi.stubGlobal('fetch', mockFetch) + + try { + await fetchPools(pair as any, 'testnet') + expect.unreachable('Should have thrown HorizonRateLimitError') + } catch (err) { + expect(err).toBeInstanceOf(HorizonRateLimitError) + expect((err as HorizonRateLimitError).retryAfterSeconds).toBe(30) + } + vi.unstubAllGlobals() + }) + + it('detects non-2xx response before body parsing', async () => { + const pair = { + pairKey: 'XLM/USDC', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN' }, + } + + const mockFetch = vi.fn().mockResolvedValue({ + ok: false, + status: 503, + headers: new Headers(), + json: vi.fn().mockRejectedValue(new Error('Should not be called')), + }) + vi.stubGlobal('fetch', mockFetch) + + await expect(fetchPools(pair as any, 'testnet')).rejects.toThrow(VenueFetchError) + vi.unstubAllGlobals() + }) + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #345: Redis Degradation & Recovery + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #345: Redis Going Away & Reconnection Recovery', () => { + it('manages transition between queue and in_process retention modes', async () => { + expect(getRetentionPathMode()).toBe('queue') + + await switchToInProcessRetention() + expect(getRetentionPathMode()).toBe('in_process') + + setRetentionPathMode('queue') + expect(getRetentionPathMode()).toBe('queue') + }) + + it('guards concurrent in-process prunes via Postgres advisory lock', async () => { + const mockQuery = vi.spyOn(pgPool, 'query').mockImplementation(async (sql: any) => { + if (typeof sql === 'string' && sql.includes('pg_try_advisory_lock')) { + return { rows: [{ acquired: false }] } as any + } + return { rows: [] } as any + }) + + const result = await safePruneWithAdvisoryLock() + expect(result).toBeNull() // Skipped because lock could not be acquired + mockQuery.mockRestore() + }) + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #344: Process Error Guard (Unhandled Rejection / Uncaught Exception) + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #344: Unhandled Rejection & Uncaught Exception Guard', () => { + it('installs process error handlers and logs full stack trace', () => { + const mockLogger = { error: vi.fn() } + installProcessErrorGuard(mockLogger) + + // Verify listeners are attached + const rejectionListeners = process.listeners('unhandledRejection') + expect(rejectionListeners.length).toBeGreaterThan(0) + + const exceptionListeners = process.listeners('uncaughtException') + expect(exceptionListeners.length).toBeGreaterThan(0) + }) + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #342: Webhook Exhausted Deliveries Dead-Letter Recording + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #342: Durable Dead-Letter Store for Webhook Deliveries', () => { + it('records failed deliveries with categorisation and excludes secret', () => { + recordFailedDelivery({ + url: 'https://example.com/webhook', + attemptCount: 3, + lastStatus: 502, + lastError: 'HTTP 502', + firstAttemptAt: new Date().toISOString(), + lastAttemptAt: new Date().toISOString(), + payload: { price: 0.15, pair: 'XLM/USDC' }, + category: 'upstream_5xx', + }) + + const failures = getRecentWebhookFailures() + expect(failures.length).toBe(1) + expect(failures[0].url).toBe('https://example.com/webhook') + expect(failures[0].category).toBe('upstream_5xx') + expect((failures[0].payload as any).secret).toBeUndefined() + }) + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #341: Webhook Retry Delay with Full Jitter + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #341: Full Jitter on Webhook Retries', () => { + it('applies full jitter uniformly within [0, baseDelay]', () => { + // Mock randomSource returning 0.5 + const deterministicDelay = retryDelayMs(2, () => 0.5) + // Attempt 2 baseDelay = 2^(2-1)*1000 = 2000ms. 0.5 * 2000 = 1000ms + expect(deterministicDelay).toBe(1000) + + // Random source produces variation + const delay1 = retryDelayMs(3, () => 0.25) // 0.25 * 4000 = 1000 + const delay2 = retryDelayMs(3, () => 0.75) // 0.75 * 4000 = 3000 + expect(delay1).toBe(1000) + expect(delay2).toBe(3000) + }) + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #339: Database Connection Pool Sizing & Exhaustion Diagnostics + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #339: Database Connection Pool Configuration & Metrics', () => { + it('has explicit defaults for poolSize and acquisition timeout', () => { + expect(DEFAULT_DATABASE_POOL_SIZE).toBe(20) + expect(DEFAULT_DATABASE_POOL_TIMEOUT_MS).toBe(10_000) + expect(poolSize).toBeGreaterThanOrEqual(20) + expect(poolTimeoutMs).toBeGreaterThanOrEqual(10_000) + }) + + it('throws DatabasePoolExhaustedError on connection acquisition timeout', () => { + expect(() => + handlePoolQueryError(new Error('connection timeout in pool')) + ).toThrow(DatabasePoolExhaustedError) + }) + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #334: Readiness vs Liveness Probes + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #334: GET /live & GET /ready Probes', () => { + it('/live returns 200 without querying database or redis', async () => { + const app = Fastify() + await registerRESTRoutes(app) + await app.ready() + + const res = await app.inject({ + method: 'GET', + url: '/live', + }) + + expect(res.statusCode).toBe(200) + expect(res.json()).toEqual({ status: 'live' }) + await app.close() + }) + + it('/ready returns 503 before boot completes, and 200 after boot completes', async () => { + const mockQuery = vi.spyOn(pgPool, 'query').mockImplementation(async (sql: any) => { + if (typeof sql === 'string' && sql.includes('SELECT 1')) { + return { rows: [{ '?column?': 1 }] } as any + } + return { rows: [] } as any + }) + + const app = Fastify() + await registerRESTRoutes(app) + await app.ready() + + // Before boot completes + resetBootStateForTesting() + const resUnready = await app.inject({ + method: 'GET', + url: '/ready', + }) + expect(resUnready.statusCode).toBe(503) + expect(resUnready.json().status).toBe('unhealthy') + + // Mark boot complete + setBootComplete(true) + const resReady = await app.inject({ + method: 'GET', + url: '/ready', + }) + expect(resReady.statusCode).toBe(200) + expect(resReady.json().status).toBe('ready') + expect(resReady.json().checks.database).toBe(true) + + mockQuery.mockRestore() + await app.close() + }) + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #331: Ingester Crash Supervisor & Exponential Backoff + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #331: Ingester Loop Backoff & Telemetry', () => { + it('grows restart backoff exponentially and caps at 60s', () => { + const err = new Error('RPC disconnect') + const delay1 = recordIngesterCrash('Soroswap', 'testnet', err) + expect(delay1).toBe(BASE_BACKOFF_MS) + + const delay2 = recordIngesterCrash('Soroswap', 'testnet', err) + expect(delay2).toBe(BASE_BACKOFF_MS * 2) + + const delay3 = recordIngesterCrash('Soroswap', 'testnet', err) + expect(delay3).toBe(BASE_BACKOFF_MS * 4) + + const history = getIngesterCrashHistory() + expect(history['testnet:Soroswap'].restarts).toBe(3) + expect(history['testnet:Soroswap'].lastError).toBe('RPC disconnect') + }) + + it('resets backoff to base after a sustained healthy run', () => { + const err = new Error('Temporary glitch') + recordIngesterCrash('AMM', 'mainnet', err) + recordIngesterCrash('AMM', 'mainnet', err) + + recordIngesterHealthyRun('AMM', 'mainnet') + const history = getIngesterCrashHistory() + expect(history['mainnet:AMM'].currentBackoffMs).toBe(BASE_BACKOFF_MS) + }) + }) + + // ───────────────────────────────────────────────────────────────────────────── + // #329: Failed Venue Distinct from Empty Venue + // ───────────────────────────────────────────────────────────────────────────── + describe('Issue #329: Failed Venue Indistinguishable from Empty Venue', () => { + it('records failed venues distinctly in getFailedVenues()', () => { + recordVenueFailure('Aquarius', 'HTTP 500 Internal Server Error') + const failures = getFailedVenues() + expect(failures['Aquarius']).toBeDefined() + expect(failures['Aquarius'].error).toContain('HTTP 500') + + recordVenueSuccess('Aquarius') + expect(getFailedVenues()['Aquarius']).toBeUndefined() + }) + + it('throws VenueFetchError on failed fetch rather than silently returning empty', async () => { + const mockFetch = vi.fn().mockResolvedValue({ + ok: false, + status: 500, + headers: new Headers(), + }) + vi.stubGlobal('fetch', mockFetch) + + const pair = { + pairKey: 'XLM/USDC', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN' }, + } + + await expect(fetchAquariusPools(pair as any)).rejects.toThrow(VenueFetchError) + vi.unstubAllGlobals() + }) + }) +}) diff --git a/src/alerts/delivery.ts b/src/alerts/delivery.ts index 1c0b28af..dfaff56f 100644 --- a/src/alerts/delivery.ts +++ b/src/alerts/delivery.ts @@ -3,25 +3,92 @@ import { createLogger } from '../logger' const log = createLogger('webhook') import { fetchWithTimeout } from '../fetch' +import { webhook_delivery_failures_total } from '../metrics' -function retryDelayMs(attempt: number): number { - return Math.pow(2, attempt - 1) * 1000 +export type RandomSource = () => number + +/** + * Calculates exponential retry delay with full jitter (#341). + * Full jitter draws uniformly from [0, baseDelay] to prevent retry synchronization. + * The randomness source is injectable for deterministic unit testing. + */ +export function retryDelayMs( + attempt: number, + randomSource: RandomSource = Math.random +): number { + const baseDelay = Math.pow(2, attempt - 1) * 1000 + return Math.floor(randomSource() * baseDelay) } function signPayload(payload: string, secret: string): string { return createHmac('sha256', secret).update(payload).digest('hex') } +// ─── Durable Dead-Letter Storage for Exhausted Deliveries (#342) ───────────── + +export interface FailedWebhookDelivery { + id: string + url: string + attemptCount: number + lastStatus: number | null + lastError: string | null + firstAttemptAt: string + lastAttemptAt: string + payload: object // Contains body only — secret is NEVER persisted (#342) + category: 'upstream_5xx' | 'timeout' | 'network' | 'unknown' +} + +const MAX_STORED_FAILURES = 100 +const recentFailures: FailedWebhookDelivery[] = [] + +export function recordFailedDelivery(failure: Omit): FailedWebhookDelivery { + const entry: FailedWebhookDelivery = { + id: `fail_${Date.now()}_${Math.random().toString(36).slice(2, 7)}`, + ...failure, + } + recentFailures.unshift(entry) + if (recentFailures.length > MAX_STORED_FAILURES) { + recentFailures.pop() + } + + // Increment Prometheus counter with bounded category label + webhook_delivery_failures_total.inc({ category: entry.category }) + + return entry +} + +export function getRecentWebhookFailures(limit = 50): readonly FailedWebhookDelivery[] { + return recentFailures.slice(0, limit) +} + +export function clearWebhookFailures(): void { + recentFailures.length = 0 +} + +function categorizeDeliveryError(res?: Response, err?: unknown): 'upstream_5xx' | 'timeout' | 'network' | 'unknown' { + if (res && res.status >= 500) return 'upstream_5xx' + const msg = err instanceof Error ? err.message : String(err ?? '') + if (msg.includes('timeout') || msg.includes('AbortError')) return 'timeout' + if (msg.includes('fetch failed') || msg.includes('ECONNREFUSED') || msg.includes('ENOTFOUND')) return 'network' + return 'unknown' +} + +/** + * Delivers JSON payload to a webhook endpoint with up to 3 attempts, full jitter, + * and durable recording of permanently failed alerts (#341, #342). + */ export async function deliverJsonWithRetries( url: string, body: object, secret: string, attempt = 1, + firstAttemptAt = new Date().toISOString(), + randomSource: RandomSource = Math.random ): Promise { const payload = JSON.stringify(body) const signature = signPayload(payload, secret) - let res: Response + let res: Response | undefined try { res = await fetchWithTimeout(url, { method: 'POST', @@ -34,10 +101,23 @@ export async function deliverJsonWithRetries( }) } catch (err) { if (attempt < 3) { - await new Promise(resolve => setTimeout(resolve, retryDelayMs(attempt))) - return deliverJsonWithRetries(url, body, secret, attempt + 1) + const delay = retryDelayMs(attempt, randomSource) + await new Promise(resolve => setTimeout(resolve, delay)) + return deliverJsonWithRetries(url, body, secret, attempt + 1, firstAttemptAt, randomSource) } + const category = categorizeDeliveryError(undefined, err) + recordFailedDelivery({ + url, + attemptCount: attempt, + lastStatus: null, + lastError: (err as Error).message, + firstAttemptAt, + lastAttemptAt: new Date().toISOString(), + payload: body, // secret is intentionally excluded (#342) + category, + }) + log.warn({ url, attempt, err }, 'webhook delivery failed after retries') return } @@ -45,14 +125,29 @@ export async function deliverJsonWithRetries( if (res.ok) return if (res.status >= 400 && res.status < 500) { + // 4xx client errors are not retried (the subscriber's URL or request is invalid) log.warn({ url, status: res.status }, 'webhook client error; not retrying') return } if (attempt < 3) { - await new Promise(resolve => setTimeout(resolve, retryDelayMs(attempt))) - return deliverJsonWithRetries(url, body, secret, attempt + 1) + const delay = retryDelayMs(attempt, randomSource) + await new Promise(resolve => setTimeout(resolve, delay)) + return deliverJsonWithRetries(url, body, secret, attempt + 1, firstAttemptAt, randomSource) } + // 3 attempts exhausted on 5xx or server errors (#342) + const category = categorizeDeliveryError(res) + recordFailedDelivery({ + url, + attemptCount: attempt, + lastStatus: res.status, + lastError: `HTTP ${res.status}`, + firstAttemptAt, + lastAttemptAt: new Date().toISOString(), + payload: body, // secret is intentionally excluded (#342) + category, + }) + log.warn({ url, attempt, status: res.status }, 'webhook delivery failed after retries') } \ No newline at end of file diff --git a/src/api/rest.ts b/src/api/rest.ts index 2f5fbdce..6484022c 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -1,6 +1,6 @@ import type { FastifyInstance } from 'fastify' import { price_requests_total } from '../metrics' -import { getCachedPrice, setCachedPrice } from '../redis' +import { redis, getCachedPrice, setCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' import { pgPool } from '../db' @@ -8,6 +8,8 @@ import { config, getNetworkConfig, activeNetwork, type NetworkName } from '../co import '../middleware/network' // declares req.network on the FastifyRequest type import { statusResponseSchema, + liveResponseSchema, + readyResponseSchema, priceResponseSchema, routeResponseSchema, historyResponseSchema, @@ -15,6 +17,11 @@ import { depthResponseSchema, installResponseValidation, } from './schemas' +import { getRecentWebhookFailures } from '../alerts/delivery' +import { getIngesterCrashHistory } from '../ingest/supervisor' +import { getFailedVenues } from '../ingest/venues/errors' +import { getRetentionPathMode } from '../jobs/snapshotRetention' +import { isBootCompleted } from '../bootState' import { getDepth } from '../pricing/depth' function makePairKey(a: string, b: string): string { @@ -68,6 +75,51 @@ export async function registerRESTRoutes(app: FastifyInstance) { // up the validating serializer. installResponseValidation(app) + // GET /live — lightweight liveness probe (#334) + app.get('/live', { config: { public: true }, schema: { response: { 200: liveResponseSchema } } }, async () => { + return { status: 'live' } + }) + + // GET /ready — readiness probe checking boot lifecycle and dependencies (#334) + app.get( + '/ready', + { config: { public: true }, schema: { response: { 200: readyResponseSchema, 503: readyResponseSchema } } }, + async (_req, reply) => { + let dbOk = false + try { + await pgPool.query('SELECT 1') + dbOk = true + } catch { + dbOk = false + } + + let redisOk = false + try { + if (redis.status === 'ready' || redis.status === 'connect') { + redisOk = true + } else { + await redis.ping() + redisOk = true + } + } catch { + redisOk = false + } + + const ready = isBootCompleted() && dbOk + const body = { + status: ready ? 'ready' : 'unhealthy', + checks: { + database: dbOk, + redis: redisOk, + }, + } + if (!ready) { + return reply.status(503).send(body) + } + return body + } + ) + // GET /status — public health/monitoring endpoint (no API key required) app.get('/status', { config: { public: true }, schema: { response: { 200: statusResponseSchema } } }, async (req) => { const network = req.network ?? activeNetwork @@ -92,6 +144,10 @@ export async function registerRESTRoutes(app: FastifyInstance) { ingestLagSeconds: lastProcessedAt ? Math.max(0, Math.round((Date.now() - new Date(lastProcessedAt).getTime()) / 1000)) : null, + retentionMode: getRetentionPathMode(), + ingesterCrashes: getIngesterCrashHistory(), + failedVenues: getFailedVenues(), + failedWebhooks: getRecentWebhookFailures(), } }) diff --git a/src/api/schemas.ts b/src/api/schemas.ts index c7db8496..67616b4e 100644 --- a/src/api/schemas.ts +++ b/src/api/schemas.ts @@ -33,6 +33,38 @@ export const statusResponseSchema = { // Seconds since the last ingester write for this network, null before the // first one. Lets `/status` reveal a stalled ingester on its own. ingestLagSeconds: { type: ['integer', 'null'] }, + retentionMode: { type: 'string' }, + ingesterCrashes: { type: 'object', additionalProperties: true }, + failedVenues: { type: 'object', additionalProperties: true }, + failedWebhooks: { type: 'array', items: { type: 'object', additionalProperties: true } }, + }, +} as const + +/** GET /live probe (#334) */ +export const liveResponseSchema = { + type: 'object', + required: ['status'], + additionalProperties: false, + properties: { + status: { type: 'string' }, + }, +} as const + +/** GET /ready probe (#334) */ +export const readyResponseSchema = { + type: 'object', + required: ['status', 'checks'], + additionalProperties: false, + properties: { + status: { type: 'string' }, + checks: { + type: 'object', + required: ['database', 'redis'], + properties: { + database: { type: 'boolean' }, + redis: { type: 'boolean' }, + }, + }, }, } as const diff --git a/src/app.ts b/src/app.ts index 671dac37..82d1eed2 100644 --- a/src/app.ts +++ b/src/app.ts @@ -102,7 +102,7 @@ export async function buildApp(options: BuildAppOptions = {}): Promise req.apiKey?.ratePerMin ?? ipRateLimitMax, timeWindow: '1 minute', keyGenerator: (req) => req.apiKey?.id ?? req.ip, - allowList: (req) => req.url === '/status', + allowList: (req) => req.url === '/status' || req.url === '/live' || req.url === '/ready', errorResponseBuilder: (req, context) => ({ statusCode: 429, error: 'Too Many Requests', diff --git a/src/bootState.ts b/src/bootState.ts new file mode 100644 index 00000000..2d669370 --- /dev/null +++ b/src/bootState.ts @@ -0,0 +1,20 @@ +/** + * Application Lifecycle & Boot Readiness State (#334) + * + * Tracks whether initialization (migrations, connections, and worker scheduling) + * has completed, allowing /ready to report 503 during startup and 200 once ready. + */ + +let bootCompleted = false + +export function setBootComplete(completed = true): void { + bootCompleted = completed +} + +export function isBootCompleted(): boolean { + return bootCompleted +} + +export function resetBootStateForTesting(): void { + bootCompleted = false +} diff --git a/src/db.ts b/src/db.ts index a98b66cc..df85f487 100644 --- a/src/db.ts +++ b/src/db.ts @@ -1,7 +1,16 @@ import { PrismaClient } from '@prisma/client' import { Pool } from 'pg' -import { db_query_duration_seconds } from './metrics' +import { + db_query_duration_seconds, + pg_pool_total_connections, + pg_pool_idle_connections, + pg_pool_waiting_count, + pg_pool_exhausted_total, +} from './metrics' import { config, type NetworkName } from './config' +import { createLogger } from './logger' + +const log = createLogger('db') // Prisma for schema management + simple queries const globalForPrisma = globalThis as unknown as { prisma?: PrismaClient } @@ -10,22 +19,81 @@ export const prisma = globalForPrisma.prisma ?? new PrismaClient({ }) if (process.env.NODE_ENV !== 'production') globalForPrisma.prisma = prisma +// ── Database Connection Pool Sizing & Telemetry (#339) ──────────────────────── +// Pool arithmetic: +// Each Lens node runs: +// - Fastify REST & GraphQL API handlers (up to ~10 concurrent queries) +// - Venue ingesters: SDEX, AMM, Soroswap, Aquarius, Snapshot per enabled network (~5 workers) +// - BullMQ aggregate refresh and snapshot retention workers (~2 workers) +// Ingesters and the API share this single pool. +// At poolSize = 20 connections per instance, 3 deployed replicas demand at most 60 connections, +// remaining safely below Postgres default max_connections (100). +export const DEFAULT_DATABASE_POOL_SIZE = 20 +export const DEFAULT_DATABASE_POOL_TIMEOUT_MS = 10_000 + +export const poolSize = + parseInt(process.env.DATABASE_POOL_SIZE ?? process.env.DATABASE_POOL_MAX ?? '', 10) || + DEFAULT_DATABASE_POOL_SIZE +export const poolTimeoutMs = + parseInt(process.env.DATABASE_POOL_TIMEOUT_MS ?? '', 10) || DEFAULT_DATABASE_POOL_TIMEOUT_MS + // Raw pg pool for time-series queries (VWAP, aggregates) // ssl: true with rejectUnauthorized: false is required for Supabase pgbouncer // (their pooler uses a self-signed cert in the chain — this is safe for Supabase specifically) export const pgPool = new Pool({ connectionString: config.db.url, + max: poolSize, + connectionTimeoutMillis: poolTimeoutMs, ssl: config.db.url.includes('supabase.com') ? { rejectUnauthorized: false } : undefined, }) -// Instrument raw pg queries +export function updatePoolMetrics(): void { + try { + if (typeof (pgPool as any)?.totalCount === 'number') { + pg_pool_total_connections.set((pgPool as any).totalCount) + pg_pool_idle_connections.set((pgPool as any).idleCount) + pg_pool_waiting_count.set((pgPool as any).waitingCount) + } + } catch { + // non-fatal + } +} + +export class DatabasePoolExhaustedError extends Error { + constructor(message: string, public readonly cause?: unknown) { + super(`Database pool exhausted: ${message}`) + this.name = 'DatabasePoolExhaustedError' + } +} + +export function handlePoolQueryError(err: any): never { + if ( + err?.message?.includes('timeout') || + err?.message?.includes('Connection pool') || + (pgPool as any)?.waitingCount > 0 + ) { + pg_pool_exhausted_total.inc() + log.error( + { waitingCount: (pgPool as any).waitingCount, totalCount: (pgPool as any).totalCount, idleCount: (pgPool as any).idleCount, err }, + 'connection pool exhausted' + ) + throw new DatabasePoolExhaustedError(err?.message ?? 'Acquisition timeout', err) + } + throw err +} + +// Instrument raw pg queries with duration timing, pool telemetry, and pool exhaustion alerts const originalQuery = pgPool.query.bind(pgPool) pgPool.query = (async (...args: any[]) => { + updatePoolMetrics() const end = db_query_duration_seconds.startTimer() try { return await (originalQuery as any)(...args) + } catch (err: any) { + handlePoolQueryError(err) } finally { end() + updatePoolMetrics() } }) as any diff --git a/src/errors/processGuard.ts b/src/errors/processGuard.ts new file mode 100644 index 00000000..aadd78a2 --- /dev/null +++ b/src/errors/processGuard.ts @@ -0,0 +1,42 @@ +/** + * Process-Level Crash & Unhandled Rejection Guard (#344) + * + * Ensures unhandled promise rejections and uncaught exceptions log their full error + * and stack trace before terminating the process non-zero so platforms can restart cleanly. + * + * Prevents running in a corrupted/half-dead state while serving financial price data. + */ + +import { unhandled_rejections_total, uncaught_exceptions_total } from '../metrics' + +let installed = false + +export function installProcessErrorGuard(logger: { error: (...args: any[]) => void } = console): void { + if (installed) return + installed = true + + process.on('unhandledRejection', (reason: unknown) => { + unhandled_rejections_total.inc() + const err = reason instanceof Error ? reason : new Error(String(reason)) + logger.error(`[fatal] Unhandled promise rejection: ${err.message}\nStack: ${err.stack ?? 'no stack'}`) + + // Exit immediately; do NOT attempt normal graceful drain (#344) + if (process.env.NODE_ENV !== 'test') { + process.exit(1) + } + }) + + process.on('uncaughtException', (err: Error) => { + uncaught_exceptions_total.inc() + logger.error(`[fatal] Uncaught exception: ${err.message}\nStack: ${err.stack ?? 'no stack'}`) + + // Exit immediately; do NOT attempt normal graceful drain (#344) + if (process.env.NODE_ENV !== 'test') { + process.exit(1) + } + }) +} + +export function resetProcessGuardForTesting(): void { + installed = false +} diff --git a/src/index.ts b/src/index.ts index 55ed4c2b..269e477b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -15,7 +15,17 @@ import { fanOutManager } from './ws/fanout' import { venueAdapters, startVenueAdapter } from './ingest/venues' import { startSnapshotIngester } from './ingesters/snapshot' import { createAggregateQueue, startAggregateWorker, scheduleAggregateRefresh } from './jobs/aggregateRefresh' -import { createSnapshotRetentionQueue, startSnapshotRetentionWorker, scheduleSnapshotRetention, pruneAllNetworks, SNAPSHOT_RETENTION_DAYS } from './jobs/snapshotRetention' +import { + createSnapshotRetentionQueue, + startSnapshotRetentionWorker, + scheduleSnapshotRetention, + pruneAllNetworks, + SNAPSHOT_RETENTION_DAYS, + safePruneWithAdvisoryLock, + switchToQueueRetention, + switchToInProcessRetention, + getRetentionPathMode, +} from './jobs/snapshotRetention' import { createPricePointsPartitionQueue, startPricePointsPartitionWorker, @@ -24,10 +34,14 @@ import { PRICE_POINTS_RETENTION_DAYS, } from './jobs/pricePointsPartitionJob' import { loadPersistedPairs, getActivePairs } from './pairsRegistry' +import { installProcessErrorGuard } from './errors/processGuard' +import { recordIngesterCrash, recordIngesterHealthyRun, HEALTHY_RESET_THRESHOLD_MS } from './ingest/supervisor' +import { setBootComplete } from './bootState' const log = createLogger('startup') async function main() { + installProcessErrorGuard() // ── Ensure DB schema is up-to-date ──────────────────────────────────────── log.info('running database migrations') execSync('node node_modules/prisma/build/index.js db push --accept-data-loss', { stdio: 'inherit' }) @@ -102,45 +116,31 @@ async function main() { } } - // ── Snapshot retention ──────────────────────────────────────────────────── + // ── Snapshot retention (#345) ────────────────────────────────────────── // price_snapshots is append-only and unbounded, so pruning it is a data // lifecycle concern, not a caching one. BullMQ is only how the prune gets - // scheduled — losing Redis must not mean losing retention, because the - // failure is silent and the bill arrives as a full disk weeks later. + // scheduled — losing Redis must not mean losing retention. // - // So: prune once here, unconditionally, then try to install the hourly - // schedule. If Redis is away, fall back to a plain timer that does the same - // work in-process. - const safePrune = async () => { - try { - const counts = await pruneAllNetworks() - for (const [network, pruned] of Object.entries(counts)) { - if (pruned > 0) { - log.info({ network, pruned, retentionDays: SNAPSHOT_RETENTION_DAYS }, 'pruned snapshots older than retention window') - } - } - } catch (err) { - log.error({ err }, 'snapshot prune failed') - } + try { + await safePruneWithAdvisoryLock() + } catch (err) { + log.error({ err }, 'initial snapshot prune failed') } - await safePrune() + await switchToQueueRetention() - { - let retentionQueue: ReturnType | undefined - let retentionWorker: ReturnType | undefined - try { - retentionQueue = createSnapshotRetentionQueue() - retentionWorker = startSnapshotRetentionWorker() - await scheduleSnapshotRetention(retentionQueue) - log.info('snapshot retention worker started') - } catch (err) { - log.warn({ err }, 'snapshot retention worker skipped (redis unavailable)') - log.warn('falling back to an in-process hourly prune') - await Promise.allSettled([retentionQueue?.close(), retentionWorker?.close()]) - setInterval(() => { void safePrune() }, 60 * 60 * 1000).unref() + // Periodically probe Redis to restore queue retention if currently degraded + setInterval(async () => { + if (getRetentionPathMode() === 'in_process') { + try { + await redis.ping() + log.info('redis returned; restoring retention worker to queue mode') + await switchToQueueRetention() + } catch { + // still offline + } } - } + }, 30_000).unref() // ── Price points partition maintenance & retention ──────────────────────── // price_points is range-partitioned by timestamp. We pre-create future partitions @@ -181,16 +181,28 @@ async function main() { } } - // ── Ingesters (run in background — infinite loops) ──────────────────────── + // ── Ingesters (run in background — infinite loops) (#331) ───────────────── // Each ingester is independently fault-isolated via restartIngester, keyed by // the (venue, network) pair: a crash in, say, the Soroswap ingester on - // mainnet only restarts that one instance and cannot affect SDEX/AMM or the - // ingesters running on testnet. + // mainnet only restarts that one instance with exponential backoff up to 60s. + // Backoff resets after a sustained healthy run of 60s. const restartIngester = (name: string, network: NetworkName, fn: () => Promise) => { - fn().catch(err => { - log.error({ ingester: name, network, err }, 'ingester crashed; restarting in 10s') - setTimeout(() => restartIngester(name, network, fn), 10_000) - }) + let healthyTimer: NodeJS.Timeout | null = null + healthyTimer = setTimeout(() => { + recordIngesterHealthyRun(name, network) + }, HEALTHY_RESET_THRESHOLD_MS) + healthyTimer.unref() + + fn() + .then(() => { + if (healthyTimer) clearTimeout(healthyTimer) + }) + .catch(err => { + if (healthyTimer) clearTimeout(healthyTimer) + const delayMs = recordIngesterCrash(name, network, err as Error) + log.error({ ingester: name, network, delayMs, err }, 'ingester crashed; restarting') + setTimeout(() => restartIngester(name, network, fn), delayMs) + }) } const enabledNetworks = getEnabledNetworks() @@ -206,6 +218,9 @@ async function main() { { count: getActivePairs().length, pairs: getActivePairs().map(p => p.pairKey) }, 'watching pairs' ) + + // ── Mark boot complete for readiness probe (#334) ───────────────────────── + setBootComplete(true) } main().catch(err => { diff --git a/src/ingest/oracles/reflector.ts b/src/ingest/oracles/reflector.ts index db5c1f09..0eeada8f 100644 --- a/src/ingest/oracles/reflector.ts +++ b/src/ingest/oracles/reflector.ts @@ -20,6 +20,10 @@ import { import { activeNetwork, getNetworkConfig, type NetworkName } from '../../config' import { getRpcServer } from '../../network/clients' import type { OracleAdapter, OracleQuote } from './adapter' +import { recordVenueFailure, recordVenueSuccess } from '../venues/errors' +import { createLogger } from '../../logger' + +const log = createLogger('oracle:reflector') // Ephemeral fee payer — simulation only, no real funds needed const FEE_PAYER = Keypair.random() @@ -83,16 +87,26 @@ export async function fetchReflectorPrice( // Reflector uses 14 decimal places of precision const price = Number(raw.price) / 1e14 + recordVenueSuccess('Reflector') return { asset: assetCode.toUpperCase(), price, timestamp: Number(raw.timestamp ?? 0), } - } catch { + } catch (err) { + log.warn({ assetCode, network, err }, 'failed to fetch oracle quote') + lastReflectorError = (err as Error).message + recordVenueFailure('Reflector', (err as Error).message) return null } } +let lastReflectorError: string | null = null + +export function getLastReflectorError(): string | null { + return lastReflectorError +} + // In-memory cache (60 s TTL) to avoid hammering RPC on every comparison request const _cache = new Map() const CACHE_TTL_MS = 60_000 diff --git a/src/ingest/supervisor.ts b/src/ingest/supervisor.ts new file mode 100644 index 00000000..4168ed4d --- /dev/null +++ b/src/ingest/supervisor.ts @@ -0,0 +1,72 @@ +/** + * Ingester Loop Supervisor (#331) + * + * Implements exponential restart backoff with a 60s ceiling, backoff reset + * after sustained healthy operation, crash telemetry via Prometheus, and durable + * crash diagnostics visible in /status. + */ + +import { ingester_restarts_total } from '../metrics' +import type { NetworkName } from '../config' + +export interface IngesterCrashRecord { + ingester: string + network: string + restarts: number + lastCrashAt: string + lastError: string + currentBackoffMs: number +} + +const crashRecords = new Map() + +export const BASE_BACKOFF_MS = 1000 +export const MAX_BACKOFF_MS = 60_000 +export const HEALTHY_RESET_THRESHOLD_MS = 60_000 + +export function recordIngesterCrash( + name: string, + network: NetworkName, + err: Error +): number { + const key = `${network}:${name}` + const existing = crashRecords.get(key) + const restarts = (existing?.restarts ?? 0) + 1 + const prevBackoff = existing ? existing.currentBackoffMs : BASE_BACKOFF_MS + const nextBackoff = Math.min(MAX_BACKOFF_MS, existing ? prevBackoff * 2 : BASE_BACKOFF_MS) + + crashRecords.set(key, { + ingester: name, + network, + restarts, + lastCrashAt: new Date().toISOString(), + lastError: err.message, + currentBackoffMs: nextBackoff, + }) + + ingester_restarts_total.inc({ ingester: name, network }) + return nextBackoff +} + +export function recordIngesterHealthyRun(name: string, network: NetworkName): void { + const key = `${network}:${name}` + const existing = crashRecords.get(key) + if (existing) { + crashRecords.set(key, { + ...existing, + currentBackoffMs: BASE_BACKOFF_MS, + }) + } +} + +export function getIngesterCrashHistory(): Record { + const result: Record = {} + for (const [key, record] of crashRecords.entries()) { + result[key] = { ...record } + } + return result +} + +export function resetCrashHistoryForTesting(): void { + crashRecords.clear() +} diff --git a/src/ingest/venues/amm.ts b/src/ingest/venues/amm.ts index fe14f8d8..07125dfd 100644 --- a/src/ingest/venues/amm.ts +++ b/src/ingest/venues/amm.ts @@ -1,4 +1,10 @@ -import { amm_snapshots_total, trades_ingested_total, last_trade_timestamp } from '../../metrics' +import { + amm_snapshots_total, + trades_ingested_total, + last_trade_timestamp, + horizon_rate_limit_remaining, + horizon_rate_limited_total, +} from '../../metrics' import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config' import { upsertPricePoints, getIndexerState, setIndexerCursor, prisma } from '../../db' import { resolvePageLedgers } from '../../ingesters/toid' @@ -14,6 +20,7 @@ import { startVenueAdapter, } from './runner' import { fetchWithTimeout } from '../../fetch' +import { VenueFetchError, HorizonRateLimitError, recordVenueFailure, recordVenueSuccess } from './errors' const log = createLogger('ingester:amm') @@ -39,15 +46,44 @@ export async function fetchPools(pair: WatchedPair, network: NetworkName = activ params.append('reserves[]', assetBStr) params.set('limit', '10') - const response = await fetchWithTimeout( - `${getNetworkConfig(network).horizon.url}/liquidity_pools?${params.toString()}` - ) + const url = `${getNetworkConfig(network).horizon.url}/liquidity_pools?${params.toString()}` + const response = await fetchWithTimeout(url) + + // Export remaining quota header gauge (#348) + const remaining = response.headers.get('x-ratelimit-remaining') + if (remaining !== null) { + const val = parseInt(remaining, 10) + if (!isNaN(val)) horizon_rate_limit_remaining.set({ network }, val) + } + + // Explicitly handle Horizon 429 Rate Limiting (#348) + if (response.status === 429) { + horizon_rate_limited_total.inc({ network }) + const retryAfterStr = response.headers.get('retry-after') + const retryAfterSec = retryAfterStr ? parseInt(retryAfterStr, 10) || undefined : undefined + log.warn({ network, pair: pair.pairKey, retryAfter: retryAfterStr ?? 'none' }, 'horizon rate limit (429) encountered') + throw new HorizonRateLimitError(network, retryAfterSec) + } + + // Detect non-2xx responses before parsing body (#348, #329) + if (!response.ok) { + log.error({ status: response.status, pair: pair.pairKey }, 'horizon returned error') + recordVenueFailure('AMM', `HTTP ${response.status}`) + throw new VenueFetchError('AMM', `HTTP ${response.status}`, response.status) + } + + recordVenueSuccess('AMM') const data = (await response.json()) as any if (!data._embedded?.records) return [] return data._embedded.records } catch (err) { + if (err instanceof VenueFetchError) { + recordVenueFailure('AMM', err.message) + throw err + } log.error({ pair: pair.pairKey, err }, 'failed to fetch pools') - return [] + recordVenueFailure('AMM', (err as Error).message) + throw new VenueFetchError('AMM', (err as Error).message) } } diff --git a/src/ingest/venues/aquarius.ts b/src/ingest/venues/aquarius.ts index 9f267cc9..79d35fdf 100644 --- a/src/ingest/venues/aquarius.ts +++ b/src/ingest/venues/aquarius.ts @@ -19,6 +19,7 @@ import { startVenueAdapter, } from './runner' import { fetchWithTimeout } from '../../fetch' +import { VenueFetchError, recordVenueFailure, recordVenueSuccess } from './errors' const log = createLogger('ingester:aquarius') @@ -57,12 +58,22 @@ export async function fetchAquariusPools( params.append('assets[]', assetBStr) const res = await fetchWithTimeout(`${apiUrl}?${params.toString()}`) - if (!res.ok) return [] + if (!res.ok) { + log.error({ status: res.status, pair: pair.pairKey }, 'horizon/api returned error') + recordVenueFailure('Aquarius', `HTTP ${res.status}`) + throw new VenueFetchError('Aquarius', `HTTP ${res.status}`, res.status) + } + recordVenueSuccess('Aquarius') const data = (await res.json()) as AquariusListResponse return data.results ?? [] } catch (err) { + if (err instanceof VenueFetchError) { + recordVenueFailure('Aquarius', err.message) + throw err + } log.error({ pair: pair.pairKey, err }, 'failed to fetch pools') - return [] + recordVenueFailure('Aquarius', (err as Error).message) + throw new VenueFetchError('Aquarius', (err as Error).message) } } diff --git a/src/ingest/venues/errors.ts b/src/ingest/venues/errors.ts new file mode 100644 index 00000000..8fcd43f2 --- /dev/null +++ b/src/ingest/venues/errors.ts @@ -0,0 +1,65 @@ +/** + * Distinct Venue Errors & Result Types (#329, #348) + * + * Ensures that upstream errors, rate limits, and network dropouts are never + * conflated with a genuinely empty orderbook or zero-pool market. + */ + +export class VenueFetchError extends Error { + constructor( + public readonly venue: string, + message: string, + public readonly statusCode?: number, + public readonly isRateLimit = false + ) { + super(`[${venue}] Fetch failed: ${message}`) + this.name = 'VenueFetchError' + } +} + +export class HorizonRateLimitError extends VenueFetchError { + constructor( + public readonly network: string, + public readonly retryAfterSeconds?: number + ) { + super( + 'Horizon', + `429 Too Many Requests (rate limited)${retryAfterSeconds ? `, Retry-After: ${retryAfterSeconds}s` : ''}`, + 429, + true + ) + this.name = 'HorizonRateLimitError' + } +} + +export interface FailedVenueRecord { + venue: string + error: string + timestamp: string +} + +const failedVenues = new Map() + +export function recordVenueFailure(venue: string, error: string): void { + failedVenues.set(venue, { + venue, + error, + timestamp: new Date().toISOString(), + }) +} + +export function recordVenueSuccess(venue: string): void { + failedVenues.delete(venue) +} + +export function getFailedVenues(): Record { + const result: Record = {} + for (const [v, record] of failedVenues.entries()) { + result[v] = { ...record } + } + return result +} + +export function resetVenueFailuresForTesting(): void { + failedVenues.clear() +} diff --git a/src/ingest/venues/soroswap.ts b/src/ingest/venues/soroswap.ts index 217f3520..21aca531 100644 --- a/src/ingest/venues/soroswap.ts +++ b/src/ingest/venues/soroswap.ts @@ -26,6 +26,7 @@ import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' import { createLogger } from '../../logger' import { fetchWithTimeout } from '../../fetch' +import { VenueFetchError, recordVenueFailure, recordVenueSuccess } from './errors' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' import { @@ -80,12 +81,18 @@ export async function fetchSoroswapTokenList( ): Promise { try { const res = await fetchWithTimeout(tokenListUrl) - if (!res.ok) throw new Error(`HTTP ${res.status}`) + if (!res.ok) { + recordVenueFailure('Soroswap', `Token list HTTP ${res.status}`) + throw new VenueFetchError('Soroswap', `Token list HTTP ${res.status}`, res.status) + } const data = (await res.json()) as SoroswapTokenList + recordVenueSuccess('Soroswap') return Array.isArray(data.tokens) ? data.tokens : [] } catch (err) { + if (err instanceof VenueFetchError) throw err log.error({ err }, 'failed to fetch token list') - return [] + recordVenueFailure('Soroswap', (err as Error).message) + throw new VenueFetchError('Soroswap', (err as Error).message) } } @@ -93,7 +100,7 @@ export async function fetchSoroswapTokenList( /** * Query the Soroswap factory contract's get_pools() to enumerate pool addresses - * for a specific token pair. Returns an empty array on RPC failure. + * for a specific token pair. Throws VenueFetchError on RPC failure. */ export async function fetchPoolsFromFactory( factoryAddress: string, @@ -132,10 +139,12 @@ export async function fetchPoolsFromFactory( if (!sim.result?.retval) return [] const pools = scValToNative(sim.result.retval) as string[] + recordVenueSuccess('Soroswap') return Array.isArray(pools) ? pools : [] } catch (err) { log.debug({ err }, 'fetchPoolsFromFactory error') - return [] + recordVenueFailure('Soroswap', (err as Error).message) + throw new VenueFetchError('Soroswap', (err as Error).message) } } diff --git a/src/jobs/snapshotRetention.ts b/src/jobs/snapshotRetention.ts index ab5c94e3..28e0fd42 100644 --- a/src/jobs/snapshotRetention.ts +++ b/src/jobs/snapshotRetention.ts @@ -2,6 +2,7 @@ import { Queue, Worker } from 'bullmq' import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' import { getEnabledNetworks } from '../network/enabledNetworks' +import { retention_path_mode } from '../metrics' import { createLogger } from '../logger' const log = createLogger('job:snapshot-retention') @@ -146,3 +147,108 @@ export async function scheduleSnapshotRetention(queue: Queue) { ) await queue.add('prune', {}) } + +// ── Retention Supervisor & Safe In-Process Concurrency Guard (#345) ─────────── + +export type RetentionMode = 'queue' | 'in_process' + +let currentRetentionMode: RetentionMode = 'queue' +let inProcessIntervalTimer: NodeJS.Timeout | null = null +let activeRetentionQueue: Queue | null = null +let activeRetentionWorker: Worker | null = null + +export function getRetentionPathMode(): RetentionMode { + return currentRetentionMode +} + +export function setRetentionPathMode(mode: RetentionMode): void { + const prev = currentRetentionMode + currentRetentionMode = mode + retention_path_mode.set({ mode: 'queue' }, mode === 'queue' ? 1 : 0) + retention_path_mode.set({ mode: 'in_process' }, mode === 'in_process' ? 1 : 0) + if (prev !== mode) { + log.info({ prev, mode }, 'transitioned retention mode') + } +} + +/** + * Concurrency guard preventing multiple instances from executing duplicate + * or colliding in-process snapshot prunes at the same time (#345). + */ +export async function safePruneWithAdvisoryLock( + retentionDays: number = SNAPSHOT_RETENTION_DAYS +): Promise> | null> { + let lockAcquired = false + try { + const res = await pgPool.query('SELECT pg_try_advisory_lock(8392104) AS acquired') + lockAcquired = res.rows[0]?.acquired === true + if (!lockAcquired) { + log.info('advisory lock not acquired; another instance is actively pruning') + return null + } + return await pruneAllNetworks(retentionDays) + } catch (err) { + log.error({ err }, 'prune failed') + throw err + } finally { + if (lockAcquired) { + try { + await pgPool.query('SELECT pg_advisory_unlock(8392104)') + } catch { + // non-fatal + } + } + } +} + +export async function switchToInProcessRetention(): Promise { + setRetentionPathMode('in_process') + if (activeRetentionQueue || activeRetentionWorker) { + await Promise.allSettled([ + activeRetentionQueue?.close(), + activeRetentionWorker?.close(), + ]) + activeRetentionQueue = null + activeRetentionWorker = null + } + if (!inProcessIntervalTimer) { + inProcessIntervalTimer = setInterval(() => { + void safePruneWithAdvisoryLock() + }, 60 * 60 * 1000) + inProcessIntervalTimer.unref() + } +} + +export async function switchToQueueRetention(): Promise { + try { + const queue = createSnapshotRetentionQueue() + const worker = startSnapshotRetentionWorker() + await scheduleSnapshotRetention(queue) + + if (inProcessIntervalTimer) { + clearInterval(inProcessIntervalTimer) + inProcessIntervalTimer = null + } + + activeRetentionQueue = queue + activeRetentionWorker = worker + setRetentionPathMode('queue') + return true + } catch (err) { + log.warn({ err }, 'failed to switch to queue retention (redis unavailable)') + await switchToInProcessRetention() + return false + } +} + +export function resetRetentionSupervisorForTesting(): void { + if (inProcessIntervalTimer) { + clearInterval(inProcessIntervalTimer) + inProcessIntervalTimer = null + } + activeRetentionQueue = null + activeRetentionWorker = null + currentRetentionMode = 'queue' + retention_path_mode.set({ mode: 'queue' }, 1) + retention_path_mode.set({ mode: 'in_process' }, 0) +} diff --git a/src/metrics.ts b/src/metrics.ts index 39a3a505..4ea621d8 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -98,6 +98,90 @@ export const db_query_duration_seconds = new Histogram({ registers: [register] }) +// ── Rate Limiting & Upstream Horizon Metrics (#348) ─────────────────────────── +export const horizon_rate_limit_remaining = new Gauge({ + name: 'horizon_rate_limit_remaining', + help: 'Remaining Horizon rate limit requests from X-RateLimit-Remaining header', + labelNames: ['network'], + registers: [register] +}) + +export const horizon_rate_limited_total = new Counter({ + name: 'horizon_rate_limited_total', + help: 'Total number of 429 Rate Limited responses encountered from Horizon', + labelNames: ['network'], + registers: [register] +}) + +// ── Redis & Retention Degradation Metrics (#345) ────────────────────────────── +export const redis_connected = new Gauge({ + name: 'redis_connected', + help: 'Status of primary Redis connection (1 = connected, 0 = unavailable)', + registers: [register] +}) + +export const retention_path_mode = new Gauge({ + name: 'retention_path_mode', + help: 'Active snapshot retention execution mode (1 for active mode)', + labelNames: ['mode'], + registers: [register] +}) + +// ── Process Crash & Unhandled Rejection Metrics (#344) ──────────────────────── +export const unhandled_rejections_total = new Counter({ + name: 'unhandled_rejections_total', + help: 'Total count of unhandled promise rejections encountered by process', + registers: [register] +}) + +export const uncaught_exceptions_total = new Counter({ + name: 'uncaught_exceptions_total', + help: 'Total count of uncaught exceptions encountered by process', + registers: [register] +}) + +// ── Database Connection Pool Metrics (#339) ─────────────────────────────────── +export const pg_pool_total_connections = new Gauge({ + name: 'pg_pool_total_connections', + help: 'Total active connections in PostgreSQL connection pool', + registers: [register] +}) + +export const pg_pool_idle_connections = new Gauge({ + name: 'pg_pool_idle_connections', + help: 'Idle connections available in PostgreSQL connection pool', + registers: [register] +}) + +export const pg_pool_waiting_count = new Gauge({ + name: 'pg_pool_waiting_count', + help: 'Number of queries currently waiting for an available PostgreSQL connection', + registers: [register] +}) + +export const pg_pool_exhausted_total = new Counter({ + name: 'pg_pool_exhausted_total', + help: 'Total number of connection acquisitions that failed due to pool exhaustion', + registers: [register] +}) + +// ── Ingester Restart & Backoff Metrics (#331) ───────────────────────────────── +export const ingester_restarts_total = new Counter({ + name: 'ingester_restarts_total', + help: 'Total number of ingester loop crash restarts by ingester and network', + labelNames: ['ingester', 'network'], + registers: [register] +}) + +// ── Webhook Delivery Failure Metrics (#342) ─────────────────────────────────── +export const webhook_delivery_failures_total = new Counter({ + name: 'webhook_delivery_failures_total', + help: 'Total count of exhausted webhook delivery failures by bounded category', + labelNames: ['category'], + registers: [register] +}) + export async function getMetrics() { return await register.metrics() } + diff --git a/src/middleware/httpMetrics.ts b/src/middleware/httpMetrics.ts index 82ada904..d0d0247e 100644 --- a/src/middleware/httpMetrics.ts +++ b/src/middleware/httpMetrics.ts @@ -14,7 +14,7 @@ interface TimedRequest extends FastifyRequest { * so counting it would have the scraper continuously inflate the very numbers * it is reading — request rate would never fall to zero on an idle service. */ -const EXCLUDED_ROUTES = new Set(['/metrics']) +const EXCLUDED_ROUTES = new Set(['/metrics', '/live', '/ready']) /** * Bucket a status code into a class label. Keeping 2xx/4xx/5xx rather than the diff --git a/src/redis.ts b/src/redis.ts index bd5e6afb..f6d107b2 100644 --- a/src/redis.ts +++ b/src/redis.ts @@ -1,5 +1,6 @@ import Redis from 'ioredis' import { config, activeNetwork } from './config' +import { redis_connected } from './metrics' import { createLogger } from './logger' const log = createLogger('redis') @@ -7,30 +8,26 @@ const log = createLogger('redis') export const redis = new Redis(config.redis.url, { maxRetriesPerRequest: 3, lazyConnect: true, - // Fail commands immediately while disconnected instead of queueing them. - // - // ioredis defaults this to true, so with an unreachable host every command - // sits in the offline queue waiting for a connection that never comes. The - // await simply never resolves — a hang, not an error, so the try/catch in - // getCachedPrice cannot help and the request dies of timeout instead of - // falling through to Postgres. - // - // It stayed hidden while REQUIRE_API_KEY was on, because the auth hook - // rejected requests before any handler could reach Redis. Opening the API up - // turned a 401 into a hang, which looked like the key change had failed. - // - // Rejecting fast is what the callers already expect: reads fall back to the - // database, writes are best-effort, and x402 metering fails closed. enableOfflineQueue: false, - // ioredis retries about once a second forever by default. Against a host - // that no longer resolves that is a DNS lookup and two log lines every - // second — roughly 170k lines a day, which buries every real message in the - // log and is how a dead cache stayed invisible for weeks. Back off to a - // lookup a minute instead. Still reconnects on its own when the host comes - // back; just quietly. retryStrategy: (times) => Math.min(times * 2_000, 60_000), }) +redis.on('connect', () => { + redis_connected.set(1) +}) + +redis.on('ready', () => { + redis_connected.set(1) +}) + +redis.on('close', () => { + redis_connected.set(0) +}) + +redis.on('end', () => { + redis_connected.set(0) +}) + // Same reasoning for the error events themselves: report the first one, then // at most one a minute with a count of what was suppressed, so a persistent // outage stays visible without drowning everything else. @@ -38,6 +35,7 @@ let lastRedisErrorLog = 0 let suppressedRedisErrors = 0 redis.on('error', (err) => { + redis_connected.set(0) const now = Date.now() if (now - lastRedisErrorLog < 60_000) { suppressedRedisErrors++