diff --git a/crates/driver/src/boundary/liquidity/uniswap/v3.rs b/crates/driver/src/boundary/liquidity/uniswap/v3.rs index 5dd505ad72..bfe8d116d1 100644 --- a/crates/driver/src/boundary/liquidity/uniswap/v3.rs +++ b/crates/driver/src/boundary/liquidity/uniswap/v3.rs @@ -114,11 +114,32 @@ async fn init_liquidity( config: &infra::liquidity::config::UniswapV3, ) -> anyhow::Result> { let web3 = eth.web3().clone(); - let source = build_pool_data_source(eth, config).await?; + let http = boundary::liquidity::http_client(); + let mut fetch_on_demand = false; + let source: Arc = match &config.pool_source { + UniswapV3PoolSource::PoolIndexer(indexer) => { + tracing::info!(url = %indexer.url, "uniswap v3: using pool-indexer as data source"); + let client = PoolIndexerClient::new(indexer.url.clone(), eth.chain(), http); + fetch_on_demand = client.fetch_on_demand(); + Arc::new(client) + } + UniswapV3PoolSource::Subgraph(subgraph) => { + tracing::info!(url = %subgraph.url, "uniswap v3: using subgraph as data source"); + let client = UniV3SubgraphClient::from_subgraph_url( + &subgraph.url, + http, + subgraph.max_pools_per_tick_query, + ) + .await + .context("failed to construct UniV3 subgraph client")?; + Arc::new(client) + } + }; let pool_fetcher = Arc::new( UniswapV3PoolFetcher::new( source, + fetch_on_demand, web3.clone(), block_retriever, config.max_pools_to_initialize, @@ -139,37 +160,3 @@ async fn init_liquidity( pool_fetcher, )) } - -/// Picks the V3 pool data source based on the configured pool source variant. -async fn build_pool_data_source( - eth: &Ethereum, - config: &infra::liquidity::config::UniswapV3, -) -> anyhow::Result> { - let http = boundary::liquidity::http_client(); - - match &config.pool_source { - UniswapV3PoolSource::PoolIndexer(indexer) => { - tracing::info!( - url = %indexer.url, - "uniswap v3: using pool-indexer as data source", - ); - Ok(Arc::new(PoolIndexerClient::new( - indexer.url.clone(), - eth.chain(), - http, - ))) - } - UniswapV3PoolSource::Subgraph(subgraph) => { - tracing::info!(url = %subgraph.url, "uniswap v3: using subgraph as data source"); - Ok(Arc::new( - UniV3SubgraphClient::from_subgraph_url( - &subgraph.url, - http, - subgraph.max_pools_per_tick_query, - ) - .await - .context("failed to construct UniV3 subgraph client")?, - )) - } - } -} diff --git a/crates/liquidity-sources/src/uniswap_v3/graph_api.rs b/crates/liquidity-sources/src/uniswap_v3/graph_api.rs index 39d0245e2f..b3d052ca6d 100644 --- a/crates/liquidity-sources/src/uniswap_v3/graph_api.rs +++ b/crates/liquidity-sources/src/uniswap_v3/graph_api.rs @@ -5,7 +5,7 @@ use { crate::{ json_map, subgraph::{ContainsId, SubgraphClient}, - uniswap_v3::V3PoolDataSource, + uniswap_v3::{BlockTarget, V3PoolDataSource}, }, alloy::primitives::{Address, U256}, anyhow::Result, @@ -168,6 +168,15 @@ impl UniV3SubgraphClient { .saturating_sub(MAX_REORG_BLOCK_COUNT)) } + /// The subgraph queries by concrete block, so `Latest` resolves to the + /// safe head. + async fn resolve_block(&self, target: BlockTarget) -> Result { + match target { + BlockTarget::Latest => self.get_safe_block().await, + BlockTarget::Number(n) => Ok(n), + } + } + fn all_pools_query(include_ticks_filter: bool) -> String { let tick_filter = if include_ticks_filter { r#"ticks_: { liquidityNet_not: "0" }"# @@ -238,7 +247,8 @@ impl UniV3SubgraphClient { impl V3PoolDataSource for UniV3SubgraphClient { /// The subgraph supports historical queries, so every returned pool is /// consistently anchored at `target_block`. - async fn get_registered_pools(&self, target_block: u64) -> Result { + async fn get_registered_pools(&self, target_block: BlockTarget) -> Result { + let target_block = self.resolve_block(target_block).await?; let variables = json_map! { "block" => target_block, }; @@ -258,8 +268,12 @@ impl V3PoolDataSource for UniV3SubgraphClient { async fn get_pools_with_ticks_by_ids( &self, ids: &[Address], - target_block: u64, + target_block: BlockTarget, ) -> Result { + if ids.is_empty() { + return Ok(PoolsWithTicks::default()); + } + let target_block = self.resolve_block(target_block).await?; let (pools, ticks) = futures::try_join!( self.get_pools_by_pool_ids(ids, target_block), self.get_ticks_by_pools_ids(ids, target_block) diff --git a/crates/liquidity-sources/src/uniswap_v3/mod.rs b/crates/liquidity-sources/src/uniswap_v3/mod.rs index 3559776950..df035e3eca 100644 --- a/crates/liquidity-sources/src/uniswap_v3/mod.rs +++ b/crates/liquidity-sources/src/uniswap_v3/mod.rs @@ -27,20 +27,29 @@ use { /// the indexer's actual block can be later than `target_block`. #[async_trait] pub trait V3PoolDataSource: Send + Sync + 'static { - /// Fetch the full set of pools the source knows about as of a block at or - /// after `target_block`. `PoolData::ticks` is always `None` here — callers - /// needing ticks must use [`Self::get_pools_with_ticks_by_ids`] separately. - /// The split lets a cheap "what pools exist?" lookup skip the expensive - /// tick fetch. - async fn get_registered_pools(&self, target_block: u64) -> Result; + /// Fetch the full set of pools the source knows about as of `target_block`. + /// `PoolData::ticks` is always `None` here — callers needing ticks must use + /// [`Self::get_pools_with_ticks_by_ids`] separately. The split lets a cheap + /// "what pools exist?" lookup skip the expensive tick fetch. + async fn get_registered_pools(&self, target_block: BlockTarget) -> Result; - /// Fetch pools + their active ticks for the given pool addresses as of a - /// block at or after `target_block`. The returned `fetched_block_number` is - /// the actual snapshot block (`>= target_block`); callers should use it as - /// the event-replay anchor. + /// Fetch pools + their active ticks for the given pool addresses as of + /// `target_block`. The returned `fetched_block_number` is the actual + /// snapshot block (`>=` a requested [`BlockTarget::Number`]); callers + /// should use it as the event-replay anchor. async fn get_pools_with_ticks_by_ids( &self, ids: &[Address], - target_block: u64, + target_block: BlockTarget, ) -> Result; } + +/// Which block a [`V3PoolDataSource`] anchors its snapshot to. +#[derive(Clone, Copy, Debug)] +pub enum BlockTarget { + /// The latest block the source can serve: the subgraph resolves this to its + /// safe head, the indexer serves at-head without waiting. + Latest, + /// A specific block; the source returns data at or after it. + Number(u64), +} diff --git a/crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs b/crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs index cc2e6c986d..4814326b31 100644 --- a/crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs +++ b/crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs @@ -1,5 +1,6 @@ use { super::{ + BlockTarget, V3PoolDataSource, event_fetching::{RecentEventsCache, UniswapV3PoolEventFetcher}, graph_api::{PoolData, Token}, @@ -20,6 +21,7 @@ use { event_handler::{EventHandler, EventStoring, MAX_REORG_BLOCK_COUNT}, maintenance::Maintaining, }, + itertools::{Either, Itertools}, model::TokenPair, num::rational::Ratio, number::serialization::HexOrDecimalU256, @@ -126,12 +128,27 @@ struct PoolsCheckpoint { struct PoolsCheckpointHandler { source: Arc, + /// Whether the source can serve cache-miss fetches cheaply on the quote + /// path (the indexer). Decided at construction; false for the subgraph. + fetch_on_demand: bool, /// Address is pool id while TokenPair is a pair or tokens for each pool. pools_by_token_pair: HashMap>, /// Pools state on a specific block number in history considered reorg safe pools_checkpoint: Mutex, } +/// What `get` finds for a set of token pairs against the checkpoint. +#[derive(Default)] +struct CachedPools { + /// Pools already present in the checkpoint cache. + pools: HashMap>, + /// Registered pools not yet cached; fetched at the current block on the + /// quote path and folded into the cache by the next maintenance run. + missing: Vec
, + /// Block the cached pools are anchored at (0 if no pools are registered). + block_number: u64, +} + impl PoolsCheckpointHandler { /// Fetches the list of existing UniswapV3 pools and their metadata (without /// state/ticks). Then fetches state/ticks for the deepest pools @@ -146,6 +163,7 @@ impl PoolsCheckpointHandler { /// it would apply that event twice. pub async fn new( source: Arc, + fetch_on_demand: bool, block_retriever: Arc, max_pools_to_initialize_cache: usize, ) -> Result { @@ -154,7 +172,9 @@ impl PoolsCheckpointHandler { .await .context("read finalized block for snapshot target_block")? .number; - let mut registered_pools = source.get_registered_pools(target_block).await?; + let mut registered_pools = source + .get_registered_pools(BlockTarget::Number(target_block)) + .await?; tracing::debug!( target_block, block = %registered_pools.fetched_block_number, @@ -196,7 +216,10 @@ impl PoolsCheckpointHandler { .take(max_pools_to_initialize_cache) .collect::>(); let pools_with_ticks = source - .get_pools_with_ticks_by_ids(&pool_ids, registered_pools.fetched_block_number) + .get_pools_with_ticks_by_ids( + &pool_ids, + BlockTarget::Number(registered_pools.fetched_block_number), + ) .await?; let pools = pools_with_ticks .pools @@ -217,86 +240,110 @@ impl PoolsCheckpointHandler { Ok(Self { source, + fetch_on_demand, pools_by_token_pair, pools_checkpoint, }) } - /// For a given list of token pairs, fetches the pools for the ones that - /// exist in the checkpoint. For the ones that don't exist, flag as - /// missing and expect to exist after the next maintenance run. - fn get(&self, token_pairs: &HashSet) -> (HashMap>, u64) { + /// Returns cached pools for the pairs, plus the ids of any that exist but + /// aren't cached yet. Misses are recorded for the next maintenance run. + fn get(&self, token_pairs: &HashSet) -> CachedPools { let mut pool_ids = token_pairs .iter() .filter_map(|pair| self.pools_by_token_pair.get(pair)) .flatten() .peekable(); - tracing::trace!("get checkpoint for pool_ids: {:?}", pool_ids); - match pool_ids.peek() { Some(_) => { let mut pools_checkpoint = self.pools_checkpoint.lock().unwrap(); - let mut existing_pools = HashMap::>::default(); - let missing_pools = pool_ids - .filter(|pool_id| match pools_checkpoint.pools.get(*pool_id) { - Some(entry) => { - existing_pools.insert(**pool_id, entry.clone()); - false - } - None => true, - }) - .collect::>(); - - tracing::trace!( - "cache hit: {:?}, cache miss: {:?}", - existing_pools.keys(), - missing_pools - ); - pools_checkpoint.missing_pools.extend(missing_pools); - (existing_pools, pools_checkpoint.block_number) + let (pools, missing): (HashMap>, Vec
) = pool_ids + .partition_map(|pool_id| match pools_checkpoint.pools.get(pool_id) { + Some(entry) => Either::Left((*pool_id, entry.clone())), + None => Either::Right(*pool_id), + }); + + tracing::trace!("cache hit: {:?}, cache miss: {:?}", pools.keys(), missing); + pools_checkpoint.missing_pools.extend(&missing); + CachedPools { + pools, + missing, + block_number: pools_checkpoint.block_number, + } } - None => Default::default(), + None => CachedPools::default(), } } - /// Fetches state/ticks for missing pools and moves them from - /// `missing_pools` to `pools` + /// Fetches and converts the given pools from the source at `target_block`, + /// skipping any that can't be converted yet (e.g. missing ticks). + async fn fetch_pools( + &self, + pool_ids: &[Address], + target_block: BlockTarget, + ) -> Result)>> { + let pools_with_ticks = self + .source + .get_pools_with_ticks_by_ids(pool_ids, target_block) + .await?; + Ok(pools_with_ticks + .pools + .into_iter() + .filter_map(|pool| { + let id = pool.id; + match PoolInfo::try_from(pool) { + Ok(info) => Some((id, Arc::new(info))), + Err(err) => { + tracing::debug!(?id, ?err, "skipping pool missing tick data"); + None + } + } + }) + .collect()) + } + + /// Fetches `missing` at the source's head. Gated on `fetch_on_demand` (set + /// at construction from the source type), so it is a no-op for the + /// subgraph; those misses wait for the maintenance run instead. It + /// fetches at the head, not the checkpoint block: the checkpoint can + /// sit ahead of the indexer's served head, so waiting on it would hang + /// the quote. A failed fetch yields fewer pools rather than failing the + /// whole quote. + async fn fetch_missing_on_demand(&self, missing: &[Address]) -> Vec<(Address, Arc)> { + if missing.is_empty() || !self.fetch_on_demand { + return Vec::new(); + } + self.fetch_pools(missing, BlockTarget::Latest) + .await + .inspect_err(|err| tracing::debug!(?err, "on-demand pool fetch failed")) + .unwrap_or_default() + } + + /// Fetches the pools flagged missing by `get` at the checkpoint block and + /// caches them. Runs from periodic maintenance. async fn update_missing_pools(&self) -> Result<()> { - let (missing_pools, block_number) = { + // Clone out and drop the lock before the async fetch. + let (missing, block_number) = { let checkpoint = self.pools_checkpoint.lock().unwrap(); - if checkpoint.missing_pools.is_empty() { - return Ok(()); - } (checkpoint.missing_pools.clone(), checkpoint.block_number) }; - tracing::debug!("currently missing pools are {:?}", missing_pools); - - let pool_ids = missing_pools.into_iter().collect::>(); - let start = std::time::Instant::now(); - let pools_with_ticks = self - .source - .get_pools_with_ticks_by_ids(&pool_ids, block_number) - .await; - tracing::debug!( - requested_pools = pool_ids.len(), - time = ?start.elapsed(), - request_successful = pools_with_ticks.is_ok(), - "fetched pool ticks" - ); - let pools_with_ticks = pools_with_ticks?; + if missing.is_empty() { + return Ok(()); + } + let fetched = self + .fetch_pools(&Vec::from_iter(missing), BlockTarget::Number(block_number)) + .await?; let mut checkpoint = self.pools_checkpoint.lock().unwrap(); - for pool in pools_with_ticks.pools { - checkpoint.missing_pools.remove(&pool.id); - checkpoint.pools.insert(pool.id, Arc::new(pool.try_into()?)); + for (id, info) in fetched { + checkpoint.missing_pools.remove(&id); + checkpoint.pools.insert(id, info); } - - tracing::debug!("number of cached pools is {}", checkpoint.pools.len()); if !checkpoint.missing_pools.is_empty() { tracing::warn!( - "not all missing pools updated: {:?}", - checkpoint.missing_pools + remaining = checkpoint.missing_pools.len(), + "not all missing pools updated" ); } Ok(()) @@ -316,14 +363,19 @@ pub struct UniswapV3PoolFetcher { impl UniswapV3PoolFetcher { pub async fn new( source: Arc, + fetch_on_demand: bool, web3: Web3, block_retriever: Arc, max_pools_to_initialize: usize, ) -> Result { let web3 = web3.labeled("uniswapV3"); - let checkpoint = - PoolsCheckpointHandler::new(source, block_retriever.clone(), max_pools_to_initialize) - .await?; + let checkpoint = PoolsCheckpointHandler::new( + source, + fetch_on_demand, + block_retriever.clone(), + max_pools_to_initialize, + ) + .await?; let init_block = checkpoint.pools_checkpoint.lock().unwrap().block_number; let init_block = block_retriever.block(init_block).await?; @@ -422,7 +474,16 @@ impl PoolFetching for UniswapV3PoolFetcher { // this is the only place where this function uses checkpoint - no data racing // between maintenance - let (mut checkpoint, checkpoint_block_number) = self.checkpoint.get(token_pairs); + let CachedPools { + pools: mut checkpoint, + missing, + block_number: checkpoint_block_number, + } = self.checkpoint.get(token_pairs); + + // No pools registered for these pairs: nothing to fetch or replay. + if checkpoint.is_empty() && missing.is_empty() { + return Ok(Vec::new()); + } if block_number > checkpoint_block_number { let block_range = RangeInclusive::try_new(checkpoint_block_number + 1, block_number)?; @@ -430,6 +491,11 @@ impl PoolFetching for UniswapV3PoolFetcher { append_events(&mut checkpoint, events); } + // The warm cache holds only the top pools by liquidity, so many + // registered pairs are absent. Resolve those misses on-demand and merge + // them after the replay; they come back current, so they aren't replayed. + checkpoint.extend(self.checkpoint.fetch_missing_on_demand(&missing).await); + // return only pools which current liquidity is positive Ok(checkpoint .into_values() @@ -749,4 +815,205 @@ mod tests { ]) ); } + + /// Serves a fixed set of pools (with ticks) from + /// `get_pools_with_ticks_by_ids` so the on-demand fetch path can be + /// exercised without a real source. `served_block` models the indexer's + /// head: a request for a higher `target_block` fails, mirroring the real + /// client's `wait_until` blocking on a block the indexer hasn't reached. + struct StubSource { + with_ticks: HashMap, + served_block: u64, + } + + impl StubSource { + fn new(pools: impl IntoIterator) -> Self { + Self { + with_ticks: pools.into_iter().map(|p| (p.id, p)).collect(), + served_block: u64::MAX, + } + } + } + + #[async_trait::async_trait] + impl V3PoolDataSource for StubSource { + async fn get_registered_pools( + &self, + _target_block: BlockTarget, + ) -> Result { + Ok(Default::default()) + } + + async fn get_pools_with_ticks_by_ids( + &self, + ids: &[Address], + target_block: BlockTarget, + ) -> Result { + let target_block = match target_block { + BlockTarget::Latest => self.served_block, + BlockTarget::Number(n) => n, + }; + anyhow::ensure!( + target_block <= self.served_block, + "indexer at {} hasn't reached target block {target_block}", + self.served_block, + ); + let pools = ids + .iter() + .filter_map(|id| self.with_ticks.get(id).cloned()) + .collect(); + Ok(crate::uniswap_v3::graph_api::PoolsWithTicks { + fetched_block_number: self.served_block, + pools, + }) + } + } + + fn pool_with_ticks(id: Address, token0: Address, token1: Address) -> PoolData { + PoolData { + id, + token0: Token { + id: token0, + decimals: 6, + }, + token1: Token { + id: token1, + decimals: 18, + }, + fee_tier: U256::from(3000), + liquidity: U256::from(1_000_000u64), + sqrt_price: U256::from(1u64), + tick: 0, + ticks: Some(vec![crate::uniswap_v3::graph_api::TickData { + tick_idx: -100, + liquidity_net: 1_000, + pool_address: id, + }]), + block_number: 100, + } + } + + fn handler(source: StubSource, checkpoint: PoolsCheckpoint) -> PoolsCheckpointHandler { + PoolsCheckpointHandler { + source: Arc::new(source), + fetch_on_demand: false, + pools_by_token_pair: HashMap::new(), + pools_checkpoint: Mutex::new(checkpoint), + } + } + + /// A pool registered for a pair but absent from the warm cache is returned + /// in `missing` (not `pools`), so the fetch path knows to resolve it. + #[test] + fn get_flags_registered_uncached_pool_as_missing() { + let token0 = Address::with_last_byte(1); + let token1 = Address::with_last_byte(2); + let pair = TokenPair::new(token0, token1).unwrap(); + let pool = Address::with_last_byte(9); + + let mut handler = handler( + StubSource::new([]), + PoolsCheckpoint { + pools: HashMap::new(), + block_number: 100, + missing_pools: HashSet::new(), + }, + ); + handler.pools_by_token_pair = HashMap::from([(pair, vec![pool])]); + + let CachedPools { pools, missing, .. } = handler.get(&HashSet::from([pair])); + assert!(pools.is_empty()); + assert_eq!(missing, vec![pool]); + } + + /// The on-demand path must not block on the checkpoint block (which can sit + /// persistently ahead of the indexer's served block); it fetches at the + /// indexer's latest block. A source that errors for any block above its + /// head still yields the pool via a `BlockTarget::Latest` fetch. + #[tokio::test] + async fn on_demand_does_not_wait_for_future_block() { + let token0 = Address::with_last_byte(1); + let token1 = Address::with_last_byte(2); + let pool = Address::with_last_byte(9); + + let mut source = StubSource::new([pool_with_ticks(pool, token0, token1)]); + source.served_block = 50; // indexer behind the checkpoint below + + let handler = handler( + source, + PoolsCheckpoint { + pools: HashMap::new(), + block_number: 100, // checkpoint ahead of the indexer + missing_pools: HashSet::new(), + }, + ); + + // Latest serves at-head, so it succeeds despite served_block < checkpoint. + let fetched = handler + .fetch_pools(&[pool], BlockTarget::Latest) + .await + .unwrap(); + assert_eq!(fetched.len(), 1); + assert_eq!(fetched[0].0, pool); + + // Fetching at the checkpoint block would fail (indexer hasn't reached it). + assert!( + handler + .fetch_pools(&[pool], BlockTarget::Number(100)) + .await + .is_err() + ); + } + + /// Unknown pairs have no registered pools, so `get` reports block 0 with + /// nothing cached or missing — the signal `fetch` uses to skip the replay. + #[test] + fn get_reports_zero_block_for_unknown_pairs() { + let handler = handler( + StubSource::new([]), + PoolsCheckpoint { + pools: HashMap::new(), + block_number: 100, + missing_pools: HashSet::new(), + }, + ); + let pair = TokenPair::new(Address::with_last_byte(1), Address::with_last_byte(2)).unwrap(); + let CachedPools { + pools, + missing, + block_number, + } = handler.get(&HashSet::from([pair])); + assert!(pools.is_empty()); + assert!(missing.is_empty()); + assert_eq!(block_number, 0); + } + + /// A pool that can't be converted (e.g. ticks not yet available) is skipped + /// rather than failing the whole batch; the convertible pool is returned. + #[tokio::test] + async fn fetch_pools_skips_unconvertible_pool() { + let token0 = Address::with_last_byte(1); + let token1 = Address::with_last_byte(2); + let good = Address::with_last_byte(9); + let bad = Address::with_last_byte(10); + + let mut bad_pool = pool_with_ticks(bad, token0, token1); + bad_pool.ticks = None; // PoolInfo::try_from fails on missing ticks + + let handler = handler( + StubSource::new([pool_with_ticks(good, token0, token1), bad_pool]), + PoolsCheckpoint { + pools: HashMap::new(), + block_number: 100, + missing_pools: HashSet::new(), + }, + ); + + let fetched = handler + .fetch_pools(&[good, bad], BlockTarget::Latest) + .await + .unwrap(); + assert_eq!(fetched.len(), 1); + assert_eq!(fetched[0].0, good); + } } diff --git a/crates/liquidity-sources/src/uniswap_v3/pool_indexer.rs b/crates/liquidity-sources/src/uniswap_v3/pool_indexer.rs index 51cefc9fa0..e0e9876d10 100644 --- a/crates/liquidity-sources/src/uniswap_v3/pool_indexer.rs +++ b/crates/liquidity-sources/src/uniswap_v3/pool_indexer.rs @@ -12,6 +12,7 @@ use { crate::uniswap_v3::{ + BlockTarget, V3PoolDataSource, graph_api::{PoolData, PoolsWithTicks, RegisteredPools, TickData, Token}, }, @@ -56,6 +57,13 @@ impl PoolIndexerClient { fn path(&self, suffix: &str) -> Url { url_join(&self.base_url, suffix) } + + /// The indexer serves at-head from its own DB, so an on-demand cache-miss + /// fetch is cheap enough for the quote path (unlike the subgraph). Read at + /// construction to gate `UniswapV3PoolFetcher`'s on-demand fetch. + pub fn fetch_on_demand(&self) -> bool { + true + } } /// Joins `path` onto `url` with exactly one slash between them. Reusing @@ -214,10 +222,15 @@ impl PoolIndexerClient { /// Polls `/pools?limit=1` every [`WAIT_UNTIL_POLL_INTERVAL`] until the /// envelope reports `block_number >= target_block`. Returns /// immediately if already there; bails after [`WAIT_UNTIL_TIMEOUT`]. + /// [`BlockTarget::Latest`] serves at-head without waiting. /// /// `503` is treated as "still bootstrapping" and the loop keeps /// polling. Other non-2xx statuses propagate as errors. - async fn wait_until(&self, target_block: u64) -> Result<()> { + async fn wait_until(&self, target_block: BlockTarget) -> Result<()> { + let target_block = match target_block { + BlockTarget::Latest => return Ok(()), + BlockTarget::Number(n) => n, + }; let deadline = std::time::Instant::now() + WAIT_UNTIL_TIMEOUT; let mut last_observed: Option = None; let mut interval = tokio::time::interval(WAIT_UNTIL_POLL_INTERVAL); @@ -280,7 +293,7 @@ impl PoolIndexerClient { #[async_trait] impl V3PoolDataSource for PoolIndexerClient { - async fn get_registered_pools(&self, target_block: u64) -> Result { + async fn get_registered_pools(&self, target_block: BlockTarget) -> Result { self.wait_until(target_block).await?; // The indexer can advance between pages, so each pool carries the // block of the page it arrived in. The envelope's @@ -321,15 +334,12 @@ impl V3PoolDataSource for PoolIndexerClient { async fn get_pools_with_ticks_by_ids( &self, ids: &[Address], - target_block: u64, + target_block: BlockTarget, ) -> Result { self.wait_until(target_block).await?; if ids.is_empty() { - return Ok(PoolsWithTicks { - fetched_block_number: target_block, - pools: Vec::new(), - }); + return Ok(PoolsWithTicks::default()); } let mut out: Vec = Vec::with_capacity(ids.len());