From a2de4f81172bbd28e1593ef40e945e386ee3c6b6 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 | 13 +- src/__tests__/schemaValidation.test.ts | 1 + src/__tests__/vwap.test.ts | 4 +- src/__tests__/vwapNetwork.test.ts | 4 +- src/aggregator/vwap.ts | 136 +++-- 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/decimalStrings.test.ts | 24 +- tests/staleness.test.ts | 10 +- tests/venueHeartbeat.test.ts | 495 ++++++++++++++++++ 24 files changed, 1128 insertions(+), 99 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 af7089b1..40a5cc82 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -1186,6 +1186,8 @@ export interface components { stalenessThresholdSeconds?: number; /** @description True if the time elapsed since newestInputObservedAt exceeds stalenessThresholdSeconds. */ stale?: boolean; + /** @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"]; @@ -2271,6 +2273,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 4ac5fa7c..36b62f25 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" + } + } + } } } } @@ -4884,6 +4919,13 @@ "stale": { "type": "boolean", "description": "True if the time elapsed since newestInputObservedAt exceeds stalenessThresholdSeconds." + }, + "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 2e67f48e..e22b3893 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: @@ -3269,6 +3296,14 @@ components: stale: type: boolean description: True if the time elapsed since newestInputObservedAt exceeds stalenessThresholdSeconds. + 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 18362b24..9dbb60b6 100644 --- a/src/__tests__/price.test.ts +++ b/src/__tests__/price.test.ts @@ -76,7 +76,7 @@ describe('GET /price/:assetA/:assetB confidence score', () => { mockQuery.mockImplementation(async (sql: string) => { if (sql.includes('AVG(spot_price::numeric)')) return { rows: [{ amm_price: '0.1' }] } - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: recent }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: recent }] } if (sql.includes('GROUP BY source')) return { rows: [{ source: 'SDEX', vol: '100' }, { source: 'AMM', vol: '50' }] } if (sql.includes('COUNT(DISTINCT COALESCE(pool_id')) return { rows: [{ sources: '2' }] } if (sql.includes('SUM(price::numeric * base_volume::numeric)')) return { rows: [{ vwap: '0.1' }] } @@ -101,7 +101,7 @@ describe('GET /price/:assetA/:assetB confidence score', () => { mockQuery.mockImplementation(async (sql: string) => { if (sql.includes('AVG(spot_price::numeric)')) return { rows: [{ amm_price: '0.1' }] } - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: twoMinAgo }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: twoMinAgo }] } 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' }] } @@ -122,7 +122,7 @@ describe('GET /price/:assetA/:assetB confidence score', () => { const tenMinAgo = new Date(now.getTime() - 600000).toISOString() // 10m ago mockQuery.mockImplementation(async (sql: string) => { - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: tenMinAgo }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: tenMinAgo }] } if (sql.includes('COUNT(DISTINCT COALESCE(pool_id')) return { rows: [{ sources: '1' }] } if (sql.includes('GROUP BY source')) return { rows: [{ source: 'SDEX', vol: '100' }] } if (sql.includes('SUM(price::numeric * base_volume::numeric)')) return { rows: [{ vwap: '0.1' }] } @@ -140,7 +140,7 @@ describe('GET /price/:assetA/:assetB confidence score', () => { it('returns unknown confidence when no trades found', async () => { mockQuery.mockImplementation(async (sql: string) => { - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: null }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: null }] } return { rows: [] } }) @@ -241,8 +241,8 @@ describe('GET /price/:assetA/:assetB network scoping', () => { mockGetBestRoute.mockResolvedValue({ route: 'SDEX' }) mockQuery.mockImplementation(async (sql: string) => { if (sql.includes('AVG(spot_price::numeric)')) return { rows: [{ amm_price: '0.1' }] } - if (sql.includes('MAX(timestamp) as last_trade')) { - return { rows: [{ last_trade: new Date(Date.now() - 10000).toISOString() }] } + if (sql.includes('as last_observed_at')) { + return { rows: [{ last_observed_at: new Date(Date.now() - 10000).toISOString() }] } } if (sql.includes('GROUP BY source')) return { rows: [{ source: 'SDEX', vol: '100' }] } if (sql.includes('COUNT(DISTINCT COALESCE(pool_id')) return { rows: [{ sources: '1' }] } @@ -307,6 +307,7 @@ describe('GET /price/:assetA/:assetB network scoping', () => { inputCount: 1, stalenessThresholdSeconds: 300, 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 f930e800..c24cc283 100644 --- a/src/__tests__/schemaValidation.test.ts +++ b/src/__tests__/schemaValidation.test.ts @@ -85,6 +85,7 @@ function validAggregate() { inputCount: 2, stalenessThresholdSeconds: 300, stale: false, + excludedVenues: [] as string[], bestRoute: 'SDEX' as const, } } diff --git a/src/__tests__/vwap.test.ts b/src/__tests__/vwap.test.ts index a9e08ae5..aef3b6c0 100644 --- a/src/__tests__/vwap.test.ts +++ b/src/__tests__/vwap.test.ts @@ -232,8 +232,8 @@ describe('aggregator/vwap.ts', () => { } // Last trade timestamp - if (sql.includes('MAX(timestamp) as last_trade')) { - return { rows: [{ last_trade: lastTradeDate ? lastTradeDate.toISOString() : null }] } + if (sql.includes('as last_observed_at')) { + return { rows: [{ last_observed_at: lastTradeDate ? lastTradeDate.toISOString() : null }] } } // Volume by source diff --git a/src/__tests__/vwapNetwork.test.ts b/src/__tests__/vwapNetwork.test.ts index f55af520..99f84e0c 100644 --- a/src/__tests__/vwapNetwork.test.ts +++ b/src/__tests__/vwapNetwork.test.ts @@ -130,10 +130,10 @@ function installSeededDb() { return { rows: [{ amm_price: String(avg) }] } } - if (sql.includes('MAX(timestamp) as last_trade')) { + if (sql.includes('as last_observed_at')) { const rows = rowsInWindow(network, pairKey, windowMinutes) const newest = rows.reduce((acc, r) => (!acc || r.ageSeconds < acc.ageSeconds ? r : acc), undefined) - return { rows: [{ last_trade: newest ? at(newest.ageSeconds) : null }] } + return { rows: [{ last_observed_at: newest ? at(newest.ageSeconds) : null }] } } if (sql.includes('COUNT(DISTINCT COALESCE(pool_id')) { diff --git a/src/aggregator/vwap.ts b/src/aggregator/vwap.ts index fb14793d..9820605f 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 } @@ -214,6 +227,7 @@ export interface AggregatedPriceNumeric { inputCount: number stalenessThresholdSeconds: number stale: boolean + excludedVenues: string[] } export interface AggregatedPriceString { @@ -235,60 +249,81 @@ export interface AggregatedPriceString { inputCount: number stalenessThresholdSeconds: number stale: boolean + excludedVenues: string[] } export async function getAggregatedPrice( pairKey: string, network: NetworkName, - options: { decimalFormat: 'string'; staleThresholdSeconds?: number } + options: { decimalFormat: 'string'; staleThresholdSeconds?: number }, + excludedSources?: string[] ): Promise export async function getAggregatedPrice( pairKey: string, network: NetworkName, - options?: { decimalFormat?: 'number'; staleThresholdSeconds?: number } + options?: { decimalFormat?: 'number'; staleThresholdSeconds?: number }, + excludedSources?: string[] ): Promise export async function getAggregatedPrice( pairKey: string, network: NetworkName, - options?: { decimalFormat?: DecimalFormat; staleThresholdSeconds?: number } + options?: { decimalFormat?: DecimalFormat; staleThresholdSeconds?: number }, + excludedSources?: string[] ): Promise export async function getAggregatedPrice( pairKey: string, network: NetworkName, - options?: { decimalFormat?: DecimalFormat; staleThresholdSeconds?: number } + options?: { decimalFormat?: DecimalFormat; staleThresholdSeconds?: number }, + 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 and input count - pgPool.query( - `SELECT MAX(timestamp) as last_trade, COUNT(*) as input_count 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] + ), + // Newest observation time + input count, excluding stale sources so the age + // reflects live data only. We measure against observed_at (the Lens-ingestion + // wall-clock, NOT NULL post-migration; COALESCE to the legacy source `timestamp` + // only for pre-migration rows) — this is a liveness clock, "when did we last + // ingest a sample", not the age of the most recent trade at its source. + excludedSources.length > 0 + ? pgPool.query( + `SELECT MAX(COALESCE(observed_at, timestamp)) as last_observed_at, COUNT(*) as input_count 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_observed_at, COUNT(*) as input_count FROM price_points + WHERE pair_key = $1 AND network = $2`, + [pairKey, network] + ), ]) const volResult = await pgPool.query( @@ -300,12 +335,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' @@ -335,21 +373,27 @@ 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') const ammPrice = parseFloat(ammPriceRaw) - const lastTrade = lastTradeResult.rows[0]?.last_trade - const lastTradeAgeSeconds = lastTrade ? Math.floor((Date.now() - new Date(lastTrade).getTime()) / 1000) : null + // Newest observation time (observed_at clock, see query above). lastTradeAgeSeconds + // keeps its public name but measures time since that newest observation — i.e. how + // long since we last ingested a live sample, which is what the stale flag enforces. + const lastObservedAt = lastTradeResult.rows[0]?.last_observed_at + const lastTradeAgeSeconds = lastObservedAt ? Math.floor((Date.now() - new Date(lastObservedAt).getTime()) / 1000) : null const sources = parseInt(sourcesResult.rows[0]?.sources ?? '0', 10) const rawInputCount = lastTradeResult.rows[0]?.input_count const inputCount = rawInputCount !== undefined ? parseInt(rawInputCount, 10) - : (lastTrade ? Math.max(1, sources) : 0) - const newestInputObservedAt = lastTrade ? new Date(lastTrade).toISOString() : null + : (lastObservedAt ? Math.max(1, sources) : 0) + const newestInputObservedAt = lastObservedAt ? new Date(lastObservedAt).toISOString() : null const staleThreshold = options?.staleThresholdSeconds ?? (config as any).pricing?.staleThresholdSeconds @@ -395,6 +439,7 @@ export async function getAggregatedPrice( inputCount, stalenessThresholdSeconds: staleThreshold, stale, + excludedVenues: excludedSources, } } @@ -417,6 +462,7 @@ export async function getAggregatedPrice( inputCount, stalenessThresholdSeconds: staleThreshold, stale, + excludedVenues: excludedSources, } } diff --git a/src/api/graphql.ts b/src/api/graphql.ts index 675a070f..5f7d4842 100644 --- a/src/api/graphql.ts +++ b/src/api/graphql.ts @@ -14,6 +14,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' @@ -149,6 +150,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 { @@ -601,7 +604,7 @@ export 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 90e3983a..9d237ff1 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, @@ -204,6 +205,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), } } ) @@ -257,7 +264,13 @@ export async function registerRESTRoutes(app: FastifyInstance) { } catch { /* fall through */ } } - const agg = await getAggregatedPrice(pair.pairKey, network, { decimalFormat, staleThresholdSeconds: staleThreshold }) + // 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, staleThresholdSeconds: staleThreshold }, 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 335fcfd3..e2b024d8 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 @@ -196,6 +213,7 @@ export const priceResponseSchema = { 'inputCount', 'stalenessThresholdSeconds', 'stale', + 'excludedVenues', 'bestRoute', 'lastUpdated', ], @@ -223,6 +241,9 @@ export const priceResponseSchema = { inputCount: { type: 'integer' }, stalenessThresholdSeconds: { type: 'integer' }, 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/decimalStrings.test.ts b/tests/decimalStrings.test.ts index c264fd18..96761a3d 100644 --- a/tests/decimalStrings.test.ts +++ b/tests/decimalStrings.test.ts @@ -97,8 +97,8 @@ describe('Issue #248: Decimal values cross API boundary as JSON numbers', () => if (sql.includes('AVG(spot_price::numeric) AS amm_price')) { return { rows: [{ amm_price: '0' }] } } - if (sql.includes('MAX(timestamp) as last_trade')) { - return { rows: [{ last_trade: new Date() }] } + if (sql.includes('as last_observed_at')) { + return { rows: [{ last_observed_at: new Date() }] } } if (sql.includes('GROUP BY source')) { return { rows: [{ source: 'SDEX', vol: '150.5' }] } @@ -133,8 +133,8 @@ describe('Issue #248: Decimal values cross API boundary as JSON numbers', () => if (sql.includes('AVG(spot_price::numeric) AS amm_price')) { return { rows: [{ amm_price: '0.000000000000000000' }] } } - if (sql.includes('MAX(timestamp) as last_trade')) { - return { rows: [{ last_trade: new Date() }] } + if (sql.includes('as last_observed_at')) { + return { rows: [{ last_observed_at: new Date() }] } } if (sql.includes('GROUP BY source')) { return { rows: [{ source: 'SDEX', vol: '500.123456789012345678' }] } @@ -176,8 +176,8 @@ describe('Issue #248: Decimal values cross API boundary as JSON numbers', () => if (sql.includes('AVG(spot_price::numeric) AS amm_price')) { return { rows: [{ amm_price: '0' }] } } - if (sql.includes('MAX(timestamp) as last_trade')) { - return { rows: [{ last_trade: new Date() }] } + if (sql.includes('as last_observed_at')) { + return { rows: [{ last_observed_at: new Date() }] } } if (sql.includes('GROUP BY source')) { return { rows: [{ source: 'SDEX', vol: '100' }] } @@ -210,8 +210,8 @@ describe('Issue #248: Decimal values cross API boundary as JSON numbers', () => if (sql.includes('AVG(spot_price::numeric) AS amm_price')) { return { rows: [{ amm_price: '0' }] } } - if (sql.includes('MAX(timestamp) as last_trade')) { - return { rows: [{ last_trade: new Date() }] } + if (sql.includes('as last_observed_at')) { + return { rows: [{ last_observed_at: new Date() }] } } if (sql.includes('GROUP BY source')) { return { rows: [{ source: 'SDEX', vol: '100' }] } @@ -241,8 +241,8 @@ describe('Issue #248: Decimal values cross API boundary as JSON numbers', () => if (sql.includes('AVG(spot_price::numeric) AS amm_price')) { return { rows: [{ amm_price: '0' }] } } - if (sql.includes('MAX(timestamp) as last_trade')) { - return { rows: [{ last_trade: new Date() }] } + if (sql.includes('as last_observed_at')) { + return { rows: [{ last_observed_at: new Date() }] } } if (sql.includes('GROUP BY source')) { return { rows: [{ source: 'SDEX', vol: '100' }] } @@ -380,8 +380,8 @@ describe('Issue #248: Decimal values cross API boundary as JSON numbers', () => if (sql.includes('AVG(spot_price::numeric) AS amm_price')) { return { rows: [{ amm_price: '0.000000000000000000' }] } } - if (sql.includes('MAX(timestamp) as last_trade')) { - return { rows: [{ last_trade: new Date() }] } + if (sql.includes('as last_observed_at')) { + return { rows: [{ last_observed_at: new Date() }] } } if (sql.includes('GROUP BY source')) { return { diff --git a/tests/staleness.test.ts b/tests/staleness.test.ts index 56fab492..8a04f63b 100644 --- a/tests/staleness.test.ts +++ b/tests/staleness.test.ts @@ -66,7 +66,7 @@ describe('stale price flag and freshness metadata', () => { mockQuery.mockImplementation(async (sql: string) => { if (sql.includes('AVG(spot_price::numeric)')) return { rows: [{ amm_price: '0.1' }] } - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: oldTrade, input_count: '5' }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: oldTrade, input_count: '5' }] } 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' }] } @@ -92,7 +92,7 @@ describe('stale price flag and freshness metadata', () => { mockQuery.mockImplementation(async (sql: string) => { if (sql.includes('AVG(spot_price::numeric)')) return { rows: [{ amm_price: '0.1' }] } - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: recentTrade, input_count: '3' }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: recentTrade, input_count: '3' }] } 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' }] } @@ -114,7 +114,7 @@ describe('stale price flag and freshness metadata', () => { it('returns stale=false when no trades exist', async () => { mockQuery.mockImplementation(async (sql: string) => { - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: null, input_count: '0' }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: null, input_count: '0' }] } return { rows: [] } }) @@ -141,7 +141,7 @@ describe('stale price flag and freshness metadata', () => { mockQuery.mockImplementation(async (sql: string) => { if (sql.includes('AVG(spot_price::numeric)')) return { rows: [{ amm_price: '0.1' }] } - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: tradeTime, input_count: '4' }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: tradeTime, input_count: '4' }] } 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' }] } @@ -184,7 +184,7 @@ describe('stale price flag and freshness metadata', () => { mockQuery.mockImplementation(async (sql: string) => { if (sql.includes('AVG(spot_price::numeric)')) return { rows: [{ amm_price: '0.1' }] } - if (sql.includes('MAX(timestamp) as last_trade')) return { rows: [{ last_trade: tradeTime, input_count: '2' }] } + if (sql.includes('as last_observed_at')) return { rows: [{ last_observed_at: tradeTime, input_count: '2' }] } 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' }] } diff --git a/tests/venueHeartbeat.test.ts b/tests/venueHeartbeat.test.ts new file mode 100644 index 00000000..79af888c --- /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('as last_observed_at')) return { rows: [{ last_observed_at: 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') + }) +})