Fetch Logs From Block Ranges in Parallel (#249)

Author: 0xNeshi

Cargo.lock

@@ -24,6 +24,7 @@ all = "warn"
 alloy = { version = "1.1.2", features = ["full"] }
 alloy-node-bindings = "1.1.2"
 tokio = { version = "1.48", features = ["full"] }
+futures = "0.3.31"
 anyhow = "1.0"
 thiserror = "2.0.17"
 tokio-stream = "0.1.17"
@@ -61,6 +62,7 @@ tokio-stream.workspace = true
 tracing.workspace = true
 backon.workspace = true
 tokio-util = "0.7.17"
+futures.workspace = true
 
 [dev-dependencies]
 tracing-subscriber.workspace = true

Cargo.toml

@@ -31,6 +31,9 @@ pub enum ScannerError {
     #[error("Max block range must be greater than 0")]
     InvalidMaxBlockRange,
 
+    #[error("Max concurrent fetches must be greater than 0")]
+    InvalidMaxConcurrentFetches,
+
     #[error("Subscription closed")]
     SubscriptionClosed,
 }

src/error.rs

@@ -6,6 +6,6 @@ mod scanner;
 pub use filter::EventFilter;
 pub use message::{EventScannerResult, Message};
 pub use scanner::{
-    EventScanner, EventScannerBuilder, Historic, LatestEvents, Live, SyncFromBlock,
-    SyncFromLatestEvents,
+    DEFAULT_MAX_CONCURRENT_FETCHES, EventScanner, EventScannerBuilder, Historic, LatestEvents,
+    Live, SyncFromBlock, SyncFromLatestEvents,
 };

src/event_scanner/mod.rs

@@ -1,7 +1,7 @@
 use std::ops::RangeInclusive;
 
 use crate::{
-    Notification, ScannerMessage,
+    Message, Notification, ScannerError, ScannerMessage,
     block_range_scanner::{BlockScannerResult, MAX_BUFFERED_MESSAGES},
     event_scanner::{filter::EventFilter, listener::EventListener},
     robust_provider::{RobustProvider, provider::Error as RobustProviderError},
@@ -11,15 +11,19 @@ use alloy::{
     network::Network,
     rpc::types::{Filter, Log},
 };
+use futures::StreamExt;
 use tokio::{
-    sync::broadcast::{self, Sender, error::RecvError},
+    sync::{
+        broadcast::{self, Sender, error::RecvError},
+        mpsc,
+    },
     task::JoinSet,
 };
-use tokio_stream::{Stream, StreamExt};
-use tracing::{error, info, warn};
+use tokio_stream::{Stream, wrappers::ReceiverStream};
+use tracing::{debug, error, info, warn};
 
 #[derive(Copy, Clone, Debug)]
-pub enum ConsumerMode {
+pub(crate) enum ConsumerMode {
     Stream,
     CollectLatest { count: usize },
 }
@@ -46,16 +50,22 @@ pub enum ConsumerMode {
 /// # Note
 ///
 /// Assumes it is running in a separate tokio task, so as to be non-blocking.
-pub async fn handle_stream<N: Network, S: Stream<Item = BlockScannerResult> + Unpin>(
+pub(crate) async fn handle_stream<N: Network, S: Stream<Item = BlockScannerResult> + Unpin>(
     mut stream: S,
     provider: &RobustProvider<N>,
     listeners: &[EventListener],
     mode: ConsumerMode,
+    max_concurrent_fetches: usize,
 ) {
     let (range_tx, _) = broadcast::channel::<BlockScannerResult>(MAX_BUFFERED_MESSAGES);
 
     let consumers = match mode {
-        ConsumerMode::Stream => spawn_log_consumers_in_stream_mode(provider, listeners, &range_tx),
+        ConsumerMode::Stream => spawn_log_consumers_in_stream_mode(
+            provider,
+            listeners,
+            &range_tx,
+            max_concurrent_fetches,
+        ),
         ConsumerMode::CollectLatest { count } => {
             spawn_log_consumers_in_collection_mode(provider, listeners, &range_tx, count)
         }
@@ -76,10 +86,11 @@ pub async fn handle_stream<N: Network, S: Stream<Item = BlockScannerResult> + Un
 }
 
 #[must_use]
-pub fn spawn_log_consumers_in_stream_mode<N: Network>(
+fn spawn_log_consumers_in_stream_mode<N: Network>(
     provider: &RobustProvider<N>,
     listeners: &[EventListener],
     range_tx: &Sender<BlockScannerResult>,
+    max_concurrent_fetches: usize,
 ) -> JoinSet<()> {
     listeners.iter().cloned().fold(JoinSet::new(), |mut set, listener| {
         let EventListener { filter, sender } = listener;
@@ -89,56 +100,72 @@ pub fn spawn_log_consumers_in_stream_mode<N: Network>(
         let mut range_rx = range_tx.subscribe();
 
         set.spawn(async move {
-            loop {
-                match range_rx.recv().await {
-                    Ok(message) => match message {
+            // We use a channel and convert the receiver to a stream because it already has a
+            // convenience function `buffered` for concurrently handling block ranges, while
+            // outputting results in the same order as they were received.
+            let (tx, rx) = mpsc::channel::<BlockScannerResult>(max_concurrent_fetches);
+
+            // Process block ranges concurrently in a separate thread so that the current thread can
+            // continue receiving and buffering subsequent block ranges while the previous ones are
+            // being processed.
+            tokio::spawn(async move {
+                let mut stream = ReceiverStream::new(rx)
+                    .map(async |message| match message {
                         Ok(ScannerMessage::Data(range)) => {
-                            match get_logs(range, &filter, &base_filter, &provider).await {
-                                Ok(logs) => {
-                                    if logs.is_empty() {
-                                        continue;
-                                    }
-
-                                    if !sender.try_stream(logs).await {
-                                        return;
-                                    }
-                                }
-                                Err(e) => {
-                                    error!(error = ?e, "Received error message");
-                                    if !sender.try_stream(e).await {
-                                        return;
-                                    }
-                                }
-                            }
-                        }
-                        Ok(ScannerMessage::Notification(notification)) => {
-                            info!(notification = ?notification, "Received notification");
-                            if !sender.try_stream(notification).await {
-                                return;
-                            }
+                            get_logs(range, &filter, &base_filter, &provider)
+                                .await
+                                .map(Message::from)
+                                .map_err(ScannerError::from)
                         }
-                        Err(e) => {
-                            error!(error = ?e, "Received error message");
-                            if !sender.try_stream(e).await {
-                                return;
-                            }
-                        }
-                    },
+                        Ok(ScannerMessage::Notification(notification)) => Ok(notification.into()),
+                        // No need to stop the stream on an error, because that decision is up to
+                        // the caller.
+                        Err(e) => Err(e),
+                    })
+                    .buffered(max_concurrent_fetches);
+
+                // process all of the buffered results
+                while let Some(result) = stream.next().await {
+                    if let Ok(ScannerMessage::Data(logs)) = result.as_ref() &&
+                        logs.is_empty()
+                    {
+                        continue;
+                    }
+
+                    if !sender.try_stream(result).await {
+                        return;
+                    }
+                }
+            });
+
+            // Receive block ranges from the broadcast channel and send them to the range processor
+            // for parallel processing.
+            loop {
+                match range_rx.recv().await {
+                    Ok(message) => {
+                        tx.send(message).await.expect("receiver dropped only if we exit this loop");
+                    }
                     Err(RecvError::Closed) => {
-                        info!("No block ranges to receive, dropping receiver.");
+                        debug!("No more block ranges to receive");
                         break;
                     }
-                    Err(RecvError::Lagged(_)) => {}
+                    Err(RecvError::Lagged(skipped)) => {
+                        debug!("Channel lagged, skipped {skipped} messages");
+                    }
                 }
             }
+
+            // Drop the local channel sender to signal to the range processor that streaming is
+            // done.
+            drop(tx);
         });
 
         set
     })
 }
 
 #[must_use]
-pub fn spawn_log_consumers_in_collection_mode<N: Network>(
+fn spawn_log_consumers_in_collection_mode<N: Network>(
     provider: &RobustProvider<N>,
     listeners: &[EventListener],
     range_tx: &Sender<BlockScannerResult>,

src/event_scanner/scanner/common.rs

@@ -36,6 +36,23 @@ impl EventScannerBuilder<Historic> {
         self
     }
 
+    /// Sets the maximum number of block-range fetches to process concurrently when
+    /// scanning a historical block range.
+    ///
+    /// Increasing this value can improve throughput by issuing multiple RPC
+    /// requests concurrently, at the cost of higher load on the provider.
+    ///
+    /// Must be greater than 0.
+    ///
+    /// Defaults to [`DEFAULT_MAX_CONCURRENT_FETCHES`][default].
+    ///
+    /// [default]: crate::event_scanner::scanner::DEFAULT_MAX_CONCURRENT_FETCHES
+    #[must_use]
+    pub fn max_concurrent_fetches(mut self, max_concurrent_fetches: usize) -> Self {
+        self.config.max_concurrent_fetches = max_concurrent_fetches;
+        self
+    }
+
     /// Connects to an existing provider with block range validation.
     ///
     /// Validates that the maximum of `from_block` and `to_block` does not exceed
@@ -51,6 +68,10 @@ impl EventScannerBuilder<Historic> {
         self,
         provider: impl IntoRobustProvider<N>,
     ) -> Result<EventScanner<Historic, N>, ScannerError> {
+        if self.config.max_concurrent_fetches == 0 {
+            return Err(ScannerError::InvalidMaxConcurrentFetches);
+        }
+
         let scanner = self.build(provider).await?;
 
         let provider = scanner.block_range_scanner.provider();
@@ -120,11 +141,19 @@ impl<N: Network> EventScanner<Historic, N> {
         let client = self.block_range_scanner.run()?;
         let stream = client.stream_historical(self.config.from_block, self.config.to_block).await?;
 
+        let max_concurrent_fetches = self.config.max_concurrent_fetches;
         let provider = self.block_range_scanner.provider().clone();
         let listeners = self.listeners.clone();
 
         tokio::spawn(async move {
-            handle_stream(stream, &provider, &listeners, ConsumerMode::Stream).await;
+            handle_stream(
+                stream,
+                &provider,
+                &listeners,
+                ConsumerMode::Stream,
+                max_concurrent_fetches,
+            )
+            .await;
         });
 
         Ok(())
@@ -133,6 +162,8 @@ impl<N: Network> EventScanner<Historic, N> {
 
 #[cfg(test)]
 mod tests {
+    use crate::event_scanner::scanner::DEFAULT_MAX_CONCURRENT_FETCHES;
+
     use super::*;
     use alloy::{
         eips::BlockNumberOrTag,
@@ -145,22 +176,25 @@ mod tests {
 
     #[test]
     fn test_historic_scanner_builder_pattern() {
-        let builder =
-            EventScannerBuilder::historic().to_block(200).max_block_range(50).from_block(100);
+        let builder = EventScannerBuilder::historic()
+            .to_block(200)
+            .max_block_range(50)
+            .max_concurrent_fetches(10)
+            .from_block(100);
 
         assert_eq!(builder.config.from_block, BlockNumberOrTag::Number(100).into());
         assert_eq!(builder.config.to_block, BlockNumberOrTag::Number(200).into());
+        assert_eq!(builder.config.max_concurrent_fetches, 10);
         assert_eq!(builder.block_range_scanner.max_block_range, 50);
     }
 
     #[test]
-    fn test_historic_scanner_builder_with_different_block_types() {
-        let builder = EventScannerBuilder::historic()
-            .from_block(BlockNumberOrTag::Earliest)
-            .to_block(BlockNumberOrTag::Latest);
+    fn test_historic_scanner_builder_with_default_values() {
+        let builder = EventScannerBuilder::historic();
 
         assert_eq!(builder.config.from_block, BlockNumberOrTag::Earliest.into());
         assert_eq!(builder.config.to_block, BlockNumberOrTag::Latest.into());
+        assert_eq!(builder.config.max_concurrent_fetches, DEFAULT_MAX_CONCURRENT_FETCHES);
     }
 
     #[test]
@@ -172,11 +206,14 @@ mod tests {
             .from_block(1)
             .from_block(2)
             .to_block(100)
-            .to_block(200);
+            .to_block(200)
+            .max_concurrent_fetches(10)
+            .max_concurrent_fetches(20);
 
         assert_eq!(builder.block_range_scanner.max_block_range, 105);
         assert_eq!(builder.config.from_block, BlockNumberOrTag::Number(2).into());
         assert_eq!(builder.config.to_block, BlockNumberOrTag::Number(200).into());
+        assert_eq!(builder.config.max_concurrent_fetches, 20);
     }
 
     #[tokio::test]
@@ -256,6 +293,15 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn returns_error_with_zero_max_concurrent_fetches() {
+        let provider = RootProvider::<Ethereum>::new(RpcClient::mocked(Asserter::new()));
+        let result =
+            EventScannerBuilder::historic().max_concurrent_fetches(0).connect(provider).await;
+
+        assert!(matches!(result, Err(ScannerError::InvalidMaxConcurrentFetches)));
+    }
+
     #[tokio::test]
     async fn test_historic_scanner_with_valid_block_hash() {
         let anvil = Anvil::new().try_spawn().unwrap();