fix(amm): retry Horizon rate limits

src/__tests__/ammIngester.test.ts

@@ -117,7 +117,10 @@ describe('AMM Ingester', () => {
     expect(record).not.toHaveProperty('ledger')
 
     global.fetch = vi.fn().mockResolvedValue({
-      json: async () => ({ _embedded: { records: [record] } })
+      ok: true,
+      status: 200,
+      headers: { get: () => null },
+      json: async () => ({ _embedded: { records: [record] } }),
     })
 
     // Deliberately a network other than the active one: the point of passing

src/__tests__/dualNetworkAlerts.test.ts

@@ -207,6 +207,9 @@ describe('Dual-network alerts and price tracking (#188)', () => {
       global.fetch = vi.fn().mockImplementation(async (url: string) => {
         if (url.includes('/trades')) {
           return {
+            ok: true,
+            status: 200,
+            headers: { get: () => null },
             json: async () => ({
               _embedded: {
                 records: [

src/ingest/venues/amm.ts

@@ -6,6 +6,7 @@ import { dispatchPriceUpdate } from '../../webhookDispatcher'
 import { publishPriceUpdate } from '../../events'
 import type { WatchedPair } from '../../types'
 import type { VenueAdapter, VenuePricePoint } from './types'
+import { fetchHorizonJson, HorizonRateLimitError } from './horizon'
 import {
   getLastPrice,
   resetLastPrice,
@@ -35,13 +36,14 @@ export async function fetchPools(pair: WatchedPair, network: NetworkName = activ
     params.append('reserves[]', assetBStr)
     params.set('limit', '10')
 
-    const response = await fetch(
-      `${getNetworkConfig(network).horizon.url}/liquidity_pools?${params.toString()}`
+    const data = await fetchHorizonJson<any>(
+      `${getNetworkConfig(network).horizon.url}/liquidity_pools?${params.toString()}`,
+      network,
     )
-    const data = (await response.json()) as any
     if (!data._embedded?.records) return []
     return data._embedded.records
   } catch (err) {
+    if (err instanceof HorizonRateLimitError) throw err
     console.error(`[amm] Failed to fetch pools for ${pair.pairKey}:`, (err as Error).message)
     return []
   }
@@ -136,10 +138,10 @@ export async function ingestPoolTrades(
   const cursor = state.cursor ?? '0'
 
   try {
-    const response = await fetch(
-      `${getNetworkConfig(network).horizon.url}/liquidity_pools/${pool.id}/trades?cursor=${cursor}&limit=${config.indexer.ammPageSize}&order=asc`
+    const data = await fetchHorizonJson<any>(
+      `${getNetworkConfig(network).horizon.url}/liquidity_pools/${pool.id}/trades?cursor=${cursor}&limit=${config.indexer.ammPageSize}&order=asc`,
+      network,
     )
-    const data = (await response.json()) as any
     const records = data._embedded?.records ?? []
 
     if (!records.length) return
@@ -213,6 +215,7 @@ export async function ingestPoolTrades(
       network,
     }).catch((err) => console.error('[amm] webhook dispatch error:', err.message))
   } catch (err) {
+    if (err instanceof HorizonRateLimitError) throw err
     console.error(`[amm] Trade ingest error for pool ${pool.id}:`, (err as Error).message)
   }
 }

src/ingest/venues/horizon.ts

@@ -0,0 +1,49 @@
+import { horizon_rate_limit_remaining } from '../../metrics'
+import type { NetworkName } from '../../config'
+
+export class HorizonRateLimitError extends Error {
+  readonly status = 429
+
+  constructor(url: string) {
+    super(`Horizon rate limit persisted after retries for ${new URL(url).pathname}`)
+    this.name = 'HorizonRateLimitError'
+  }
+}
+
+function retryDelay(response: Response, attempt: number): number {
+  const retryAfter = response.headers.get('retry-after')
+  if (retryAfter) {
+    const seconds = Number(retryAfter)
+    if (Number.isFinite(seconds) && seconds >= 0) return seconds * 1000
+    const date = Date.parse(retryAfter)
+    if (Number.isFinite(date)) return Math.max(0, date - Date.now())
+  }
+  const exponential = Math.min(1_000 * 2 ** attempt, 30_000)
+  return Math.round(exponential * (0.75 + Math.random() * 0.5))
+}
+
+/** Fetch and validate Horizon JSON, retrying 429s without treating them as empty data. */
+export async function fetchHorizonJson<T>(
+  url: string,
+  network: NetworkName,
+  attempts = 3,
+): Promise<T> {
+  for (let attempt = 0; attempt < attempts; attempt++) {
+    const response = await fetch(url)
+    const remaining = response.headers.get('x-ratelimit-remaining')
+    if (remaining !== null && Number.isFinite(Number(remaining))) {
+      horizon_rate_limit_remaining.set({ network }, Number(remaining))
+    }
+
+    if (response.status === 429) {
+      if (attempt + 1 === attempts) throw new HorizonRateLimitError(url)
+      await new Promise((resolve) => setTimeout(resolve, retryDelay(response, attempt)))
+      continue
+    }
+    if (!response.ok) {
+      throw new Error(`Horizon request failed with HTTP ${response.status} for ${new URL(url).pathname}`)
+    }
+    return await response.json() as T
+  }
+  throw new HorizonRateLimitError(url)
+}

src/metrics.ts

@@ -59,6 +59,13 @@ export const last_trade_timestamp = new Gauge({
   registers: [register]
 })
 
+export const horizon_rate_limit_remaining = new Gauge({
+  name: 'horizon_rate_limit_remaining',
+  help: 'Remaining Horizon API requests in the current rate-limit window',
+  labelNames: ['network'],
+  registers: [register],
+})
+
 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',

tests/ingestMetrics.test.ts

@@ -188,6 +188,9 @@ describe('last_trade_timestamp', () => {
 
   it('tracks the AMM loop per network too', async () => {
     global.fetch = vi.fn().mockResolvedValue({
+      ok: true,
+      status: 200,
+      headers: { get: () => null },
       json: async () => ({
         _embedded: {
           records: [