event_scanner/

block_range_scanner.rs

//! Example usage:
//!
//! ```rust,no_run
//! use alloy::{eips::BlockNumberOrTag, network::Ethereum, primitives::BlockNumber};
//! use std::ops::RangeInclusive;
//! use tokio_stream::{StreamExt, wrappers::ReceiverStream};
//!
//! use alloy::providers::{Provider, ProviderBuilder};
//! use event_scanner::{
//!     ScannerError,
//!     block_range_scanner::{
//!         BlockRangeScanner, BlockRangeScannerClient, DEFAULT_BLOCK_CONFIRMATIONS,
//!         DEFAULT_MAX_BLOCK_RANGE, Message,
//!     },
//!     robust_provider::RobustProviderBuilder,
//! };
//! use tokio::time::Duration;
//! use tracing::{error, info};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     // Initialize logging
//!     tracing_subscriber::fmt::init();
//!
//!     // Configuration
//!     let provider = ProviderBuilder::new().connect("ws://localhost:8546").await?;
//!     let robust_provider = RobustProviderBuilder::new(provider).build().await?;
//!     let block_range_scanner = BlockRangeScanner::new().connect(robust_provider).await?;
//!
//!     // Create client to send subscribe command to block scanner
//!     let client: BlockRangeScannerClient = block_range_scanner.run()?;
//!
//!     let mut stream =
//!         client.stream_from(BlockNumberOrTag::Number(5), DEFAULT_BLOCK_CONFIRMATIONS).await?;
//!
//!     while let Some(message) = stream.next().await {
//!         match message {
//!             Message::Data(range) => {
//!                 // process range
//!             }
//!             Message::Error(e) => {
//!                 error!("Received error from subscription: {e}");
//!                 match e {
//!                     ScannerError::ServiceShutdown => break,
//!                     _ => {
//!                         error!("Non-fatal error, continuing: {e}");
//!                     }
//!                 }
//!             }
//!             Message::Status(status) => {
//!                 info!("Received status message: {:?}", status);
//!             }
//!         }
//!     }
//!
//!     info!("Data processing stopped.");
//!
//!     Ok(())
//! }
//! ```

use std::{cmp::Ordering, ops::RangeInclusive};
use tokio::{
    sync::{mpsc, oneshot},
    try_join,
};
use tokio_stream::{StreamExt, wrappers::ReceiverStream};

use crate::{
    ScannerMessage,
    error::ScannerError,
    robust_provider::{Error as RobustProviderError, IntoRobustProvider, RobustProvider},
    types::{ScannerStatus, TryStream},
};
use alloy::{
    consensus::BlockHeader,
    eips::BlockNumberOrTag,
    network::{BlockResponse, Network, primitives::HeaderResponse},
    primitives::{B256, BlockNumber},
    pubsub::Subscription,
    transports::{RpcError, TransportErrorKind},
};
use tracing::{debug, error, info, warn};

pub const DEFAULT_MAX_BLOCK_RANGE: u64 = 1000;
// copied form https://github.com/taikoxyz/taiko-mono/blob/f4b3a0e830e42e2fee54829326389709dd422098/packages/taiko-client/pkg/chain_iterator/block_batch_iterator.go#L19
pub const DEFAULT_BLOCK_CONFIRMATIONS: u64 = 0;

pub const MAX_BUFFERED_MESSAGES: usize = 50000;

// Maximum amount of reorged blocks on Ethereum (after this amount of block confirmations, a block
// is considered final)
pub const DEFAULT_REORG_REWIND_DEPTH: u64 = 64;

pub type Message = ScannerMessage<RangeInclusive<BlockNumber>, ScannerError>;

impl From<RangeInclusive<BlockNumber>> for Message {
    fn from(logs: RangeInclusive<BlockNumber>) -> Self {
        Message::Data(logs)
    }
}

impl PartialEq<RangeInclusive<BlockNumber>> for Message {
    fn eq(&self, other: &RangeInclusive<BlockNumber>) -> bool {
        if let Message::Data(range) = self { range.eq(other) } else { false }
    }
}

impl From<RobustProviderError> for Message {
    fn from(error: RobustProviderError) -> Self {
        Message::Error(error.into())
    }
}

impl From<RpcError<TransportErrorKind>> for Message {
    fn from(error: RpcError<TransportErrorKind>) -> Self {
        Message::Error(error.into())
    }
}

