feat(metrics): track venue ingestion health

src/ingest/venues/amm.ts

@@ -1,4 +1,4 @@
-import { amm_snapshots_total, trades_ingested_total, last_trade_timestamp } from '../../metrics'
+import { amm_snapshots_total, trades_ingested_total, last_trade_timestamp, markIngestSuccess } from '../../metrics'
 import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config'
 import { upsertPricePoints, getIndexerState, setIndexerCursor, prisma } from '../../db'
 import { resolvePageLedgers } from '../../ingesters/toid'
@@ -106,6 +106,7 @@ export async function snapshotPool(
         ],
         network
       )
+      markIngestSuccess('AMM', network)
 
       const trackerKey = pair.pairKey
       const { previousPrice } = setLastPrice('AMM', network, trackerKey, spotPrice)
@@ -192,6 +193,7 @@ export async function ingestPoolTrades(
     )
 
     await upsertPricePoints(points, network)
+    markIngestSuccess('AMM', network)
 
     trades_ingested_total.inc({ pair: pair.pairKey, network }, points.length)
     last_trade_timestamp.set(

src/ingest/venues/runner.ts

@@ -6,6 +6,7 @@ import { publishPriceUpdate } from '../../events'
 import { createLogger } from '../../logger'
 import type { WatchedPair } from '../../types'
 import type { VenueAdapter, VenuePricePoint } from './types'
+import { ingest_cycle_duration_seconds, ingest_errors_total, markIngestSuccess } from '../../metrics'
 
 const log = createLogger('ingester:runner')
 
@@ -52,6 +53,7 @@ export async function writeVenuePricePoints(
   if (points.length === 0) return
 
   await upsertPricePoints(points as any, network)
+  markIngestSuccess(venue, network)
 
   const latest = points[points.length - 1]
   const { previousPrice, currentPrice } = setLastPrice(venue, network, pair.pairKey, latest.price)
@@ -104,13 +106,26 @@ export async function startVenueAdapter(
 
   let consecutiveErrors = 0
   while (true) {
+    const cycleStartedAt = Date.now()
     try {
       const pairs = getActivePairs()
       for (const pair of pairs) {
         await adapter.ingestPair(pair, network)
       }
+      ingest_cycle_duration_seconds.observe(
+        { venue: adapter.name, network },
+        (Date.now() - cycleStartedAt) / 1000,
+      )
       consecutiveErrors = 0
     } catch (err) {
+      const error = err as Error
+      const category = /429|rate.?limit/i.test(error.message) ? 'rate_limit' :
+        /timeout|timed out|abort/i.test(error.message) ? 'timeout' : 'other'
+      ingest_errors_total.inc({ venue: adapter.name, network, category })
+      ingest_cycle_duration_seconds.observe(
+        { venue: adapter.name, network },
+        (Date.now() - cycleStartedAt) / 1000,
+      )
       consecutiveErrors++
       const backoffMs = Math.min(10_000 * Math.pow(2, consecutiveErrors - 1), 60_000)
       log.error(

src/ingest/venues/sdex.ts

@@ -1,5 +1,5 @@
 import { Asset } from '@stellar/stellar-sdk'
-import { trades_ingested_total, last_trade_timestamp } from '../../metrics'
+import { trades_ingested_total, last_trade_timestamp, markIngestSuccess } from '../../metrics'
 import { config, activeNetwork, type NetworkName } from '../../config'
 import { getHorizonServer } from '../../network/clients'
 import { upsertPricePoints, getIndexerState, setIndexerCursor } from '../../db'
@@ -135,6 +135,7 @@ export async function writeSdexPricePoints(
   const { previousPrice, currentPrice } = setLastPrice('SDEX', network, trackerKey, points[points.length - 1].price)
 
   await upsertPricePoints(points, network)
+  markIngestSuccess('SDEX', network)
 
   trades_ingested_total.inc({ pair: pair.pairKey, network }, points.length)
   last_trade_timestamp.set(

src/ingest/venues/soroswap.ts

@@ -22,6 +22,7 @@ import {
 import { activeNetwork, getNetworkConfig, type NetworkName } from '../../config'
 import { getRpcServer } from '../../network/clients'
 import { upsertPricePoints } from '../../db'
+import { markIngestSuccess } from '../../metrics'
 import { dispatchPriceUpdate } from '../../webhookDispatcher'
 import { publishPriceUpdate } from '../../events'
 import { createLogger } from '../../logger'
@@ -241,6 +242,7 @@ export async function ingestPool(
       ],
       network
     )
+    markIngestSuccess('Soroswap', network)
 
     const trackerKey = pair.pairKey
     const { previousPrice } = setLastPrice('Soroswap', network, trackerKey, spotPrice)

src/metrics.ts

@@ -85,6 +85,38 @@ export const last_trade_timestamp = new Gauge({
   registers: [register]
 })
 
+// Ingestion health is labelled only by the finite adapter registry and the
+// two supported networks. Do not add pair or pool identifiers here: they make
+// the series count grow with user configuration and on-chain pool creation.
+export const ingest_last_success_timestamp = new Gauge({
+  name: 'ingest_last_success_timestamp',
+  help: 'Unix timestamp of the last successful price write by venue and network',
+  labelNames: ['venue', 'network'],
+  registers: [register],
+})
+
+export const ingest_cycle_duration_seconds = new Histogram({
+  name: 'ingest_cycle_duration_seconds',
+  help: 'Duration of a venue ingestion cycle in seconds',
+  labelNames: ['venue', 'network'],
+  // Venue polls are mostly sub-second, but Horizon and Soroswap requests can
+  // take several seconds. These buckets expose normal cycles and slow calls.
+  buckets: [0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60],
+  registers: [register],
+})
+
+export const ingest_errors_total = new Counter({
+  name: 'ingest_errors_total',
+  help: 'Ingestion cycle errors by venue, network and bounded error category',
+  labelNames: ['venue', 'network', 'category'],
+  registers: [register],
+})
+
+/** Record a successful persisted price write without high-cardinality labels. */
+export function markIngestSuccess(venue: string, network: string): void {
+  ingest_last_success_timestamp.set({ venue, network }, Math.floor(Date.now() / 1000))
+}
+
 export const http_requests_total = new Counter({
   name: 'http_requests_total',
   help: 'Total number of HTTP requests served, by method, route template and status class',