From b257679b3c8d9239fe7078c9102e44b3c426fcf9 Mon Sep 17 00:00:00 2001 From: gavin Date: Fri, 29 May 2026 08:22:23 +0800 Subject: [PATCH] fix reorg detection for canonical blocks --- src/block_range_scanner/reorg_handler.rs | 93 +++++++++++++++++++----- 1 file changed, 76 insertions(+), 17 deletions(-) diff --git a/src/block_range_scanner/reorg_handler.rs b/src/block_range_scanner/reorg_handler.rs index aff904a5..c97e016b 100644 --- a/src/block_range_scanner/reorg_handler.rs +++ b/src/block_range_scanner/reorg_handler.rs @@ -2,7 +2,7 @@ use alloy::{ consensus::BlockHeader, eips::BlockNumberOrTag, network::{BlockResponse, Ethereum, Network, primitives::HeaderResponse}, - primitives::BlockHash, + primitives::{BlockHash, BlockNumber}, }; use robust_provider::RobustProvider; @@ -33,7 +33,7 @@ pub trait ReorgHandler { #[derive(Clone, Debug)] pub(crate) struct DefaultReorgHandler { provider: RobustProvider, - buffer: RingBuffer, + buffer: RingBuffer<(BlockNumber, BlockHash)>, } impl ReorgHandler for DefaultReorgHandler { @@ -81,12 +81,13 @@ impl ReorgHandler for DefaultReorgHandler { let block = block.header(); if !self.reorg_detected(block).await? { + let block_number = block.number(); let block_hash = block.hash(); // 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); + if !matches!(self.buffer.back(), Some(&(_, hash)) if hash == block_hash) { + self.buffer.push((block_number, block_hash)); trace!( - block_number = block.number(), + block_number, block_hash = %block_hash, "Block hash added to reorg buffer" ); @@ -100,20 +101,27 @@ impl ReorgHandler for DefaultReorgHandler { "Reorg detected, searching for common ancestor" ); - while let Some(&block_hash) = self.buffer.back() { - trace!(block_hash = %block_hash, "Checking if buffered block exists on chain"); - match self.provider.get_block_by_hash(block_hash).await { - Ok(common_ancestor) => { + while let Some(&(candidate_number, candidate_hash)) = self.buffer.back() { + trace!( + block_number = candidate_number, + block_hash = %candidate_hash, + "Checking if buffered block is canonical" + ); + match self.provider.get_block_by_number(candidate_number.into()).await { + Ok(canonical) if canonical.header().hash() == candidate_hash => { debug!( - common_ancestor_hash = %block_hash, - common_ancestor_number = common_ancestor.header().number(), + common_ancestor_hash = %candidate_hash, + common_ancestor_number = candidate_number, "Found common ancestor" ); - return self.return_common_ancestor(common_ancestor).await; + return self.return_common_ancestor(canonical).await; } - Err(robust_provider::Error::BlockNotFound) => { - // block was reorged - trace!(block_hash = %block_hash, "Buffered block was reorged, removing from buffer"); + Ok(_) | Err(robust_provider::Error::BlockNotFound) => { + trace!( + block_number = candidate_number, + block_hash = %candidate_hash, + "Buffered block is no longer canonical, removing from buffer" + ); _ = self.buffer.pop_back(); } Err(e) => return Err(e.into()), @@ -139,8 +147,8 @@ impl DefaultReorgHandler { } async fn reorg_detected(&self, block: &N::HeaderResponse) -> Result { - match self.provider.get_block_by_hash(block.hash()).await { - Ok(_) => Ok(false), + match self.provider.get_block_by_number(block.number().into()).await { + Ok(canonical_block) => Ok(canonical_block.header().hash() != block.hash()), Err(robust_provider::Error::BlockNotFound) => Ok(true), Err(e) => Err(e.into()), } @@ -177,3 +185,54 @@ impl DefaultReorgHandler { Ok(Some(common_ancestor)) } } + +#[cfg(test)] +mod tests { + use super::*; + use alloy::{ + consensus::Header as ConsensusHeader, + network::Ethereum, + providers::{RootProvider, mock::Asserter}, + rpc::{client::RpcClient, types::Block}, + }; + use robust_provider::RobustProviderBuilder; + + fn block(number: u64, hash: BlockHash) -> Block { + Block::empty(alloy::rpc::types::Header { + hash, + inner: ConsensusHeader { number, ..Default::default() }, + total_difficulty: None, + size: None, + }) + } + + #[tokio::test] + async fn detects_reorg_when_old_hash_is_available_but_not_canonical() -> anyhow::Result<()> { + let parent_hash = BlockHash::repeat_byte(0x09); + let old_hash = BlockHash::repeat_byte(0x0a); + let new_hash = BlockHash::repeat_byte(0x0b); + let parent_block = block(9, parent_hash); + let old_block = block(10, old_hash); + let new_block = block(10, new_hash); + + let asserter = Asserter::new(); + asserter.push_success(&Some(new_block.clone())); + asserter.push_success(&Some(new_block)); + asserter.push_success(&Some(parent_block.clone())); + asserter.push_success(&Some(parent_block.clone())); + asserter.push_success(&Some(parent_block)); + + let provider = RootProvider::::new(RpcClient::mocked(asserter)); + let provider = RobustProviderBuilder::fragile(provider).build().await?; + let mut handler = DefaultReorgHandler::new(provider, RingBufferCapacity::Limited(10)); + handler.buffer.push((9, parent_hash)); + handler.buffer.push((10, old_hash)); + + let common_ancestor = handler.check(&old_block).await?.expect("reorg should be detected"); + + assert_eq!(common_ancestor.header().number(), 9); + assert_eq!(common_ancestor.header().hash(), parent_hash); + + Ok(()) + } +}