impl From<ScannerError> for Message {
    fn from(error: ScannerError) -> Self {
        Message::Error(error)
    }
}

#[derive(Clone)]
pub struct BlockRangeScanner {
    pub max_block_range: u64,
}

impl Default for BlockRangeScanner {
    fn default() -> Self {
        Self::new()
    }
}

impl BlockRangeScanner {
    #[must_use]
    pub fn new() -> Self {
        Self { max_block_range: DEFAULT_MAX_BLOCK_RANGE }
    }

    #[must_use]
    pub fn max_block_range(mut self, max_block_range: u64) -> Self {
        self.max_block_range = max_block_range;
        self
    }

    /// Connects to an existing provider
    ///
    /// # Errors
    ///
    /// Returns an error if the provider connection fails.
    pub async fn connect<N: Network>(
        self,
        provider: impl IntoRobustProvider<N>,
    ) -> Result<ConnectedBlockRangeScanner<N>, ScannerError> {
        let provider = provider.into_robust_provider().await?;
        Ok(ConnectedBlockRangeScanner { provider, max_block_range: self.max_block_range })
    }
}

pub struct ConnectedBlockRangeScanner<N: Network> {
    provider: RobustProvider<N>,
    max_block_range: u64,
}

impl<N: Network> ConnectedBlockRangeScanner<N> {
    /// Returns the `RobustProvider`
    #[must_use]
    pub fn provider(&self) -> &RobustProvider<N> {
        &self.provider
    }

    /// Starts the subscription service and returns a client for sending commands.
    ///
    /// # Errors
    ///
    /// Returns an error if the subscription service fails to start.
    pub fn run(&self) -> Result<BlockRangeScannerClient, ScannerError> {
        let (service, cmd_tx) = Service::new(self.provider.clone(), self.max_block_range);
        tokio::spawn(async move {
            service.run().await;
        });
        Ok(BlockRangeScannerClient::new(cmd_tx))
    }
}

#[derive(Debug)]
pub enum Command {
    StreamLive {
        sender: mpsc::Sender<Message>,
        block_confirmations: u64,
        response: oneshot::Sender<Result<(), ScannerError>>,
    },
    StreamHistorical {
        sender: mpsc::Sender<Message>,
        start_height: BlockNumberOrTag,
        end_height: BlockNumberOrTag,
        response: oneshot::Sender<Result<(), ScannerError>>,
    },
    StreamFrom {
        sender: mpsc::Sender<Message>,
        start_height: BlockNumberOrTag,
        block_confirmations: u64,
        response: oneshot::Sender<Result<(), ScannerError>>,
    },
    Rewind {
        sender: mpsc::Sender<Message>,
        start_height: BlockNumberOrTag,
        end_height: BlockNumberOrTag,
        response: oneshot::Sender<Result<(), ScannerError>>,
    },
}

struct Service<N: Network> {
    provider: RobustProvider<N>,
    max_block_range: u64,
    error_count: u64,
    command_receiver: mpsc::Receiver<Command>,
    shutdown: bool,
}

impl<N: Network> Service<N> {
    pub fn new(provider: RobustProvider<N>, max_block_range: u64) -> (Self, mpsc::Sender<Command>) {
        let (cmd_tx, cmd_rx) = mpsc::channel(100);

        let service = Self {
            provider,
            max_block_range,
            error_count: 0,
            command_receiver: cmd_rx,
            shutdown: false,
        };

        (service, cmd_tx)
    }

    pub async fn run(mut self) {
        info!("Starting subscription service");

        while !self.shutdown {
            tokio::select! {
                cmd = self.command_receiver.recv() => {
                    if let Some(command) = cmd {
                        if let Err(e) = self.handle_command(command).await {
                            error!("Command handling error: {}", e);
                            self.error_count += 1;
                        }
                    } else {
                        warn!("Command channel closed, shutting down");
                        break;
                    }
                }
            }
        }

        info!("Subscription service stopped");
    }

