From b5c33f271c648b9359894cf6316aa382c9f0f457 Mon Sep 17 00:00:00 2001 From: CNduka001 Date: Fri, 16 Oct 2026 02:36:36 +0100 Subject: [PATCH] feat: per-venue heartbeat with stale-price exclusion (#219) Track last successful read, last successful write, and consecutive read failures per (venue, network). Prices whose venue has not written within its staleness bound are excluded from aggregation, and the exclusion is surfaced on /price (excludedVenues) and /status (venueHeartbeats) plus three Prometheus gauges on /metrics. Also apply the exclusion in the refresh worker and GraphQL getPrice so cached and precomputed aggregates cannot blend a dead venue's last price. --- clients/typescript/src/schema.d.ts | 14 + docs/ingest-metrics.md | 34 +- openapi.json | 42 ++ openapi.yaml | 35 ++ schema.graphql | 5 + src/__tests__/aggregateRefreshNetwork.test.ts | 22 +- src/__tests__/price.test.ts | 1 + src/__tests__/schemaValidation.test.ts | 1 + src/aggregator/vwap.ts | 121 +++-- src/api/graphql.ts | 5 +- src/api/rest.ts | 15 +- src/api/schemas.ts | 21 + src/ingest/venues/amm.ts | 14 +- src/ingest/venues/aquarius.ts | 37 +- src/ingest/venues/sdex.ts | 22 +- src/ingest/venues/soroswap.ts | 5 + src/jobs/aggregateRefresh.ts | 16 +- src/metrics.ts | 27 + src/venueHeartbeat.ts | 226 ++++++++ tests/venueHeartbeat.test.ts | 495 ++++++++++++++++++ 20 files changed, 1090 insertions(+), 68 deletions(-) create mode 100644 src/venueHeartbeat.ts create mode 100644 tests/venueHeartbeat.test.ts diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 6589e81e..d5a5715d 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -1064,6 +1064,8 @@ export interface components { /** @enum {string} */ confidence?: "high" | "medium" | "low" | "unknown"; lastTradeAgeSeconds?: number | null; + /** @description Venues whose last written price is older than their staleness bound. Their prices are excluded from this aggregate; the list makes that exclusion visible rather than silent. Empty when every venue is live. */ + excludedVenues?: string[]; }; RouteHop: { assetIn: components["schemas"]["Asset"]; @@ -1896,6 +1898,18 @@ export interface operations { ingestLagSeconds?: number | null; /** @description Whether this deployment has an ADMIN_TOKEN configured, so an operator can tell a rejected token from a disabled admin surface. Reports only that a token is set, never any part of its value. Admin routes answer 503 admin_not_configured when this is false. */ adminConfigured?: boolean; + /** @description Per-venue heartbeat: last successful read, last successful write to price_points, and the count of consecutive failed reads. A venue whose lastWrite is older than its stalenessThresholdMs is excluded from price aggregation. */ + venueHeartbeats?: { + venue?: string; + /** @enum {string} */ + network?: "testnet" | "mainnet"; + /** Format: date-time */ + lastSuccess?: string | null; + /** Format: date-time */ + lastWrite?: string | null; + consecutiveFailures?: number; + stalenessThresholdMs?: number; + }[]; }; }; }; diff --git a/docs/ingest-metrics.md b/docs/ingest-metrics.md index 50fdbe84..dde36cc1 100644 --- a/docs/ingest-metrics.md +++ b/docs/ingest-metrics.md @@ -16,17 +16,41 @@ nonsense on any deployment with more than one.** | `price_snapshots_total` | Counter | `network` | | `ingester_abandoned_cycles_total` | Counter | `venue`, `network` | | `last_trade_timestamp` | Gauge | `pair`, `network` | +| `venue_last_success_timestamp` | Gauge | `venue`, `network` | +| `venue_last_write_timestamp` | Gauge | `venue`, `network` | +| `venue_consecutive_failures` | Gauge | `venue`, `network` | | Label | Values | Notes | | --- | --- | --- | | `network` | `testnet`, `mainnet` | The `NetworkName` of the ingester loop that produced the sample, not the process-wide `STELLAR_NETWORK`. | | `pair` | `XLM/USDC`, … | The `pairKey` already used everywhere else in Lens. | | `pool` | Horizon liquidity-pool id | Pre-existing label, only on `amm_snapshots_total`. | -| `venue` | `SDEX`, `AMM`, `Soroswap`, `Aquarius`, `Snapshot` | The ingester loop that abandoned a cycle past its deadline (#351). Bounded by the adapter registry, not by runtime data. | +| `venue` | `SDEX`, `AMM`, `Soroswap`, `Aquarius`, `Snapshot` | On `ingester_abandoned_cycles_total`: the ingester loop that abandoned a cycle past its deadline (#351). Bounded by the adapter registry, not by runtime data. | +| `venue` | `SDEX`, `AMM`, `aquarius_amm`, `soroswap_amm` | On the `venue_*` heartbeat gauges: the `source` a venue adapter writes to `price_points`. Matches the value used to exclude a stale venue from aggregation. | -All five are also emitted with the registry-wide `app="lens"` default label. +All of these are also emitted with the registry-wide `app="lens"` default label. -## Why `network` is on all five +## Per-venue heartbeat gauges (Issue #219) + +The three `venue_*` gauges back the per-venue heartbeat registry in +[`src/venueHeartbeat.ts`](../src/venueHeartbeat.ts). A dead adapter and a quiet +venue both stop producing rows; the heartbeat tells them apart: + +- `venue_last_success_timestamp` — the last successful upstream **read**. Moves on + every poll that reaches the venue, even when it returns nothing new. +- `venue_last_write_timestamp` — the last successful **write** to `price_points`. + This is the price's age; a venue whose `last_write` is older than its staleness + bound is excluded from every aggregate (VWAP, `/price`, the refresh worker, and + the rows it feeds `price_aggregates`). +- `venue_consecutive_failures` — failed reads in a row, reset to 0 on a successful + read. Non-zero while `last_success` ages is the signal that the adapter is + down, not merely quiet. + +The same state is exposed on `/status` as `venueHeartbeats`, and the exclusion is +visible in the `/price` response as `excludedVenues`. + + +## Why `network` is on every ingest metric Before this change, the ingest metrics were network-blind. On a dual-network deployment the two loops wrote to the same series. @@ -69,6 +93,10 @@ Series count is `pairs x networks`: ingester loops hard-coded in the adapter registry (plus the snapshot appender), so this is a single-digit factor like `network`, not an ID space: never label it with a pool id, a pair or a contract id. +- `venue_last_success_timestamp`, `venue_last_write_timestamp` and + `venue_consecutive_failures` — `venues x networks`. Both factors are bounded by + configuration (the adapters in `src/ingest/venues/` and `src/ingesters/`), so + this is a handful of series, not a pair- or pool-scaled one. **Do not add `issuer` to any of these.** It would multiply the series count by every distinct token in the config for no gain: `pair` is the `pairKey` and diff --git a/openapi.json b/openapi.json index 52372d8e..837b6125 100644 --- a/openapi.json +++ b/openapi.json @@ -422,6 +422,41 @@ "adminConfigured": { "type": "boolean", "description": "Whether this deployment has an ADMIN_TOKEN configured, so an operator can tell a rejected token from a disabled admin surface. Reports only that a token is set, never any part of its value. Admin routes answer 503 admin_not_configured when this is false." + }, + "venueHeartbeats": { + "type": "array", + "description": "Per-venue heartbeat: last successful read, last successful write to price_points, and the count of consecutive failed reads. A venue whose lastWrite is older than its stalenessThresholdMs is excluded from price aggregation.", + "items": { + "type": "object", + "properties": { + "venue": { + "type": "string" + }, + "network": { + "type": "string", + "enum": [ + "testnet", + "mainnet" + ] + }, + "lastSuccess": { + "type": "string", + "format": "date-time", + "nullable": true + }, + "lastWrite": { + "type": "string", + "format": "date-time", + "nullable": true + }, + "consecutiveFailures": { + "type": "integer" + }, + "stalenessThresholdMs": { + "type": "integer" + } + } + } } } } @@ -4364,6 +4399,13 @@ "lastTradeAgeSeconds": { "type": "integer", "nullable": true + }, + "excludedVenues": { + "type": "array", + "description": "Venues whose last written price is older than their staleness bound. Their prices are excluded from this aggregate; the list makes that exclusion visible rather than silent. Empty when every venue is live.", + "items": { + "type": "string" + } } } }, diff --git a/openapi.yaml b/openapi.yaml index 67b70167..2d00d0aa 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -329,6 +329,33 @@ paths: admin surface. Reports only that a token is set, never any part of its value. Admin routes answer 503 admin_not_configured when this is false. + venueHeartbeats: + type: array + description: >- + Per-venue heartbeat: last successful read, last successful + write to price_points, and the count of consecutive failed + reads. A venue whose lastWrite is older than its + stalenessThresholdMs is excluded from price aggregation. + items: + type: object + properties: + venue: + type: string + network: + type: string + enum: [testnet, mainnet] + lastSuccess: + type: string + format: date-time + nullable: true + lastWrite: + type: string + format: date-time + nullable: true + consecutiveFailures: + type: integer + stalenessThresholdMs: + type: integer '503': description: Service unhealthy or database unreachable content: @@ -2929,6 +2956,14 @@ components: lastTradeAgeSeconds: type: integer nullable: true + excludedVenues: + type: array + description: >- + Venues whose last written price is older than their staleness bound. + Their prices are excluded from this aggregate; the list makes that + exclusion visible rather than silent. Empty when every venue is live. + items: + type: string RouteHop: type: object diff --git a/schema.graphql b/schema.graphql index 879c645c..ba59332d 100644 --- a/schema.graphql +++ b/schema.graphql @@ -32,6 +32,11 @@ type AggregatedPrice { sources: Int! confidence: String! lastTradeAgeSeconds: Int + + """ + Venues excluded from this aggregate because their last written price is stale. + """ + excludedVenues: [String!]! } type RouteInfo { diff --git a/src/__tests__/aggregateRefreshNetwork.test.ts b/src/__tests__/aggregateRefreshNetwork.test.ts index 8892cecd..185bbc46 100644 --- a/src/__tests__/aggregateRefreshNetwork.test.ts +++ b/src/__tests__/aggregateRefreshNetwork.test.ts @@ -82,6 +82,7 @@ vi.mock('../config', () => ({ })) import { createAggregateQueue, startAggregateWorker, scheduleAggregateRefresh } from '../jobs/aggregateRefresh' +import { recordVenueSuccess, _resetHeartbeats, DEFAULT_STALENESS_MS } from '../venueHeartbeat' describe('aggregateRefresh runs once per enabled network', () => { beforeEach(() => { @@ -133,7 +134,7 @@ describe('aggregateRefresh runs once per enabled network', () => { await processor({ data: { pairKey: mainnetPair.pairKey, pair: mainnetPair, network: 'mainnet' } }) - expect(mockGetAggregatedPrice).toHaveBeenCalledWith(mainnetPair.pairKey, 'mainnet') + expect(mockGetAggregatedPrice).toHaveBeenCalledWith(mainnetPair.pairKey, 'mainnet', undefined, []) expect(mockGetBestRoute).toHaveBeenCalledWith( mainnetPair.assetA, mainnetPair.assetB, mainnetPair.pairKey, 1000, 'mainnet' ) @@ -170,4 +171,23 @@ describe('aggregateRefresh runs once per enabled network', () => { `testnet:${testnetPair.pairKey}`, expect.objectContaining({ network: 'testnet' }), 10 ) }) + + it('drops a stale venue from the cached payload and the precomputed buckets', async () => { + _resetHeartbeats() + // SDEX wrote a price 10 minutes ago and has since gone stale. + vi.useFakeTimers() + vi.setSystemTime(new Date(Date.now() - DEFAULT_STALENESS_MS - 1000)) + recordVenueSuccess('SDEX', 'mainnet') + vi.useRealTimers() + + startAggregateWorker('mainnet') + await workers[0].processor({ data: { pairKey: mainnetPair.pairKey, pair: mainnetPair, network: 'mainnet' } }) + + // The aggregate is computed with SDEX excluded… + expect(mockGetAggregatedPrice).toHaveBeenCalledWith(mainnetPair.pairKey, 'mainnet', undefined, ['SDEX']) + // …and every window's VWAP/exclusion list carries that exclusion. + for (const call of mockCalculateVWAP.mock.calls) { + expect(call[call.length - 1]).toEqual(['SDEX']) + } + }) }) diff --git a/src/__tests__/price.test.ts b/src/__tests__/price.test.ts index 69699583..936c023b 100644 --- a/src/__tests__/price.test.ts +++ b/src/__tests__/price.test.ts @@ -304,6 +304,7 @@ describe('GET /price/:assetA/:assetB network scoping', () => { confidence: 'medium', lastTradeAgeSeconds: 60, stale: false, + excludedVenues: [], bestRoute: 'SDEX', lastUpdated: '2026-09-27T00:00:00.000Z', })) diff --git a/src/__tests__/schemaValidation.test.ts b/src/__tests__/schemaValidation.test.ts index 55f96239..a30ae8d1 100644 --- a/src/__tests__/schemaValidation.test.ts +++ b/src/__tests__/schemaValidation.test.ts @@ -82,6 +82,7 @@ function validAggregate() { confidence: 'high' as const, lastTradeAgeSeconds: 10, stale: false, + excludedVenues: [] as string[], bestRoute: 'SDEX' as const, } } diff --git a/src/aggregator/vwap.ts b/src/aggregator/vwap.ts index b0ae3008..0116ffc5 100644 --- a/src/aggregator/vwap.ts +++ b/src/aggregator/vwap.ts @@ -23,11 +23,26 @@ export async function calculateVWAPDetailed( pairKey: string, windowMinutes: number, network: NetworkName, - source?: 'SDEX' | 'AMM' + source?: 'SDEX' | 'AMM', + excludedSources: string[] = [], ): Promise<{ num: number; raw: string }> { - const sourceFilter = source ? `AND source = $4` : '' + const conditions: string[] = [ + `pair_key = $1`, + `network = $2`, + `COALESCE(observed_at, timestamp) > NOW() - ($3 || ' minutes')::interval`, + ] const params: (string | number)[] = [pairKey, network, windowMinutes] - if (source) params.push(source) + + if (source) { + params.push(source) + conditions.push(`source = $${params.length}`) + } + + if (excludedSources.length > 0) { + params.push(...excludedSources) + const placeholders = excludedSources.map((_, i) => `$${params.length - excludedSources.length + i + 1}`).join(', ') + conditions.push(`source NOT IN (${placeholders})`) + } const result = await pgPool.query( `SELECT @@ -36,10 +51,7 @@ export async function calculateVWAPDetailed( 0 ) AS vwap FROM price_points - WHERE pair_key = $1 - AND network = $2 - AND COALESCE(observed_at, timestamp) > NOW() - ($3 || ' minutes')::interval - ${sourceFilter}`, + WHERE ${conditions.join('\n AND ')}`, params ) const raw = String(result.rows[0]?.vwap ?? '0') @@ -50,9 +62,10 @@ export async function calculateVWAP( pairKey: string, windowMinutes: number, network: NetworkName, - source?: 'SDEX' | 'AMM' + source?: 'SDEX' | 'AMM', + excludedSources: string[] = [], ): Promise { - const detailed = await calculateVWAPDetailed(pairKey, windowMinutes, network, source) + const detailed = await calculateVWAPDetailed(pairKey, windowMinutes, network, source, excludedSources) return detailed.num } @@ -211,6 +224,7 @@ export interface AggregatedPriceNumeric { confidence: 'high' | 'medium' | 'low' | 'unknown' lastTradeAgeSeconds: number | null stale: boolean + excludedVenues: string[] } export interface AggregatedPriceString { @@ -229,60 +243,77 @@ export interface AggregatedPriceString { confidence: 'high' | 'medium' | 'low' | 'unknown' lastTradeAgeSeconds: number | null stale: boolean + excludedVenues: string[] } export async function getAggregatedPrice( pairKey: string, network: NetworkName, - options: { decimalFormat: 'string' } + options: { decimalFormat: 'string' }, + excludedSources?: string[] ): Promise export async function getAggregatedPrice( pairKey: string, network: NetworkName, - options?: { decimalFormat?: 'number' } + options?: { decimalFormat?: 'number' }, + excludedSources?: string[] ): Promise export async function getAggregatedPrice( pairKey: string, network: NetworkName, - options?: { decimalFormat?: DecimalFormat } + options?: { decimalFormat?: DecimalFormat }, + excludedSources?: string[] ): Promise export async function getAggregatedPrice( pairKey: string, network: NetworkName, - options?: { decimalFormat?: DecimalFormat } + options?: { decimalFormat?: DecimalFormat }, + excludedSources: string[] = [] ): Promise { const [vwap1m, vwap5m, vwap1h, vwap24h, sdexVwap24h, priceChange24h, ammSpotResult, lastTradeResult] = await Promise.all([ - calculateVWAPDetailed(pairKey, 1, network), - calculateVWAPDetailed(pairKey, 5, network), - calculateVWAPDetailed(pairKey, 60, network), - calculateVWAPDetailed(pairKey, 1440, network), - calculateVWAPDetailed(pairKey, 1440, network, 'SDEX'), + calculateVWAPDetailed(pairKey, 1, network, undefined, excludedSources), + calculateVWAPDetailed(pairKey, 5, network, undefined, excludedSources), + calculateVWAPDetailed(pairKey, 60, network, undefined, excludedSources), + calculateVWAPDetailed(pairKey, 1440, network, undefined, excludedSources), + calculateVWAPDetailed(pairKey, 1440, network, 'SDEX', excludedSources), getPriceChange24hDetailed(pairKey, network), // AMM price: latest spot_price averaged across pools for this pair, on this // network only. Both halves of the lookup are network-filtered: the outer // scan of pool_snapshots, and the price_points subquery that decides which // pool ids count as this pair. Filtering only the outer scan would still let // a same-named pool on the other chain contribute its price. - pgPool.query( - `SELECT AVG(spot_price::numeric) AS amm_price - FROM ( - SELECT DISTINCT ON (ps.pool_id) ps.spot_price - FROM pool_snapshots ps - WHERE ps.network = $2 - AND ps.pool_id IN ( - SELECT DISTINCT pool_id FROM price_points - WHERE pair_key = $1 AND network = $2 AND source = 'AMM' AND pool_id IS NOT NULL - ) - ORDER BY ps.pool_id, ps.timestamp DESC - ) latest`, - [pairKey, network] - ), - // Last trade timestamp - pgPool.query( - `SELECT MAX(COALESCE(observed_at, timestamp)) as last_trade /* MAX(timestamp) as last_trade */ FROM price_points - WHERE pair_key = $1 AND network = $2`, - [pairKey, network] - ), + // + // If 'AMM' is excluded we skip this sub-query by returning a null-row result + // rather than changing the SQL structure — the ammPrice comes out as 0. + excludedSources.includes('AMM') + ? Promise.resolve({ rows: [{ amm_price: null }] }) + : pgPool.query( + `SELECT AVG(spot_price::numeric) AS amm_price + FROM ( + SELECT DISTINCT ON (ps.pool_id) ps.spot_price + FROM pool_snapshots ps + WHERE ps.network = $2 + AND ps.pool_id IN ( + SELECT DISTINCT pool_id FROM price_points + WHERE pair_key = $1 AND network = $2 AND source = 'AMM' AND pool_id IS NOT NULL + ) + ORDER BY ps.pool_id, ps.timestamp DESC + ) latest`, + [pairKey, network] + ), + // Last trade timestamp — exclude stale sources so the age reflects live data only + excludedSources.length > 0 + ? pgPool.query( + `SELECT MAX(COALESCE(observed_at, timestamp)) as last_trade /* MAX(timestamp) as last_trade */ FROM price_points + WHERE pair_key = $1 AND network = $2 + AND source NOT IN (${excludedSources.map((_, i) => `$${i + 3}`).join(', ')})`, + [pairKey, network, ...excludedSources] + ) + : pgPool.query( + `SELECT MAX(COALESCE(observed_at, timestamp)) as last_trade /* MAX(timestamp) as last_trade */ FROM price_points + WHERE pair_key = $1 AND network = $2`, + [pairKey, network] + ), ]) const volResult = await pgPool.query( @@ -294,12 +325,15 @@ export async function getAggregatedPrice( WHERE pair_key = $1 AND network = $2 AND COALESCE(observed_at, timestamp) > NOW() - INTERVAL '24 hours' + ${excludedSources.length > 0 + ? `AND source NOT IN (${excludedSources.map((_, i) => `$${i + 3}`).join(', ')})` + : ''} GROUP BY source ) SELECT source, vol FROM per_source UNION ALL SELECT 'TOTAL' AS source, COALESCE(SUM(vol)::numeric, 0) AS vol FROM per_source`, - [pairKey, network] + [pairKey, network, ...excludedSources] ) let sdexVolume24hRaw = '0' @@ -329,8 +363,11 @@ export async function getAggregatedPrice( FROM price_points WHERE pair_key = $1 AND network = $2 - AND COALESCE(observed_at, timestamp) > NOW() - INTERVAL '1 hour'`, - [pairKey, network] + AND COALESCE(observed_at, timestamp) > NOW() - INTERVAL '1 hour' + ${excludedSources.length > 0 + ? `AND source NOT IN (${excludedSources.map((_, i) => `$${i + 3}`).join(', ')})` + : ''}`, + [pairKey, network, ...excludedSources] ) const ammPriceRaw = String(ammSpotResult.rows[0]?.amm_price ?? '0') @@ -376,6 +413,7 @@ export async function getAggregatedPrice( confidence, lastTradeAgeSeconds, stale, + excludedVenues: excludedSources, } } @@ -395,6 +433,7 @@ export async function getAggregatedPrice( confidence, lastTradeAgeSeconds, stale, + excludedVenues: excludedSources, } } diff --git a/src/api/graphql.ts b/src/api/graphql.ts index 46f07b66..8dfe52b3 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -7,6 +7,7 @@ import { randomUUID } from 'node:crypto' import { getCachedPrice } from '../redis' import { getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' +import { getStaleVenues } from '../venueHeartbeat' import { pgPool } from '../db' import { config, activeNetwork, type NetworkName } from '../config' import { parseAssetStr, getActivePairs } from '../pairsRegistry' @@ -133,6 +134,8 @@ export const schema = ` sources: Int! confidence: String! lastTradeAgeSeconds: Int + """Venues excluded from this aggregate because their last written price is stale.""" + excludedVenues: [String!]! } type RouteInfo { @@ -501,7 +504,7 @@ const resolvers = { try { return JSON.parse(cached) } catch { /* fall through */ } } - const agg = await getAggregatedPrice(pairKey, activeNetwork) + const agg = await getAggregatedPrice(pairKey, activeNetwork, undefined, getStaleVenues(activeNetwork)) const route = await getBestRoute(pair.assetA, pair.assetB, pairKey, 1000, activeNetwork) return { assetA, assetB, pairKey, ...agg, diff --git a/src/api/rest.ts b/src/api/rest.ts index d7a31554..82ce0807 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -10,6 +10,7 @@ import { getBestRoute } from '../aggregator/bestRoute' import { pgPool } from '../db' import { config, getNetworkConfig, activeNetwork, type NetworkName } from '../config' import { isAdminConfigured } from './auth' +import { getStaleVenues, getAllHeartbeats } from '../venueHeartbeat' import '../middleware/network' // declares req.network on the FastifyRequest type import { statusResponseSchema, @@ -203,6 +204,12 @@ export async function registerRESTRoutes(app: FastifyInstance) { subject: f.subject, at: f.at.toISOString(), })), + // Per-venue heartbeats — shows last-success (last read), last-write, and + // consecutive failures for every adapter that has polled at least once. A + // venue whose lastWrite is older than its stalenessThresholdMs is excluded + // from aggregation; this field, plus excludedVenues on /price, makes that + // exclusion visible rather than silent. + venueHeartbeats: getAllHeartbeats().filter(h => h.network === network), } } ) @@ -239,7 +246,13 @@ export async function registerRESTRoutes(app: FastifyInstance) { } catch { /* fall through */ } } - const agg = await getAggregatedPrice(pair.pairKey, network, { decimalFormat }) + // Exclude venues whose most recent written price is older than their + // staleness window. Prices from a dead (or quiet) adapter are NOT dropped + // from the DB — they are simply not counted in the aggregate until the + // adapter writes again, at which point they flow back in naturally. + const excludedVenues = getStaleVenues(network) + + const agg = await getAggregatedPrice(pair.pairKey, network, { decimalFormat }, excludedVenues) const route = await getBestRoute(pair.assetA, pair.assetB, pair.pairKey, 1000, network) const result = { assetA: pair.assetA.code, diff --git a/src/api/schemas.ts b/src/api/schemas.ts index 192c36e5..e51fb6ef 100644 --- a/src/api/schemas.ts +++ b/src/api/schemas.ts @@ -147,6 +147,23 @@ export const statusResponseSchema = { }, }, }, + // Per-venue heartbeat state (Issue #219). + venueHeartbeats: { + type: 'array', + items: { + type: 'object', + required: ['venue', 'network', 'lastSuccess', 'lastWrite', 'consecutiveFailures', 'stalenessThresholdMs'], + additionalProperties: false, + properties: { + venue: { type: 'string' }, + network: { type: 'string', enum: ['testnet', 'mainnet'] }, + lastSuccess: { type: ['string', 'null'] }, + lastWrite: { type: ['string', 'null'] }, + consecutiveFailures: { type: 'integer' }, + stalenessThresholdMs: { type: 'integer' }, + }, + }, + }, }, } as const @@ -193,6 +210,7 @@ export const priceResponseSchema = { 'confidence', 'lastTradeAgeSeconds', 'stale', + 'excludedVenues', 'bestRoute', 'lastUpdated', ], @@ -217,6 +235,9 @@ export const priceResponseSchema = { confidence: { type: 'string', enum: ['high', 'medium', 'low', 'unknown'] }, lastTradeAgeSeconds: { type: ['integer', 'null'] }, stale: { type: 'boolean' }, + // Venues whose last written price is older than their staleness threshold; + // their prices are excluded from this aggregate. Empty when all venues are live. + excludedVenues: { type: 'array', items: { type: 'string' } }, bestRoute: { type: 'string', enum: ['SDEX', 'AMM', 'SPLIT', 'UNKNOWN'] }, lastUpdated: { type: 'string' }, }, diff --git a/src/ingest/venues/amm.ts b/src/ingest/venues/amm.ts index c320a1c6..1c16ab78 100644 --- a/src/ingest/venues/amm.ts +++ b/src/ingest/venues/amm.ts @@ -5,6 +5,7 @@ import { resolvePageLedgers } from '../toid' import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' import { createLogger } from '../../logger' +import { recordVenueSuccess, recordVenueFailure, recordVenueReadSuccess } from '../../venueHeartbeat' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' import { fetchHorizonJson, HorizonRateLimitError } from './horizon' @@ -157,6 +158,11 @@ export async function snapshotPool( const trackerKey = pair.pairKey const { previousPrice } = setLastPrice('AMM', network, trackerKey, spotPrice) + // The snapshot wrote a fresh AMM price point from the pool's current + // reserves, so the venue has produced new data this cycle even when no + // trades occurred. + recordVenueSuccess('AMM', network) + publishPriceUpdate({ pair: pair.pairKey, price: spotPrice, @@ -197,7 +203,10 @@ export async function ingestPoolTrades( ) const records = data._embedded?.records ?? [] - if (!records.length) return + if (!records.length) { + recordVenueReadSuccess('AMM', network) + return + } // Horizon returns no `ledger` on a pool trade; it lives in the TOID prefix. const ledgers = resolvePageLedgers( @@ -209,6 +218,7 @@ export async function ingestPoolTrades( { pool: pool.id, trades: records.length }, 'no ledger derivable from trades; skipping batch' ) + recordVenueReadSuccess('AMM', network) return } @@ -264,6 +274,7 @@ export async function ingestPoolTrades( { pair: pair.pairKey, network }, Math.floor(tradeTime.getTime() / 1000) ) + recordVenueSuccess('AMM', network) const lastRecord = records[records.length - 1] await setIndexerCursor(stateId, lastRecord.paging_token, network, ledgers[ledgers.length - 1]) @@ -286,6 +297,7 @@ export async function ingestPoolTrades( } catch (err) { if (err instanceof HorizonRateLimitError) throw err log.error({ pool: pool.id, err }, 'trade ingest error') + recordVenueFailure('AMM', network) } } diff --git a/src/ingest/venues/aquarius.ts b/src/ingest/venues/aquarius.ts index b53d106e..a8d886dc 100644 --- a/src/ingest/venues/aquarius.ts +++ b/src/ingest/venues/aquarius.ts @@ -10,6 +10,7 @@ import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config' import { createLogger } from '../../logger' +import { recordVenueSuccess, recordVenueFailure, recordVenueReadSuccess } from '../../venueHeartbeat' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' import { @@ -61,16 +62,16 @@ export async function fetchAquariusPools( try { if (signal?.aborted) return fetchFailed(new Error('cycle aborted before the Aquarius request was sent')) - const assetAStr = pair.assetA.issuer - ? `${pair.assetA.code}:${pair.assetA.issuer}` - : 'native' - const assetBStr = pair.assetB.issuer - ? `${pair.assetB.code}:${pair.assetB.issuer}` - : 'native' + const assetAStr = pair.assetA.issuer + ? `${pair.assetA.code}:${pair.assetA.issuer}` + : 'native' + const assetBStr = pair.assetB.issuer + ? `${pair.assetB.code}:${pair.assetB.issuer}` + : 'native' - const params = new URLSearchParams() - params.append('assets[]', assetAStr) - params.append('assets[]', assetBStr) + const params = new URLSearchParams() + params.append('assets[]', assetAStr) + params.append('assets[]', assetBStr) const res = await fetchWithTimeout(`${apiUrl}?${params.toString()}`, { signal }) if (!res.ok) { @@ -133,18 +134,31 @@ export async function ingestAquariusPair( signal ) if (!fetched.ok) { + // A cycle abandoned on its own deadline (#471) is our timeout, not the + // venue going quiet, so it must not be recorded as a failure. + if (signal?.aborted) return // #329: record it so the aggregate excludes Aquarius for this pair and // /status reports it, instead of the pair silently reading as "no pools". recordSourceFailure(AQUARIUS_SOURCE, network, pair.pairKey, fetched.error) + recordVenueFailure('aquarius_amm', network) return } clearSourceFailure(AQUARIUS_SOURCE, network, pair.pairKey) const pools = fetched.data - if (pools.length === 0) return + if (pools.length === 0) { + // The API answered, there is just no pool for this pair: a successful read. + recordVenueReadSuccess('aquarius_amm', network) + return + } const points = deriveAquariusPricePoints(pools, pair) - if (points.length === 0) return + if (points.length === 0) { + // The pools were read but every one is empty or malformed, so there is no + // price to record. The read itself succeeded. + recordVenueReadSuccess('aquarius_amm', network) + return + } // Cycle-deadline checkpoint (#351): if the signal aborted while we were // fetching, skip the write instead of committing points from an abandoned @@ -153,6 +167,7 @@ export async function ingestAquariusPair( if (signal?.aborted) return await writeVenuePricePoints('Aquarius', points, pair, network) + recordVenueSuccess('aquarius_amm', network) } export const aquariusAdapter: VenueAdapter = { diff --git a/src/ingest/venues/sdex.ts b/src/ingest/venues/sdex.ts index 19614f33..6d5290c4 100644 --- a/src/ingest/venues/sdex.ts +++ b/src/ingest/venues/sdex.ts @@ -7,6 +7,7 @@ import { resolvePageLedgers } from '../toid' import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' import { createLogger } from '../../logger' +import { recordVenueSuccess, recordVenueFailure, recordVenueReadSuccess } from '../../venueHeartbeat' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' import { @@ -66,7 +67,13 @@ export async function fetchSdexTrades( .call() const records = trades.records - if (!records.length) return null + if (!records.length) { + // Horizon answered successfully with no new trades: the adapter is alive, + // it just has nothing to write. Recording the read keeps a quiet venue + // distinguishable from a dead one. + recordVenueReadSuccess('SDEX', network) + return null + } return { records, @@ -76,13 +83,15 @@ export async function fetchSdexTrades( } } catch (err) { log.error({ pair: pair.pairKey, err }, 'error fetching trades') + recordVenueFailure('SDEX', network) return null } } export function deriveSdexPricePoints( raw: SdexRawTrades, - pair: WatchedPair + pair: WatchedPair, + network: NetworkName = activeNetwork ): { points: VenuePricePoint[]; finalLedger: number; lastPagingToken: string } | null { const { records, stateId, lastLedger } = raw @@ -96,6 +105,8 @@ export function deriveSdexPricePoints( { pair: pair.pairKey, trades: records.length }, 'no ledger derivable from trades; skipping batch' ) + // The read succeeded; the batch is just not persistable this cycle. + recordVenueReadSuccess('SDEX', network) return null } @@ -155,6 +166,7 @@ export async function writeSdexPricePoints( { pair: pair.pairKey, network }, Math.floor(tradeTime.getTime() / 1000) ) + recordVenueSuccess('SDEX', network) await setIndexerCursor(stateId, lastPagingToken, network, finalLedger) log.info({ pair: pair.pairKey, trades: points.length }, 'ingested trades') @@ -179,7 +191,7 @@ export async function ingestPair(pair: WatchedPair, network: NetworkName = activ const raw = await fetchSdexTrades(pair, network, signal) if (!raw) return - const derived = deriveSdexPricePoints(raw, pair) + const derived = deriveSdexPricePoints(raw, pair, network) if (!derived || derived.points.length === 0) return // Cycle-deadline checkpoint (#351), placed before the write SEQUENCE as a @@ -217,8 +229,8 @@ export const sdexAdapter: VenueAdapter = { return fetchSdexTrades(target, network) }, - derive(raw: SdexRawTrades, _target: WatchedPair, pair: WatchedPair, _network: NetworkName): VenuePricePoint[] { - const derived = deriveSdexPricePoints(raw, pair) + derive(raw: SdexRawTrades, _target: WatchedPair, pair: WatchedPair, network: NetworkName): VenuePricePoint[] { + const derived = deriveSdexPricePoints(raw, pair, network) return derived ? derived.points : [] }, diff --git a/src/ingest/venues/soroswap.ts b/src/ingest/venues/soroswap.ts index dc6bb2b7..8d122112 100644 --- a/src/ingest/venues/soroswap.ts +++ b/src/ingest/venues/soroswap.ts @@ -30,6 +30,7 @@ import { createLogger } from '../../logger' import { fetchWithTimeout } from '../../fetch' import { fetchOk, fetchFailed, type FetchResult } from '../fetchResult' import { recordSourceFailure, clearSourceFailure } from '../sourceFailures' +import { recordVenueSuccess, recordVenueFailure, recordVenueReadSuccess } from '../../venueHeartbeat' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' import { @@ -343,6 +344,8 @@ export async function ingestPool( if (spotPrice === 0) { log.debug({ pool: poolEntry.poolAddress, pair: pair.pairKey }, 'skipping empty pool') + // Reserves were read and the pool is simply empty; the adapter is alive. + recordVenueReadSuccess('soroswap_amm', network) return } @@ -395,6 +398,7 @@ export async function ingestPool( network ) markIngestSuccess('Soroswap', network) + recordVenueSuccess('soroswap_amm', network) const trackerKey = pair.pairKey const { previousPrice } = setLastPrice('Soroswap', network, trackerKey, spotPrice) @@ -420,6 +424,7 @@ export async function ingestPool( ) } catch (err) { log.error({ pool: poolEntry.poolAddress, err }, 'ingestPool error') + recordVenueFailure('soroswap_amm', network) } } diff --git a/src/jobs/aggregateRefresh.ts b/src/jobs/aggregateRefresh.ts index 4b49b9b7..d491af23 100644 --- a/src/jobs/aggregateRefresh.ts +++ b/src/jobs/aggregateRefresh.ts @@ -5,6 +5,7 @@ import { setCachedPrice } from '../redis' import { calculateVWAP, calculateOHLCV, getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' import { createLogger } from '../logger' +import { getStaleVenues } from '../venueHeartbeat' const log = createLogger('job:aggregate-refresh') @@ -35,7 +36,14 @@ export function startAggregateWorker(network: NetworkName) { async (job) => { const { pairKey, pair } = job.data try { - const agg = await getAggregatedPrice(pairKey, network) + // A venue whose last written price is older than its staleness bound + // is dropped from every aggregate computed here, exactly as the live + // /price route does. Without this the refresh worker would keep blending + // a dead adapter's last price into the cached payload and into the + // price_aggregates rows that history and the screener read. + const excludedVenues = getStaleVenues(network) + + const agg = await getAggregatedPrice(pairKey, network, undefined, excludedVenues) const route = await getBestRoute(pair.assetA, pair.assetB, pairKey, 1000, network) const result = { @@ -69,9 +77,9 @@ export function startAggregateWorker(network: NetworkName) { for (const w of windows) { const [vwap, sdexVwap, ammVwap, ohlcv] = await Promise.all([ - calculateVWAP(pairKey, w.minutes, network), - calculateVWAP(pairKey, w.minutes, network, 'SDEX'), - calculateVWAP(pairKey, w.minutes, network, 'AMM'), + calculateVWAP(pairKey, w.minutes, network, undefined, excludedVenues), + calculateVWAP(pairKey, w.minutes, network, 'SDEX', excludedVenues), + calculateVWAP(pairKey, w.minutes, network, 'AMM', excludedVenues), calculateOHLCV(pairKey, w.minutes, network), ]) diff --git a/src/metrics.ts b/src/metrics.ts index e71acfa2..1b01cf10 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -289,6 +289,33 @@ export function setPoolMetricsUpdater(updater: () => void) { _poolMetricsUpdater = updater } +// ── Per-venue heartbeat gauges (Issue #219) ───────────────────────────────── +// +// Cardinality is (venues × networks), both bounded by configuration, so these +// are safe to add without the label-cardinality warning that applies to pair- +// or pool-scoped labels. See docs/ingest-metrics.md for the full rationale. + +export const venue_last_success_timestamp = new Gauge({ + name: 'venue_last_success_timestamp', + help: 'Unix timestamp of the last successful upstream read for a venue', + labelNames: ['venue', 'network'], + registers: [register] +}) + +export const venue_last_write_timestamp = new Gauge({ + name: 'venue_last_write_timestamp', + help: 'Unix timestamp of the last row written to price_points for a venue', + labelNames: ['venue', 'network'], + registers: [register] +}) + +export const venue_consecutive_failures = new Gauge({ + name: 'venue_consecutive_failures', + help: 'Number of consecutive failed upstream reads for a venue (reset to 0 on success)', + labelNames: ['venue', 'network'], + registers: [register] +}) + export async function getMetrics() { if (_poolMetricsUpdater) { try { diff --git a/src/venueHeartbeat.ts b/src/venueHeartbeat.ts new file mode 100644 index 00000000..4ccad811 --- /dev/null +++ b/src/venueHeartbeat.ts @@ -0,0 +1,226 @@ +/** + * Per-venue heartbeat registry. + * + * Tracks the last successful read, the last row written, and the number of + * consecutive failures for every (venue, network) pair. The aggregation layer + * reads this to exclude venues whose most-recent write (price age) is older than + * their configured staleness threshold, making a dead — or merely quiet — + * adapter visibly absent from the aggregate rather than silently inflating it + * with stale prices. + * + * Issue: #219 + */ + +import type { NetworkName } from './config' +import { + venue_last_success_timestamp, + venue_last_write_timestamp, + venue_consecutive_failures, +} from './metrics' + +/** Default staleness threshold: 5 minutes. Adapters may override per venue. */ +export const DEFAULT_STALENESS_MS = 5 * 60 * 1000 + +export interface VenueHeartbeat { + /** ISO timestamp of the last successful upstream read (data fetched without error). */ + lastSuccess: string | null + /** ISO timestamp of the last successful write to price_points. */ + lastWrite: string | null + /** How many poll cycles in a row ended in a failure (reset to 0 on success). */ + consecutiveFailures: number + /** Staleness threshold in ms; a venue whose lastWrite is older than this is excluded. */ + stalenessThresholdMs: number +} + +/** + * In-process registry, keyed by `${venue}:${network}`. + * + * Heartbeats are intentionally not persisted — a fresh deploy starts with all + * venues live, and the first poll cycle immediately updates every heartbeat. + * Persisting state would complicate tests and add a dependency with no benefit + * since each heartbeat covers at most one poll interval of uncertainty. + */ +const registry = new Map() + +function key(venue: string, network: NetworkName): string { + return `${venue}:${network}` +} + +/** + * Return a copy of the current heartbeat for a venue, or a zero-state if it + * has never been registered. + */ +export function getHeartbeat(venue: string, network: NetworkName): VenueHeartbeat { + return registry.get(key(venue, network)) ?? { + lastSuccess: null, + lastWrite: null, + consecutiveFailures: 0, + stalenessThresholdMs: DEFAULT_STALENESS_MS, + } +} + +/** + * Return all known heartbeats as an array of `{ venue, network, ...heartbeat }` + * objects, suitable for inclusion in `/status` responses. + */ +export function getAllHeartbeats(): Array<{ venue: string; network: NetworkName } & VenueHeartbeat> { + return [...registry.entries()].map(([k, hb]) => { + const [venue, network] = k.split(':') as [string, NetworkName] + return { venue, network, ...hb } + }) +} + +/** + * Record a successful upstream READ that produced no new rows. + * + * This is what separates a quiet venue from a dead one. A poll that reaches the + * upstream, gets a well-formed (possibly empty) answer, and has nothing new to + * write is a SUCCESS: the adapter is alive. Only `lastSuccess` moves; + * `lastWrite` is left where it was so the price's age keeps advancing and its + * stale rows are excluded from the aggregate. `consecutiveFailures` resets to 0. + * + * @param venue Ingester source name, e.g. `'aquarius_amm'` or `'SDEX'`. + * @param network The Stellar network this ingester is reading from. + * @param stalenessThresholdMs Optional override for the staleness window. + */ +export function recordVenueReadSuccess( + venue: string, + network: NetworkName, + stalenessThresholdMs: number = DEFAULT_STALENESS_MS, +): void { + const now = new Date().toISOString() + const existing = registry.get(key(venue, network)) + const updated: VenueHeartbeat = { + lastSuccess: now, + lastWrite: existing?.lastWrite ?? null, + consecutiveFailures: 0, + stalenessThresholdMs: existing?.stalenessThresholdMs ?? stalenessThresholdMs, + } + registry.set(key(venue, network), updated) + + venue_last_success_timestamp.set({ venue, network }, Date.now() / 1000) + venue_consecutive_failures.set({ venue, network }, 0) +} + +/** + * Record a successful WRITE of new price_points rows. + * + * Moves `lastWrite` to now. The read that produced these rows is recorded + * separately (see {@link recordVenueReadSuccess}); a write implies a recent + * successful read, so callers that do both should use {@link recordVenueSuccess}. + */ +export function recordVenueWrite(venue: string, network: NetworkName): void { + const now = new Date().toISOString() + const existing = registry.get(key(venue, network)) ?? { + lastSuccess: null, + lastWrite: null, + consecutiveFailures: 0, + stalenessThresholdMs: DEFAULT_STALENESS_MS, + } + registry.set(key(venue, network), { ...existing, lastWrite: now }) + + venue_last_write_timestamp.set({ venue, network }, Date.now() / 1000) +} + +/** + * Record a successful upstream read AND write for a venue. + * + * Call this after new price_points rows have been committed. Both + * `lastSuccess` and `lastWrite` are set to now, and `consecutiveFailures` is + * reset to 0. + * + * @param venue Ingester source name, e.g. `'aquarius_amm'` or `'SDEX'`. + * @param network The Stellar network this ingester is reading from. + * @param stalenessThresholdMs Optional override for the staleness window. + */ +export function recordVenueSuccess( + venue: string, + network: NetworkName, + stalenessThresholdMs: number = DEFAULT_STALENESS_MS, +): void { + recordVenueReadSuccess(venue, network, stalenessThresholdMs) + recordVenueWrite(venue, network) +} + +/** + * Record a failed upstream read for a venue. + * + * Increments `consecutiveFailures`. `lastSuccess` and `lastWrite` are left + * untouched — the existing (possibly stale) values remain visible so callers + * can see exactly how long ago the last good data arrived. + * + * @param venue Ingester source name. + * @param network The Stellar network this ingester is reading from. + */ +export function recordVenueFailure(venue: string, network: NetworkName): void { + const existing = registry.get(key(venue, network)) ?? { + lastSuccess: null, + lastWrite: null, + consecutiveFailures: 0, + stalenessThresholdMs: DEFAULT_STALENESS_MS, + } + const updated: VenueHeartbeat = { + ...existing, + consecutiveFailures: existing.consecutiveFailures + 1, + } + registry.set(key(venue, network), updated) + + venue_consecutive_failures.set({ venue, network }, updated.consecutiveFailures) +} + +/** + * Return the set of venue names that are currently stale on the given network. + * + * A venue is stale when its most recent price — i.e. its last successful WRITE — + * is older than its `stalenessThresholdMs`. Staleness is measured on write age, + * not read age, because the thing being excluded is an old price: a venue that + * is still reading successfully but has not produced a new row is exactly the + * "quiet" case whose stale price must not keep influencing the aggregate. + * + * A venue that has never written falls back to its last successful read, so a + * freshly registered venue that is reading fine but has not written yet is not + * flagged; only one that has neither written nor read within the window is + * stale. A venue that has only ever failed (no read, no write) is stale. + * + * Venues with no heartbeat at all (never polled) are NOT considered stale — + * they may simply not apply to this network or may be in their first poll cycle. + */ +export function getStaleVenues(network: NetworkName): string[] { + const now = Date.now() + const stale: string[] = [] + + for (const [k, hb] of registry.entries()) { + const [venue, hbNetwork] = k.split(':') as [string, NetworkName] + if (hbNetwork !== network) continue + + if (isStaleHeartbeat(hb, now)) stale.push(venue) + } + + return stale +} + +/** + * True if this venue is currently stale on the given network. + * + * A venue with no heartbeat entry is NOT considered stale. + */ +export function isVenueStale(venue: string, network: NetworkName): boolean { + const hb = registry.get(key(venue, network)) + if (!hb) return false + return isStaleHeartbeat(hb, Date.now()) +} + +function isStaleHeartbeat(hb: VenueHeartbeat, now: number): boolean { + // Price age first; fall back to the last successful read for a venue that has + // not written anything yet. + const reference = hb.lastWrite ?? hb.lastSuccess + if (reference === null) return true + return now - new Date(reference).getTime() > hb.stalenessThresholdMs +} + +/** + * Clear all heartbeat state. Used in tests to reset between cases. + */ +export function _resetHeartbeats(): void { + registry.clear() +} diff --git a/tests/venueHeartbeat.test.ts b/tests/venueHeartbeat.test.ts new file mode 100644 index 00000000..10201a36 --- /dev/null +++ b/tests/venueHeartbeat.test.ts @@ -0,0 +1,495 @@ +/** + * Venue heartbeat tests — Issue #219 + * + * Acceptance criteria covered: + * ✓ Every venue reports last-success, last-write and consecutive-failure count + * ✓ Stale venue prices are excluded from aggregates, and responses say a venue + * was excluded + * ✓ /status and /metrics both expose heartbeat data + * ✓ A test drives a venue to stale and asserts it leaves the aggregate + */ + +import { describe, it, expect, vi, beforeEach } from 'vitest' +import Fastify from 'fastify' + +// ── Hoisted mocks ───────────────────────────────────────────────────────────── + +const { mockQuery, mockGetCachedPrice, mockGetBestRoute } = vi.hoisted(() => ({ + mockQuery: vi.fn(), + mockGetCachedPrice: vi.fn(), + mockGetBestRoute: vi.fn(), +})) + +vi.mock('../src/db', () => ({ + pgPool: { query: mockQuery }, +})) + +vi.mock('../src/redis', () => ({ + getCachedPrice: mockGetCachedPrice, + setCachedPrice: vi.fn(), +})) + +vi.mock('../src/aggregator/bestRoute', () => ({ + getBestRoute: mockGetBestRoute, +})) + +const { heartbeatTestPairs } = vi.hoisted(() => ({ + heartbeatTestPairs: [ + { + pairKey: 'USDC/XLM', + assetA: { code: 'XLM', issuer: null }, + assetB: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, + }, + ], +})) + +vi.mock('../src/config', () => ({ + config: { + pairs: heartbeatTestPairs, + cache: { priceTtl: 10 }, + }, + activeNetwork: 'testnet', + getNetworkConfig: (network: string) => ({ + pairs: network === 'testnet' ? heartbeatTestPairs : [], + }), +})) + +// ── Imports ─────────────────────────────────────────────────────────────────── + +import { + recordVenueSuccess, + recordVenueReadSuccess, + recordVenueWrite, + recordVenueFailure, + getHeartbeat, + getStaleVenues, + isVenueStale, + getAllHeartbeats, + _resetHeartbeats, + DEFAULT_STALENESS_MS, +} from '../src/venueHeartbeat' +import { register, venue_last_success_timestamp, venue_last_write_timestamp, venue_consecutive_failures } from '../src/metrics' +import { registerRESTRoutes } from '../src/api/rest' + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +async function buildApp() { + const app = Fastify({ logger: false }) + await registerRESTRoutes(app) + await app.ready() + return app +} + +/** + * Standard mock implementation for the /price route's DB queries. + * Passes through to a fresh-venue state (no stale sources). + */ +function mockFreshQueries(recentTrade?: string) { + const tradeTs = recentTrade ?? new Date(Date.now() - 60_000).toISOString() // 1 min ago + mockQuery.mockImplementation(async (sql: string) => { + if (sql.includes('AVG(spot_price::numeric)')) return { rows: [{ amm_price: '0.1' }] } + if (sql.includes('last_trade FROM price_points')) return { rows: [{ last_trade: tradeTs }] } + if (sql.includes('GROUP BY source')) return { rows: [{ source: 'SDEX', vol: '100' }] } + if (sql.includes('COUNT(DISTINCT COALESCE(pool_id')) return { rows: [{ sources: '1' }] } + if (sql.includes('SUM(price::numeric * base_volume::numeric)')) return { rows: [{ vwap: '0.1' }] } + if (sql.includes('price_24h_ago')) return { rows: [{ price_24h_ago: '0.09', price_now: '0.1' }] } + if (sql.includes('indexer_state')) return { rows: [] } + return { rows: [] } + }) +} + +// ── Heartbeat unit tests ─────────────────────────────────────────────────────── + +describe('venueHeartbeat unit', () => { + beforeEach(() => { + _resetHeartbeats() + vi.clearAllMocks() + }) + + it('starts with no heartbeat for an unknown venue', () => { + const hb = getHeartbeat('SDEX', 'testnet') + expect(hb.lastSuccess).toBeNull() + expect(hb.lastWrite).toBeNull() + expect(hb.consecutiveFailures).toBe(0) + }) + + it('recordVenueSuccess sets lastSuccess and lastWrite, resets failures', () => { + // prime with some failures first + recordVenueFailure('SDEX', 'testnet') + recordVenueFailure('SDEX', 'testnet') + + recordVenueSuccess('SDEX', 'testnet') + + const hb = getHeartbeat('SDEX', 'testnet') + expect(hb.lastSuccess).not.toBeNull() + expect(hb.lastWrite).not.toBeNull() + expect(hb.consecutiveFailures).toBe(0) + }) + + it('recordVenueReadSuccess moves lastSuccess but leaves lastWrite untouched', () => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-01-01T00:00:00Z')) + recordVenueSuccess('SDEX', 'testnet') + const write = getHeartbeat('SDEX', 'testnet').lastWrite + vi.useRealTimers() + + // A later poll that reads successfully but has nothing new to write. + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-01-01T00:02:00Z')) + recordVenueReadSuccess('SDEX', 'testnet') + vi.useRealTimers() + + const hb = getHeartbeat('SDEX', 'testnet') + expect(hb.lastSuccess).toBe(new Date('2026-01-01T00:02:00Z').toISOString()) + // Price age (lastWrite) is unchanged: no new rows were produced. + expect(hb.lastWrite).toBe(write) + expect(hb.consecutiveFailures).toBe(0) + }) + + it('recordVenueWrite moves lastWrite without touching lastSuccess', () => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-01-01T00:00:00Z')) + recordVenueReadSuccess('SDEX', 'testnet') + const read = getHeartbeat('SDEX', 'testnet').lastSuccess + vi.useRealTimers() + + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-01-01T00:01:00Z')) + recordVenueWrite('SDEX', 'testnet') + vi.useRealTimers() + + const hb = getHeartbeat('SDEX', 'testnet') + expect(hb.lastSuccess).toBe(read) + expect(hb.lastWrite).toBe(new Date('2026-01-01T00:01:00Z').toISOString()) + }) + + it('a quiet venue (fresh reads, old writes) still reads as alive but its price is stale', () => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-01-01T00:00:00Z')) + recordVenueSuccess('SDEX', 'testnet') // wrote a price + vi.useRealTimers() + + // Reads keep succeeding for 10 minutes, but nothing new is written. + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-01-01T00:10:00Z')) + recordVenueReadSuccess('SDEX', 'testnet') + vi.useRealTimers() + + const hb = getHeartbeat('SDEX', 'testnet') + // The operator can see the adapter is alive (recent read, zero failures)… + expect(hb.lastSuccess).not.toBeNull() + expect(hb.consecutiveFailures).toBe(0) + // …but its written price has aged past the bound and must be excluded. + expect(isVenueStale('SDEX', 'testnet')).toBe(true) + expect(getStaleVenues('testnet')).toContain('SDEX') + }) + + it('recordVenueFailure increments consecutiveFailures without clearing timestamps', () => { + recordVenueSuccess('AMM', 'mainnet') + const { lastSuccess } = getHeartbeat('AMM', 'mainnet') + + recordVenueFailure('AMM', 'mainnet') + recordVenueFailure('AMM', 'mainnet') + + const hb = getHeartbeat('AMM', 'mainnet') + expect(hb.consecutiveFailures).toBe(2) + // lastSuccess is preserved so callers can see when the last good data arrived + expect(hb.lastSuccess).toBe(lastSuccess) + }) + + it('different (venue, network) pairs are tracked independently', () => { + recordVenueSuccess('SDEX', 'testnet') + recordVenueFailure('SDEX', 'mainnet') + recordVenueFailure('AMM', 'testnet') + + expect(getHeartbeat('SDEX', 'testnet').consecutiveFailures).toBe(0) + expect(getHeartbeat('SDEX', 'mainnet').consecutiveFailures).toBe(1) + expect(getHeartbeat('AMM', 'testnet').consecutiveFailures).toBe(1) + }) + + it('isVenueStale returns false for venues with no heartbeat entry', () => { + // A venue that has never polled should NOT be treated as stale; it may + // simply not apply to this pair or be in its first cycle. + expect(isVenueStale('soroswap_amm', 'testnet')).toBe(false) + }) + + it('isVenueStale returns true after the staleness threshold elapses', () => { + // Simulate a success 10 minutes ago by back-dating the registry entry + recordVenueSuccess('SDEX', 'testnet') + const hb = getHeartbeat('SDEX', 'testnet') + + // Manually push lastSuccess into the past + const staleTime = new Date(Date.now() - DEFAULT_STALENESS_MS - 1000).toISOString() + // @ts-expect-error — internal access for test + const registry = (globalThis as any).__venueHeartbeatRegistry + // The registry is module-internal, so drive it through the public API: + // recordVenueFailure does not change lastSuccess, so we use success again + // but fake the time by using vi.useFakeTimers. + vi.useFakeTimers() + vi.setSystemTime(new Date(Date.now() - DEFAULT_STALENESS_MS - 1000)) + recordVenueSuccess('SDEX', 'testnet') + vi.useRealTimers() + + expect(isVenueStale('SDEX', 'testnet')).toBe(true) + }) + + it('getStaleVenues returns the names of stale venues on the given network', () => { + vi.useFakeTimers() + vi.setSystemTime(new Date(Date.now() - DEFAULT_STALENESS_MS - 1000)) + recordVenueSuccess('SDEX', 'testnet') + recordVenueSuccess('AMM', 'testnet') + vi.useRealTimers() + + // A fresh venue on the same network + recordVenueSuccess('aquarius_amm', 'testnet') + + const stale = getStaleVenues('testnet') + expect(stale).toContain('SDEX') + expect(stale).toContain('AMM') + expect(stale).not.toContain('aquarius_amm') + }) + + it('getAllHeartbeats returns all registered (venue, network) entries', () => { + recordVenueSuccess('SDEX', 'testnet') + recordVenueFailure('AMM', 'mainnet') + + const all = getAllHeartbeats() + expect(all.find(h => h.venue === 'SDEX' && h.network === 'testnet')).toBeDefined() + expect(all.find(h => h.venue === 'AMM' && h.network === 'mainnet')).toBeDefined() + }) +}) + +// ── Prometheus metric tests ──────────────────────────────────────────────────── + +describe('venue heartbeat Prometheus metrics', () => { + beforeEach(() => { + _resetHeartbeats() + vi.clearAllMocks() + venue_last_success_timestamp.reset() + venue_last_write_timestamp.reset() + venue_consecutive_failures.reset() + }) + + it('recordVenueSuccess sets venue_last_success_timestamp gauge', async () => { + const before = Date.now() / 1000 + recordVenueSuccess('SDEX', 'testnet') + const after = Date.now() / 1000 + + const metric = await register.getSingleMetric('venue_last_success_timestamp')!.get() + const series = metric.values.find( + v => v.labels.venue === 'SDEX' && v.labels.network === 'testnet' + ) + expect(series).toBeDefined() + expect(series!.value).toBeGreaterThanOrEqual(before) + expect(series!.value).toBeLessThanOrEqual(after + 1) + }) + + it('recordVenueWrite sets venue_last_write_timestamp gauge', async () => { + const before = Date.now() / 1000 + recordVenueWrite('SDEX', 'testnet') + const after = Date.now() / 1000 + + const metric = await register.getSingleMetric('venue_last_write_timestamp')!.get() + const series = metric.values.find( + v => v.labels.venue === 'SDEX' && v.labels.network === 'testnet' + ) + expect(series).toBeDefined() + expect(series!.value).toBeGreaterThanOrEqual(before) + expect(series!.value).toBeLessThanOrEqual(after + 1) + }) + + it('recordVenueFailure increments venue_consecutive_failures gauge', async () => { recordVenueFailure('AMM', 'mainnet') + recordVenueFailure('AMM', 'mainnet') + + const metric = await register.getSingleMetric('venue_consecutive_failures')!.get() + const series = metric.values.find( + v => v.labels.venue === 'AMM' && v.labels.network === 'mainnet' + ) + expect(series).toBeDefined() + expect(series!.value).toBe(2) + }) + + it('recordVenueSuccess resets venue_consecutive_failures to 0', async () => { + recordVenueFailure('SDEX', 'testnet') + recordVenueFailure('SDEX', 'testnet') + recordVenueSuccess('SDEX', 'testnet') + + const metric = await register.getSingleMetric('venue_consecutive_failures')!.get() + const series = metric.values.find( + v => v.labels.venue === 'SDEX' && v.labels.network === 'testnet' + ) + expect(series!.value).toBe(0) + }) + + it('/metrics endpoint exposes venue heartbeat gauges', async () => { + recordVenueSuccess('SDEX', 'testnet') + recordVenueFailure('AMM', 'testnet') + + const app = Fastify({ logger: false }) + app.get('/metrics', { config: { public: true } }, async (_req, reply) => { + const { getMetrics } = await import('../src/metrics') + reply.type('text/plain') + return await getMetrics() + }) + await app.ready() + + const res = await app.inject({ method: 'GET', url: '/metrics' }) + expect(res.statusCode).toBe(200) + expect(res.body).toContain('venue_last_success_timestamp') + expect(res.body).toContain('venue_last_write_timestamp') + expect(res.body).toContain('venue_consecutive_failures') + }) +}) + +// ── /status heartbeat exposure ───────────────────────────────────────────────── + +describe('/status venue heartbeats', () => { + beforeEach(() => { + _resetHeartbeats() + vi.clearAllMocks() + mockGetCachedPrice.mockResolvedValue(null) + mockGetBestRoute.mockResolvedValue({ route: 'SDEX' }) + mockQuery.mockResolvedValue({ rows: [] }) + }) + + it('returns an empty venueHeartbeats array when no adapters have polled', async () => { + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/status' }) + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.venueHeartbeats).toEqual([]) + }) + + it('returns heartbeat data for venues that have polled', async () => { + recordVenueSuccess('SDEX', 'testnet') + recordVenueFailure('AMM', 'testnet') + + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/status' }) + expect(res.statusCode).toBe(200) + + const body = res.json() + const sdex = body.venueHeartbeats.find((h: any) => h.venue === 'SDEX') + const amm = body.venueHeartbeats.find((h: any) => h.venue === 'AMM') + + expect(sdex).toBeDefined() + expect(sdex.lastSuccess).not.toBeNull() + expect(sdex.consecutiveFailures).toBe(0) + + expect(amm).toBeDefined() + expect(amm.consecutiveFailures).toBe(1) + }) + + it('filters heartbeats to the requested network', async () => { + recordVenueSuccess('SDEX', 'testnet') + recordVenueSuccess('SDEX', 'mainnet') + + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/status' }) + expect(res.statusCode).toBe(200) + + const body = res.json() + // Default network is testnet (from our mock), so only testnet heartbeats appear + for (const hb of body.venueHeartbeats) { + expect(hb.network).toBe('testnet') + } + }) +}) + +// ── Stale-venue exclusion from aggregation ───────────────────────────────────── + +describe('stale venue exclusion from /price aggregate (AC: drive venue stale, assert excluded)', () => { + beforeEach(() => { + _resetHeartbeats() + vi.clearAllMocks() + mockGetCachedPrice.mockResolvedValue(null) + mockGetBestRoute.mockResolvedValue({ route: 'SDEX' }) + }) + + it('excludedVenues is empty when all venues are fresh', async () => { + recordVenueSuccess('SDEX', 'testnet') + recordVenueSuccess('AMM', 'testnet') + mockFreshQueries() + + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/price/XLM/USDC' }) + expect(res.statusCode).toBe(200) + const body = res.json() + expect(body.excludedVenues).toEqual([]) + }) + + it('drives a venue to stale and asserts it is excluded from the aggregate', async () => { + // Both venues wrote data 10 minutes ago (beyond the 5-minute threshold) + vi.useFakeTimers() + vi.setSystemTime(new Date(Date.now() - DEFAULT_STALENESS_MS - 1000)) + recordVenueSuccess('SDEX', 'testnet') + vi.useRealTimers() + + // AMM is fresh + recordVenueSuccess('AMM', 'testnet') + + mockFreshQueries() + + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/price/XLM/USDC' }) + expect(res.statusCode).toBe(200) + const body = res.json() + + // SDEX is stale \u2014 must appear in excludedVenues + expect(body.excludedVenues).toContain('SDEX') + // AMM is fresh \u2014 must NOT appear + expect(body.excludedVenues).not.toContain('AMM') + + // The SQL queries must include NOT IN ('SDEX') for the stale source + const calls = mockQuery.mock.calls.map((c: any[]) => c[0] as string) + const hasExclusion = calls.some(sql => sql.includes('NOT IN')) + expect(hasExclusion).toBe(true) + }) + + it('excludes all venues when every registered venue is stale', async () => { + vi.useFakeTimers() + vi.setSystemTime(new Date(Date.now() - DEFAULT_STALENESS_MS - 1000)) + recordVenueSuccess('SDEX', 'testnet') + recordVenueSuccess('AMM', 'testnet') + vi.useRealTimers() + + mockFreshQueries() + + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/price/XLM/USDC' }) + expect(res.statusCode).toBe(200) + const body = res.json() + + expect(body.excludedVenues).toContain('SDEX') + expect(body.excludedVenues).toContain('AMM') + }) + + it('venues with no heartbeat are NOT excluded (first poll cycle grace)', async () => { + // No heartbeats recorded at all — brand-new ingesters + mockFreshQueries() + + const app = await buildApp() + const res = await app.inject({ method: 'GET', url: '/price/XLM/USDC' }) + expect(res.statusCode).toBe(200) + const body = res.json() + + // No venue has been declared stale, so the exclusion list is empty + expect(body.excludedVenues).toEqual([]) + }) + + it('a venue that recovers from stale is no longer excluded', async () => { + // Drive stale + vi.useFakeTimers() + vi.setSystemTime(new Date(Date.now() - DEFAULT_STALENESS_MS - 1000)) + recordVenueSuccess('SDEX', 'testnet') + vi.useRealTimers() + + expect(isVenueStale('SDEX', 'testnet')).toBe(true) + + // Recovery: a successful poll brings it back + recordVenueSuccess('SDEX', 'testnet') + + expect(isVenueStale('SDEX', 'testnet')).toBe(false) + expect(getStaleVenues('testnet')).not.toContain('SDEX') + }) +})