From e794d6ecad7fa6a5d22beafcc5d5d7bb6d2f7701 Mon Sep 17 00:00:00 2001 From: TateB Date: Tue, 21 Jul 2026 12:06:58 +0000 Subject: [PATCH] fix(indexer): retry raced reth head refresh Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01C9RTnbCyhqEUSPc7xCK76p --- apps/indexer/src/provider/reth_db/enabled.rs | 20 +- apps/indexer/src/provider/reth_db/head.rs | 242 +++++++++++++++++++ 2 files changed, 247 insertions(+), 15 deletions(-) create mode 100644 apps/indexer/src/provider/reth_db/head.rs diff --git a/apps/indexer/src/provider/reth_db/enabled.rs b/apps/indexer/src/provider/reth_db/enabled.rs index c036f9d0..6997dff2 100644 --- a/apps/indexer/src/provider/reth_db/enabled.rs +++ b/apps/indexer/src/provider/reth_db/enabled.rs @@ -12,8 +12,8 @@ use reth_ethereum::{ node::{EthereumNode, api::NodeTypesWithDBAdapter}, primitives::Block as _, provider::{ - BlockHashReader, BlockNumReader, BlockReader, ChainStateBlockReader, HeaderProvider, - ProviderError, ProviderFactory, ReceiptProvider, TransactionVariant, db::DatabaseEnv, + BlockHashReader, BlockReader, ChainStateBlockReader, HeaderProvider, ProviderError, + ProviderFactory, ReceiptProvider, TransactionVariant, db::DatabaseEnv, providers::ReadOnlyConfig, }, }; @@ -24,6 +24,8 @@ mod api; mod code; #[path = "convert.rs"] mod convert; +#[path = "head.rs"] +mod head; #[path = "logs.rs"] mod logs; #[path = "receipts.rs"] @@ -96,23 +98,11 @@ impl RethDbReader { fn fetch_chain_heads_sync(&self) -> Result { let factory = self.factory()?; - let chain_info = factory.chain_info()?; + let canonical_hash = head::fetch_canonical_head_with_retry(&self.chain, &factory)?; let provider = factory.provider()?; let safe_number = nonzero_checkpoint_number(provider.last_safe_block_number()?); let finalized_number = nonzero_checkpoint_number(provider.last_finalized_block_number()?); drop(provider); - let canonical_hash = if chain_info.best_hash == B256::ZERO { - factory - .block_hash(chain_info.best_number)? - .with_context(|| { - format!( - "Reth DB did not return canonical block hash for best number {}", - chain_info.best_number - ) - })? - } else { - chain_info.best_hash - }; Ok(ProviderHeadSnapshot { canonical: self.fetch_block_by_b256(&factory, canonical_hash)?, diff --git a/apps/indexer/src/provider/reth_db/head.rs b/apps/indexer/src/provider/reth_db/head.rs new file mode 100644 index 00000000..4990e0dd --- /dev/null +++ b/apps/indexer/src/provider/reth_db/head.rs @@ -0,0 +1,242 @@ +use std::{fmt, thread, time::Duration}; + +use alloy_primitives::B256; +use anyhow::Result; +use reth_ethereum::provider::{BlockHashReader, BlockNumReader}; +use tracing::{debug, info}; + +use super::EthereumRethProviderFactory; + +const HEAD_REFRESH_MAX_ATTEMPTS: usize = 5; +const HEAD_REFRESH_RETRY_BACKOFF: Duration = Duration::from_millis(150); + +#[derive(Debug, Eq, PartialEq)] +struct CanonicalHead { + best_number: u64, + block_hash: B256, +} + +#[derive(Debug)] +struct MissingCanonicalBlockHash { + best_number: u64, +} + +impl fmt::Display for MissingCanonicalBlockHash { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "Reth DB did not return canonical block hash for best number {}", + self.best_number + ) + } +} + +impl std::error::Error for MissingCanonicalBlockHash {} + +pub(super) fn fetch_canonical_head_with_retry( + chain: &str, + factory: &EthereumRethProviderFactory, +) -> Result { + retry_canonical_head_read(chain, || fetch_canonical_head_once(factory), thread::sleep) + .map(|head| head.block_hash) +} + +fn fetch_canonical_head_once(factory: &EthereumRethProviderFactory) -> Result { + let chain_info = factory.chain_info()?; + let block_hash = if chain_info.best_hash == B256::ZERO { + factory + .block_hash(chain_info.best_number)? + .ok_or(MissingCanonicalBlockHash { + best_number: chain_info.best_number, + })? + } else { + chain_info.best_hash + }; + + Ok(CanonicalHead { + best_number: chain_info.best_number, + block_hash, + }) +} + +fn retry_canonical_head_read( + chain: &str, + mut read: impl FnMut() -> Result, + mut backoff: impl FnMut(Duration), +) -> Result { + let mut attempt_count = 1usize; + + loop { + match read() { + Ok(head) => { + if attempt_count > 1 { + info!( + service = "indexer", + command = "provider", + chain, + canonical_best_number = head.best_number, + attempt_count, + retry_count = attempt_count - 1, + "Reth DB head refresh recovered after canonical block hash retry" + ); + } + return Ok(head); + } + Err(error) => { + let Some(raced_best_number) = missing_canonical_block_hash_best_number(&error) + else { + return Err(error); + }; + if attempt_count >= HEAD_REFRESH_MAX_ATTEMPTS { + return Err(error); + } + + debug!( + service = "indexer", + command = "provider", + chain, + raced_best_number, + attempt_count, + max_attempts = HEAD_REFRESH_MAX_ATTEMPTS, + backoff_ms = HEAD_REFRESH_RETRY_BACKOFF.as_millis() as u64, + "Reth DB canonical block hash was absent for the resolved best number; retrying head refresh" + ); + backoff(HEAD_REFRESH_RETRY_BACKOFF); + attempt_count += 1; + } + } + } +} + +fn missing_canonical_block_hash_best_number(error: &anyhow::Error) -> Option { + error.chain().find_map(|cause| { + cause + .downcast_ref::() + .map(|missing| missing.best_number) + }) +} + +#[cfg(test)] +mod tests { + use std::collections::VecDeque; + + use reth_ethereum::provider::ProviderError; + + use super::*; + + #[test] + fn classifier_accepts_only_typed_missing_canonical_hash_error() { + let typed = anyhow::Error::new(MissingCanonicalBlockHash { best_number: 42 }); + assert_eq!(missing_canonical_block_hash_best_number(&typed), Some(42)); + + let matching_text = + anyhow::anyhow!("Reth DB did not return canonical block hash for best number 42"); + assert_eq!( + missing_canonical_block_hash_best_number(&matching_text), + None + ); + + let other_provider_error = anyhow::Error::new(ProviderError::StateForNumberNotFound(42)); + assert_eq!( + missing_canonical_block_hash_best_number(&other_provider_error), + None + ); + } + + #[test] + fn absent_then_present_retries_with_a_fresh_best_number() -> Result<()> { + let resolved_hash = B256::repeat_byte(0x22); + let mut outcomes = VecDeque::from([ + Err(anyhow::Error::new(MissingCanonicalBlockHash { + best_number: 100, + })), + Ok(CanonicalHead { + best_number: 101, + block_hash: resolved_hash, + }), + ]); + let mut read_count = 0usize; + let mut backoffs = Vec::new(); + + let head = retry_canonical_head_read( + "ethereum-mainnet", + || { + read_count += 1; + outcomes.pop_front().expect("test read outcome") + }, + |duration| backoffs.push(duration), + )?; + + assert_eq!(read_count, 2); + assert!(outcomes.is_empty()); + assert_eq!( + head, + CanonicalHead { + best_number: 101, + block_hash: resolved_hash, + } + ); + assert_eq!(backoffs, vec![HEAD_REFRESH_RETRY_BACKOFF]); + Ok(()) + } + + #[test] + fn persistent_absence_exhausts_attempt_cap_and_preserves_error() { + let mut read_count = 0usize; + let mut backoffs = Vec::new(); + + let error = retry_canonical_head_read( + "ethereum-mainnet", + || { + let best_number = 200 + read_count as u64; + read_count += 1; + Err(anyhow::Error::new(MissingCanonicalBlockHash { + best_number, + })) + }, + |duration| backoffs.push(duration), + ) + .expect_err("persistent missing canonical hash must fail"); + + assert_eq!(read_count, HEAD_REFRESH_MAX_ATTEMPTS); + assert_eq!( + backoffs, + vec![HEAD_REFRESH_RETRY_BACKOFF; HEAD_REFRESH_MAX_ATTEMPTS - 1] + ); + assert_eq!( + error.to_string(), + "Reth DB did not return canonical block hash for best number 204" + ); + assert_eq!( + error + .downcast_ref::() + .map(|missing| missing.best_number), + Some(204) + ); + } + + #[test] + fn unrelated_error_does_not_retry() { + let mut read_count = 0usize; + let mut backoffs = Vec::new(); + + let error = retry_canonical_head_read( + "ethereum-mainnet", + || { + read_count += 1; + Err(anyhow::Error::new(ProviderError::StateForNumberNotFound( + 42, + ))) + }, + |duration| backoffs.push(duration), + ) + .expect_err("unrelated head-refresh error must fail immediately"); + + assert_eq!(read_count, 1); + assert!(backoffs.is_empty()); + assert!(matches!( + error.downcast_ref::(), + Some(ProviderError::StateForNumberNotFound(42)) + )); + } +}