    async fn handle_command(&mut self, command: Command) -> Result<(), ScannerError> {
        match command {
            Command::StreamLive { sender, block_confirmations, response } => {
                info!("Starting live stream");
                let result = self.handle_live(block_confirmations, sender).await;
                let _ = response.send(result);
            }
            Command::StreamHistorical { sender, start_height, end_height, response } => {
                info!(start_height = ?start_height, end_height = ?end_height, "Starting historical stream");
                let result = self.handle_historical(start_height, end_height, sender).await;
                let _ = response.send(result);
            }
            Command::StreamFrom { sender, start_height, block_confirmations, response } => {
                info!(start_height = ?start_height, "Starting streaming from");
                let result = self.handle_sync(start_height, block_confirmations, sender).await;
                let _ = response.send(result);
            }
            Command::Rewind { sender, start_height, end_height, response } => {
                info!(start_height = ?start_height, end_height = ?end_height, "Starting rewind");
                let result = self.handle_rewind(start_height, end_height, sender).await;
                let _ = response.send(result);
            }
        }
        Ok(())
    }

    async fn handle_live(
        &mut self,
        block_confirmations: u64,
        sender: mpsc::Sender<Message>,
    ) -> Result<(), ScannerError> {
        let max_block_range = self.max_block_range;
        let latest = self.provider.get_block_number().await?;

        // the next block returned by the underlying subscription will always be `latest + 1`,
        // because `latest` was already mined and subscription by definition only streams after new
        // blocks have been mined
        let range_start = (latest + 1).saturating_sub(block_confirmations);

        let subscription = self.provider.subscribe_blocks().await?;

        info!("WebSocket connected for live blocks");

        tokio::spawn(async move {
            Self::stream_live_blocks(
                range_start,
                subscription,
                sender,
                block_confirmations,
                max_block_range,
            )
            .await;
        });

        Ok(())
    }

