tracing fixes proposal

Author: 0xNeshi

src/block_range_scanner.rs

@@ -73,11 +73,10 @@ use crate::{
     types::{IntoScannerResult, Notification, ScannerResult, TryStream},
 };
 
-#[allow(unused_imports)]
 use alloy::{
     consensus::BlockHeader,
     eips::{BlockId, BlockNumberOrTag},
-    network::{BlockResponse, Network, primitives::HeaderResponse},
+    network::{BlockResponse, Network},
     primitives::BlockNumber,
 };
 
@@ -288,47 +287,46 @@ impl<N: Network> Service<N> {
     }
 
     pub async fn run(mut self) {
-        opt_info!("Starting subscription service");
+        info!("Starting subscription service");
 
         while !self.shutdown {
             tokio::select! {
                 cmd = self.command_receiver.recv() => {
                     if let Some(command) = cmd {
-                        #[allow(clippy::used_underscore_binding)]
-                        if let Err(_e) = self.handle_command(command).await {
-                            opt_error!("Command handling error: {}", _e);
+                                                if let Err(e) = self.handle_command(command).await {
+                            error!(error = %e, "Command handling error");
                             self.error_count += 1;
                         }
                     } else {
-                        opt_warn!("Command channel closed, shutting down");
+                        warn!("Command channel closed, shutting down");
                         break;
                     }
                 }
             }
         }
 
-        opt_info!("Subscription service stopped");
+        info!("Subscription service stopped");
     }
 
     async fn handle_command(&mut self, command: Command) -> Result<(), ScannerError> {
         match command {
             Command::StreamLive { sender, block_confirmations, response } => {
-                opt_info!("Starting live stream");
+                info!("Starting live stream");
                 let result = self.handle_live(block_confirmations, sender).await;
                 let _ = response.send(result);
             }
             Command::StreamHistorical { sender, start_id, end_id, response } => {
-                opt_info!(start_id = ?start_id, end_id = ?end_id, "Starting historical stream");
+                info!(start_id = ?start_id, end_id = ?end_id, "Starting historical stream");
                 let result = self.handle_historical(start_id, end_id, sender).await;
                 let _ = response.send(result);
             }
             Command::StreamFrom { sender, start_id, block_confirmations, response } => {
-                opt_info!(start_id = ?start_id, "Starting streaming from");
+                info!(start_id = ?start_id, "Starting streaming from");
                 let result = self.handle_sync(start_id, block_confirmations, sender).await;
                 let _ = response.send(result);
             }
             Command::Rewind { sender, start_id, end_id, response } => {
-                opt_info!(start_id = ?start_id, end_id = ?end_id, "Starting rewind");
+                info!(start_id = ?start_id, end_id = ?end_id, "Starting rewind");
                 let result = self.handle_rewind(start_id, end_id, sender).await;
                 let _ = response.send(result);
             }
@@ -353,7 +351,7 @@ impl<N: Network> Service<N> {
 
         let subscription = self.provider.subscribe_blocks().await?;
 
-        opt_info!("WebSocket connected for live blocks");
+        info!("WebSocket connected for live blocks");
 
         tokio::spawn(async move {
             let mut reorg_handler =
@@ -396,7 +394,7 @@ impl<N: Network> Service<N> {
             _ => (start_block_num, end_block_num),
         };
 
-        opt_info!(
+        info!(
             start_block = start_block_num,
             end_block = end_block_num,
             "Normalized the block range"
@@ -486,7 +484,7 @@ impl<N: Network> Service<N> {
         {
             Ok(block) => block,
             Err(e) => {
-                opt_error!(error = %e, "Failed to get finalized block");
+                error!(error = %e, "Failed to get finalized block");
                 _ = sender.try_stream(e).await;
                 return;
             }
@@ -508,7 +506,7 @@ impl<N: Network> Service<N> {
                 let reorg = match reorg_handler.check(&tip).await {
                     Ok(opt) => opt,
                     Err(e) => {
-                        opt_error!(error = %e, "Terminal RPC call error, shutting down");
+                        error!(error = %e, "Terminal RPC call error, shutting down");
                         _ = sender.try_stream(e).await;
                         return;
                     }
@@ -529,7 +527,7 @@ impl<N: Network> Service<N> {
             }
         }
 
-        opt_info!(batch_count = iter.batch_count(), "Rewind completed");
+        info!(batch_count = iter.batch_count(), "Rewind completed");
     }
 
     /// Handles re-scanning of reorged blocks.
@@ -544,9 +542,9 @@ impl<N: Network> Service<N> {
     ) -> bool {
         let tip_number = tip.header().number();
         let common_ancestor = common_ancestor.header().number();
-        opt_info!(
+        info!(
             block_number = %tip_number,
-            hash = %tip.header().hash(),
+            hash = %alloy::network::primitives::HeaderResponse::hash(tip.header()),
             common_ancestor = %common_ancestor,
             "Reorg detected"
         );
@@ -559,13 +557,10 @@ impl<N: Network> Service<N> {
         *tip = match provider.get_block_by_number(tip_number.into()).await {
             Ok(block) => block,
             Err(e) => {
-                #[allow(clippy::if_same_then_else)]
                 if matches!(e, crate::robust_provider::Error::BlockNotFound(_)) {
-                    opt_error!(
-                        "Unexpected error: pre-reorg chain tip should exist on a reorged chain"
-                    );
+                    error!("Unexpected error: pre-reorg chain tip should exist on a reorged chain");
                 } else {
-                    opt_error!(error = %e, "Terminal RPC call error, shutting down");
+                    error!(error = %e, "Terminal RPC call error, shutting down");
                 }
                 _ = sender.try_stream(e).await;
                 return false;

src/block_range_scanner/common.rs

@@ -66,7 +66,7 @@ pub(crate) async fn stream_live_blocks<N: Network>(
     )
     .await;
 
-    opt_warn!("Live block subscription ended");
+    warn!("Live block subscription ended");
 }
 
 async fn get_first_block<
@@ -84,9 +84,7 @@ async fn get_first_block<
                     subscription::Error::Lagged(_) => {
                         // scanner already accounts for skipped block numbers
                         // next block will be the actual incoming block
-                        opt_info!(
-                            "Skipping Error::Lagged, next block should be the first live block"
-                        );
+                        info!("Skipping Error::Lagged, next block should be the first live block");
                     }
                     subscription::Error::Timeout => {
                         _ = sender.try_stream(ScannerError::Timeout).await;
@@ -133,7 +131,7 @@ async fn initialize_live_streaming_state<N: Network>(
     reorg_handler: &mut ReorgHandler<N>,
 ) -> Option<LiveStreamingState<N>> {
     let incoming_block_num = first_block.number();
-    opt_info!(block_number = incoming_block_num, "Received first block header");
+    info!(block_number = incoming_block_num, "Received first block header");
 
     let confirmed = incoming_block_num.saturating_sub(block_confirmations);
 
@@ -177,7 +175,7 @@ async fn stream_blocks_continuously<
         let incoming_block = match incoming_block {
             Ok(block) => block,
             Err(e) => {
-                opt_error!(error = %e, "Error receiving block from stream");
+                error!(error = %e, "Error receiving block from stream");
                 match e {
                     subscription::Error::Lagged(_) => {
                         // scanner already accounts for skipped block numbers
@@ -201,7 +199,7 @@ async fn stream_blocks_continuously<
         };
 
         let incoming_block_num = incoming_block.number();
-        opt_info!(block_number = incoming_block_num, "Received block header");
+        info!(block_number = incoming_block_num, "Received block header");
 
         let Some(previous_batch_end) = state.previous_batch_end.as_ref() else {
             // previously detected reorg wasn't fully handled
@@ -211,7 +209,7 @@ async fn stream_blocks_continuously<
         let common_ancestor = match reorg_handler.check(previous_batch_end).await {
             Ok(reorg_opt) => reorg_opt,
             Err(e) => {
-                opt_error!(error = %e, "Failed to perform reorg check");
+                error!(error = %e, "Failed to perform reorg check");
                 _ = sender.try_stream(e).await;
                 return;
             }
@@ -261,7 +259,7 @@ async fn handle_reorg_detected<N: Network>(
     // Reset streaming position based on common ancestor
     if ancestor_num < stream_start {
         // Reorg went before our starting point - restart from stream_start
-        opt_info!(
+        info!(
             ancestor_block = ancestor_num,
             stream_start = stream_start,
             "Reorg detected before stream start, resetting to stream start"
@@ -270,7 +268,7 @@ async fn handle_reorg_detected<N: Network>(
         state.previous_batch_end = None;
     } else {
         // Resume from after the common ancestor
-        opt_info!(ancestor_block = ancestor_num, "Reorg detected, resuming from common ancestor");
+        info!(ancestor_block = ancestor_num, "Reorg detected, resuming from common ancestor");
         state.batch_start = ancestor_num + 1;
         state.previous_batch_end = Some(common_ancestor);
     }
@@ -336,12 +334,12 @@ pub(crate) async fn stream_historical_range<N: Network>(
     provider: &RobustProvider<N>,
     reorg_handler: &mut ReorgHandler<N>,
 ) -> Option<()> {
-    opt_info!("Getting finalized block number");
+    info!("Getting finalized block number");
     let finalized = match provider.get_block_number_by_id(BlockNumberOrTag::Finalized.into()).await
     {
         Ok(block) => block,
         Err(e) => {
-            opt_error!(error = %e, "Failed to get finalized block");
+            error!(error = %e, "Failed to get finalized block");
             _ = sender.try_stream(e).await;
             return None;
         }
@@ -407,7 +405,7 @@ pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
         let batch_end = match provider.get_block_by_number(batch_end_num.into()).await {
             Ok(block) => block,
             Err(e) => {
-                opt_error!(batch_start = batch.start(), batch_end = batch_end_num, error = %e, "Failed to get ending block of the current batch");
+                error!(batch_start = batch.start(), batch_end = batch_end_num, error = %e, "Failed to get ending block of the current batch");
                 _ = sender.try_stream(e).await;
                 return None;
             }
@@ -420,7 +418,7 @@ pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
         let reorged_opt = match reorg_handler.check(&batch_end).await {
             Ok(opt) => opt,
             Err(e) => {
-                opt_error!(error = %e, "Failed to perform reorg check");
+                error!(error = %e, "Failed to perform reorg check");
                 _ = sender.try_stream(e).await;
                 return None;
             }
@@ -437,6 +435,6 @@ pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
         last_batch_end = Some(batch_end);
     }
 
-    opt_error!(batch_count = iter.batch_count(), "Historical sync completed");
+    error!(batch_count = iter.batch_count(), "Historical sync completed");
     last_batch_end
 }

src/block_range_scanner/reorg_handler.rs

@@ -62,22 +62,22 @@ impl<N: Network> ReorgHandler<N> {
         block: &N::BlockResponse,
     ) -> Result<Option<N::BlockResponse>, ScannerError> {
         let block = block.header();
-        opt_info!(block_hash = %block.hash(), block_number = block.number(), "Checking if block was reorged");
+        info!(block_hash = %block.hash(), block_number = block.number(), "Checking if block was reorged");
 
         if !self.reorg_detected(block).await? {
             let block_hash = block.hash();
-            opt_info!(block_hash = %block_hash, block_number = block.number(), "No reorg detected");
+            info!(block_hash = %block_hash, block_number = block.number(), "No reorg detected");
             // store the incoming block's hash for future reference
             if !matches!(self.buffer.back(), Some(&hash) if hash == block_hash) {
                 self.buffer.push(block_hash);
             }
             return Ok(None);
         }
 
-        opt_info!("Reorg detected, searching for common ancestor");
+        info!("Reorg detected, searching for common ancestor");
 
         while let Some(&block_hash) = self.buffer.back() {
-            opt_info!(block_hash = %block_hash, "Checking if block exists on-chain");
+            info!(block_hash = %block_hash, "Checking if block exists on-chain");
             match self.provider.get_block_by_hash(block_hash).await {
                 Ok(common_ancestor) => return self.return_common_ancestor(common_ancestor).await,
                 Err(robust_provider::Error::BlockNotFound(_)) => {
@@ -93,13 +93,12 @@ impl<N: Network> ReorgHandler<N> {
         // no need to store finalized block's hash in the buffer, as it is returned by default only
         // if not buffered hashes exist on-chain
 
-        opt_warn!("Possible deep reorg detected, setting finalized block as common ancestor");
+        warn!("Possible deep reorg detected, setting finalized block as common ancestor");
 
         let finalized = self.provider.get_block_by_number(BlockNumberOrTag::Finalized).await?;
 
-        #[allow(clippy::used_underscore_binding)]
-        let _header = finalized.header();
-        opt_info!(finalized_hash = %_header.hash(), block_number = _header.number(), "Finalized block set as common ancestor");
+        let header = finalized.header();
+        info!(finalized_hash = %header.hash(), block_number = header.number(), "Finalized block set as common ancestor");
 
         Ok(Some(finalized))
     }
@@ -120,10 +119,10 @@ impl<N: Network> ReorgHandler<N> {
         let finalized = self.provider.get_block_by_number(BlockNumberOrTag::Finalized).await?;
         let finalized_header = finalized.header();
         let common_ancestor = if finalized_header.number() <= common_ancestor_header.number() {
-            opt_info!(common_ancestor = %common_ancestor_header.hash(), block_number = common_ancestor_header.number(), "Common ancestor found");
+            info!(common_ancestor = %common_ancestor_header.hash(), block_number = common_ancestor_header.number(), "Common ancestor found");
             common_ancestor
         } else {
-            opt_warn!(
+            warn!(
                 finalized_hash = %finalized_header.hash(), block_number = finalized_header.number(), "Possible deep reorg detected, using finalized block as common ancestor"
             );
             // all buffered blocks are finalized, so no more need to track them

src/block_range_scanner/sync_handler.rs

@@ -45,14 +45,14 @@ impl<N: Network> SyncHandler<N> {
 
         match sync_state {
             SyncState::AlreadyLive { start_block } => {
-                opt_info!(
+                info!(
                     start_block = start_block,
                     "Start block is beyond confirmed tip, waiting until starting block is confirmed before starting live stream"
                 );
                 self.spawn_live_only(start_block).await?;
             }
             SyncState::NeedsCatchup { start_block, confirmed_tip } => {
-                opt_info!(
+                info!(
                     start_block = start_block,
                     confirmed_tip = confirmed_tip,
                     "Start block is behind confirmed tip, catching up then transitioning to live"
@@ -131,7 +131,7 @@ impl<N: Network> SyncHandler<N> {
                     return; // channel closed
                 }
                 Err(e) => {
-                    opt_error!(error = %e, "Error during historical catchup, shutting down");
+                    error!(error = %e, "Error during historical catchup, shutting down");
                     _ = sender.try_stream(e).await;
                     return;
                 }
@@ -182,7 +182,7 @@ impl<N: Network> SyncHandler<N> {
             confirmed_tip = latest.saturating_sub(block_confirmations);
         }
 
-        opt_info!("Historical catchup complete, ready to transition to live");
+        info!("Historical catchup complete, ready to transition to live");
 
         Ok(Some(start_block))
     }
@@ -199,7 +199,7 @@ impl<N: Network> SyncHandler<N> {
         let subscription = match provider.subscribe_blocks().await {
             Ok(sub) => sub,
             Err(e) => {
-                opt_error!(error = %e, "Error subscribing to live blocks, shutting down");
+                error!(error = %e, "Error subscribing to live blocks, shutting down");
                 _ = sender.try_stream(e).await;
                 return;
             }
@@ -209,7 +209,7 @@ impl<N: Network> SyncHandler<N> {
             return;
         }
 
-        opt_info!("Successfully transitioned from historical to live streaming");
+        info!("Successfully transitioned from historical to live streaming");
 
         common::stream_live_blocks(
             start_block,