feat: support cast logs query chunking (#12692)

* feat: support cast logs query chunking

* comments

* comments

* docs

* comments

Author: stevencartavia

crates/cast/src/cmd/logs.rs

@@ -45,14 +45,27 @@ pub struct LogsArgs {
     #[arg(long)]
     subscribe: bool,
 
+    /// Number of blocks to query in each chunk when the provider has range limits.
+    /// Defaults to 10000 blocks per chunk.
+    #[arg(long, default_value_t = 10000)]
+    query_size: u64,
+
     #[command(flatten)]
     eth: EthereumOpts,
 }
 
 impl LogsArgs {
     pub async fn run(self) -> Result<()> {
-        let Self { from_block, to_block, address, sig_or_topic, topics_or_args, subscribe, eth } =
-            self;
+        let Self {
+            from_block,
+            to_block,
+            address,
+            sig_or_topic,
+            topics_or_args,
+            subscribe,
+            query_size,
+            eth,
+        } = self;
 
         let config = eth.load_config()?;
         let provider = utils::get_provider(&config)?;
@@ -77,7 +90,7 @@ impl LogsArgs {
         let filter = build_filter(from_block, to_block, addresses, sig_or_topic, topics_or_args)?;
 
         if !subscribe {
-            let logs = cast.filter_logs(filter).await?;
+            let logs = cast.filter_logs_chunked(filter, query_size).await?;
             sh_println!("{logs}")?;
             return Ok(());
         }

crates/cast/src/lib.rs

@@ -19,7 +19,8 @@ use alloy_provider::{
 };
 use alloy_rlp::Decodable;
 use alloy_rpc_types::{
-    BlockId, BlockNumberOrTag, BlockOverrides, Filter, TransactionRequest, state::StateOverride,
+    BlockId, BlockNumberOrTag, BlockOverrides, Filter, FilterBlockOption, Log, TransactionRequest,
+    state::StateOverride,
 };
 use alloy_serde::WithOtherFields;
 use base::{Base, NumberWithBase, ToBase};
@@ -76,7 +77,7 @@ pub struct Cast<P> {
     provider: P,
 }
 
-impl<P: Provider<AnyNetwork>> Cast<P> {
+impl<P: Provider<AnyNetwork> + Clone + Unpin> Cast<P> {
     /// Creates a new Cast instance from the provided client
     ///
     /// # Example
@@ -964,7 +965,19 @@ impl<P: Provider<AnyNetwork>> Cast<P> {
 
     pub async fn filter_logs(&self, filter: Filter) -> Result<String> {
         let logs = self.provider.get_logs(&filter).await?;
+        Self::format_logs(logs)
+    }
 
+    /// Retrieves logs using chunked requests to handle large block ranges.
+    ///
+    /// Automatically divides large block ranges into smaller chunks to avoid provider limits
+    /// and processes them with controlled concurrency to prevent rate limiting.
+    pub async fn filter_logs_chunked(&self, filter: Filter, chunk_size: u64) -> Result<String> {
+        let logs = self.get_logs_chunked(&filter, chunk_size).await?;
+        Self::format_logs(logs)
+    }
+
+    fn format_logs(logs: Vec<Log>) -> Result<String> {
         let res = if shell::is_json() {
             serde_json::to_string(&logs)?
         } else {
@@ -981,6 +994,100 @@ impl<P: Provider<AnyNetwork>> Cast<P> {
         Ok(res)
     }
 
+    fn extract_block_range(filter: &Filter) -> (Option<u64>, Option<u64>) {
+        let FilterBlockOption::Range { from_block, to_block } = &filter.block_option else {
+            return (None, None);
+        };
+
+        (from_block.and_then(|b| b.as_number()), to_block.and_then(|b| b.as_number()))
+    }
+
+    /// Retrieves logs with automatic chunking fallback.
+    ///
+    /// First tries to fetch logs for the entire range. If that fails,
+    /// falls back to concurrent chunked requests with rate limiting.
+    async fn get_logs_chunked(&self, filter: &Filter, chunk_size: u64) -> Result<Vec<Log>>
+    where
+        P: Clone + Unpin,
+    {
+        // Try the full range first
+        if let Ok(logs) = self.provider.get_logs(filter).await {
+            return Ok(logs);
+        }
+
+        // Fallback: use concurrent chunked approach
+        self.get_logs_chunked_concurrent(filter, chunk_size).await
+    }
+
+    /// Retrieves logs using concurrent chunked requests with rate limiting.
+    ///
+    /// Divides the block range into chunks and processes them with a maximum of
+    /// 5 concurrent requests. Falls back to single-block queries if chunks fail.
+    async fn get_logs_chunked_concurrent(
+        &self,
+        filter: &Filter,
+        chunk_size: u64,
+    ) -> Result<Vec<Log>>
+    where
+        P: Clone + Unpin,
+    {
+        let (from_block, to_block) = Self::extract_block_range(filter);
+        let (Some(from), Some(to)) = (from_block, to_block) else {
+            return self.provider.get_logs(filter).await.map_err(Into::into);
+        };
+
+        if from >= to {
+            return Ok(vec![]);
+        }
+
+        // Create chunk ranges using iterator
+        let chunk_ranges: Vec<(u64, u64)> = (from..to)
+            .step_by(chunk_size as usize)
+            .map(|chunk_start| (chunk_start, (chunk_start + chunk_size).min(to)))
+            .collect();
+
+        // Process chunks with controlled concurrency using buffered stream
+        let mut all_results: Vec<(u64, Vec<Log>)> = futures::stream::iter(chunk_ranges)
+            .map(|(start_block, chunk_end)| {
+                let chunk_filter = filter.clone().from_block(start_block).to_block(chunk_end - 1);
+                let provider = self.provider.clone();
+
+                async move {
+                    // Try direct chunk request with simplified fallback
+                    match provider.get_logs(&chunk_filter).await {
+                        Ok(logs) => (start_block, logs),
+                        Err(_) => {
+                            // Simple fallback: try individual blocks in this chunk
+                            let mut fallback_logs = Vec::new();
+                            for single_block in start_block..chunk_end {
+                                let single_filter = chunk_filter
+                                    .clone()
+                                    .from_block(single_block)
+                                    .to_block(single_block);
+                                if let Ok(logs) = provider.get_logs(&single_filter).await {
+                                    fallback_logs.extend(logs);
+                                }
+                            }
+                            (start_block, fallback_logs)
+                        }
+                    }
+                }
+            })
+            .buffered(5) // Limit to 5 concurrent requests to avoid rate limits
+            .collect()
+            .await;
+
+        // Sort once at the end by block number and flatten
+        all_results.sort_by_key(|(block_num, _)| *block_num);
+
+        let mut all_logs = Vec::new();
+        for (_, logs) in all_results {
+            all_logs.extend(logs);
+        }
+
+        Ok(all_logs)
+    }
+
     /// Converts a block identifier into a block number.
     ///
     /// If the block identifier is a block number, then this function returns the block number. If

crates/cast/tests/cli/main.rs

@@ -1611,6 +1611,24 @@ casttest!(logs_sig_2, |_prj, cmd| {
     .stdout_eq(file!["../fixtures/cast_logs.stdout"]);
 });
 
+casttest!(logs_chunked_large_range, |_prj, cmd| {
+    let rpc = next_http_archive_rpc_url();
+    cmd.args([
+        "logs",
+        "--rpc-url",
+        rpc.as_str(),
+        "--from-block",
+        "18000000",
+        "--to-block",
+        "18050000",
+        "--query-size",
+        "1000",
+        "Transfer(address indexed from, address indexed to, uint256 value)",
+        "0xA0b86a33E6441d02dd8C6B2b7E5D1E3eD7F73b4b",
+    ])
+    .assert_success();
+});
+
 casttest!(mktx, |_prj, cmd| {
     cmd.args([
         "mktx",