    async fn handle_historical(
        &mut self,
        start_height: BlockNumberOrTag,
        end_height: BlockNumberOrTag,
        sender: mpsc::Sender<Message>,
    ) -> Result<(), ScannerError> {
        let max_block_range = self.max_block_range;

        let (start_block, end_block) = tokio::try_join!(
            self.provider.get_block_by_number(start_height),
            self.provider.get_block_by_number(end_height)
        )?;

        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),
            _ => (start_block_num, end_block_num),
        };

        info!(start_block = start_block_num, end_block = end_block_num, "Syncing historical data");

        tokio::spawn(async move {
            Self::stream_historical_blocks(
                start_block_num,
                end_block_num,
                max_block_range,
                &sender,
            )
            .await;
        });

        Ok(())
    }

    async fn handle_sync(
        &mut self,
        start_height: BlockNumberOrTag,
        block_confirmations: u64,
        sender: mpsc::Sender<Message>,
    ) -> Result<(), ScannerError> {
        let provider = self.provider.clone();
        let max_block_range = self.max_block_range;

        let get_start_block = async || -> Result<BlockNumber, ScannerError> {
            let block = match start_height {
                BlockNumberOrTag::Number(num) => num,
                block_tag => provider.get_block_by_number(block_tag).await?.header().number(),
            };
            Ok(block)
        };

        let get_latest_block = async || -> Result<BlockNumber, ScannerError> {
            let block =
                provider.get_block_by_number(BlockNumberOrTag::Latest).await?.header().number();
            Ok(block)
        };

        // Step 1:
        // Fetches the starting block and end block for historical sync in parallel
        let (start_block, latest_block) = tokio::try_join!(get_start_block(), get_latest_block())?;

        let confirmed_tip = latest_block.saturating_sub(block_confirmations);

        let subscription = self.provider.subscribe_blocks().await?;
        info!("Buffering live blocks");

        // If start is beyond confirmed tip, skip historical and go straight to live
        if start_block > confirmed_tip {
            info!(
                start_block = start_block,
                confirmed_tip = confirmed_tip,
                "Start block is beyond confirmed tip, starting live stream"
            );

            tokio::spawn(async move {
                Self::stream_live_blocks(
                    start_block,
                    subscription,
                    sender,
                    block_confirmations,
                    max_block_range,
                )
                .await;
            });

            return Ok(());
        }

        info!(start_block = start_block, end_block = confirmed_tip, "Syncing historical data");

        // Step 2: Setup the live streaming buffer
        // This channel will accumulate while historical sync is running
        let (live_block_buffer_sender, live_block_buffer_receiver) =
            mpsc::channel::<Message>(MAX_BUFFERED_MESSAGES);

        // The cutoff is the last block we have synced historically
        // Any block > cutoff will come from the live stream
        let cutoff = confirmed_tip;

        // This task runs independently, accumulating new blocks while wehistorical data is syncing
        tokio::spawn(async move {
            Self::stream_live_blocks(
                cutoff + 1,
                subscription,
                live_block_buffer_sender,
                block_confirmations,
                max_block_range,
            )
            .await;
        });

        tokio::spawn(async move {
            // Step 4: Perform historical synchronization
            // This processes blocks from start_block to end_block (cutoff)
            // If this fails, we need to abort the live streaming task
            Self::stream_historical_blocks(start_block, confirmed_tip, max_block_range, &sender)
                .await;

            info!("Chain tip reached, switching to live");

            if !sender.try_stream(ScannerStatus::SwitchingToLive).await {
                return;
            }

            info!("Successfully transitioned from historical to live data");

            // Step 5:
            // Spawn the buffer processor task
            // This will:
            // 1. Process all buffered blocks, filtering out any ≤ cutoff
            // 2. Forward blocks > cutoff to the user
            // 3. Continue forwarding until the buffer if exhausted (waits for new blocks from live
            //    stream)
            Self::process_live_block_buffer(live_block_buffer_receiver, sender, cutoff).await;
        });

        Ok(())
    }

    async fn handle_rewind(
        &mut self,
        start_height: BlockNumberOrTag,
        end_height: BlockNumberOrTag,
        sender: mpsc::Sender<Message>,
    ) -> Result<(), ScannerError> {
        let max_block_range = self.max_block_range;
        let provider = self.provider.clone();

        let (start_block, end_block) = try_join!(
            self.provider.get_block_by_number(start_height),
            self.provider.get_block_by_number(end_height),
        )?;

        // normalize block range
        let (from, to) = match start_block.header().number().cmp(&end_block.header().number()) {
            Ordering::Greater => (start_block, end_block),
            _ => (end_block, start_block),
        };

        tokio::spawn(async move {
            Self::stream_rewind(from, to, max_block_range, &sender, &provider).await;
        });

        Ok(())
    }

    /// Streams blocks in reverse order from `from` to `to`.
    ///
    /// The `from` block is assumed to be greater than or equal to the `to` block.
    ///
    /// # Errors
    ///
    /// Returns an error if the stream fails
    async fn stream_rewind(
        from: N::BlockResponse,
        to: N::BlockResponse,
        max_block_range: u64,
        sender: &mpsc::Sender<Message>,
        provider: &RobustProvider<N>,
    ) {
        let mut batch_count = 0;

        // for checking whether reorg occurred
        let mut tip_hash = from.header().hash();

        let from = from.header().number();
        let to = to.header().number();

        // we're iterating in reverse
        let mut batch_from = from;

        while batch_from >= to {
            let batch_to = batch_from.saturating_sub(max_block_range - 1).max(to);

            // stream the range regularly, i.e. from smaller block number to greater
            if !sender.try_stream(batch_to..=batch_from).await {
                break;
            }

            batch_count += 1;
            if batch_count % 10 == 0 {
                debug!(batch_count = batch_count, "Processed rewind batches");
            }

            // check early if end of stream achieved to avoid subtraction overflow when `to
            // == 0`
            if batch_to == to {
                break;
            }

            let reorged = match reorg_detected(provider, tip_hash).await {
                Ok(detected) => {
                    info!(block_number = %from, hash = %tip_hash, "Reorg detected");
                    detected
                }
                Err(e) => {
                    error!(error = %e, "Terminal RPC call error, shutting down");
                    _ = sender.try_stream(e);
                    return;
                }
            };

            if reorged {
                info!(block_number = %from, hash = %tip_hash, "Reorg detected");

                if !sender.try_stream(ScannerStatus::ReorgDetected).await {
                    break;
                }

                // restart rewind
                batch_from = from;
                // store the updated end block hash
                tip_hash = match provider.get_block_by_number(from.into()).await {
                    Ok(block) => block.header().hash(),
                    Err(RobustProviderError::BlockNotFound(_)) => {
                        panic!("Block with number '{from}' should exist post-reorg");
                    }
                    Err(e) => {
                        error!(error = %e, "Terminal RPC call error, shutting down");
                        _ = sender.try_stream(e);
                        return;
                    }
                };
            } else {
                // `batch_to` is always greater than `to`, so `batch_to - 1` is always a valid
                // unsigned integer
                batch_from = batch_to - 1;
            }
        }

        info!(batch_count = batch_count, "Rewind completed");
    }

    async fn stream_historical_blocks(
        start: BlockNumber,
        end: BlockNumber,
        max_block_range: u64,
        sender: &mpsc::Sender<Message>,
    ) {
        let mut batch_count = 0;

        let mut next_start_block = start;

        // must be <= to include the edge case when start == end (i.e. return the single block
        // range)
        while next_start_block <= end {
            let batch_end_block_number =
                next_start_block.saturating_add(max_block_range - 1).min(end);

            if !sender.try_stream(next_start_block..=batch_end_block_number).await {
                break;
            }

            batch_count += 1;
            if batch_count % 10 == 0 {
                debug!(batch_count = batch_count, "Processed historical batches");
            }

            if batch_end_block_number == end {
                break;
            }

            // Next block number always exists as we checked end block previously
            let next_start_block_number = batch_end_block_number.saturating_add(1);

            next_start_block = next_start_block_number;
        }

        info!(batch_count = batch_count, "Historical sync completed");
    }

    async fn stream_live_blocks(
        mut range_start: BlockNumber,
        subscription: Subscription<N::HeaderResponse>,
        sender: mpsc::Sender<Message>,
        block_confirmations: u64,
        max_block_range: u64,
    ) {
        // ensure we start streaming only after the expected_next_block cutoff
        let cutoff = range_start;
        let mut stream = subscription.into_stream().skip_while(|header| header.number() < cutoff);

        while let Some(incoming_block) = stream.next().await {
            let incoming_block_num = incoming_block.number();
            info!(block_number = incoming_block_num, "Received block header");

            if incoming_block_num < range_start {
                warn!("Reorg detected: sending forked range");
                if !sender.try_stream(ScannerStatus::ReorgDetected).await {
                    return;
                }

                // Calculate the confirmed block position for the incoming block
                let incoming_confirmed = incoming_block_num.saturating_sub(block_confirmations);

                // updated expected block to updated confirmed
                range_start = incoming_confirmed;
            }

            let confirmed = incoming_block_num.saturating_sub(block_confirmations);
            if confirmed >= range_start {
                // NOTE: Edge case when difference between range end and range start >= max
                // reads
                let range_end = confirmed.min(range_start.saturating_add(max_block_range - 1));

                info!(range_start = range_start, range_end = range_end, "Sending live block range");

                if !sender.try_stream(range_start..=range_end).await {
                    return;
                }

                // Overflow can not realistically happen
                range_start = range_end + 1;
            }
        }
    }

    async fn process_live_block_buffer(
        mut buffer_rx: mpsc::Receiver<Message>,
        sender: mpsc::Sender<Message>,
        cutoff: BlockNumber,
    ) {
        let mut processed = 0;
        let mut discarded = 0;

        // Process all buffered messages
        while let Some(data) = buffer_rx.recv().await {
            match data {
                Message::Data(range) => {
                    let (start, end) = (*range.start(), *range.end());
                    if start >= cutoff {
                        if !sender.try_stream(range).await {
                            break;
                        }
                        processed += end - start;
                    } else if end >= cutoff {
                        discarded += cutoff - start;

                        let start = cutoff;
                        if !sender.try_stream(start..=end).await {
                            break;
                        }
                        processed += end - start;
                    } else {
                        discarded += end - start;
                    }
                }
                other => {
                    // Could be error or status
                    if !sender.try_stream(other).await {
                        break;
                    }
                }
            }
        }

        info!(processed = processed, discarded = discarded, "Processed buffered messages");
    }
}

