feat: move block not found to provider

Author: LeoPatOZ

src/block_range_scanner.rs

@@ -65,8 +65,8 @@
 
 use std::{cmp::Ordering, ops::RangeInclusive, time::Duration};
 use tokio::{
-    join,
     sync::{mpsc, oneshot},
+    try_join,
 };
 use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 
@@ -398,10 +398,8 @@ impl<N: Network> Service<N> {
             self.provider.get_block_by_number(end_height)
         )?;
 
-        let start_block_num =
-            start_block.ok_or_else(|| ScannerError::BlockNotFound(start_height))?.header().number();
-        let end_block_num =
-            end_block.ok_or_else(|| ScannerError::BlockNotFound(end_height))?.header().number();
+        let start_block_num = start_block.header().number();
+        let end_block_num = end_block.header().number();
 
         let (start_block_num, end_block_num) = match start_block_num.cmp(&end_block_num) {
             Ordering::Greater => (end_block_num, start_block_num),
@@ -435,12 +433,8 @@ impl<N: Network> Service<N> {
             self.provider.get_block_by_number(BlockNumberOrTag::Latest)
         )?;
 
-        let start_block_num =
-            start_block.ok_or_else(|| ScannerError::BlockNotFound(start_height))?.header().number();
-        let latest_block = latest_block
-            .ok_or_else(|| ScannerError::BlockNotFound(BlockNumberOrTag::Latest))?
-            .header()
-            .number();
+        let start_block_num = start_block.header().number();
+        let latest_block = latest_block.header().number();
 
         let confirmed_tip_num = latest_block.saturating_sub(block_confirmations);
 
@@ -536,13 +530,10 @@ impl<N: Network> Service<N> {
         start_height: BlockNumberOrTag,
         end_height: BlockNumberOrTag,
     ) -> Result<(), ScannerError> {
-        let (start_block, end_block) = join!(
+        let (start_block, end_block) = try_join!(
             self.provider.get_block_by_number(start_height),
             self.provider.get_block_by_number(end_height),
-        );
-
-        let start_block = start_block?.ok_or(ScannerError::BlockNotFound(start_height))?;
-        let end_block = end_block?.ok_or(ScannerError::BlockNotFound(end_height))?;
+        )?;
 
         // normalize block range
         let (from, to) = match start_block.header().number().cmp(&end_block.header().number()) {
@@ -606,13 +597,7 @@ impl<N: Network> Service<N> {
                 // restart rewind
                 batch_from = from;
                 // store the updated end block hash
-                tip_hash = self
-                    .provider
-                    .get_block_by_number(from.into())
-                    .await?
-                    .expect("Chain should have the same height post-reorg")
-                    .header()
-                    .hash();
+                tip_hash = self.provider.get_block_by_number(from.into()).await?.header().hash();
             } else {
                 // SAFETY: `batch_to` is always greater than `to`, so `batch_to - 1` is always
                 // a valid unsigned integer
@@ -785,11 +770,7 @@ impl<N: Network> Service<N> {
     async fn get_block_subscription(
         provider: &RobustProvider<N>,
     ) -> Result<Subscription<N::HeaderResponse>, ScannerError> {
-        let ws_stream = provider
-            .subscribe_blocks()
-            .await
-            .map_err(|_| ScannerError::WebSocketConnectionFailed(1))?;
-
+        let ws_stream = provider.subscribe_blocks().await?;
         Ok(ws_stream)
     }
 
@@ -1656,13 +1637,11 @@ mod tests {
         let (tx, mut rx) = mpsc::channel(1);
         service.subscriber = Some(tx);
 
-        service
-            .send_to_subscriber(Message::Error(ScannerError::WebSocketConnectionFailed(4)))
-            .await;
+        service.send_to_subscriber(Message::Error(ScannerError::BlockNotFound(4.into()))).await;
 
         match rx.recv().await.expect("subscriber should stay open") {
-            Message::Error(ScannerError::WebSocketConnectionFailed(attempts)) => {
-                assert_eq!(attempts, 4);
+            Message::Error(ScannerError::BlockNotFound(attempts)) => {
+                assert_eq!(attempts, 4.into());
             }
             other => panic!("unexpected message: {other:?}"),
         }

src/error.rs

@@ -3,17 +3,14 @@ use std::{ops::RangeInclusive, sync::Arc};
 use alloy::{
     eips::BlockNumberOrTag,
     primitives::BlockNumber,
-    transports::{RpcError, TransportErrorKind, http::reqwest},
+    transports::{RpcError, TransportErrorKind},
 };
 use thiserror::Error;
 
 use crate::{block_range_scanner::Message, robust_provider::RobustProviderError};
 
 #[derive(Error, Debug, Clone)]
 pub enum ScannerError {
-    #[error("HTTP request failed: {0}")]
-    HttpError(Arc<reqwest::Error>),
-
     // #[error("WebSocket error: {0}")]
     // WebSocketError(#[from] tokio_tungstenite::tungstenite::Error),
     #[error("Serialization error: {0}")]
@@ -37,9 +34,6 @@ pub enum ScannerError {
     #[error("Historical sync failed: {0}")]
     HistoricalSyncError(String),
 
-    #[error("WebSocket connection failed after {0} attempts")]
-    WebSocketConnectionFailed(usize),
-
     #[error("Block not found, block number: {0}")]
     BlockNotFound(BlockNumberOrTag),
 
@@ -56,6 +50,7 @@ impl From<RobustProviderError> for ScannerError {
             RobustProviderError::RpcError(err) => ScannerError::RpcError(err),
             RobustProviderError::Timeout => ScannerError::Timeout,
             RobustProviderError::RetryFail(num) => ScannerError::RetryFail(num),
+            RobustProviderError::BlockNotFound(block) => ScannerError::BlockNotFound(block),
         }
     }
 }
@@ -69,12 +64,6 @@ impl From<Result<RangeInclusive<BlockNumber>, ScannerError>> for Message {
     }
 }
 
-impl From<reqwest::Error> for ScannerError {
-    fn from(error: reqwest::Error) -> Self {
-        ScannerError::HttpError(Arc::new(error))
-    }
-}
-
 impl From<serde_json::Error> for ScannerError {
     fn from(error: serde_json::Error) -> Self {
         ScannerError::SerializationError(Arc::new(error))

src/robust_provider.rs

@@ -20,6 +20,8 @@ pub enum RobustProviderError {
     Timeout,
     #[error("Retry failed after {0} tries")]
     RetryFail(usize),
+    #[error("Block not found, block number: {0}")]
+    BlockNotFound(BlockNumberOrTag),
 }
 
 impl From<RpcError<TransportErrorKind>> for RobustProviderError {
@@ -87,7 +89,7 @@ impl<N: Network> RobustProvider<N> {
     pub async fn get_block_by_number(
         &self,
         number: BlockNumberOrTag,
-    ) -> Result<Option<N::BlockResponse>, RobustProviderError> {
+    ) -> Result<N::BlockResponse, RobustProviderError> {
         info!("eth_getBlockByNumber called");
         let operation = async || {
             self.provider.get_block_by_number(number).await.map_err(RobustProviderError::from)
@@ -96,7 +98,11 @@ impl<N: Network> RobustProvider<N> {
         if let Err(e) = &result {
             error!(error = %e, "eth_getByBlockNumber failed");
         }
-        result
+
+        match result? {
+            Some(block) => Ok(block),
+            None => Err(RobustProviderError::BlockNotFound(number)),
+        }
     }
 
     /// Fetch the latest block number with retry and timeout.