[felixumeh]

Closes #351

Summary of Changes

Bounds every ingester cycle with a deadline so one slow upstream can no longer stall an ingester while every health signal still reports the service as healthy. Issues below are fixed by this PR:

• Closes Ingester loops have no upper bound on a single cycle #351: Each round of an ingester loop (SDEX, AMM, Soroswap, Aquarius venue adapters, plus the 1-minute snapshot appender) now runs under runCycleWithDeadline() in the new src/ingest/cycleDeadline.ts. A cycle that exceeds its deadline is aborted at its next unit checkpoint, counted in the new ingester_abandoned_cycles_total metric labelled venue + network, logged, and abandoned — the loop resumes after teardown of work already in flight instead of awaiting the abandoned cycle's remaining work.

All five acceptance criteria from the issue:

• A cycle that exceeds its deadline is abandoned, not awaited indefinitely — the cycle races a timer; on expiry its AbortSignal fires and every checkpoint stops further pairs/pools from starting, so the abandoned cycle stops at its in-flight unit rather than grinding through the rest of the round.
• The deadline is configurable per ingester — INGEST_CYCLE_DEADLINE_MS with per-venue (INGEST_CYCLE_DEADLINE_MS_SOROSWAP) and per-venue-per-network (INGEST_CYCLE_DEADLINE_MS_SOROSWAP_MAINNET) overrides, precedence documented in getCycleDeadlineMs() and .env.example; default 60s, 0 explicitly disables the deadline. Follows the paired-var convention used everywhere else in src/config.ts.
• Abandoned cycles are counted in a metric — ingester_abandoned_cycles_total{venue,network} (prom-client Counter in src/metrics.ts); cardinality is venues x networks, bounded by the adapter registry, documented in docs/ingest-metrics.md alongside the existing metrics, with a promql query example.
• No partial write survives an abandoned cycle — and, per the issue's "check how each ingester writes before changing the control flow", the checkpoints were placed to match the existing write units: upsertPricePoints is one createMany(skipDuplicates); the SDEX/AMM page upsert and its cursor advance are checked before the sequence as a whole, never between the two statements; snapshotPool, Soroswap's per-pool write and the snapshot appender are each either skipped whole or run whole. Skipping costs only a refetch from the same cursor, and skipDuplicates makes the replay idempotent.
• A test with a slow upstream proves the loop keeps making progress — src/__tests__/ingestCycleDeadline.test.ts drives the real startVenueAdapter loop with a 250ms upstream under a 50ms deadline and asserts ≥3 cycles start, maxActive === 1 (no two cycles ever overlap), and ≥2 abandonments land on the counter with venue/network labels.

Guarantees in detail

• Overlap is impossible, structurally: runCycleWithDeadline() never resolves while the previous cycle's promise is unsettled — abandonment awaits unwind, not work. The unwind is prompt because the signal is passed to fetchWithTimeout({ signal }) (the house pattern from src/fetch.ts/src/alerts/delivery.ts) and checked at every unit boundary. If a single in-flight call ignores even its own timeout, the loop still refuses to start an overlapping cycle and escalates with a loud [cycle-deadline] … still has not unwound error after a 10s grace (CYCLE_UNWIND_GRACE_MS), rather than silently double-running a venue.
• Failures keep their backoff unchanged: cycle errors are rethrown untouched after clearTimeout, so startVenueAdapter's exponential backoff and restartIngester fault isolation behave exactly as before; only deadline expiry counts as an abandonment (metric, no error backoff), so a slow venue is not also punished with retry delay.
• Adapter contract stays backward compatible: ingestPair(pair, network, signal?) adds an optional third parameter; all four adapters, both writeVenuePricePoints and the snapshot loop thread it through; existing custom adapters and tests compile and run unmodified.
• Testability without touching production semantics: startVenueAdapter(adapter, network, { stopSignal }) lets a loop stop at a cycle boundary — used by the new test; unset in production where loops legitimately run until process exit.

Files changed

