Improve Logs + Reduce Logging Noise (#281)

Author: 0xNeshi

src/block_range_scanner/common.rs

@@ -50,6 +50,10 @@ impl IntoScannerResult<RangeInclusive<BlockNumber>> for RangeInclusive<BlockNumb
 }
 
 #[allow(clippy::too_many_arguments)]
+#[cfg_attr(
+    feature = "tracing",
+    tracing::instrument(level = "trace", skip(subscription, sender, provider, reorg_handler))
+)]
 pub(crate) async fn stream_live_blocks<N: Network>(
     stream_start: BlockNumber,
     subscription: RobustSubscription<N>,
@@ -69,6 +73,16 @@ pub(crate) async fn stream_live_blocks<N: Network>(
         return;
     };
 
+    debug!(
+        first_block = first_block.number(),
+        stream_start = stream_start,
+        "Received first relevant block, starting live streaming"
+    );
+
+    // This check is necessary when running `sync` modes. It makes sense to stream this notification
+    // only once the first relevant block is received from the subscription, and not before that;
+    // otherwise callers might perform certain operations expecting the relevant blocks to start
+    // coming, when in fact they are not.
     if notify_after_first_block && !sender.try_stream(Notification::SwitchingToLive).await {
         return;
     }
@@ -100,8 +114,6 @@ pub(crate) async fn stream_live_blocks<N: Network>(
         reorg_handler,
     )
     .await;
-
-    warn!("Live block subscription ended");
 }
 
 async fn get_first_block<
@@ -119,7 +131,6 @@ async fn get_first_block<
                     subscription::Error::Lagged(_) => {
                         // scanner already accounts for skipped block numbers
                         // next block will be the actual incoming block
-                        info!("Skipping Error::Lagged, next block should be the first live block");
                     }
                     subscription::Error::Timeout => {
                         _ = sender.try_stream(ScannerError::Timeout).await;
@@ -154,8 +165,8 @@ fn skip_to_first_relevant_block<N: Network>(
     })
 }
 
