feat: per-venue heartbeat with stale-price exclusion (#219)

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;
+                        }[];
                     };
                 };
             };

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

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"
+            }
           }
         }
       },

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

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 {

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

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

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,
   }
 }