merge main

Author: 0xNeshi

.cargo/config.toml

@@ -0,0 +1,2 @@
+[alias]
+nt = "nextest run --all-features"

src/block_range_scanner.rs

@@ -384,7 +384,7 @@ impl<N: Network> Service<N> {
         let provider = self.provider.clone();
 
         let (start_block, end_block) =
-            try_join!(self.provider.get_block(start_id), self.provider.get_block(end_id),)?;
+            try_join!(self.provider.get_block(start_id), self.provider.get_block(end_id))?;
 
         // normalize block range
         let (from, to) = match start_block.header().number().cmp(&end_block.header().number()) {

src/block_range_scanner/sync_handler.rs

@@ -1,16 +1,11 @@
-use alloy::{
-    consensus::BlockHeader,
-    eips::{BlockId, BlockNumberOrTag},
-    network::{BlockResponse, Network},
-    primitives::BlockNumber,
-};
+use alloy::{eips::BlockId, network::Network, primitives::BlockNumber};
 use tokio::sync::mpsc;
 use tracing::{error, info};
 
 use crate::{
     Notification, ScannerError,
     block_range_scanner::{BlockScannerResult, common, reorg_handler::ReorgHandler},
-    robust_provider::{self, RobustProvider},
+    robust_provider::RobustProvider,
     types::TryStream,
 };
 
@@ -69,16 +64,8 @@ impl<N: Network> SyncHandler<N> {
 
     /// Determines whether we need to catch up or can start live immediately
     async fn determine_sync_state(&self) -> Result<SyncState, ScannerError> {
-        let get_start_block = async || -> Result<BlockNumber, robust_provider::Error> {
-            let block = match self.start_id {
-                BlockId::Number(BlockNumberOrTag::Number(num)) => num,
-                _ => self.provider.get_block(self.start_id).await?.header().number(),
-            };
-            Ok(block)
-        };
-
         let (start_block, confirmed_tip) = tokio::try_join!(
-            get_start_block(),
+            self.provider.get_block_number_by_id(self.start_id),
             self.provider.get_latest_confirmed(self.block_confirmations)
         )?;
 

src/event_scanner/scanner/common.rs

@@ -1,7 +1,7 @@
 use std::ops::RangeInclusive;
 
 use crate::{
-    ScannerMessage,
+    Notification, ScannerMessage,
     block_range_scanner::{BlockScannerResult, MAX_BUFFERED_MESSAGES},
     event_scanner::{EventScannerResult, filter::EventFilter, listener::EventListener},
     robust_provider::{Error as RobustProviderError, RobustProvider},
@@ -120,10 +120,15 @@ pub fn spawn_log_consumers<N: Network>(
             }
 
             if let ConsumerMode::CollectLatest { .. } = mode {
-                if !collected.is_empty() {
-                    collected.reverse(); // restore chronological order
+                if collected.is_empty() {
+                    info!("No logs found");
+                    _ = sender.try_stream(Notification::NoPastLogsFound).await;
+                    return;
                 }
 
+                info!(count = collected.len(), "Logs found");
+                collected.reverse(); // restore chronological order
+
                 info!("Sending collected logs to consumer");
                 _ = sender.try_stream(collected).await;
             }

src/event_scanner/scanner/historic.rs

@@ -51,17 +51,17 @@ impl EventScannerBuilder<Historic> {
             }
         };
 
+        if from_num > latest_block {
+            Err(ScannerError::BlockExceedsLatest("from_block", from_num, latest_block))?;
+        }
+
         let to_num = match scanner.config.to_block {
             BlockId::Number(to_block) => to_block.as_number().unwrap_or(0),
             BlockId::Hash(to_hash) => {
                 provider.get_block_by_hash(to_hash.into()).await?.header().number()
             }
         };
 
-        if from_num > latest_block {
-            Err(ScannerError::BlockExceedsLatest("from_block", from_num, latest_block))?;
-        }
-
         if to_num > latest_block {
             Err(ScannerError::BlockExceedsLatest("to_block", to_num, latest_block))?;
         }

src/event_scanner/scanner/latest.rs

@@ -1,4 +1,8 @@
-use alloy::{eips::BlockId, network::Network};
+use alloy::{
+    consensus::BlockHeader,
+    eips::BlockId,
+    network::{BlockResponse, Network},
+};
 
 use super::common::{ConsumerMode, handle_stream};
 use crate::{
@@ -45,13 +49,28 @@ impl EventScannerBuilder<LatestEvents> {
         let scanner = self.build(provider).await?;
 
         let provider = scanner.block_range_scanner.provider();
+        let latest_block = provider.get_block_number().await?;
+
+        let from_num = match scanner.config.from_block {
+            BlockId::Number(from_block) => from_block.as_number().unwrap_or(0),
+            BlockId::Hash(from_hash) => {
+                provider.get_block_by_hash(from_hash.into()).await?.header().number()
+            }
+        };
 
-        if let BlockId::Hash(from_hash) = scanner.config.from_block {
-            provider.get_block_by_hash(from_hash.into()).await?;
+        if from_num > latest_block {
+            Err(ScannerError::BlockExceedsLatest("from_block", from_num, latest_block))?;
         }
 
-        if let BlockId::Hash(to_hash) = scanner.config.to_block {
-            provider.get_block_by_hash(to_hash.into()).await?;
+        let to_num = match scanner.config.to_block {
+            BlockId::Number(to_block) => to_block.as_number().unwrap_or(0),
+            BlockId::Hash(to_hash) => {
+                provider.get_block_by_hash(to_hash.into()).await?.header().number()
+            }
+        };
+
+        if to_num > latest_block {
+            Err(ScannerError::BlockExceedsLatest("to_block", to_num, latest_block))?;
         }
 
         Ok(scanner)
@@ -281,4 +300,70 @@ mod tests {
 
         assert!(result.is_ok());
     }
+
+    #[tokio::test]
+    async fn test_from_block_above_latest_returns_error() {
+        let anvil = Anvil::new().try_spawn().unwrap();
+        let provider = ProviderBuilder::new().connect_http(anvil.endpoint_url());
+
+        let latest_block = provider.get_block_number().await.unwrap();
+
+        let result = EventScannerBuilder::latest(1)
+            .from_block(latest_block + 100)
+            .to_block(latest_block)
+            .connect(provider)
+            .await;
+
+        match result {
+            Err(ScannerError::BlockExceedsLatest("from_block", max, latest)) => {
+                assert_eq!(max, latest_block + 100);
+                assert_eq!(latest, latest_block);
+            }
+            _ => panic!("Expected BlockExceedsLatest error"),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_to_block_above_latest_returns_error() {
+        let anvil = Anvil::new().try_spawn().unwrap();
+        let provider = ProviderBuilder::new().connect_http(anvil.endpoint_url());
+
+        let latest_block = provider.get_block_number().await.unwrap();
+
+        let result = EventScannerBuilder::latest(1)
+            .from_block(0)
+            .to_block(latest_block + 100)
+            .connect(provider)
+            .await;
+
+        match result {
+            Err(ScannerError::BlockExceedsLatest("to_block", max, latest)) => {
+                assert_eq!(max, latest_block + 100);
+                assert_eq!(latest, latest_block);
+            }
+            _ => panic!("Expected BlockExceedsLatest error"),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_to_and_from_block_above_latest_returns_error() {
+        let anvil = Anvil::new().try_spawn().unwrap();
+        let provider = ProviderBuilder::new().connect_http(anvil.endpoint_url());
+
+        let latest_block = provider.get_block_number().await.unwrap();
+
+        let result = EventScannerBuilder::latest(1)
+            .from_block(latest_block + 50)
+            .to_block(latest_block + 100)
+            .connect(provider)
+            .await;
+
+        match result {
+            Err(ScannerError::BlockExceedsLatest("from_block", max, latest)) => {
+                assert_eq!(max, latest_block + 50);
+                assert_eq!(latest, latest_block);
+            }
+            _ => panic!("Expected BlockExceedsLatest error for 'from_block'"),
+        }
+    }
 }

src/event_scanner/scanner/mod.rs

@@ -299,6 +299,13 @@ impl EventScannerBuilder<Unspecified> {
     /// - **Default range**: By default, scans from `Earliest` to `Latest` block
     /// - **Reorg handling**: Periodically checks the tip to detect reorgs during the scan
     ///
+    /// # Notifications
+    ///
+    /// The scanner emits the following notification before delivering log data:
+    ///
+    /// - **[`Notification::NoPastLogsFound`][no_logs]**: Emitted when no matching logs are found in
+    ///   the scanned range.
+    ///
     /// # Arguments
     ///
     /// * `count` - Maximum number of recent events to collect per listener (must be greater than 0)
@@ -323,6 +330,7 @@ impl EventScannerBuilder<Unspecified> {
     /// [start]: EventScanner::start
     /// [sync_from_latest]: EventScannerBuilder::from_latest
     /// [reorg]: crate::Notification::ReorgDetected
+    /// [no_logs]: crate::Notification::NoPastLogsFound
     #[must_use]
     pub fn latest(count: usize) -> EventScannerBuilder<LatestEvents> {
         EventScannerBuilder::<LatestEvents>::new(count)

src/event_scanner/scanner/sync/from_latest.rs

@@ -1,8 +1,4 @@
-use alloy::{
-    consensus::BlockHeader,
-    eips::BlockNumberOrTag,
-    network::{BlockResponse, Network},
-};
+use alloy::{eips::BlockNumberOrTag, network::Network};
 
 use tracing::{error, info};
 
@@ -73,8 +69,7 @@ impl<N: Network> EventScanner<SyncFromLatestEvents, N> {
         // This is used to determine the starting point for the rewind stream and the live
         // stream. We do this before starting the streams to avoid a race condition
         // where the latest block changes while we're setting up the streams.
-        let latest_block =
-            provider.get_block_by_number(BlockNumberOrTag::Latest).await?.header().number();
+        let latest_block = provider.get_block_number().await?;
 
         // Setup rewind and live streams to run in parallel.
         let rewind_stream = client.rewind(BlockNumberOrTag::Earliest, latest_block).await?;

src/event_scanner/scanner/sync/mod.rs

@@ -79,6 +79,17 @@ impl EventScannerBuilder<Synchronize> {
     ///     block confirmations
     /// - **Continuous operation**: Live phase continues indefinitely until the scanner is dropped
     ///
+    /// # Notifications
+    ///
+    /// During the **latest events phase**, the scanner can emit the following notification
+    /// before transitioning to live mode:
+    ///
+    /// - **[`Notification::NoPastLogsFound`][no_logs]**: Emitted when no matching logs are found in
+    ///   the scanned range
+    ///
+    /// After the latest events phase completes, [`Notification::SwitchingToLive`][switch_to_live]
+    /// is emitted before transitioning to the live streaming phase.
+    ///
     /// # Arguments
     ///
     /// * `count` - Maximum number of recent events to collect per listener before switching to live
@@ -102,6 +113,7 @@ impl EventScannerBuilder<Synchronize> {
     /// [start]: crate::event_scanner::EventScanner::start
     /// [reorg]: crate::types::Notification::ReorgDetected
     /// [switch_to_live]: crate::types::Notification::SwitchingToLive
+    /// [no_logs]: crate::types::Notification::NoPastLogsFound
     #[must_use]
     pub fn from_latest(self, count: usize) -> EventScannerBuilder<SyncFromLatestEvents> {
         EventScannerBuilder::<SyncFromLatestEvents>::new(count)

src/robust_provider/provider.rs

@@ -42,6 +42,8 @@ impl<N: Network> RobustProvider<N> {
 
     /// Fetch a block by [`BlockNumberOrTag`] with retry and timeout.
     ///
+    /// This is a wrapper function for [`Provider::get_block_by_number`].
+    ///
     /// # Errors
     ///
     /// See [retry errors](#retry-errors).
@@ -65,6 +67,8 @@ impl<N: Network> RobustProvider<N> {
 
     /// Fetch a block number by [`BlockId`]  with retry and timeout.
     ///
+    /// This is a wrapper function for [`Provider::get_block`].
+    ///
     /// # Errors
     ///
     /// See [retry errors](#retry-errors).
@@ -81,6 +85,8 @@ impl<N: Network> RobustProvider<N> {
 
     /// Fetch the latest block number with retry and timeout.
     ///
+    /// This is a wrapper function for [`Provider::get_block_number`].
+    ///
     /// # Errors
     ///
     /// See [retry errors](#retry-errors).
@@ -98,6 +104,31 @@ impl<N: Network> RobustProvider<N> {
         result
     }
 
+    /// Get the block number for a given block identifier.
+    ///
+    /// This is a wrapper function for [`Provider::get_block_number_by_id`].
+    ///
+    /// # Arguments
+    ///
+    /// * `block_id` - The block identifier to fetch the block number for.
+    ///
+    /// # Errors
+    ///
+    /// See [retry errors](#retry-errors).
+    pub async fn get_block_number_by_id(&self, block_id: BlockId) -> Result<u64, Error> {
+        info!("get_block_number_by_id called");
+        let result = self
+            .retry_with_total_timeout(
+                move |provider| async move { provider.get_block_number_by_id(block_id).await },
+                false,
+            )
+            .await;
+        if let Err(e) = &result {
+            error!(error = %e, "get_block_number_by_id failed");
+        }
+        result?.ok_or_else(|| Error::BlockNotFound(block_id))
+    }
+
     /// Fetch the latest confirmed block number with retry and timeout.
     ///
     /// This method fetches the latest block number and subtracts the specified
@@ -120,6 +151,8 @@ impl<N: Network> RobustProvider<N> {
 
     /// Fetch a block by [`BlockHash`] with retry and timeout.
     ///
+    /// This is a wrapper function for [`Provider::get_block_by_hash`].
+    ///
     /// # Errors
     ///
     /// See [retry errors](#retry-errors).
@@ -140,6 +173,8 @@ impl<N: Network> RobustProvider<N> {
 
     /// Fetch logs for the given [`Filter`] with retry and timeout.
     ///
+    /// This is a wrapper function for [`Provider::get_logs`].
+    ///
     /// # Errors
     ///
     /// See [retry errors](#retry-errors).
@@ -159,6 +194,8 @@ impl<N: Network> RobustProvider<N> {
 
     /// Subscribe to new block headers with retry and timeout.
     ///
+    /// This is a wrapper function for [`Provider::subscribe_blocks`].
+    ///
     /// # Errors
     ///
     /// see [retry errors](#retry-errors).