Description
Supervisor::tick in crates/ingest/src/lib.rs polls every matching wallet's Horizon account sequentially in a for loop, awaiting each poll_once before starting the next. With one wallet this is fine, but as the number of master wallets on a network grows, a single slow or hanging Horizon request for one wallet delays polling for every wallet behind it in the loop, directly increasing deposit-detection latency for unrelated customers.
Requirements and Context
- Poll wallets concurrently, bounded by a configurable concurrency limit (default a sane value like 10), using something like
futures::stream::iter(...).for_each_concurrent(limit, ...) (add futures as a dependency if not already available) rather than unbounded tokio::spawn per wallet, to avoid overwhelming Horizon with simultaneous requests from a large wallet set.
- Add the concurrency limit as a
Supervisor::new parameter (or a builder-style setter, matching the existing with_webhooks pattern on Ingestor), and thread a corresponding env var through bin/server/src/main.rs::Config (e.g. INGEST_CONCURRENCY, defaulting to today's effectively-sequential behavior of 1 only if the team prefers a conservative default — otherwise pick and justify a higher default in the PR description).
- Preserve today's per-wallet error handling (
tracing::warn! and continue) — one wallet's failure must still not abort the others.
Suggested Execution
Branch: feat/ingest/concurrent-wallet-polling
Implement Changes
- Update
Supervisor::tick in crates/ingest/src/lib.rs to use bounded concurrent polling.
- Update
Supervisor::new's signature (or add a builder method) and bin/server/src/main.rs to wire the new config value through.
Test and Commit
tick_polls_wallets_concurrently_up_to_the_configured_limit (using a mock Horizon with an artificial per-request delay, asserting wall-clock time for polling N wallets is much closer to one request's latency than N times it).
tick_still_isolates_failures_between_wallets_under_concurrency.
- Run
cargo test -p octo-ingest locally before committing.
Example Commit Message
feat(ingest): poll wallets concurrently in Supervisor::tick, bounded by a configurable limit
Supervisor::tick polled every matching wallet sequentially, so one slow or
hanging Horizon request delayed deposit detection for every other wallet
behind it. Adds bounded concurrent polling with a configurable limit, wired
through INGEST_CONCURRENCY, while preserving today's per-wallet error
isolation.
Guidelines
- Bound the concurrency — do not fan out unboundedly, since a large wallet set hitting Horizon all at once risks the ingest worker itself getting rate-limited (see the related Horizon 429-handling issue).
- Reference this issue with
Closes #<issue-number> in the PR description.
Description
Supervisor::tickincrates/ingest/src/lib.rspolls every matching wallet's Horizon account sequentially in aforloop, awaiting eachpoll_oncebefore starting the next. With one wallet this is fine, but as the number of master wallets on a network grows, a single slow or hanging Horizon request for one wallet delays polling for every wallet behind it in the loop, directly increasing deposit-detection latency for unrelated customers.Requirements and Context
futures::stream::iter(...).for_each_concurrent(limit, ...)(addfuturesas a dependency if not already available) rather than unboundedtokio::spawnper wallet, to avoid overwhelming Horizon with simultaneous requests from a large wallet set.Supervisor::newparameter (or a builder-style setter, matching the existingwith_webhookspattern onIngestor), and thread a corresponding env var throughbin/server/src/main.rs::Config(e.g.INGEST_CONCURRENCY, defaulting to today's effectively-sequential behavior of1only if the team prefers a conservative default — otherwise pick and justify a higher default in the PR description).tracing::warn!and continue) — one wallet's failure must still not abort the others.Suggested Execution
Branch:
feat/ingest/concurrent-wallet-pollingImplement Changes
Supervisor::tickincrates/ingest/src/lib.rsto use bounded concurrent polling.Supervisor::new's signature (or add a builder method) andbin/server/src/main.rsto wire the new config value through.Test and Commit
tick_polls_wallets_concurrently_up_to_the_configured_limit(using a mock Horizon with an artificial per-request delay, asserting wall-clock time for polling N wallets is much closer to one request's latency than N times it).tick_still_isolates_failures_between_wallets_under_concurrency.cargo test -p octo-ingestlocally before committing.Example Commit Message
Guidelines
Closes #<issue-number>in the PR description.