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') + }) +})