-/// Initializes the streaming state after receiving the first block
-/// Returns None if the channel is closed
+/// Initializes the streaming state after receiving the first block.
+/// Returns None if the channel is closed.
 async fn initialize_live_streaming_state<N: Network>(
     first_block: N::HeaderResponse,
     stream_start: BlockNumber,
@@ -165,10 +176,7 @@ async fn initialize_live_streaming_state<N: Network>(
     provider: &RobustProvider<N>,
     reorg_handler: &mut ReorgHandler<N>,
 ) -> Option<LiveStreamingState<N>> {
-    let incoming_block_num = first_block.number();
-    info!(block_number = incoming_block_num, "Received first block header");
-
-    let confirmed = incoming_block_num.saturating_sub(block_confirmations);
+    let confirmed = first_block.number().saturating_sub(block_confirmations);
 
     // The minimum common ancestor is the block before the stream start
     let min_common_ancestor = stream_start.saturating_sub(1);
@@ -210,10 +218,9 @@ async fn stream_blocks_continuously<
         let incoming_block = match incoming_block {
             Ok(block) => block,
             Err(e) => {
-                error!(error = %e, "Error receiving block from stream");
                 match e {
                     subscription::Error::Lagged(_) => {
-                        // scanner already accounts for skipped block numbers
+                        // scanner already accounts for skipped block numbers,
                         // next block will be the actual incoming block
                         continue;
                     }
@@ -233,8 +240,8 @@ async fn stream_blocks_continuously<
             }
         };
 
-        let incoming_block_num = incoming_block.number();
-        info!(block_number = incoming_block_num, "Received block header");
+        let incoming_block = incoming_block.number();
+        trace!(received = incoming_block, "Received item from block subscription");
 
         let Some(previous_batch_end) = state.previous_batch_end.as_ref() else {
             // previously detected reorg wasn't fully handled
@@ -244,7 +251,7 @@ async fn stream_blocks_continuously<
         let common_ancestor = match reorg_handler.check(previous_batch_end).await {
             Ok(reorg_opt) => reorg_opt,
             Err(e) => {
-                error!(error = %e, "Failed to perform reorg check");
+                error!("Failed to perform reorg check");
                 _ = sender.try_stream(e).await;
                 return;
             }
@@ -260,7 +267,7 @@ async fn stream_blocks_continuously<
         }
 
         // Stream the next batch of confirmed blocks
-        let batch_end_num = incoming_block_num.saturating_sub(block_confirmations);
+        let batch_end_num = incoming_block.saturating_sub(block_confirmations);
         if !stream_next_batch(
             batch_end_num,
             state,
@@ -287,23 +294,33 @@ async fn handle_reorg_detected<N: Network>(
 ) -> bool {
     let ancestor_num = common_ancestor.header().number();
 
+    info!(
+        common_ancestor = ancestor_num,
+        stream_start = stream_start,
+        "Reorg detected during live streaming"
+    );
+
     if !sender.try_stream(Notification::ReorgDetected { common_ancestor: ancestor_num }).await {
         return false;
     }
 
     // Reset streaming position based on common ancestor
     if ancestor_num < stream_start {
         // Reorg went before our starting point - restart from stream_start
-        info!(
-            ancestor_block = ancestor_num,
+        debug!(
+            common_ancestor = ancestor_num,
             stream_start = stream_start,
-            "Reorg detected before stream start, resetting to stream start"
+            "Reorg predates stream start, restarting from stream_start"
         );
         state.batch_start = stream_start;
         state.previous_batch_end = None;
     } else {
         // Resume from after the common ancestor
-        info!(ancestor_block = ancestor_num, "Reorg detected, resuming from common ancestor");
+        debug!(
+            common_ancestor = ancestor_num,
+            resume_from = ancestor_num + 1,
+            "Resuming from after common ancestor"
+        );
         state.batch_start = ancestor_num + 1;
         state.previous_batch_end = Some(common_ancestor);
     }
@@ -361,6 +378,10 @@ struct LiveStreamingState<N: Network> {
 }
 
 #[must_use]
+#[cfg_attr(
+    feature = "tracing",
+    tracing::instrument(level = "trace", skip(sender, provider, reorg_handler))
+)]
 pub(crate) async fn stream_historical_range<N: Network>(
     start: BlockNumber,
     end: BlockNumber,
@@ -369,20 +390,31 @@ pub(crate) async fn stream_historical_range<N: Network>(
     provider: &RobustProvider<N>,
     reorg_handler: &mut ReorgHandler<N>,
 ) -> Option<()> {
-    info!("Getting finalized block number");
     let finalized = match provider.get_block_number_by_id(BlockNumberOrTag::Finalized.into()).await
     {
         Ok(block) => block,
         Err(e) => {
-            error!(error = %e, "Failed to get finalized block");
+            warn!("Failed to get finalized block");
             _ = sender.try_stream(e).await;
             return None;
         }
     };
 
+    debug!(finalized_block = finalized, "Got finalized block for historical range");
+
     // no reorg check for finalized blocks
     let finalized_batch_end = finalized.min(end);
+    let finalized_range_count =
+        RangeIterator::forward(start, finalized_batch_end, max_block_range).count();
+    trace!(
+        start = start,
+        finalized_batch_end = finalized_batch_end,
+        batch_count = finalized_range_count,
+        "Streaming finalized blocks (no reorg check)"
+    );
+
     for range in RangeIterator::forward(start, finalized_batch_end, max_block_range) {
+        trace!(range_start = *range.start(), range_end = *range.end(), "Streaming finalized range");
         if !sender.try_stream(range).await {
             return None; // channel closed
         }
@@ -423,6 +455,10 @@ pub(crate) async fn stream_historical_range<N: Network>(
 }
 
 /// Assumes that `min_common_ancestor <= next_start_block <= end`, performs no internal checks.
+#[cfg_attr(
+    feature = "tracing",
+    tracing::instrument(level = "trace", skip(sender, provider, reorg_handler))
+)]
 pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
     min_common_ancestor: BlockNumber,
     next_start_block: BlockNumber,
@@ -440,7 +476,11 @@ 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) => {
-                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,
+                    "Failed to get ending block of the current batch"
+                );
                 _ = sender.try_stream(e).await;
                 return None;
             }
@@ -453,23 +493,28 @@ 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) => {
-                error!(error = %e, "Failed to perform reorg check");
+                error!("Failed to perform reorg check");
                 _ = sender.try_stream(e).await;
                 return None;
             }
         };
 
         if let Some(common_ancestor) = reorged_opt {
             let common_ancestor = common_ancestor.header().number();
+            info!(
+                common_ancestor = common_ancestor,
+                "Reorg detected during historical streaming, resetting range iterator"
+            );
             if !sender.try_stream(Notification::ReorgDetected { common_ancestor }).await {
                 return None;
             }
-            iter.reset_to((common_ancestor + 1).max(min_common_ancestor));
+            let reset_to = (common_ancestor + 1).max(min_common_ancestor);
+            debug!(reset_to = reset_to, "Resetting range iterator after reorg");
+            iter.reset_to(reset_to);
         }
 
         last_batch_end = Some(batch_end);
     }
 
-    info!(batch_count = iter.batch_count(), "Historical sync completed");
     last_batch_end
 }

