refactor: Introduced the ReorgHandler trait (#233)

<!-- Append the issue number -->
Resolves #226 

This PR refactors the `ReorgHandler` struct into a trait-based design to
enable better testability.

### Changes

- **Added `ReorgHandler` trait** - Defines the interface for chain
reorganization detection with a single `check` method
- **Renamed `ReorgHandler` struct to `DefaultReorgHandler`** - The
concrete implementation using RPC providers now implements the new trait
- **Updated function signatures** - All functions that accept a reorg
handler now use generic bounds (`R: ReorgHandler<N>`) instead of the
concrete type

Author: yug49

src/block_range_scanner/common.rs

@@ -54,14 +54,14 @@ impl IntoScannerResult<RangeInclusive<BlockNumber>> for RangeInclusive<BlockNumb
     feature = "tracing",
     tracing::instrument(level = "trace", skip(subscription, sender, provider, reorg_handler))
 )]
-pub(crate) async fn stream_live_blocks<N: Network>(
+pub(crate) async fn stream_live_blocks<N: Network, R: ReorgHandler<N>>(
     stream_start: BlockNumber,
     subscription: RobustSubscription<N>,
     sender: &mpsc::Sender<BlockScannerResult>,
     provider: &RobustProvider<N>,
     block_confirmations: u64,
     max_block_range: u64,
-    reorg_handler: &mut ReorgHandler<N>,
+    reorg_handler: &mut R,
     notify_after_first_block: bool,
 ) {
     // Phase 1: Wait for first relevant block
@@ -169,14 +169,14 @@ fn skip_to_first_relevant_block<N: Network>(
 
 /// Initializes the streaming state after receiving the first block.
 /// Returns None if the channel is closed.
-async fn initialize_live_streaming_state<N: Network>(
+async fn initialize_live_streaming_state<N: Network, R: ReorgHandler<N>>(
     first_block: N::HeaderResponse,
     stream_start: BlockNumber,
     block_confirmations: u64,
     max_block_range: u64,
     sender: &mpsc::Sender<BlockScannerResult>,
     provider: &RobustProvider<N>,
-    reorg_handler: &mut ReorgHandler<N>,
+    reorg_handler: &mut R,
 ) -> Option<LiveStreamingState<N>> {
     let confirmed = first_block.number().saturating_sub(block_confirmations);
 
@@ -206,6 +206,7 @@ async fn initialize_live_streaming_state<N: Network>(
 async fn stream_blocks_continuously<
     N: Network,
     S: tokio_stream::Stream<Item = Result<N::HeaderResponse, subscription::Error>> + Unpin,
+    R: ReorgHandler<N>,
 >(
     stream: &mut S,
     state: &mut LiveStreamingState<N>,
@@ -214,7 +215,7 @@ async fn stream_blocks_continuously<
     max_block_range: u64,
     sender: &mpsc::Sender<BlockScannerResult>,
     provider: &RobustProvider<N>,
-    reorg_handler: &mut ReorgHandler<N>,
+    reorg_handler: &mut R,
 ) {
     while let Some(incoming_block) = stream.next().await {
         let incoming_block = match incoming_block {
@@ -337,14 +338,14 @@ async fn handle_reorg_detected<N: Network>(
 
 /// Streams the next batch of blocks up to `batch_end_num`.
 /// Returns `ChannelState::Closed` if the channel is closed, `ChannelState::Open` otherwise.
-async fn stream_next_batch<N: Network>(
+async fn stream_next_batch<N: Network, R: ReorgHandler<N>>(
     batch_end_num: BlockNumber,
     state: &mut LiveStreamingState<N>,
     stream_start: BlockNumber,
     max_block_range: u64,
     sender: &mpsc::Sender<BlockScannerResult>,
     provider: &RobustProvider<N>,
-    reorg_handler: &mut ReorgHandler<N>,
+    reorg_handler: &mut R,
 ) -> ChannelState {
     if batch_end_num < state.batch_start {
         // No new confirmed blocks to stream yet
@@ -389,13 +390,13 @@ struct LiveStreamingState<N: Network> {
     feature = "tracing",
     tracing::instrument(level = "trace", skip(sender, provider, reorg_handler))
 )]
-pub(crate) async fn stream_historical_range<N: Network>(
+pub(crate) async fn stream_historical_range<N: Network, R: ReorgHandler<N>>(
     start: BlockNumber,
     end: BlockNumber,
     max_block_range: u64,
     sender: &mpsc::Sender<BlockScannerResult>,
     provider: &RobustProvider<N>,
-    reorg_handler: &mut ReorgHandler<N>,
+    reorg_handler: &mut R,
 ) -> Option<()> {
     // NOTE: Edge case - If the chain is too young to expose finalized blocks (height < finalized
     // depth) just use zero.
@@ -464,14 +465,14 @@ pub(crate) async fn stream_historical_range<N: Network>(
     feature = "tracing",
     tracing::instrument(level = "trace", skip(sender, provider, reorg_handler))
 )]
-pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
+pub(crate) async fn stream_range_with_reorg_handling<N: Network, R: ReorgHandler<N>>(
     min_common_ancestor: BlockNumber,
     next_start_block: BlockNumber,
     end: BlockNumber,
     max_block_range: u64,
     sender: &mpsc::Sender<BlockScannerResult>,
     provider: &RobustProvider<N>,
-    reorg_handler: &mut ReorgHandler<N>,
+    reorg_handler: &mut R,
 ) -> Option<N::BlockResponse> {
     let mut last_batch_end: Option<N::BlockResponse> = None;
     let mut iter = RangeIterator::forward(next_start_block, end, max_block_range);

src/block_range_scanner/mod.rs

@@ -9,6 +9,7 @@ mod sync_handler;
 
 pub use builder::BlockRangeScannerBuilder;
 pub use common::BlockScannerResult;
+pub use reorg_handler::ReorgHandler;
 pub use ring_buffer::RingBufferCapacity;
 pub use scanner::BlockRangeScanner;
 

src/block_range_scanner/reorg_handler.rs

@@ -9,17 +9,34 @@ use robust_provider::RobustProvider;
 use super::ring_buffer::RingBuffer;
 use crate::{ScannerError, block_range_scanner::ring_buffer::RingBufferCapacity};
 
+/// Trait for handling chain reorganizations.
+#[allow(async_fn_in_trait)]
+pub trait ReorgHandler<N: Network> {
+    /// Checks if a block was reorged and returns the common ancestor if found.
+    ///
+    /// # Arguments
+    ///
+    /// * `block` - The block to check for reorg.
+    ///
+    /// # Returns
+    ///
+    /// * `Ok(Some(common_ancestor))` - If a reorg was detected, returns the common ancestor block.
+    /// * `Ok(None)` - If no reorg was detected, returns `None`.
+    /// * `Err(e)` - If an error occurred while checking for reorg.
+    async fn check(
+        &mut self,
+        block: &N::BlockResponse,
+    ) -> Result<Option<N::BlockResponse>, ScannerError>;
+}
+
+/// Default implementation of [`ReorgHandler`] that uses an RPC provider.
 #[derive(Clone, Debug)]
-pub(crate) struct ReorgHandler<N: Network = Ethereum> {
+pub(crate) struct DefaultReorgHandler<N: Network = Ethereum> {
     provider: RobustProvider<N>,
     buffer: RingBuffer<BlockHash>,
 }
 
-impl<N: Network> ReorgHandler<N> {
-    pub fn new(provider: RobustProvider<N>, capacity: RingBufferCapacity) -> Self {
-        Self { provider, buffer: RingBuffer::new(capacity) }
-    }
-
+impl<N: Network> ReorgHandler<N> for DefaultReorgHandler<N> {
     /// Checks if a block was reorged and returns the common ancestor if found.
     ///
     /// # Arguments
@@ -57,7 +74,7 @@ impl<N: Network> ReorgHandler<N> {
         feature = "tracing",
         tracing::instrument(level = "trace", fields(block.hash = %block.header().hash(), block.number = block.header().number()))
     )]
-    pub async fn check(
+    async fn check(
         &mut self,
         block: &N::BlockResponse,
     ) -> Result<Option<N::BlockResponse>, ScannerError> {
@@ -114,6 +131,12 @@ impl<N: Network> ReorgHandler<N> {
 
         Ok(Some(finalized))
     }
+}
+
+impl<N: Network> DefaultReorgHandler<N> {
+    pub fn new(provider: RobustProvider<N>, capacity: RingBufferCapacity) -> Self {
+        Self { provider, buffer: RingBuffer::new(capacity) }
+    }
 
     async fn reorg_detected(&self, block: &N::HeaderResponse) -> Result<bool, ScannerError> {
         match self.provider.get_block_by_hash(block.hash()).await {

src/block_range_scanner/rewind_handler.rs

@@ -12,7 +12,9 @@ use robust_provider::RobustProvider;
 use crate::{
     Notification, ScannerError,
     block_range_scanner::{
-        common::BlockScannerResult, range_iterator::RangeIterator, reorg_handler::ReorgHandler,
+        common::BlockScannerResult,
+        range_iterator::RangeIterator,
+        reorg_handler::{DefaultReorgHandler, ReorgHandler},
         ring_buffer::RingBufferCapacity,
     },
     types::{ChannelState, TryStream},
@@ -24,7 +26,7 @@ pub(crate) struct RewindHandler<N: Network> {
     start_id: BlockId,
     end_id: BlockId,
     sender: mpsc::Sender<BlockScannerResult>,
-    reorg_handler: ReorgHandler<N>,
+    reorg_handler: DefaultReorgHandler<N>,
 }
 
 impl<N: Network> RewindHandler<N> {
@@ -36,7 +38,8 @@ impl<N: Network> RewindHandler<N> {
         past_blocks_storage_capacity: RingBufferCapacity,
         sender: mpsc::Sender<BlockScannerResult>,
     ) -> Self {
-        let reorg_handler = ReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
+        let reorg_handler =
+            DefaultReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
         Self { provider, max_block_range, start_id, end_id, sender, reorg_handler }
     }
 
@@ -89,13 +92,13 @@ impl<N: Network> RewindHandler<N> {
         feature = "tracing",
         tracing::instrument(level = "trace", skip(sender, provider, reorg_handler))
     )]
-    async fn handle_stream_rewind(
+    async fn handle_stream_rewind<R: ReorgHandler<N>>(
         from: N::BlockResponse,
         to: N::BlockResponse,
         max_block_range: u64,
         sender: &mpsc::Sender<BlockScannerResult>,
         provider: &RobustProvider<N>,
-        reorg_handler: &mut ReorgHandler<N>,
+        reorg_handler: &mut R,
     ) {
         // for checking whether reorg occurred
         let mut tip = from;

src/block_range_scanner/scanner.rs

@@ -87,7 +87,7 @@ use crate::{
     block_range_scanner::{
         RingBufferCapacity,
         common::{self, BlockScannerResult},
-        reorg_handler::ReorgHandler,
+        reorg_handler::DefaultReorgHandler,
         rewind_handler::RewindHandler,
         sync_handler::SyncHandler,
     },
@@ -178,7 +178,7 @@ impl<N: Network> BlockRangeScanner<N> {
 
         tokio::spawn(async move {
             let mut reorg_handler =
-                ReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
+                DefaultReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
 
             common::stream_live_blocks(
                 start_block,
@@ -246,7 +246,7 @@ impl<N: Network> BlockRangeScanner<N> {
 
         tokio::spawn(async move {
             let mut reorg_handler =
-                ReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
+                DefaultReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
 
             _ = common::stream_historical_range(
                 start_block_num,

src/block_range_scanner/sync_handler.rs

@@ -6,7 +6,7 @@ use crate::{
     Notification, ScannerError,
     block_range_scanner::{
         common::{self, BlockScannerResult},
-        reorg_handler::ReorgHandler,
+        reorg_handler::{DefaultReorgHandler, ReorgHandler},
         ring_buffer::RingBufferCapacity,
     },
     types::TryStream,
@@ -18,7 +18,7 @@ pub(crate) struct SyncHandler<N: Network> {
     start_id: BlockId,
     block_confirmations: u64,
     sender: mpsc::Sender<BlockScannerResult>,
-    reorg_handler: ReorgHandler<N>,
+    reorg_handler: DefaultReorgHandler<N>,
 }
 
 impl<N: Network> SyncHandler<N> {
@@ -30,7 +30,8 @@ impl<N: Network> SyncHandler<N> {
         past_blocks_storage_capacity: RingBufferCapacity,
         sender: mpsc::Sender<BlockScannerResult>,
     ) -> Self {
-        let reorg_handler = ReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
+        let reorg_handler =
+            DefaultReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
         Self { provider, max_block_range, start_id, block_confirmations, sender, reorg_handler }
     }
 
@@ -147,14 +148,14 @@ impl<N: Network> SyncHandler<N> {
 
     /// Catches up on historical blocks until we reach the chain tip
     /// Returns the block number where live streaming should begin
-    async fn catchup_historical_blocks(
+    async fn catchup_historical_blocks<R: ReorgHandler<N>>(
         mut start_block: BlockNumber,
         mut confirmed_tip: BlockNumber,
         block_confirmations: u64,
         max_block_range: u64,
         sender: &mpsc::Sender<BlockScannerResult>,
         provider: &RobustProvider<N>,
-        reorg_handler: &mut ReorgHandler<N>,
+        reorg_handler: &mut R,
     ) -> Result<Option<BlockNumber>, ScannerError> {
         while start_block < confirmed_tip {
             if common::stream_historical_range(
@@ -181,13 +182,13 @@ impl<N: Network> SyncHandler<N> {
     }
 
     /// Subscribes to live blocks and begins streaming
-    async fn transition_to_live(
+    async fn transition_to_live<R: ReorgHandler<N>>(
         start_block: BlockNumber,
         block_confirmations: u64,
         max_block_range: u64,
         sender: &mpsc::Sender<BlockScannerResult>,
         provider: &RobustProvider<N>,
-        reorg_handler: &mut ReorgHandler<N>,
+        reorg_handler: &mut R,
     ) {
         let subscription = match provider.subscribe_blocks().await {
             Ok(sub) => sub,

src/lib.rs

@@ -63,7 +63,7 @@ mod types;
 
 pub use block_range_scanner::{
     BlockRangeScanner, BlockRangeScannerBuilder, BlockScannerResult, DEFAULT_BLOCK_CONFIRMATIONS,
-    DEFAULT_MAX_BLOCK_RANGE, DEFAULT_STREAM_BUFFER_CAPACITY, RingBufferCapacity,
+    DEFAULT_MAX_BLOCK_RANGE, DEFAULT_STREAM_BUFFER_CAPACITY, ReorgHandler, RingBufferCapacity,
     RingBufferCapacity as PastBlocksStorageCapacity,
 };