async fn reorg_detected<N: Network>(
    provider: &RobustProvider<N>,
    hash_to_check: B256,
) -> Result<bool, ScannerError> {
    match provider.get_block_by_hash(hash_to_check).await {
        Ok(_) => Ok(false),
        Err(RobustProviderError::BlockNotFound(_)) => Ok(true),
        Err(e) => Err(e.into()),
    }
}

pub struct BlockRangeScannerClient {
    command_sender: mpsc::Sender<Command>,
}

impl BlockRangeScannerClient {
    /// Creates a new subscription client.
    ///
    /// # Arguments
    ///
    /// * `command_sender` - The sender for sending commands to the subscription service.
    #[must_use]
    pub fn new(command_sender: mpsc::Sender<Command>) -> Self {
        Self { command_sender }
    }

    /// Streams live blocks starting from the latest block.
    ///
    /// # Arguments
    ///
    /// * `block_confirmations` - Number of confirmations to apply once in live mode.
    ///
    /// # Errors
    ///
    /// * `ScannerError::ServiceShutdown` - if the service is already shutting down.
    pub async fn stream_live(
        &self,
        block_confirmations: u64,
    ) -> Result<ReceiverStream<Message>, ScannerError> {
        let (blocks_sender, blocks_receiver) = mpsc::channel(MAX_BUFFERED_MESSAGES);
        let (response_tx, response_rx) = oneshot::channel();

        let command = Command::StreamLive {
            sender: blocks_sender,
            block_confirmations,
            response: response_tx,
        };

        self.command_sender.send(command).await.map_err(|_| ScannerError::ServiceShutdown)?;

        response_rx.await.map_err(|_| ScannerError::ServiceShutdown)??;

        Ok(ReceiverStream::new(blocks_receiver))
    }