File	Change
src/ingest/cycleDeadline.ts	New — runCycleWithDeadline(), getCycleDeadlineMs() precedence chain, CycleDeadlineExceeded, grace-window logging
src/metrics.ts	New ingester_abandoned_cycles_total counter (venue, network)
src/ingest/venues/runner.ts	Cycle wrapped in the deadline; per-pair abandonment checkpoint; writeVenuePricePoints pre-write checkpoint; VenueAdapterLoopOptions
src/ingest/venues/types.ts	ingestPair gains optional signal
src/ingest/venues/{sdex,amm,soroswap,aquarius}.ts	Signal threaded from ingestPair → outbound fetches (fetchWithTimeout({ signal })) and checkpoints before each write unit
src/ingesters/snapshot.ts	Minute-append wrapped in the deadline (sleep excluded; single atomic INSERT … ON CONFLICT)
.env.example	INGEST_CYCLE_DEADLINE_MS[_<NETWORK>] documented (drift guard)
docs/ingest-metrics.md	Metric table, label table, cardinality bounds, query example
.changeset/ingest-cycle-deadline.md	Minor bumpsheet entry
src/__tests__/ingestCycleDeadline.test.ts	New — 8 tests: slow-upstream progress, no-partial-write, precedence/invalid/0-disables, error propagation

The branch also carries one pre-existing test-hygiene commit (test: reset upsertPricePoints mock between soroswap ingester cases) — clearAllMocks retains implementations, so a rejection stub could leak into the next test under a shuffled run.

Verification (all run locally, mirroring .github/workflows/ci.yml)