src/block_range_scanner/range_iterator.rs

@@ -74,14 +74,6 @@ impl RangeIterator<Reverse> {
     }
 }
 
-impl<D> RangeIterator<D> {
-    /// Returns the number of batches yielded so far.
-    #[must_use]
-    pub fn batch_count(&self) -> u64 {
-        self.batch_count
-    }
-}
-
 impl Iterator for RangeIterator<Forward> {
     type Item = RangeInclusive<BlockNumber>;
 
@@ -90,14 +82,10 @@ impl Iterator for RangeIterator<Forward> {
             return None;
         }
 
-        self.batch_count += 1;
-        if self.batch_count % 10 == 0 {
-            debug!(batch_count = self.batch_count, "Processed batches");
-        }
-
         let batch_start = self.current;
         let batch_end = batch_start.saturating_add(self.range_size - 1).min(self.end);
         self.current = batch_end + 1;
+        self.batch_count += 1;
 
         Some(batch_start..=batch_end)
     }
@@ -117,14 +105,10 @@ impl Iterator for RangeIterator<Reverse> {
             return None;
         }
 
-        self.batch_count += 1;
-        if self.batch_count % 10 == 0 {
-            debug!(batch_count = self.batch_count, "Processed batches");
-        }
-
         let batch_high = self.current;
         let batch_low = batch_high.saturating_sub(self.range_size - 1).max(self.end);
         self.current = batch_low.saturating_sub(1);
+        self.batch_count += 1;
 
         Some(batch_low..=batch_high)
     }
@@ -302,35 +286,4 @@ mod tests {
     fn reverse_zero_max_block_range_panics() {
         let _ = RangeIterator::reverse(200, 100, 0);
     }
-
-    #[test]
-    fn batch_count_increments() {
-        let mut iter = RangeIterator::forward(100, 300, 50);
-        assert_eq!(iter.batch_count(), 0);
-
-        iter.next();
-        assert_eq!(iter.batch_count(), 1);
-
-        iter.next();
-        assert_eq!(iter.batch_count(), 2);
-
-        iter.next();
-        assert_eq!(iter.batch_count(), 3);
-    }
-
-    #[test]
-    fn batch_count_persists_after_reset() {
-        let mut iter = RangeIterator::forward(100, 300, 50);
-        iter.next();
-        iter.next();
-        assert_eq!(iter.batch_count(), 2);
-
-        iter.reset_to(150);
-
-        // batch_count is not reset
-        assert_eq!(iter.batch_count(), 2);
-
-        iter.next();
-        assert_eq!(iter.batch_count(), 3);
-    }
 }