Re-introduce on-demand V3 pool fetch, gated to the indexer

crates/driver/src/boundary/liquidity/uniswap/v3.rs

@@ -114,11 +114,32 @@ async fn init_liquidity(
     config: &infra::liquidity::config::UniswapV3,
 ) -> anyhow::Result<impl LiquidityCollecting + use<>> {
     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<dyn V3PoolDataSource> = 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<Arc<dyn V3PoolDataSource>> {
-    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")?,
-            ))
-        }
-    }
-}

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<u64> {
+        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<RegisteredPools> {
+    async fn get_registered_pools(&self, target_block: BlockTarget) -> Result<RegisteredPools> {
+        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<PoolsWithTicks> {
+        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)

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<RegisteredPools>;
+    /// 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<RegisteredPools>;
 
-    /// 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<PoolsWithTicks>;
 }
+
+/// 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),
+}