From 46b261176ca1e11956f5d3cb562c404868b0cbd8 Mon Sep 17 00:00:00 2001
From: abdulrahmanj383-creator
 <292028682+abdulrahmanj383-creator@users.noreply.github.com>
Date: Sat, 17 Oct 2026 01:54:38 +0100
Subject: [PATCH 1/3] fix(amm): retry Horizon rate limits

---
 src/ingest/venues/amm.ts     | 15 ++++++-----
 src/ingest/venues/horizon.ts | 49 ++++++++++++++++++++++++++++++++++++
 src/metrics.ts               |  7 ++++++
 3 files changed, 65 insertions(+), 6 deletions(-)
 create mode 100644 src/ingest/venues/horizon.ts

diff --git a/src/ingest/venues/amm.ts b/src/ingest/venues/amm.ts
index 4c825a7f..81666cea 100644
--- a/src/ingest/venues/amm.ts
+++ b/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)
   }
 }
diff --git a/src/ingest/venues/horizon.ts b/src/ingest/venues/horizon.ts
new file mode 100644
index 00000000..a36b66dd
--- /dev/null
+++ b/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)
+}
diff --git a/src/metrics.ts b/src/metrics.ts
index 39a3a505..77380530 100644
--- a/src/metrics.ts
+++ b/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',

From 383d0fb63c03dbf56e97ff8a10502b9b50ba88bd Mon Sep 17 00:00:00 2001
From: abdulrahmanj383-creator
 <292028682+abdulrahmanj383-creator@users.noreply.github.com>
Date: Sat, 17 Oct 2026 07:46:11 +0100
Subject: [PATCH 2/3] test(amm): return successful Horizon responses in mocks

---
 src/__tests__/ammIngester.test.ts       | 4 +++-
 src/__tests__/dualNetworkAlerts.test.ts | 2 ++
 tests/ingestMetrics.test.ts             | 2 ++
 3 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/src/__tests__/ammIngester.test.ts b/src/__tests__/ammIngester.test.ts
index fe41051d..1dc14e06 100644
--- a/src/__tests__/ammIngester.test.ts
+++ b/src/__tests__/ammIngester.test.ts
@@ -117,7 +117,9 @@ describe('AMM Ingester', () => {
     expect(record).not.toHaveProperty('ledger')
 
     global.fetch = vi.fn().mockResolvedValue({
-      json: async () => ({ _embedded: { records: [record] } })
+      ok: true,
+      status: 200,
+      json: async () => ({ _embedded: { records: [record] } }),
     })
 
     // Deliberately a network other than the active one: the point of passing
diff --git a/src/__tests__/dualNetworkAlerts.test.ts b/src/__tests__/dualNetworkAlerts.test.ts
index 2729c0d7..ccc867d3 100644
--- a/src/__tests__/dualNetworkAlerts.test.ts
+++ b/src/__tests__/dualNetworkAlerts.test.ts
@@ -207,6 +207,8 @@ 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,
             json: async () => ({
               _embedded: {
                 records: [
diff --git a/tests/ingestMetrics.test.ts b/tests/ingestMetrics.test.ts
index b80e0d94..579f28e9 100644
--- a/tests/ingestMetrics.test.ts
+++ b/tests/ingestMetrics.test.ts
@@ -188,6 +188,8 @@ describe('last_trade_timestamp', () => {
 
   it('tracks the AMM loop per network too', async () => {
     global.fetch = vi.fn().mockResolvedValue({
+      ok: true,
+      status: 200,
       json: async () => ({
         _embedded: {
           records: [

From 183ba62180bd5f45731e143aa281cb0f608a5431 Mon Sep 17 00:00:00 2001
From: abdulrahmanj383-creator
 <292028682+abdulrahmanj383-creator@users.noreply.github.com>
Date: Sat, 17 Oct 2026 07:51:27 +0100
Subject: [PATCH 3/3] test(amm): include response headers in Horizon mocks

---
 src/__tests__/ammIngester.test.ts       | 1 +
 src/__tests__/dualNetworkAlerts.test.ts | 1 +
 tests/ingestMetrics.test.ts             | 1 +
 3 files changed, 3 insertions(+)

diff --git a/src/__tests__/ammIngester.test.ts b/src/__tests__/ammIngester.test.ts
index 1dc14e06..22ca7cd9 100644
--- a/src/__tests__/ammIngester.test.ts
+++ b/src/__tests__/ammIngester.test.ts
@@ -119,6 +119,7 @@ describe('AMM Ingester', () => {
     global.fetch = vi.fn().mockResolvedValue({
       ok: true,
       status: 200,
+      headers: { get: () => null },
       json: async () => ({ _embedded: { records: [record] } }),
     })
 
diff --git a/src/__tests__/dualNetworkAlerts.test.ts b/src/__tests__/dualNetworkAlerts.test.ts
index ccc867d3..34977b1e 100644
--- a/src/__tests__/dualNetworkAlerts.test.ts
+++ b/src/__tests__/dualNetworkAlerts.test.ts
@@ -209,6 +209,7 @@ describe('Dual-network alerts and price tracking (#188)', () => {
           return {
             ok: true,
             status: 200,
+            headers: { get: () => null },
             json: async () => ({
               _embedded: {
                 records: [
diff --git a/tests/ingestMetrics.test.ts b/tests/ingestMetrics.test.ts
index 579f28e9..a94d69e4 100644
--- a/tests/ingestMetrics.test.ts
+++ b/tests/ingestMetrics.test.ts
@@ -190,6 +190,7 @@ describe('last_trade_timestamp', () => {
     global.fetch = vi.fn().mockResolvedValue({
       ok: true,
       status: 200,
+      headers: { get: () => null },
       json: async () => ({
         _embedded: {
           records: [