    /// Streams a batch of historical blocks from `start_height` to `end_height`.
    ///
    /// # Arguments
    ///
    /// * `start_height` - The starting block number or tag.
    /// * `end_height` - The ending block number or tag.
    ///
    /// # Errors
    ///
    /// * `ScannerError::ServiceShutdown` - if the service is already shutting down.
    pub async fn stream_historical(
        &self,
        start_height: impl Into<BlockNumberOrTag>,
        end_height: impl Into<BlockNumberOrTag>,
    ) -> Result<ReceiverStream<Message>, ScannerError> {
        let (blocks_sender, blocks_receiver) = mpsc::channel(MAX_BUFFERED_MESSAGES);
        let (response_tx, response_rx) = oneshot::channel();

        let command = Command::StreamHistorical {
            sender: blocks_sender,
            start_height: start_height.into(),
            end_height: end_height.into(),
            response: response_tx,
        };

        self.command_sender.send(command).await.map_err(|_| ScannerError::ServiceShutdown)?;

        response_rx.await.map_err(|_| ScannerError::ServiceShutdown)??;

        Ok(ReceiverStream::new(blocks_receiver))
    }

    /// Streams blocks starting from `start_height` and transitions to live mode.
    ///
    /// # Arguments
    ///
    /// * `start_height` - The starting block number or tag.
    /// * `block_confirmations` - Number of confirmations to apply once in live mode.
    ///
    /// # Errors
    ///
    /// * `ScannerError::ServiceShutdown` - if the service is already shutting down.
    pub async fn stream_from(
        &self,
        start_height: impl Into<BlockNumberOrTag>,
        block_confirmations: u64,
    ) -> Result<ReceiverStream<Message>, ScannerError> {
        let (blocks_sender, blocks_receiver) = mpsc::channel(MAX_BUFFERED_MESSAGES);
        let (response_tx, response_rx) = oneshot::channel();

        let command = Command::StreamFrom {
            sender: blocks_sender,
            start_height: start_height.into(),
            block_confirmations,
            response: response_tx,
        };

        self.command_sender.send(command).await.map_err(|_| ScannerError::ServiceShutdown)?;

        response_rx.await.map_err(|_| ScannerError::ServiceShutdown)??;

        Ok(ReceiverStream::new(blocks_receiver))
    }

