feat(ingest): bound every ingester cycle with a deadline, abandoned-cycle metric, and overlap-proof abandonment (#351)

.changeset/ingest-cycle-deadline.md

@@ -0,0 +1,5 @@
+---
+"lens": minor
+---
+
+Bound every ingester cycle with a deadline so one slow upstream can no longer stall an ingester while every health signal still reports it as healthy. Each round of a venue loop (SDEX, AMM, Soroswap, Aquarius) or the snapshot appender now runs under `runCycleWithDeadline`: a cycle over its deadline is aborted at its next unit checkpoint, counted in the new `ingester_abandoned_cycles_total` metric (`venue`, `network` labels) and logged, and the following cycle starts only after the abandoned one has unwound — two cycles against the same venue can never overlap, and because every checkpoint sits *before* a write unit rather than inside one, an abandoned cycle can never leave a partial write. The per-ingester deadline is configurable via `INGEST_CYCLE_DEADLINE_MS` with `<VENUE>`/`<NETWORK>` overrides (0 disables it), venue adapters receive the cycle `AbortSignal` as an optional third `ingestPair` argument, and `startVenueAdapter` accepts an optional `stopSignal` so a loop can be stopped cleanly at a cycle boundary.

.env.example

@@ -107,6 +107,16 @@ SDEX_PAGE_SIZE=200
 AMM_PAGE_SIZE=200
 # How often to poll Soroswap pool reserves in milliseconds (default: 60000 = 60s)
 SOROSWAP_POLL_INTERVAL_MS=60000
+# Per-ingester cycle deadline in milliseconds (default: 60000 = 60s). A poll
+# cycle that overruns it is aborted and counted in the
+# ingester_abandoned_cycles_total metric instead of stalling the ingester loop.
+# Set 0 to disable the deadline. Precedence, most specific first:
+#   INGEST_CYCLE_DEADLINE_MS_<VENUE>_<NETWORK>  e.g. INGEST_CYCLE_DEADLINE_MS_SOROSWAP_MAINNET
+#   INGEST_CYCLE_DEADLINE_MS_<VENUE>            venues: SDEX, AMM, Soroswap, Aquarius, Snapshot
+#   INGEST_CYCLE_DEADLINE_MS_<NETWORK>, then INGEST_CYCLE_DEADLINE_MS
+INGEST_CYCLE_DEADLINE_MS=60000
+INGEST_CYCLE_DEADLINE_MS_TESTNET=60000
+INGEST_CYCLE_DEADLINE_MS_MAINNET=60000
 
 # --- Auth & Security ---
 # Secret key required for admin endpoints (e.g., adding/removing pairs)

docs/ingest-metrics.md

@@ -14,17 +14,19 @@ nonsense on any deployment with more than one.**
 | `trades_ingested_total` | Counter | `pair`, `network` |
 | `amm_snapshots_total` | Counter | `pool`, `network` |
 | `price_snapshots_total` | Counter | `network` |
+| `ingester_abandoned_cycles_total` | Counter | `venue`, `network` |
 | `last_trade_timestamp` | Gauge | `pair`, `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. |
 
-All four are also emitted with the registry-wide `app="lens"` default label.
+All five are also emitted with the registry-wide `app="lens"` default label.
 
-## Why `network` is on all four
+## Why `network` is on all five
 
 Before this change, the ingest metrics were network-blind. On a dual-network
 deployment the two loops wrote to the same series.
@@ -63,6 +65,10 @@ Series count is `pairs x networks`:
 - `amm_snapshots_total` — `pools x networks`. Pools are discovered from Horizon
   for the configured pairs only, so this stays bounded by the same watch list.
 - `price_snapshots_total` — `networks`. One series per network, full stop.
+- `ingester_abandoned_cycles_total` — `venues x networks`. `venue` is the five
+  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.
 
 **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
@@ -96,6 +102,9 @@ time() - last_trade_timestamp{pair="XLM/USDC", network="mainnet"}
 
 # Pairs that have gone quiet in the last 15 minutes
 time() - last_trade_timestamp > 900
+
+# Which ingester loop is overrunning its cycle deadline, on which network
+sum(rate(ingester_abandoned_cycles_total[15m])) by (venue, network)
 ```
 
 ## Suggested alerts

src/__tests__/ingestCycleDeadline.test.ts

@@ -0,0 +1,258 @@
+/**
+ * Issue #351 — per-cycle deadline acceptance tests.
+ *
+ * The headline case is the one the issue exists for: a slow upstream must not
+ * stall an ingester loop. The loop keeps ticking, cycles that overrun their
+ * deadline are abandoned and counted (`venue`, `network`), no two cycles
+ * against the same venue ever overlap, and an abandoned cycle never gets to
+ * start a write it would leave half-done.
+ */
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  upsertPricePoints: vi.fn(),
+  getIndexerState: vi.fn(),
+  setIndexerCursor: vi.fn(),
+  getActivePairs: vi.fn(),
+  dispatchPriceUpdate: vi.fn(),
+  publishPriceUpdate: vi.fn(),
+  sdexCall: vi.fn(),
+}))
+
+vi.mock('../db', () => ({
+  upsertPricePoints: mocks.upsertPricePoints,
+  getIndexerState: mocks.getIndexerState,
+  setIndexerCursor: mocks.setIndexerCursor,
+  prisma: { poolSnapshot: { create: vi.fn() } },
+  pgPool: { query: vi.fn() },
+}))
+vi.mock('../pairsRegistry', () => ({ getActivePairs: mocks.getActivePairs }))
+vi.mock('../webhookDispatcher', () => ({ dispatchPriceUpdate: mocks.dispatchPriceUpdate }))
+vi.mock('../events', () => ({ publishPriceUpdate: mocks.publishPriceUpdate }))
+vi.mock('../network/clients', () => {
+  const server: any = {
+    trades: () => server,
+    forAssetPair: () => server,
+    cursor: () => server,
+    limit: () => server,
+    order: () => server,
+    call: () => mocks.sdexCall(),
+  }
+  return { getHorizonServer: () => server, getRpcServer: () => ({}) }
+})
+
+import {
+  runCycleWithDeadline,
+  getCycleDeadlineMs,
+  DEFAULT_CYCLE_DEADLINE_MS,
+} from '../ingest/cycleDeadline'
+import { startVenueAdapter, type VenueAdapter } from '../ingest/venues/runner'
+import { ingestPair as sdexIngestPair } from '../ingest/venues/sdex'
+import { ingester_abandoned_cycles_total } from '../metrics'
+
+// Full issuer — never elide an address in a fixture (#351 house rule).
+const USDC_ISSUER = 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN'
+
+const pair = {
+  pairKey: 'XLM/USDC',
+  assetA: { code: 'XLM', issuer: null },
+  assetB: { code: 'USDC', issuer: USDC_ISSUER },
+}
+
+/** One Horizon SDEX trade record; synthetic paging tokens fall back to the stored ledger. */
+function trade(iso: string, id: string) {
+  return {
+    id,
+    paging_token: `p-${id}`,
+    base_asset_type: 'native',
+    base_asset_code: 'XLM',
+    base_amount: '100.0',
+    counter_amount: '20.0',
+    price: { n: 1, d: 5 },
+    ledger_close_time: iso,
+  }
+}
+
+/** Sum of all samples on the abandoned-cycle counter (0 if it never fired). */
+async function abandonedTotal(): Promise<number> {
+  const metric = await ingester_abandoned_cycles_total.get()
+  return metric.values.reduce((sum, v) => sum + v.value, 0)
+}
+
+async function waitFor(condition: () => boolean, timeoutMs: number): Promise<void> {
+  const start = Date.now()
+  while (!condition()) {
+    if (Date.now() - start > timeoutMs) {
+      throw new Error(`condition not met within ${timeoutMs}ms`)
+    }
+    await new Promise((resolve) => setTimeout(resolve, 10))
+  }
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  ingester_abandoned_cycles_total.reset()
+  mocks.getIndexerState.mockResolvedValue({ cursor: '0', ledger: 64702061 })
+  mocks.getActivePairs.mockReturnValue([pair])
+  mocks.upsertPricePoints.mockResolvedValue(1)
+  mocks.setIndexerCursor.mockResolvedValue(undefined)
+  mocks.dispatchPriceUpdate.mockResolvedValue(undefined)
+})
+
+describe('slow upstream (#351 acceptance)', () => {
+  it('keeps the loop making progress when an upstream is slower than the cycle deadline', async () => {
+    // Per-ingester deadline through the real resolution path: venue key
+    // SLOWDEX (adapter name SlowDEX), testnet network.
+    process.env.INGEST_CYCLE_DEADLINE_MS_SLOWDEX = '50'
+
+    const stop = new AbortController()
+    let started = 0
+    let active = 0
+    let maxActive = 0
+
+    const slowAdapter: VenueAdapter = {
+      name: 'SlowDEX',
+      source: 'slow_dex',
+      isEnabled: () => true,
+      getPollIntervalMs: () => 10,
+      discover: async () => [],
+      read: async () => null,
+      derive: () => [],
+      write: async () => {},
+      ingestPair: async () => {
+        started++
+        active++
+        maxActive = Math.max(maxActive, active)
+        await new Promise((resolve) => setTimeout(resolve, 250))
+        active--
+      },
+    }
+
+    const loop = startVenueAdapter(slowAdapter, 'testnet', { stopSignal: stop.signal })
+
+    // Cycle 1 abandons at 50ms but the loop must resume and start cycle 2,
+    // then cycle 3: progress with an upstream that is 5x over its deadline.
+    await waitFor(() => started >= 3, 5_000)
+
+    stop.abort()
+    await loop
+
+    expect(started).toBeGreaterThanOrEqual(3)
+    // Abandonment must be strictly sequential — never two at once.
+    expect(maxActive).toBe(1)
+    expect(await abandonedTotal()).toBeGreaterThanOrEqual(2)
+
+    const metric = await ingester_abandoned_cycles_total.get()
+    expect(metric.values.length).toBeGreaterThan(0)
+    for (const v of metric.values) {
+      // The labels #351 asks for: which loop, on which network.
+      expect(v.labels).toMatchObject({ venue: 'SlowDEX', network: 'testnet' })
+    }
+  }, 15_000)
+
+  it('never lets an abandoned cycle start a write, and never splits a unit that did start', async () => {
+    // (a) The round trip is still in flight when the deadline fires. When it
+    // finally returns, the write checkpoint must skip the whole unit: no
+    // upsert, no cursor advance — nothing partial, nothing post-abandon.
+    mocks.sdexCall.mockImplementation(
+      () =>
+        new Promise((resolve) =>
+          setTimeout(() => resolve({ records: [trade('2024-06-01T00:00:00Z', 't-1')] }), 200)
+        )
+    )
+
+    const slow = await runCycleWithDeadline({
+      venue: 'SDEX',
+      network: 'testnet',
+      deadlineMs: 50,
+      cycle: (signal) => sdexIngestPair(pair as never, 'testnet', signal),
+    })
+
+    expect(slow.status).toBe('abandoned')
+    expect(await abandonedTotal()).toBe(1)
+    // The unit's write sequence commits upsert + cursor together or not at all.
+    expect(mocks.upsertPricePoints).not.toHaveBeenCalled()
+    expect(mocks.setIndexerCursor).not.toHaveBeenCalled()
+
+    // (b) A unit that starts before its deadline commits both statements —
+    // cursor advance included, never one without the other.
+    mocks.sdexCall.mockResolvedValue({ records: [trade('2024-06-01T00:00:00Z', 't-2')] })
+
+    const fast = await runCycleWithDeadline({
+      venue: 'SDEX',
+      network: 'testnet',
+      deadlineMs: 5_000,
+      cycle: (signal) => sdexIngestPair(pair as never, 'testnet', signal),
+    })
+
+    expect(fast.status).toBe('completed')
+    expect(mocks.upsertPricePoints).toHaveBeenCalledTimes(1)
+    expect(mocks.setIndexerCursor).toHaveBeenCalledTimes(1)
+    expect(await abandonedTotal()).toBe(1) // (b) must not count as abandoned
+  }, 10_000)
+})
+
+describe('getCycleDeadlineMs — configurable per ingester (#351)', () => {
+  it('falls back to the default with no env set', () => {
+    expect(getCycleDeadlineMs('SDEX', 'testnet')).toBe(DEFAULT_CYCLE_DEADLINE_MS)
+  })
+
+  it('resolves venue, network and global overrides with documented precedence', () => {
+    process.env.INGEST_CYCLE_DEADLINE_MS = '1000'
+    expect(getCycleDeadlineMs('SDEX', 'testnet')).toBe(1000)
+
+    process.env.INGEST_CYCLE_DEADLINE_MS_TESTNET = '2000'
+    expect(getCycleDeadlineMs('SDEX', 'testnet')).toBe(2000)
+    // The other network still sees the global value.
+    expect(getCycleDeadlineMs('SDEX', 'mainnet')).toBe(1000)
+
+    process.env.INGEST_CYCLE_DEADLINE_MS_SDEX = '3000'
+    expect(getCycleDeadlineMs('SDEX', 'testnet')).toBe(3000)
+
+    process.env.INGEST_CYCLE_DEADLINE_MS_SDEX_TESTNET = '4000'
+    expect(getCycleDeadlineMs('SDEX', 'testnet')).toBe(4000)
+  })
+
+  it('folds adapter names into env-safe keys', () => {
+    process.env.INGEST_CYCLE_DEADLINE_MS_AQUARIUS = '777'
+    expect(getCycleDeadlineMs('Aquarius', 'mainnet')).toBe(777)
+  })
+
+  it('warns and falls through on an invalid value instead of disabling the deadline', () => {
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
+    process.env.INGEST_CYCLE_DEADLINE_MS = 'soon'
+    expect(getCycleDeadlineMs('SDEX', 'testnet')).toBe(DEFAULT_CYCLE_DEADLINE_MS)
+    expect(warn).toHaveBeenCalledTimes(1)
+    warn.mockRestore()
+  })
+
+  it('treats 0 as an explicit opt-out that runs the cycle unbounded and uncounted', async () => {
+    const result = await runCycleWithDeadline({
+      venue: 'SDEX',
+      network: 'testnet',
+      deadlineMs: 0,
+      cycle: async () => {
+        await new Promise((resolve) => setTimeout(resolve, 120))
+      },
+    })
+    expect(result.status).toBe('completed')
+    expect(await abandonedTotal()).toBe(0)
+  })
+})
+
+describe('error propagation', () => {
+  it('rethrows cycle errors untouched so the runner keeps its backoff', async () => {
+    await expect(
+      runCycleWithDeadline({
+        venue: 'AMM',
+        network: 'testnet',
+        deadlineMs: 5_000,
+        cycle: async () => {
+          throw new Error('upstream exploded')
+        },
+      })
+    ).rejects.toThrow('upstream exploded')
+    // A failing cycle is a failure, not an abandonment — no metric, no backoff change.
+    expect(await abandonedTotal()).toBe(0)
+  })
+})

src/__tests__/soroswapIngester.test.ts

@@ -100,7 +100,14 @@ describe('ingestPool', () => {
   }
 
   beforeEach(() => {
+    // `clearAllMocks` clears calls but leaves implementations in place, so the
+    // `mockRejectedValue` set by the "does not throw when upsertPricePoints
+    // rejects" case below would otherwise survive into whichever test runs
+    // next. Under a shuffled run that poisons this test (upsertPricePoints
+    // rejects, ingestPool bails early, dispatchPriceUpdate is never called),
+    // so re-establish the default resolution every time.
     vi.clearAllMocks()
+    ;(upsertPricePoints as ReturnType<typeof vi.fn>).mockResolvedValue(undefined)
   })
 
   it('stores a price point when reserves are non-zero', async () => {