• npm ci → up to date, exit 0
• prisma generate → exit 0
• npx tsc --noEmit → exit 0, zero errors
• npm run build → exit 0 (emits dist/)
• npm test → exit 0 — 67 files passed, 673 tests passed, 1 skipped (includes the 8 new Ingester loops have no upper bound on a single cycle #351 tests; .env.example drift guard and OpenAPI sync green)
• Targeted run of ingestCycleDeadline.test.ts → 8/8 pass with live [cycle-deadline] … abandoned after exceeding its 50ms deadline logs proving the behavior end-to-end

Notes for reviewers

• Overlap with feat(ingest): add schema verification, cycle deadlines, lag telemetry, and /status health breakdown #463: that open PR also proposes cycle deadlines (among schema verification and lag telemetry). This PR implements Ingester loops have no upper bound on a single cycle #351 standalone against current main; happy to reconcile whichever direction you prefer — the runCycleWithDeadline() seam is small and would slot under whatever scheduling feat(ingest): add schema verification, cycle deadlines, lag telemetry, and /status health breakdown #463 lands.
• Known-unfixed (deliberately, out of Ingester loops have no upper bound on a single cycle #351's scope): the stellar-sdk Horizon call itself cannot be interrupted mid-round-trip — the timeout there belongs to fetchWithTimeout-style per-call deadlines (fix: enforce deadline timeouts on all outbound fetch calls #379); the cycle deadline ensures we refuse to begin work for an abandoned cycle and to overlap cycles regardless.
• The k6 prices load test is already failing on main (per the issue text) and is not part of PR CI; unaffected here.
• npm test includes one intentionally skipped integration e2e test (tests/integration/e2e.test.ts), pre-existing and unrelated.

[drips-wave]

@felixumeh Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits.

You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀

Learn more about application limits

[Miracle656]

Merged as bc1eb53. This is the best-engineered PR in this batch, and the issue called it correctly — it was the most dangerous shape in the set, because the failure mode is silence. A cycle that never finishes never crashes, so the restart logic never fires and the venue just stops producing prices with nothing in the logs.

First, a correction on my side: #351 had been closed in error. I closed it when merging #479, which added exponential backoff and crash reporting for an ingester that crashes — a different failure entirely. I have reopened it and this PR closes it properly.

What is right

I read runCycleWithDeadline closely because a timeout wrapper is easy to write and easy to get subtly wrong. Four things it gets right that are commonly missed:

The cycle promise is always attached to a handler.

const cycleSettled: Promise<CycleSettled> = cycle(controller.signal).then(
  () => ({ outcome: 'fulfilled' as const }),
  (error) => ({ outcome: 'rejected' as const, error })
)

When the deadline wins the race, the abandoned cycle may still reject later. Without this it becomes an unhandled rejection, which in Node takes the process down — so the naive version of this wrapper turns a slow venue into a crash. Converting rejection into a value at the point of creation is the right fix.

Both timers are cleared in finally, including the unwind grace timer in its own nested try/finally. A deadline wrapper that leaks a setTimeout per cycle leaks a handle every poll interval, forever.

Overlap is impossible by construction. On expiry you await cycleSettled before returning, so the next cycle cannot start while the previous one is still in flight. That is the correct priority: two concurrent cycles against one venue double-write.

0 is an opt-out that still passes a signal.

if (deadlineMs <= 0) {
  await cycle(new AbortController().signal)
  return { status: 'completed', durationMs: Date.now() - startedAt }
}

Adapter code paths are identical whether or not a deadline is configured, so disabling it in one environment cannot change behaviour anywhere else. Easy to skip, and skipping it is how you get a bug that only appears in the environment that turned the feature off.

And the env precedence (_VENUE_NETWORK → _VENUE → _NETWORK → bare) with an invalid value falling through rather than disabling the deadline is the safe direction: a typo must never read as "unbounded".

The merge

Seven conflicts, all from work landed after your branch was cut. Resolved as unions; the one worth flagging is soroswap.ts, where main added _fetchDecimals and you added signal, both as trailing parameters. I kept _fetchDecimals in its position so existing positional callers are untouched, which means the internal call site had to start passing it explicitly:

ingestPool(
  { poolAddress: addr, tokenA, tokenB },
  pair,
  fetchPoolReserves,
  network,
  fetchTokenDecimals,   // omitting this put the AbortSignal in its place
  signal
)

tsc caught that one (Argument of type 'AbortSignal | undefined' is not assignable to parameter of type '(tokenAddress: string, ...) => Promise<number | null>'), which is a good argument for an options object rather than a sixth positional parameter if this signature grows again.

In runner.ts I kept main's ingest_cycle_duration_seconds observation and moved it to cover the deadline-bounded cycle, including the abandoned case — an abandoned cycle is precisely the one whose duration an operator wants in the histogram, and since runCycleWithDeadline resolves rather than throwing on expiry, the success-path observation is the right place for it.

Two fixes on top

1. cycleDeadline.ts used console.warn/console.error — three sites, which src/__tests__/logger.test.ts rejects by design:

+ [ "src\\ingest\\cycleDeadline.ts:90", "src\\ingest\\cycleDeadline.ts:191", "src\\ingest\\cycleDeadline.ts:203" ]

That guard exists because buildLogger pipes every line through redactSecrets, and an ingest failure is exactly where a connection string or an API key surfaces. Switched to createLogger('cycle-deadline') with structured fields rather than interpolated strings, so { venue, network, deadlineMs, elapsedMs } is queryable instead of being parsed back out of prose.

2. A test spied on console.warn to prove an invalid deadline falls through.

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)

It broke with the logger change, but it was testing the wrong thing anyway: the behaviour that matters is the fallback, not that a particular function was called once. Rewritten to assert the value, plus a second case for precedence around a bad value — a broken INGEST_CYCLE_DEADLINE_MS_SDEX must not shadow a working INGEST_CYCLE_DEADLINE_MS, which nothing covered.

One limitation worth stating plainly

On expiry the loop waits for the abandoned cycle to unwind, and the grace timer only logs:

const graceTimer = setTimeout(() => {
  log.error({ venue, network, graceMs: CYCLE_UNWIND_GRACE_MS }, 'abandoned cycle has not unwound …')
}, CYCLE_UNWIND_GRACE_MS)

So an upstream call that genuinely never settles still parks the loop — the original failure, strictly speaking. What has changed is that it is no longer silent: ingester_abandoned_cycles_total increments and an error is logged with the venue and network, which is what #351 actually asked for ("the venue goes quiet without any error being raised anywhere"). And it requires an outbound call with no timeout of its own, which HTTP_TIMEOUT_MS and fetchWithTimeout already rule out.

Your comment says this outright, which is why I merged it rather than asking for a hard abort. Refusing to overlap beats double-writing, and a documented trade-off is worth more than a wrapper that quietly starts a second cycle.

128 test files / 1,530 tests pass. tsc --noEmit and npm run build clean.