    /// Streams blocks in reverse order from `start_height` to `end_height`.
    ///
    /// # Arguments
    ///
    /// * `start_height` - The starting block number or tag (defaults to Latest if None).
    /// * `end_height` - The ending block number or tag (defaults to Earliest if None).
    ///
    /// # Errors
    ///
    /// * `ScannerError::ServiceShutdown` - if the service is already shutting down.
    pub async fn rewind(
        &self,
        start_height: impl Into<BlockNumberOrTag>,
        end_height: impl Into<BlockNumberOrTag>,
    ) -> Result<ReceiverStream<Message>, ScannerError> {
        let (blocks_sender, blocks_receiver) = mpsc::channel(MAX_BUFFERED_MESSAGES);
        let (response_tx, response_rx) = oneshot::channel();

        let command = Command::Rewind {
            sender: blocks_sender,
            start_height: start_height.into(),
            end_height: end_height.into(),
            response: response_tx,
        };

        self.command_sender.send(command).await.map_err(|_| ScannerError::ServiceShutdown)?;

        response_rx.await.map_err(|_| ScannerError::ServiceShutdown)??;

        Ok(ReceiverStream::new(blocks_receiver))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{assert_closed, assert_next};
    use alloy::{eips::BlockId, network::Ethereum};
    use tokio::sync::mpsc;

    #[test]
    fn block_range_scanner_defaults_match_constants() {
        let scanner = BlockRangeScanner::new();

        assert_eq!(scanner.max_block_range, DEFAULT_MAX_BLOCK_RANGE);
    }

    #[test]
    fn builder_methods_update_configuration() {
        let max_block_range = 42;

        let scanner = BlockRangeScanner::new().max_block_range(max_block_range);

        assert_eq!(scanner.max_block_range, max_block_range);
    }

    #[tokio::test]
    async fn buffered_messages_after_cutoff_are_all_passed() {
        let cutoff = 50;
        let (buffer_tx, buffer_rx) = mpsc::channel(8);
        buffer_tx.send(Message::Data(51..=55)).await.unwrap();
        buffer_tx.send(Message::Data(56..=60)).await.unwrap();
        buffer_tx.send(Message::Data(61..=70)).await.unwrap();
        drop(buffer_tx);

        let (out_tx, out_rx) = mpsc::channel(8);
        Service::<Ethereum>::process_live_block_buffer(buffer_rx, out_tx, cutoff).await;

        let mut stream = ReceiverStream::new(out_rx);

        assert_next!(stream, 51..=55);
        assert_next!(stream, 56..=60);
        assert_next!(stream, 61..=70);
        assert_closed!(stream);
    }

    #[tokio::test]
    async fn ranges_entirely_before_cutoff_are_discarded() {
        let cutoff = 100;

        let (buffer_tx, buffer_rx) = mpsc::channel(8);
        buffer_tx.send(Message::Data(40..=50)).await.unwrap();
        buffer_tx.send(Message::Data(51..=60)).await.unwrap();
        buffer_tx.send(Message::Data(61..=70)).await.unwrap();
        drop(buffer_tx);

        let (out_tx, out_rx) = mpsc::channel(8);
        Service::<Ethereum>::process_live_block_buffer(buffer_rx, out_tx, cutoff).await;

        let mut stream = ReceiverStream::new(out_rx);

        assert_closed!(stream);
    }

    #[tokio::test]
    async fn ranges_overlapping_cutoff_are_trimmed() {
        let cutoff = 75;

        let (buffer_tx, buffer_rx) = mpsc::channel(8);
        buffer_tx.send(Message::Data(60..=70)).await.unwrap();
        buffer_tx.send(Message::Data(71..=80)).await.unwrap();
        buffer_tx.send(Message::Data(81..=86)).await.unwrap();
        drop(buffer_tx);

        let (out_tx, out_rx) = mpsc::channel(8);
        Service::<Ethereum>::process_live_block_buffer(buffer_rx, out_tx, cutoff).await;

        let mut stream = ReceiverStream::new(out_rx);

        assert_next!(stream, 75..=80);
        assert_next!(stream, 81..=86);
        assert_closed!(stream);
    }

    #[tokio::test]
    async fn edge_case_range_exactly_at_cutoff() {
        let cutoff = 100;

        let (buffer_tx, buffer_rx) = mpsc::channel(8);
        buffer_tx.send(Message::Data(98..=98)).await.unwrap(); // Just before: discard
        buffer_tx.send(Message::Data(99..=100)).await.unwrap(); // Includes cutoff: trim to 100..=100
        buffer_tx.send(Message::Data(100..=100)).await.unwrap(); // Exactly at: forward
        buffer_tx.send(Message::Data(100..=101)).await.unwrap(); // Starts at cutoff: forward
        buffer_tx.send(Message::Data(102..=102)).await.unwrap(); // After cutoff: forward
        drop(buffer_tx);

        let (out_tx, out_rx) = mpsc::channel(8);
        Service::<Ethereum>::process_live_block_buffer(buffer_rx, out_tx, cutoff).await;

        let mut stream = ReceiverStream::new(out_rx);

        assert_next!(stream, 100..=100);
        assert_next!(stream, 100..=100);
        assert_next!(stream, 100..=101);
        assert_next!(stream, 102..=102);
        assert_closed!(stream);
    }

    #[tokio::test]
    async fn try_send_forwards_errors_to_subscribers() {
        let (tx, mut rx) = mpsc::channel::<Message>(1);

        _ = tx.try_stream(ScannerError::BlockNotFound(4.into())).await;

        assert!(matches!(
            rx.recv().await,
            Some(ScannerMessage::Error(ScannerError::BlockNotFound(BlockId::Number(
                BlockNumberOrTag::Number(4)
            ))))
        ));
    }
}