diff --git a/.changeset/ingest-cycle-deadline.md b/.changeset/ingest-cycle-deadline.md new file mode 100644 index 00000000..75c21c16 --- /dev/null +++ b/.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 ``/`` 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. \ No newline at end of file diff --git a/.env.example b/.env.example index 41e4465f..be85820d 100644 --- a/.env.example +++ b/.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__ e.g. INGEST_CYCLE_DEADLINE_MS_SOROSWAP_MAINNET +# INGEST_CYCLE_DEADLINE_MS_ venues: SDEX, AMM, Soroswap, Aquarius, Snapshot +# INGEST_CYCLE_DEADLINE_MS_, 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) diff --git a/docs/ingest-metrics.md b/docs/ingest-metrics.md index ed7fc973..5ec5eedf 100644 --- a/docs/ingest-metrics.md +++ b/docs/ingest-metrics.md @@ -14,6 +14,7 @@ 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 | @@ -21,10 +22,11 @@ nonsense on any deployment with more than one.** | `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 diff --git a/src/__tests__/ingestCycleDeadline.test.ts b/src/__tests__/ingestCycleDeadline.test.ts new file mode 100644 index 00000000..08fa3e0a --- /dev/null +++ b/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 { + 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 { + 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) + }) +}) \ No newline at end of file diff --git a/src/__tests__/soroswapIngester.test.ts b/src/__tests__/soroswapIngester.test.ts index 9ae7358b..b7435fff 100644 --- a/src/__tests__/soroswapIngester.test.ts +++ b/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).mockResolvedValue(undefined) }) it('stores a price point when reserves are non-zero', async () => { diff --git a/src/ingest/cycleDeadline.ts b/src/ingest/cycleDeadline.ts new file mode 100644 index 00000000..d44a14e0 --- /dev/null +++ b/src/ingest/cycleDeadline.ts @@ -0,0 +1,226 @@ +/** + * Per-cycle deadline for ingester loops. Issue: #351. + * + * Every ingester is an infinite loop (`startVenueAdapter`, the snapshot + * ingester) restarted on crash by `restartIngester`. Each pass of such a loop + * is one *cycle*: for a venue adapter that is one round over all active pairs, + * for the snapshot ingester one minute-append. Before this module a single + * slow upstream call could stretch a cycle arbitrarily long. The loop has not + * "crashed", so nothing restarts it, and every monitoring signal the service + * has still reports the ingester as healthy while it silently stops producing. + * + * This module bounds one cycle: + * + * 1. The cycle races a timer of `deadlineMs`. When the timer wins, the cycle's + * `AbortSignal` fires and the cycle is **abandoned**: counted in + * `ingester_abandoned_cycles_total` (labels: `venue`, `network`) and logged. + * It is abandoned, not awaited indefinitely — the abandoned cycle stops at + * its next checkpoint instead of grinding through every remaining pair or + * pool, so the loop resumes after teardown rather than after the slow work. + * 2. Cycles stay strictly sequential. The abandoned cycle is always awaited + * until it *unwinds* before the next one may start, so abandoning a cycle + * never starts a second concurrent cycle against the same venue. + * 3. Checkpoints are placed **before a unit's write sequence starts, never in + * the middle of one** — that is the "check how each ingester writes before + * changing the control flow" part of #351. Writes in Lens are already + * atomic units: `upsertPricePoints` is one `createMany(skipDuplicates)`, the + * SDEX/AMM cursor advance rides together with its page upsert, the snapshot + * append is one `INSERT … ON CONFLICT … DO NOTHING`. So an abandoned cycle + * either has not started a unit — the unit is skipped and its data re-read + * next cycle, where `skipDuplicates` makes the replay idempotent — or it + * runs the unit whole. A partial write cannot survive abandonment because + * no checkpoint sits between statements a unit commits together. + */ + +import type { NetworkName } from '../config' +import { ingester_abandoned_cycles_total } from '../metrics' + +/** + * Default cycle deadline (ms) when no environment override is set. + * + * 60s: a healthy cycle is sub-second (pairs come from an in-memory registry, + * upstream calls are individually bounded by `fetchWithTimeout`'s default + * timeout), so a minute means something is genuinely stuck, while staying + * generous enough that a cold Horizon/RPC can never trip it under normal load. + */ +export const DEFAULT_CYCLE_DEADLINE_MS = 60_000 + +/** + * After an abandoned cycle is aborted, how long its unwind may take before we + * escalate to an error log. The loop still never starts an overlapping cycle — + * it waits — but a cycle that fails to unwind within this window means one + * in-flight call ignored its own timeout, and that must be visible in logs + * rather than an ingester that quietly stopped ticking. + */ +export const CYCLE_UNWIND_GRACE_MS = 10_000 + +/** The abort reason attached to a cycle's signal when its deadline fires. */ +export class CycleDeadlineExceeded extends Error { + constructor(venue: string, network: NetworkName, deadlineMs: number) { + super(`${venue}/${network} cycle exceeded its ${deadlineMs}ms deadline`) + this.name = 'CycleDeadlineExceeded' + } +} + +export type CycleStatus = 'completed' | 'abandoned' + +export interface CycleResult { + status: CycleStatus + /** Wall time from cycle start to the outcome being decided. */ + durationMs: number +} + +type CycleSettled = + | { outcome: 'fulfilled' } + | { outcome: 'rejected'; error: unknown } + +/** + * Parse one deadline environment value. + * + * Returns `undefined` for unset/empty so the next candidate in the precedence + * chain is tried. An unparseable value never silently disables the deadline — + * it is warned about and skipped, so the fallback still bounds the cycle. + * `0` is valid and means "deadline disabled": an explicit operator decision, + * unlike a typo, which must not become one. + */ +function readDeadlineMs(raw: string | undefined, envKey: string): number | undefined { + if (raw === undefined || raw.trim() === '') return undefined + const parsed = Number(raw) + if (!Number.isFinite(parsed) || parsed < 0) { + console.warn( + `[cycle-deadline] ${envKey}="${raw}" is not a non-negative number of milliseconds; ignoring it` + ) + return undefined + } + return parsed +} + +/** + * Resolve the cycle deadline for one ingester ("venue" means the loop + * identity here: `SDEX`, `AMM`, `Soroswap`, `Aquarius`, `Snapshot`). + * + * Precedence, most specific first — mirroring the paired-var convention used + * everywhere else in `src/config.ts`: + * + * 1. `INGEST_CYCLE_DEADLINE_MS__` e.g. `INGEST_CYCLE_DEADLINE_MS_SOROSWAP_MAINNET` + * 2. `INGEST_CYCLE_DEADLINE_MS_` e.g. `INGEST_CYCLE_DEADLINE_MS_SOROSWAP` + * 3. `INGEST_CYCLE_DEADLINE_MS_` `INGEST_CYCLE_DEADLINE_MS_TESTNET` / `INGEST_CYCLE_DEADLINE_MS_MAINNET` + * 4. `INGEST_CYCLE_DEADLINE_MS` + * 5. `DEFAULT_CYCLE_DEADLINE_MS` + * + * Venue names are upper-cased with non-alphanumeric characters folded to `_`, + * so `Aquarius` → `AQUARIUS` and any future adapter follows the same rule. + * `.env.example` documents levels 1, 3 and 4 — level 1 as the documented + * pattern, because the venue dimension is bounded by the adapter registry but + * cannot be enumerated by the env drift guard the way per-network suffixes can. + */ +export function getCycleDeadlineMs(venue: string, network: NetworkName): number { + const venueKey = venue.toUpperCase().replace(/[^A-Z0-9]+/g, '_') + const suffix = network.toUpperCase() as 'TESTNET' | 'MAINNET' + + const venueNetworkKey = `INGEST_CYCLE_DEADLINE_MS_${venueKey}_${suffix}` + const venueKeyEnv = `INGEST_CYCLE_DEADLINE_MS_${venueKey}` + const networkKey = `INGEST_CYCLE_DEADLINE_MS_${suffix}` + + const resolved = + readDeadlineMs(process.env[venueNetworkKey], venueNetworkKey) ?? + readDeadlineMs(process.env[venueKeyEnv], venueKeyEnv) ?? + readDeadlineMs(process.env[networkKey], networkKey) ?? + readDeadlineMs(process.env.INGEST_CYCLE_DEADLINE_MS, 'INGEST_CYCLE_DEADLINE_MS') ?? + DEFAULT_CYCLE_DEADLINE_MS + + return resolved +} + +/** + * Run exactly one ingester cycle under `deadlineMs` (defaulting to + * `getCycleDeadlineMs(venue, network)`). + * + * - finishes in time → resolves `{ status: 'completed' }`; a cycle error is + * rethrown untouched so the caller's existing backoff keeps working. + * - exceeds it → the cycle's signal aborts, the abandonment is counted and + * logged, the cycle is awaited only until it unwinds, then resolves + * `{ status: 'abandoned' }`. + * - `deadlineMs <= 0` → deadline disabled: plain await, no timer. + * + * Never resolves while the previous cycle's function is still running, which + * is what keeps two cycles against the same venue from ever overlapping. + */ +export async function runCycleWithDeadline(opts: { + venue: string + network: NetworkName + /** Override the resolved environment deadline (tests / special callers). */ + deadlineMs?: number + cycle: (signal: AbortSignal) => Promise +}): Promise { + const { venue, network, cycle } = opts + const deadlineMs = opts.deadlineMs ?? getCycleDeadlineMs(venue, network) + const startedAt = Date.now() + + if (deadlineMs <= 0) { + // Deadline disabled by configuration. Hand the cycle a signal anyway so + // adapter code paths are identical with and without the deadline. + await cycle(new AbortController().signal) + return { status: 'completed', durationMs: Date.now() - startedAt } + } + + const controller = new AbortController() + const deadlineError = new CycleDeadlineExceeded(venue, network, deadlineMs) + + let timer: NodeJS.Timeout | undefined + const deadlineFired = new Promise<'deadline'>((resolve) => { + timer = setTimeout(() => { + controller.abort(deadlineError) + resolve('deadline') + }, deadlineMs) + }) + + // Bridge the cycle into a promise that is always handled: when the deadline + // wins the race, a later rejection from the abandoned cycle must not become + // an unhandled rejection — it is awaited (and discarded) below instead. + const cycleSettled: Promise = cycle(controller.signal).then( + () => ({ outcome: 'fulfilled' as const }), + (error) => ({ outcome: 'rejected' as const, error }) + ) + + try { + const first = await Promise.race([cycleSettled, deadlineFired]) + + if (first === 'deadline') { + ingester_abandoned_cycles_total.inc({ venue, network }) + console.error( + `[cycle-deadline] ${venue}/${network} cycle abandoned after exceeding its ${deadlineMs}ms deadline ` + + `(elapsed ${Date.now() - startedAt}ms)` + ) + + // Start the next cycle only once this one has unwound — overlap must + // remain impossible. This awaits teardown, not the abandoned work: the + // signal has already stopped every remaining unit at its checkpoint, so + // what is left is the single in-flight call the current unit sits on + // (bounded by that call's own timeout). If even that does not settle + // within the grace window we refuse to overlap anyway — and say so. + const graceTimer = setTimeout(() => { + console.error( + `[cycle-deadline] ${venue}/${network} abandoned cycle still has not unwound ` + + `${CYCLE_UNWIND_GRACE_MS}ms after its abort — holding the loop open rather than overlapping it; ` + + `an in-flight upstream call is ignoring its own timeout` + ) + }, CYCLE_UNWIND_GRACE_MS) + + try { + await cycleSettled + } finally { + clearTimeout(graceTimer) + } + + return { status: 'abandoned', durationMs: Date.now() - startedAt } + } + + clearTimeout(timer) + + if (first.outcome === 'rejected') throw first.error + return { status: 'completed', durationMs: Date.now() - startedAt } + } finally { + clearTimeout(timer) + } +} \ No newline at end of file diff --git a/src/ingest/venues/amm.ts b/src/ingest/venues/amm.ts index b715adb0..b7c2d945 100644 --- a/src/ingest/venues/amm.ts +++ b/src/ingest/venues/amm.ts @@ -22,8 +22,10 @@ export function _getLastPrice(network: NetworkName, pairKey: string): number | u return getLastPrice('AMM', network, pairKey) } -export async function fetchPools(pair: WatchedPair, network: NetworkName = activeNetwork): Promise { +export async function fetchPools(pair: WatchedPair, network: NetworkName = activeNetwork, signal?: AbortSignal): Promise { try { + if (signal?.aborted) return [] + const assetAStr = pair.assetA.issuer ? `${pair.assetA.code}:${pair.assetA.issuer}` : 'native' @@ -37,7 +39,8 @@ export async function fetchPools(pair: WatchedPair, network: NetworkName = activ params.set('limit', '10') const response = await fetchWithTimeout( - `${getNetworkConfig(network).horizon.url}/liquidity_pools?${params.toString()}` + `${getNetworkConfig(network).horizon.url}/liquidity_pools?${params.toString()}`, + { signal } ) const data = (await response.json()) as any if (!data._embedded?.records) return [] @@ -51,9 +54,17 @@ export async function fetchPools(pair: WatchedPair, network: NetworkName = activ export async function snapshotPool( pool: any, pair: WatchedPair, - network: NetworkName = activeNetwork + network: NetworkName = activeNetwork, + signal?: AbortSignal ): Promise { try { + // Cycle-deadline checkpoint (#351): this unit makes no upstream call — it + // is two writes (pool snapshot + its zero-volume price point) — so an + // abandoned cycle skips the whole unit instead of starting it. Once begun + // it runs to completion; each statement is atomic and the checkpoint never + // sits between them. + if (signal?.aborted) return + const r0 = pool.reserves[0] const r1 = pool.reserves[1] @@ -130,15 +141,19 @@ export async function snapshotPool( export async function ingestPoolTrades( pool: any, pair: WatchedPair, - network: NetworkName = activeNetwork + network: NetworkName = activeNetwork, + signal?: AbortSignal ): Promise { const stateId = `amm:${network}:${pool.id}` const state = await getIndexerState(stateId, network) const cursor = state.cursor ?? '0' try { + if (signal?.aborted) return + const response = await fetchWithTimeout( - `${getNetworkConfig(network).horizon.url}/liquidity_pools/${pool.id}/trades?cursor=${cursor}&limit=${config.indexer.ammPageSize}&order=asc` + `${getNetworkConfig(network).horizon.url}/liquidity_pools/${pool.id}/trades?cursor=${cursor}&limit=${config.indexer.ammPageSize}&order=asc`, + { signal } ) const data = (await response.json()) as any const records = data._embedded?.records ?? [] @@ -180,6 +195,14 @@ export async function ingestPoolTrades( }) const trackerKey = pair.pairKey + + // Cycle-deadline checkpoint (#351), ahead of the write SEQUENCE: the page + // upsert and the cursor advance below commit together, so a cycle + // abandoned here skips both — never one without the other. The page is + // re-read from the same cursor next cycle and upsertPricePoints dedupes + // on conflict, so dropping out early costs nothing but a refetch. + if (signal?.aborted) return + const { previousPrice, currentPrice } = setLastPrice( 'AMM', network, @@ -218,14 +241,19 @@ export async function ingestPoolTrades( } } -export async function ingestPair(pair: WatchedPair, network: NetworkName = activeNetwork): Promise { - const pools = await fetchPools(pair, network) +export async function ingestPair(pair: WatchedPair, network: NetworkName = activeNetwork, signal?: AbortSignal): Promise { + const pools = await fetchPools(pair, network, signal) console.log(`[amm] ${pair.pairKey}: found ${pools.length} AMM pools`) await Promise.all( pools.map(async (pool) => { - await snapshotPool(pool, pair, network) - await ingestPoolTrades(pool, pair, network) + // Cycle-deadline checkpoint (#351): skip pools whose unit has not + // started because the cycle was abandoned mid-flight. Each pool is one + // write unit, so skipping whole pools never tears a unit in half. + if (signal?.aborted) return + await snapshotPool(pool, pair, network, signal) + if (signal?.aborted) return + await ingestPoolTrades(pool, pair, network, signal) }) ) } @@ -259,8 +287,8 @@ export const ammAdapter: VenueAdapter = { await ingestPoolTrades(pool, pair, network) }, - async ingestPair(pair: WatchedPair, network: NetworkName): Promise { - await ingestPair(pair, network) + async ingestPair(pair: WatchedPair, network: NetworkName, signal?: AbortSignal): Promise { + await ingestPair(pair, network, signal) }, _resetLastPrice, diff --git a/src/ingest/venues/aquarius.ts b/src/ingest/venues/aquarius.ts index 3c2255cd..b86dbeb3 100644 --- a/src/ingest/venues/aquarius.ts +++ b/src/ingest/venues/aquarius.ts @@ -39,9 +39,12 @@ export interface AquariusListResponse { export async function fetchAquariusPools( pair: WatchedPair, - apiUrl: string = config.aquarius.apiUrl + apiUrl: string = config.aquarius.apiUrl, + signal?: AbortSignal ): Promise { try { + if (signal?.aborted) return [] + const assetAStr = pair.assetA.issuer ? `${pair.assetA.code}:${pair.assetA.issuer}` : 'native' @@ -53,7 +56,7 @@ export async function fetchAquariusPools( params.append('assets[]', assetAStr) params.append('assets[]', assetBStr) - const res = await fetchWithTimeout(`${apiUrl}?${params.toString()}`) + const res = await fetchWithTimeout(`${apiUrl}?${params.toString()}`, { signal }) if (!res.ok) return [] const data = (await res.json()) as AquariusListResponse return data.results ?? [] @@ -94,14 +97,25 @@ export function deriveAquariusPricePoints( export async function ingestAquariusPair( pair: WatchedPair, - network: NetworkName = activeNetwork + network: NetworkName = activeNetwork, + signal?: AbortSignal ): Promise { - const pools = await fetchAquariusPools(pair, getNetworkConfig(network).aquarius.apiUrl) + const pools = await fetchAquariusPools( + pair, + getNetworkConfig(network).aquarius.apiUrl, + signal + ) if (pools.length === 0) return const points = deriveAquariusPricePoints(pools, pair) if (points.length === 0) return + // Cycle-deadline checkpoint (#351): if the signal aborted while we were + // fetching, skip the write instead of committing points from an abandoned + // cycle. Nothing has been written yet, so skipping leaves no partial state + // — the same pools are re-read next cycle. + if (signal?.aborted) return + await writeVenuePricePoints('Aquarius', points, pair, network) } @@ -134,8 +148,8 @@ export const aquariusAdapter: VenueAdapter = { await writeVenuePricePoints('Aquarius', points, pair, network) }, - async ingestPair(pair: WatchedPair, network: NetworkName): Promise { - await ingestAquariusPair(pair, network) + async ingestPair(pair: WatchedPair, network: NetworkName, signal?: AbortSignal): Promise { + await ingestAquariusPair(pair, network, signal) }, _resetLastPrice, diff --git a/src/ingest/venues/runner.ts b/src/ingest/venues/runner.ts index ca2f1cf2..f07ef08b 100644 --- a/src/ingest/venues/runner.ts +++ b/src/ingest/venues/runner.ts @@ -3,6 +3,7 @@ import { getActivePairs } from '../../pairsRegistry' import { upsertPricePoints } from '../../db' import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' +import { runCycleWithDeadline } from '../cycleDeadline' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' @@ -44,10 +45,18 @@ export async function writeVenuePricePoints( venue: string, points: VenuePricePoint[], pair: WatchedPair, - network: NetworkName + network: NetworkName, + signal?: AbortSignal ): Promise { if (points.length === 0) return + // Cycle-deadline checkpoint (#351): an abandoned cycle skips the write it + // has not started yet. The checkpoint is before the unit's only statement, + // never between statements — `upsertPricePoints` is a single + // `createMany(skipDuplicates)`, so it writes all of these points or none of + // them, and the events below fire only for points that actually landed. + if (signal?.aborted) return + await upsertPricePoints(points as any, network) const latest = points[points.length - 1] @@ -77,18 +86,33 @@ export async function sleep(ms: number): Promise { // ── Scheduling, Retry & Error Handling Runner ───────────────────────────────── +/** + * Options for a venue adapter's polling loop. + */ +export interface VenueAdapterLoopOptions { + /** + * Stops the loop cleanly at the next cycle boundary instead of running + * forever. Production leaves it unset (loops live until the process exits); + * tests use it to return from the otherwise-infinite loop. + */ + stopSignal?: AbortSignal +} + /** * Runs an individual venue adapter polling loop. * * Implements: * - Network-enable checking * - Scheduling loop based on adapter's getPollIntervalMs() + * - A per-cycle deadline: a cycle over its deadline is aborted, abandoned and + * counted (#351) rather than stalling the loop until the upstream recovers * - Exponential backoff retry on cycle failure * - Fault isolation: errors are caught, logged, and backed off without crashing the host */ export async function startVenueAdapter( adapter: VenueAdapter, - network: NetworkName = activeNetwork + network: NetworkName = activeNetwork, + options: VenueAdapterLoopOptions = {} ): Promise { if (!adapter.isEnabled(network)) { console.log(`[${adapter.name}] ${adapter.name} is disabled on ${network} — ingester not started`) @@ -98,12 +122,28 @@ export async function startVenueAdapter( console.log(`[${adapter.name}] Starting ${adapter.name} venue adapter on ${network}`) let consecutiveErrors = 0 - while (true) { + while (!options.stopSignal?.aborted) { try { - const pairs = getActivePairs() - for (const pair of pairs) { - await adapter.ingestPair(pair, network) - } + // One cycle = one round over every active pair, bounded by the venue's + // per-cycle deadline (#351). See src/ingest/cycleDeadline.ts: on + // expiry the cycle is aborted and counted, and the next cycle starts + // only after this one has unwound, so two cycles against the same venue + // can never overlap. + await runCycleWithDeadline({ + venue: adapter.name, + network, + cycle: async (signal) => { + const pairs = getActivePairs() + for (const pair of pairs) { + // Abandonment checkpoint (#351): the deadline has fired — do not + // start another pair. Pairs already in flight keep their own + // checkpoints inside the adapter (one write unit each), so the + // cycle stops at a unit boundary and no write is left half-done. + if (signal.aborted) break + await adapter.ingestPair(pair, network, signal) + } + }, + }) consecutiveErrors = 0 } catch (err) { consecutiveErrors++ diff --git a/src/ingest/venues/sdex.ts b/src/ingest/venues/sdex.ts index e8adc156..f6fabb9c 100644 --- a/src/ingest/venues/sdex.ts +++ b/src/ingest/venues/sdex.ts @@ -37,13 +37,20 @@ export interface SdexRawTrades { export async function fetchSdexTrades( pair: WatchedPair, - network: NetworkName = activeNetwork + network: NetworkName = activeNetwork, + signal?: AbortSignal ): Promise { const stateId = `sdex:${network}:${pair.pairKey}` const state = await getIndexerState(stateId, network) const cursor = state.cursor ?? '0' try { + // Cycle-deadline checkpoint (#351): an already-abandoned cycle must not + // start a fresh round trip. The SDK call below cannot take our signal + // mid-flight, but it is one bounded HTTP request — we only ever refuse to + // *begin* one on behalf of an abandoned cycle. + if (signal?.aborted) return null + const assetA = toAsset(pair.assetA) const assetB = toAsset(pair.assetB) @@ -155,13 +162,20 @@ export async function writeSdexPricePoints( }).catch((err) => console.error('[sdex] webhook dispatch error:', err.message)) } -export async function ingestPair(pair: WatchedPair, network: NetworkName = activeNetwork): Promise { - const raw = await fetchSdexTrades(pair, network) +export async function ingestPair(pair: WatchedPair, network: NetworkName = activeNetwork, signal?: AbortSignal): Promise { + const raw = await fetchSdexTrades(pair, network, signal) if (!raw) return const derived = deriveSdexPricePoints(raw, pair) if (!derived || derived.points.length === 0) return + // Cycle-deadline checkpoint (#351), placed before the write SEQUENCE as a + // whole: `writeSdexPricePoints` commits the page upsert and the cursor + // advance together, so a cycle abandoned here skips both — never one + // without the other. Skipping merely re-reads the same page next cycle; + // `upsertPricePoints` uses skipDuplicates, so the replay is idempotent. + if (signal?.aborted) return + await writeSdexPricePoints(derived, pair, network) } @@ -194,8 +208,8 @@ export const sdexAdapter: VenueAdapter = { await ingestPair(pair, network) }, - async ingestPair(pair: WatchedPair, network: NetworkName): Promise { - await ingestPair(pair, network) + async ingestPair(pair: WatchedPair, network: NetworkName, signal?: AbortSignal): Promise { + await ingestPair(pair, network, signal) }, _resetLastPrice, diff --git a/src/ingest/venues/soroswap.ts b/src/ingest/venues/soroswap.ts index 1b2b2b42..2f36168d 100644 --- a/src/ingest/venues/soroswap.ts +++ b/src/ingest/venues/soroswap.ts @@ -73,10 +73,11 @@ export interface PoolEntry { * degrades gracefully without affecting other ingesters. */ export async function fetchSoroswapTokenList( - tokenListUrl: string = getNetworkConfig(activeNetwork).soroswap.tokenListUrl + tokenListUrl: string = getNetworkConfig(activeNetwork).soroswap.tokenListUrl, + signal?: AbortSignal ): Promise { try { - const res = await fetchWithTimeout(tokenListUrl) + const res = await fetchWithTimeout(tokenListUrl, { signal }) if (!res.ok) throw new Error(`HTTP ${res.status}`) const data = (await res.json()) as SoroswapTokenList return Array.isArray(data.tokens) ? data.tokens : [] @@ -218,9 +219,14 @@ export async function ingestPool( poolEntry: PoolEntry, pair: WatchedPair, _fetchReserves: (poolAddress: string, network?: NetworkName) => Promise<[bigint, bigint] | null> = fetchPoolReserves, - network: NetworkName = activeNetwork + network: NetworkName = activeNetwork, + signal?: AbortSignal ): Promise { try { + // Cycle-deadline checkpoint (#351): an abandoned cycle does not start + // another RPC simulation for a pool whose unit has not begun. + if (signal?.aborted) return + const reserves = await _fetchReserves(poolEntry.poolAddress, network) if (!reserves) return @@ -234,6 +240,13 @@ export async function ingestPool( return } + // Cycle-deadline checkpoint (#351), ahead of the write: nothing has been + // written yet, so skipping leaves no partial state — the reserves are + // re-read next cycle. The write itself is one upsert, so it commits whole + // or not at all, and Soroswap keeps no cursor whose advance could be + // separated from it. + if (signal?.aborted) return + await upsertPricePoints( [ { @@ -300,7 +313,8 @@ export async function ingestPair( tokenB: string, network?: NetworkName ) => Promise = fetchPoolsFromFactory, - network: NetworkName = activeNetwork + network: NetworkName = activeNetwork, + signal?: AbortSignal ): Promise { const tokenA = tokens.find( (t) => t.symbol.toUpperCase() === pair.assetA.code.toUpperCase() @@ -330,13 +344,19 @@ export async function ingestPair( return } + // Cycle-deadline checkpoint (#351): discovery above is one bounded RPC call + // we cannot interrupt mid-flight; this is where an abandoned cycle stops + // instead of scheduling a reserves read per pool after its deadline. + if (signal?.aborted) return + await Promise.all( poolAddresses.map((addr) => ingestPool( { poolAddress: addr, tokenA, tokenB }, pair, fetchPoolReserves, - network + network, + signal ) ) ) @@ -400,11 +420,11 @@ export const soroswapAdapter: VenueAdapter = { await ingestPool(target, pair, fetchPoolReserves, network) }, - async ingestPair(pair: WatchedPair, network: NetworkName): Promise { + async ingestPair(pair: WatchedPair, network: NetworkName, signal?: AbortSignal): Promise { const netConfig = getNetworkConfig(network) - const tokens = await fetchSoroswapTokenList(netConfig.soroswap.tokenListUrl) + const tokens = await fetchSoroswapTokenList(netConfig.soroswap.tokenListUrl, signal) if (tokens.length === 0) return - await ingestPair(pair, tokens, netConfig.soroswap.factoryAddress, fetchPoolsFromFactory, network) + await ingestPair(pair, tokens, netConfig.soroswap.factoryAddress, fetchPoolsFromFactory, network, signal) }, _resetLastPrice, diff --git a/src/ingest/venues/types.ts b/src/ingest/venues/types.ts index c42dee85..41f3274c 100644 --- a/src/ingest/venues/types.ts +++ b/src/ingest/venues/types.ts @@ -60,8 +60,14 @@ export interface VenueAdapter { /** * Optional full cycle override for a pair, if the adapter handles compound execution. + * + * `signal` is the cycle deadline's AbortSignal (issue #351): adapters check + * it at their checkpoints — before starting a unit's writes, and via + * `fetchWithTimeout({ signal })` for outbound HTTP — so an abandoned cycle + * stops at a unit boundary and its writes are never torn. It is optional to + * stay compatible with adapters (and tests) that have no I/O to cancel. */ - ingestPair(pair: WatchedPair, network: NetworkName): Promise + ingestPair(pair: WatchedPair, network: NetworkName, signal?: AbortSignal): Promise /** * Resets internal in-memory state (such as last-price cache) for testing. diff --git a/src/ingesters/snapshot.ts b/src/ingesters/snapshot.ts index 84224373..fdecc1b5 100644 --- a/src/ingesters/snapshot.ts +++ b/src/ingesters/snapshot.ts @@ -2,6 +2,7 @@ import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' import { getActivePairs } from '../pairsRegistry' import { price_snapshots_total } from '../metrics' +import { runCycleWithDeadline } from '../ingest/cycleDeadline' /** * Floors a Date down to the start of its minute (seconds/millis zeroed). @@ -86,8 +87,23 @@ export async function startSnapshotIngester(network: NetworkName = activeNetwork await sleep(msToNextMinute) try { - const n = await appendSnapshots(new Date(), network) - if (n > 0) console.log(`[snapshot] appended ${n} snapshot(s) on ${network}`) + // Sleep to the minute boundary is not part of the cycle — the deadline + // bounds the append itself, exactly like one venue-adapter round. + // + // On expiry (#351) the append is counted as abandoned and the loop + // resumes. `appendSnapshots` is a single `INSERT … ON CONFLICT … DO + // NOTHING`, so the worst an abandoned snapshot cycle can leave behind + // is that fully-committed statement or nothing at all: there is no + // sequence of writes for a checkpoint to land in the middle of. + await runCycleWithDeadline({ + venue: 'Snapshot', + network, + cycle: async (signal) => { + if (signal.aborted) return + const n = await appendSnapshots(new Date(), network) + if (n > 0) console.log(`[snapshot] appended ${n} snapshot(s) on ${network}`) + }, + }) } catch (err) { console.error(`[snapshot] Error appending snapshots on ${network}:`, (err as Error).message) } diff --git a/src/metrics.ts b/src/metrics.ts index 39a3a505..f7f99e73 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -40,6 +40,18 @@ export const price_snapshots_total = new Counter({ registers: [register] }) +export const ingester_abandoned_cycles_total = new Counter({ + name: 'ingester_abandoned_cycles_total', + help: 'Total number of ingester cycles abandoned after exceeding their per-cycle deadline', + // `venue` is the ingester loop's own name (SDEX, AMM, Soroswap, Aquarius, + // Snapshot) — a bounded set hard-coded in the adapter registry, exactly like + // the network factor above. Together they give `venues x networks` series: + // this is the label #351 asks for, so an on-call can tell WHICH loop is + // stalling on WHICH network without an unbounded dimension ever appearing. + labelNames: ['venue', 'network'], + registers: [register] +}) + export const price_requests_total = new Counter({ name: 'price_requests_total', help: 'Total number of price API requests served',