jleony/nanoreth
1
0
Fork 0
mirror of https://github.com/hl-archive-node/nanoreth.git synced 2025-12-06 10:59:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: use BlockWithParent for StageError (#13198)

Browse Source
This commit is contained in:
Dan Cline
2024-12-09 19:21:46 -05:00
committed by GitHub
parent 980e62a5b8
commit c9bd64018a
15 changed files with 117 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -9293,6 +9293,7 @@ dependencies = [
name = "reth-stages-api" name = "reth-stages-api"
version = "1.1.2" version = "1.1.2"
dependencies = [ dependencies = [
"alloy-eips",
"alloy-primitives", "alloy-primitives",
"aquamarine", "aquamarine",
"assert_matches", "assert_matches",

41
crates/consensus/beacon/src/engine/invalid_headers.rs
View File

@ -1,12 +1,11 @@
use alloy_consensus::Header; use alloy_eips::eip1898::BlockWithParent;
use alloy_primitives::B256; use alloy_primitives::B256;
use reth_metrics::{ use reth_metrics::{
metrics::{Counter, Gauge}, metrics::{Counter, Gauge},
Metrics, Metrics,
}; };
use reth_primitives::SealedHeader;
use schnellru::{ByLength, LruMap}; use schnellru::{ByLength, LruMap};
use std::{fmt::Debug, sync::Arc}; use std::fmt::Debug;
use tracing::warn; use tracing::warn;
/// The max hit counter for invalid headers in the cache before it is forcefully evicted. /// The max hit counter for invalid headers in the cache before it is forcefully evicted.
@ -17,20 +16,20 @@ const INVALID_HEADER_HIT_EVICTION_THRESHOLD: u8 = 128;
/// Keeps track of invalid headers. /// Keeps track of invalid headers.
#[derive(Debug)] #[derive(Debug)]
pub struct InvalidHeaderCache<H = Header> { pub struct InvalidHeaderCache {
/// This maps a header hash to a reference to its invalid ancestor. /// This maps a header hash to a reference to its invalid ancestor.
headers: LruMap<B256, HeaderEntry<H>>, headers: LruMap<B256, HeaderEntry>,
/// Metrics for the cache. /// Metrics for the cache.
metrics: InvalidHeaderCacheMetrics, metrics: InvalidHeaderCacheMetrics,
} }
impl<H: Debug> InvalidHeaderCache<H> { impl InvalidHeaderCache {
/// Invalid header cache constructor. /// Invalid header cache constructor.
pub fn new(max_length: u32) -> Self { pub fn new(max_length: u32) -> Self {
Self { headers: LruMap::new(ByLength::new(max_length)), metrics: Default::default() } Self { headers: LruMap::new(ByLength::new(max_length)), metrics: Default::default() }
} }
fn insert_entry(&mut self, hash: B256, header: Arc<H>) { fn insert_entry(&mut self, hash: B256, header: BlockWithParent) {
self.headers.insert(hash, HeaderEntry { header, hit_count: 0 }); self.headers.insert(hash, HeaderEntry { header, hit_count: 0 });
} }
@ -38,7 +37,7 @@ impl<H: Debug> InvalidHeaderCache<H> {
/// ///
/// If this is called, the hit count for the entry is incremented. /// If this is called, the hit count for the entry is incremented.
/// If the hit count exceeds the threshold, the entry is evicted and `None` is returned. /// If the hit count exceeds the threshold, the entry is evicted and `None` is returned.
pub fn get(&mut self, hash: &B256) -> Option<Arc<H>> { pub fn get(&mut self, hash: &B256) -> Option<BlockWithParent> {
{ {
let entry = self.headers.get(hash)?; let entry = self.headers.get(hash)?;
entry.hit_count += 1; entry.hit_count += 1;
@ -53,7 +52,11 @@ impl<H: Debug> InvalidHeaderCache<H> {
} }
/// Inserts an invalid block into the cache, with a given invalid ancestor. /// Inserts an invalid block into the cache, with a given invalid ancestor.
pub fn insert_with_invalid_ancestor(&mut self, header_hash: B256, invalid_ancestor: Arc<H>) { pub fn insert_with_invalid_ancestor(
&mut self,
header_hash: B256,
invalid_ancestor: BlockWithParent,
) {
if self.get(&header_hash).is_none() { if self.get(&header_hash).is_none() {
warn!(target: "consensus::engine", hash=?header_hash, ?invalid_ancestor, "Bad block with existing invalid ancestor"); warn!(target: "consensus::engine", hash=?header_hash, ?invalid_ancestor, "Bad block with existing invalid ancestor");
self.insert_entry(header_hash, invalid_ancestor); self.insert_entry(header_hash, invalid_ancestor);
@ -65,12 +68,10 @@ impl<H: Debug> InvalidHeaderCache<H> {
} }
/// Inserts an invalid ancestor into the map. /// Inserts an invalid ancestor into the map.
pub fn insert(&mut self, invalid_ancestor: SealedHeader<H>) { pub fn insert(&mut self, invalid_ancestor: BlockWithParent) {
if self.get(&invalid_ancestor.hash()).is_none() { if self.get(&invalid_ancestor.block.hash).is_none() {
let hash = invalid_ancestor.hash(); warn!(target: "consensus::engine", ?invalid_ancestor, "Bad block with hash");
let header = invalid_ancestor.unseal(); self.insert_entry(invalid_ancestor.block.hash, invalid_ancestor);
warn!(target: "consensus::engine", ?hash, ?header, "Bad block with hash");
self.insert_entry(hash, Arc::new(header));
// update metrics // update metrics
self.metrics.unique_inserts.increment(1); self.metrics.unique_inserts.increment(1);
@ -79,11 +80,11 @@ impl<H: Debug> InvalidHeaderCache<H> {
} }
} }
struct HeaderEntry<H> { struct HeaderEntry {
/// Keeps track how many times this header has been hit. /// Keeps track how many times this header has been hit.
hit_count: u8, hit_count: u8,
/// The actually header entry /// The actual header entry
header: Arc<H>, header: BlockWithParent,
} }
/// Metrics for the invalid headers cache. /// Metrics for the invalid headers cache.
@ -103,13 +104,15 @@ struct InvalidHeaderCacheMetrics {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use alloy_consensus::Header;
use reth_primitives::SealedHeader;
#[test] #[test]
fn test_hit_eviction() { fn test_hit_eviction() {
let mut cache = InvalidHeaderCache::new(10); let mut cache = InvalidHeaderCache::new(10);
let header = Header::default(); let header = Header::default();
let header = SealedHeader::seal(header); let header = SealedHeader::seal(header);
cache.insert(header.clone()); cache.insert(header.block_with_parent());
assert_eq!(cache.headers.get(&header.hash()).unwrap().hit_count, 0); assert_eq!(cache.headers.get(&header.hash()).unwrap().hit_count, 0);
for hit in 1..INVALID_HEADER_HIT_EVICTION_THRESHOLD { for hit in 1..INVALID_HEADER_HIT_EVICTION_THRESHOLD {

30
crates/consensus/beacon/src/engine/mod.rs
View File

@ -760,14 +760,14 @@ where
// iterate over ancestors in the invalid cache // iterate over ancestors in the invalid cache
// until we encounter the first valid ancestor // until we encounter the first valid ancestor
let mut current_hash = parent_hash; let mut current_hash = parent_hash;
let mut current_header = self.invalid_headers.get(&current_hash); let mut current_block = self.invalid_headers.get(&current_hash);
while let Some(header) = current_header { while let Some(block) = current_block {
current_hash = header.parent_hash; current_hash = block.parent;
current_header = self.invalid_headers.get(&current_hash); current_block = self.invalid_headers.get(&current_hash);
// If current_header is None, then the current_hash does not have an invalid // If current_header is None, then the current_hash does not have an invalid
// ancestor in the cache, check its presence in blockchain tree // ancestor in the cache, check its presence in blockchain tree
if current_header.is_none() && if current_block.is_none() &&
self.blockchain.find_block_by_hash(current_hash, BlockSource::Any)?.is_some() self.blockchain.find_block_by_hash(current_hash, BlockSource::Any)?.is_some()
{ {
return Ok(Some(current_hash)) return Ok(Some(current_hash))
@ -806,13 +806,13 @@ where
head: B256, head: B256,
) -> ProviderResult<Option<PayloadStatus>> { ) -> ProviderResult<Option<PayloadStatus>> {
// check if the check hash was previously marked as invalid // check if the check hash was previously marked as invalid
let Some(header) = self.invalid_headers.get(&check) else { return Ok(None) }; let Some(block) = self.invalid_headers.get(&check) else { return Ok(None) };
// populate the latest valid hash field // populate the latest valid hash field
let status = self.prepare_invalid_response(header.parent_hash)?; let status = self.prepare_invalid_response(block.parent)?;
// insert the head block into the invalid header cache // insert the head block into the invalid header cache
self.invalid_headers.insert_with_invalid_ancestor(head, header); self.invalid_headers.insert_with_invalid_ancestor(head, block);
Ok(Some(status)) Ok(Some(status))
} }
@ -821,10 +821,10 @@ where
/// to a forkchoice update. /// to a forkchoice update.
fn check_invalid_ancestor(&mut self, head: B256) -> ProviderResult<Option<PayloadStatus>> { fn check_invalid_ancestor(&mut self, head: B256) -> ProviderResult<Option<PayloadStatus>> {
// check if the head was previously marked as invalid // check if the head was previously marked as invalid
let Some(header) = self.invalid_headers.get(&head) else { return Ok(None) }; let Some(block) = self.invalid_headers.get(&head) else { return Ok(None) };
// populate the latest valid hash field // populate the latest valid hash field
Ok(Some(self.prepare_invalid_response(header.parent_hash)?)) Ok(Some(self.prepare_invalid_response(block.parent)?))
} }
/// Record latency metrics for one call to make a block canonical /// Record latency metrics for one call to make a block canonical
@ -1454,7 +1454,7 @@ where
fn on_pipeline_outcome(&mut self, ctrl: ControlFlow) -> RethResult<()> { fn on_pipeline_outcome(&mut self, ctrl: ControlFlow) -> RethResult<()> {
// Pipeline unwound, memorize the invalid block and wait for CL for next sync target. // Pipeline unwound, memorize the invalid block and wait for CL for next sync target.
if let ControlFlow::Unwind { bad_block, .. } = ctrl { if let ControlFlow::Unwind { bad_block, .. } = ctrl {
warn!(target: "consensus::engine", invalid_hash=?bad_block.hash(), invalid_number=?bad_block.number, "Bad block detected in unwind"); warn!(target: "consensus::engine", invalid_num_hash=?bad_block.block, "Bad block detected in unwind");
// update the `invalid_headers` cache with the new invalid header // update the `invalid_headers` cache with the new invalid header
self.invalid_headers.insert(*bad_block); self.invalid_headers.insert(*bad_block);
return Ok(()) return Ok(())
@ -1673,7 +1673,7 @@ where
self.latest_valid_hash_for_invalid_payload(block.parent_hash)? self.latest_valid_hash_for_invalid_payload(block.parent_hash)?
}; };
// keep track of the invalid header // keep track of the invalid header
self.invalid_headers.insert(block.header); self.invalid_headers.insert(block.header.block_with_parent());
PayloadStatus::new( PayloadStatus::new(
PayloadStatusEnum::Invalid { validation_error: error.to_string() }, PayloadStatusEnum::Invalid { validation_error: error.to_string() },
latest_valid_hash, latest_valid_hash,
@ -1782,7 +1782,7 @@ where
let (block, err) = err.split(); let (block, err) = err.split();
warn!(target: "consensus::engine", invalid_number=?block.number, invalid_hash=?block.hash(), %err, "Marking block as invalid"); warn!(target: "consensus::engine", invalid_number=?block.number, invalid_hash=?block.hash(), %err, "Marking block as invalid");
self.invalid_headers.insert(block.header); self.invalid_headers.insert(block.header.block_with_parent());
} }
} }
} }
@ -2035,7 +2035,7 @@ mod tests {
.await; .await;
assert_matches!( assert_matches!(
res.await, res.await,
Ok(Err(BeaconConsensusEngineError::Pipeline(n))) if matches!(*n.as_ref(),PipelineError::Stage(StageError::ChannelClosed)) Ok(Err(BeaconConsensusEngineError::Pipeline(n))) if matches!(*n.as_ref(), PipelineError::Stage(StageError::ChannelClosed))
); );
} }
@ -2141,7 +2141,7 @@ mod tests {
assert_matches!( assert_matches!(
rx.await, rx.await,
Ok(Err(BeaconConsensusEngineError::Pipeline(n))) if matches!(*n.as_ref(),PipelineError::Stage(StageError::ChannelClosed)) Ok(Err(BeaconConsensusEngineError::Pipeline(n))) if matches!(*n.as_ref(), PipelineError::Stage(StageError::ChannelClosed))
); );
} }

18
crates/engine/tree/src/tree/mod.rs
View File

@ -1328,7 +1328,7 @@ where
// Pipeline unwound, memorize the invalid block and wait for CL for next sync target. // Pipeline unwound, memorize the invalid block and wait for CL for next sync target.
if let ControlFlow::Unwind { bad_block, .. } = ctrl { if let ControlFlow::Unwind { bad_block, .. } = ctrl {
warn!(target: "engine::tree", invalid_hash=?bad_block.hash(), invalid_number=?bad_block.number, "Bad block detected in unwind"); warn!(target: "engine::tree", invalid_block=?bad_block, "Bad block detected in unwind");
// update the `invalid_headers` cache with the new invalid header // update the `invalid_headers` cache with the new invalid header
self.state.invalid_headers.insert(*bad_block); self.state.invalid_headers.insert(*bad_block);
return Ok(()) return Ok(())
@ -1678,14 +1678,14 @@ where
// iterate over ancestors in the invalid cache // iterate over ancestors in the invalid cache
// until we encounter the first valid ancestor // until we encounter the first valid ancestor
let mut current_hash = parent_hash; let mut current_hash = parent_hash;
let mut current_header = self.state.invalid_headers.get(&current_hash); let mut current_block = self.state.invalid_headers.get(&current_hash);
while let Some(header) = current_header { while let Some(block_with_parent) = current_block {
current_hash = header.parent_hash; current_hash = block_with_parent.parent;
current_header = self.state.invalid_headers.get(&current_hash); current_block = self.state.invalid_headers.get(&current_hash);
// If current_header is None, then the current_hash does not have an invalid // If current_header is None, then the current_hash does not have an invalid
// ancestor in the cache, check its presence in blockchain tree // ancestor in the cache, check its presence in blockchain tree
if current_header.is_none() && self.block_by_hash(current_hash)?.is_some() { if current_block.is_none() && self.block_by_hash(current_hash)?.is_some() {
return Ok(Some(current_hash)) return Ok(Some(current_hash))
} }
} }
@ -1735,7 +1735,7 @@ where
let Some(header) = self.state.invalid_headers.get(&check) else { return Ok(None) }; let Some(header) = self.state.invalid_headers.get(&check) else { return Ok(None) };
// populate the latest valid hash field // populate the latest valid hash field
let status = self.prepare_invalid_response(header.parent_hash)?; let status = self.prepare_invalid_response(header.parent)?;
// insert the head block into the invalid header cache // insert the head block into the invalid header cache
self.state.invalid_headers.insert_with_invalid_ancestor(head, header); self.state.invalid_headers.insert_with_invalid_ancestor(head, header);
@ -1749,7 +1749,7 @@ where
// check if the head was previously marked as invalid // check if the head was previously marked as invalid
let Some(header) = self.state.invalid_headers.get(&head) else { return Ok(None) }; let Some(header) = self.state.invalid_headers.get(&head) else { return Ok(None) };
// populate the latest valid hash field // populate the latest valid hash field
Ok(Some(self.prepare_invalid_response(header.parent_hash)?)) Ok(Some(self.prepare_invalid_response(header.parent)?))
} }
/// Validate if block is correct and satisfies all the consensus rules that concern the header /// Validate if block is correct and satisfies all the consensus rules that concern the header
@ -2395,7 +2395,7 @@ where
}; };
// keep track of the invalid header // keep track of the invalid header
self.state.invalid_headers.insert(block.header); self.state.invalid_headers.insert(block.header.block_with_parent());
Ok(PayloadStatus::new( Ok(PayloadStatus::new(
PayloadStatusEnum::Invalid { validation_error: validation_err.to_string() }, PayloadStatusEnum::Invalid { validation_error: validation_err.to_string() },
latest_valid_hash, latest_valid_hash,

7
crates/primitives-traits/src/header/sealed.rs
View File

@ -1,7 +1,7 @@
use crate::InMemorySize; use crate::InMemorySize;
pub use alloy_consensus::Header; pub use alloy_consensus::Header;
use alloy_consensus::Sealed; use alloy_consensus::Sealed;
use alloy_eips::BlockNumHash; use alloy_eips::{eip1898::BlockWithParent, BlockNumHash};
use alloy_primitives::{keccak256, BlockHash, Sealable}; use alloy_primitives::{keccak256, BlockHash, Sealable};
use alloy_rlp::{Decodable, Encodable}; use alloy_rlp::{Decodable, Encodable};
use bytes::BufMut; use bytes::BufMut;
@ -65,6 +65,11 @@ impl<H: alloy_consensus::BlockHeader> SealedHeader<H> {
pub fn num_hash(&self) -> BlockNumHash { pub fn num_hash(&self) -> BlockNumHash {
BlockNumHash::new(self.number(), self.hash) BlockNumHash::new(self.number(), self.hash)
} }
/// Return a [`BlockWithParent`] for this header.
pub fn block_with_parent(&self) -> BlockWithParent {
BlockWithParent { parent: self.parent_hash(), block: self.num_hash() }
}
} }
impl<H: InMemorySize> InMemorySize for SealedHeader<H> { impl<H: InMemorySize> InMemorySize for SealedHeader<H> {

2
crates/stages/api/Cargo.toml
View File

@ -23,7 +23,9 @@ reth-errors.workspace = true
reth-stages-types.workspace = true reth-stages-types.workspace = true
reth-static-file-types.workspace = true reth-static-file-types.workspace = true
# alloy
alloy-primitives.workspace = true alloy-primitives.workspace = true
alloy-eips.workspace = true
# metrics # metrics
reth-metrics.workspace = true reth-metrics.workspace = true

22
crates/stages/api/src/error.rs
View File

@ -1,8 +1,8 @@
use crate::PipelineEvent; use crate::PipelineEvent;
use alloy_eips::eip1898::BlockWithParent;
use reth_consensus::ConsensusError; use reth_consensus::ConsensusError;
use reth_errors::{BlockExecutionError, DatabaseError, RethError}; use reth_errors::{BlockExecutionError, DatabaseError, RethError};
use reth_network_p2p::error::DownloadError; use reth_network_p2p::error::DownloadError;
use reth_primitives_traits::SealedHeader;
use reth_provider::ProviderError; use reth_provider::ProviderError;
use reth_prune::{PruneSegment, PruneSegmentError, PrunerError}; use reth_prune::{PruneSegment, PruneSegmentError, PrunerError};
use reth_static_file_types::StaticFileSegment; use reth_static_file_types::StaticFileSegment;
@ -34,10 +34,10 @@ impl BlockErrorKind {
#[derive(Error, Debug)] #[derive(Error, Debug)]
pub enum StageError { pub enum StageError {
/// The stage encountered an error related to a block. /// The stage encountered an error related to a block.
#[error("stage encountered an error in block #{number}: {error}", number = block.number)] #[error("stage encountered an error in block #{number}: {error}", number = block.block.number)]
Block { Block {
/// The block that caused the error. /// The block that caused the error.
block: Box<SealedHeader>, block: Box<BlockWithParent>,
/// The specific error type, either consensus or execution error. /// The specific error type, either consensus or execution error.
#[source] #[source]
error: BlockErrorKind, error: BlockErrorKind,
@ -48,16 +48,16 @@ pub enum StageError {
"stage encountered inconsistent chain: \ "stage encountered inconsistent chain: \
downloaded header #{header_number} ({header_hash}) is detached from \ downloaded header #{header_number} ({header_hash}) is detached from \
local head #{head_number} ({head_hash}): {error}", local head #{head_number} ({head_hash}): {error}",
header_number = header.number, header_number = header.block.number,
header_hash = header.hash(), header_hash = header.block.hash,
head_number = local_head.number, head_number = local_head.block.number,
head_hash = local_head.hash(), head_hash = local_head.block.hash,
)] )]
DetachedHead { DetachedHead {
/// The local head we attempted to attach to. /// The local head we attempted to attach to.
local_head: Box<SealedHeader>, local_head: Box<BlockWithParent>,
/// The header we attempted to attach. /// The header we attempted to attach.
header: Box<SealedHeader>, header: Box<BlockWithParent>,
/// The error that occurred when attempting to attach the header. /// The error that occurred when attempting to attach the header.
#[source] #[source]
error: Box<ConsensusError>, error: Box<ConsensusError>,
@ -92,10 +92,10 @@ pub enum StageError {
#[error("invalid download response: {0}")] #[error("invalid download response: {0}")]
Download(#[from] DownloadError), Download(#[from] DownloadError),
/// Database is ahead of static file data. /// Database is ahead of static file data.
#[error("missing static file data for block number: {number}", number = block.number)] #[error("missing static file data for block number: {number}", number = block.block.number)]
MissingStaticFileData { MissingStaticFileData {
/// Starting block with missing data. /// Starting block with missing data.
block: Box<SealedHeader>, block: Box<BlockWithParent>,
/// Static File segment /// Static File segment
segment: StaticFileSegment, segment: StaticFileSegment,
}, },

4
crates/stages/api/src/pipeline/ctrl.rs
View File

@ -1,5 +1,5 @@
use alloy_eips::eip1898::BlockWithParent;
use alloy_primitives::BlockNumber; use alloy_primitives::BlockNumber;
use reth_primitives_traits::SealedHeader;
/// Determines the control flow during pipeline execution. /// Determines the control flow during pipeline execution.
/// ///
@ -11,7 +11,7 @@ pub enum ControlFlow {
/// The block to unwind to. /// The block to unwind to.
target: BlockNumber, target: BlockNumber,
/// The block that caused the unwind. /// The block that caused the unwind.
bad_block: Box<SealedHeader>, bad_block: Box<BlockWithParent>,
}, },
/// The pipeline made progress. /// The pipeline made progress.
Continue { Continue {

16
crates/stages/api/src/pipeline/mod.rs
View File

@ -223,7 +223,7 @@ impl<N: ProviderNodeTypes> Pipeline<N> {
} }
ControlFlow::Continue { block_number } => self.progress.update(block_number), ControlFlow::Continue { block_number } => self.progress.update(block_number),
ControlFlow::Unwind { target, bad_block } => { ControlFlow::Unwind { target, bad_block } => {
self.unwind(target, Some(bad_block.number))?; self.unwind(target, Some(bad_block.block.number))?;
return Ok(ControlFlow::Unwind { target, bad_block }) return Ok(ControlFlow::Unwind { target, bad_block })
} }
} }
@ -505,7 +505,7 @@ fn on_stage_error<N: ProviderNodeTypes>(
// We unwind because of a detached head. // We unwind because of a detached head.
let unwind_to = let unwind_to =
local_head.number.saturating_sub(BEACON_CONSENSUS_REORG_UNWIND_DEPTH).max(1); local_head.block.number.saturating_sub(BEACON_CONSENSUS_REORG_UNWIND_DEPTH).max(1);
Ok(Some(ControlFlow::Unwind { target: unwind_to, bad_block: local_head })) Ok(Some(ControlFlow::Unwind { target: unwind_to, bad_block: local_head }))
} else if let StageError::Block { block, error } = err { } else if let StageError::Block { block, error } = err {
match error { match error {
@ -513,7 +513,7 @@ fn on_stage_error<N: ProviderNodeTypes>(
error!( error!(
target: "sync::pipeline", target: "sync::pipeline",
stage = %stage_id, stage = %stage_id,
bad_block = %block.number, bad_block = %block.block.number,
"Stage encountered a validation error: {validation_error}" "Stage encountered a validation error: {validation_error}"
); );
@ -542,7 +542,7 @@ fn on_stage_error<N: ProviderNodeTypes>(
error!( error!(
target: "sync::pipeline", target: "sync::pipeline",
stage = %stage_id, stage = %stage_id,
bad_block = %block.number, bad_block = %block.block.number,
"Stage encountered an execution error: {execution_error}" "Stage encountered an execution error: {execution_error}"
); );
@ -560,12 +560,12 @@ fn on_stage_error<N: ProviderNodeTypes>(
error!( error!(
target: "sync::pipeline", target: "sync::pipeline",
stage = %stage_id, stage = %stage_id,
bad_block = %block.number, bad_block = %block.block.number,
segment = %segment, segment = %segment,
"Stage is missing static file data." "Stage is missing static file data."
); );
Ok(Some(ControlFlow::Unwind { target: block.number - 1, bad_block: block })) Ok(Some(ControlFlow::Unwind { target: block.block.number - 1, bad_block: block }))
} else if err.is_fatal() { } else if err.is_fatal() {
error!(target: "sync::pipeline", stage = %stage_id, "Stage encountered a fatal error: {err}"); error!(target: "sync::pipeline", stage = %stage_id, "Stage encountered a fatal error: {err}");
Err(err.into()) Err(err.into())
@ -603,7 +603,7 @@ mod tests {
use reth_errors::ProviderError; use reth_errors::ProviderError;
use reth_provider::test_utils::{create_test_provider_factory, MockNodeTypesWithDB}; use reth_provider::test_utils::{create_test_provider_factory, MockNodeTypesWithDB};
use reth_prune::PruneModes; use reth_prune::PruneModes;
use reth_testing_utils::{generators, generators::random_header}; use reth_testing_utils::generators::{self, random_block_with_parent};
use tokio_stream::StreamExt; use tokio_stream::StreamExt;
#[test] #[test]
@ -975,7 +975,7 @@ mod tests {
.add_stage( .add_stage(
TestStage::new(StageId::Other("B")) TestStage::new(StageId::Other("B"))
.add_exec(Err(StageError::Block { .add_exec(Err(StageError::Block {
block: Box::new(random_header( block: Box::new(random_block_with_parent(
&mut generators::rng(), &mut generators::rng(),
5, 5,
Default::default(), Default::default(),

15
crates/stages/stages/src/stages/execution.rs
View File

@ -1,5 +1,6 @@
use crate::stages::MERKLE_STAGE_DEFAULT_CLEAN_THRESHOLD; use crate::stages::MERKLE_STAGE_DEFAULT_CLEAN_THRESHOLD;
use alloy_consensus::{BlockHeader, Header}; use alloy_consensus::{BlockHeader, Header};
use alloy_eips::{eip1898::BlockWithParent, NumHash};
use alloy_primitives::BlockNumber; use alloy_primitives::BlockNumber;
use num_traits::Zero; use num_traits::Zero;
use reth_config::config::ExecutionConfig; use reth_config::config::ExecutionConfig;
@ -11,7 +12,7 @@ use reth_evm::{
}; };
use reth_execution_types::Chain; use reth_execution_types::Chain;
use reth_exex::{ExExManagerHandle, ExExNotification, ExExNotificationSource}; use reth_exex::{ExExManagerHandle, ExExNotification, ExExNotificationSource};
use reth_primitives::{SealedHeader, StaticFileSegment}; use reth_primitives::StaticFileSegment;
use reth_primitives_traits::{format_gas_throughput, Block, BlockBody, NodePrimitives}; use reth_primitives_traits::{format_gas_throughput, Block, BlockBody, NodePrimitives};
use reth_provider::{ use reth_provider::{
providers::{StaticFileProvider, StaticFileWriter}, providers::{StaticFileProvider, StaticFileWriter},
@ -359,9 +360,15 @@ where
let execute_start = Instant::now(); let execute_start = Instant::now();
self.metrics.metered_one((&block, td).into(), |input| { self.metrics.metered_one((&block, td).into(), |input| {
executor.execute_and_verify_one(input).map_err(|error| StageError::Block { executor.execute_and_verify_one(input).map_err(|error| {
block: Box::new(SealedHeader::seal(block.header().clone())), let header = block.header();
error: BlockErrorKind::Execution(error), StageError::Block {
block: Box::new(BlockWithParent::new(
header.parent_hash(),
NumHash::new(header.number(), header.hash_slow()),
)),
error: BlockErrorKind::Execution(error),
}
}) })
})?; })?;

12
crates/stages/stages/src/stages/headers.rs
View File

@ -1,4 +1,5 @@
use alloy_consensus::BlockHeader; use alloy_consensus::BlockHeader;
use alloy_eips::{eip1898::BlockWithParent, NumHash};
use alloy_primitives::{BlockHash, BlockNumber, Bytes, B256}; use alloy_primitives::{BlockHash, BlockNumber, Bytes, B256};
use futures_util::StreamExt; use futures_util::StreamExt;
use reth_config::config::EtlConfig; use reth_config::config::EtlConfig;
@ -143,7 +144,10 @@ where
// Header validation // Header validation
self.consensus.validate_header_with_total_difficulty(&header, td).map_err(|error| { self.consensus.validate_header_with_total_difficulty(&header, td).map_err(|error| {
StageError::Block { StageError::Block {
block: Box::new(SealedHeader::new(header.clone(), header_hash)), block: Box::new(BlockWithParent::new(
header.parent_hash,
NumHash::new(header.number, header_hash),
)),
error: BlockErrorKind::Validation(error), error: BlockErrorKind::Validation(error),
} }
})?; })?;
@ -272,7 +276,11 @@ where
} }
Some(Err(HeadersDownloaderError::DetachedHead { local_head, header, error })) => { Some(Err(HeadersDownloaderError::DetachedHead { local_head, header, error })) => {
error!(target: "sync::stages::headers", %error, "Cannot attach header to head"); error!(target: "sync::stages::headers", %error, "Cannot attach header to head");
return Poll::Ready(Err(StageError::DetachedHead { local_head, header, error })) return Poll::Ready(Err(StageError::DetachedHead {
local_head: Box::new(local_head.block_with_parent()),
header: Box::new(header.block_with_parent()),
error,
}))
} }
None => return Poll::Ready(Err(StageError::ChannelClosed)), None => return Poll::Ready(Err(StageError::ChannelClosed)),
} }

2
crates/stages/stages/src/stages/merkle.rs
View File

@ -357,7 +357,7 @@ fn validate_state_root(
error: BlockErrorKind::Validation(ConsensusError::BodyStateRootDiff( error: BlockErrorKind::Validation(ConsensusError::BodyStateRootDiff(
GotExpected { got, expected: expected.state_root }.into(), GotExpected { got, expected: expected.state_root }.into(),
)), )),
block: Box::new(expected), block: Box::new(expected.block_with_parent()),
}) })
} }
} }

2
crates/stages/stages/src/stages/sender_recovery.rs
View File

@ -192,7 +192,7 @@ where
})?; })?;
Err(StageError::Block { Err(StageError::Block {
block: Box::new(sealed_header), block: Box::new(sealed_header.block_with_parent()),
error: BlockErrorKind::Validation( error: BlockErrorKind::Validation(
ConsensusError::TransactionSignerRecoveryError, ConsensusError::TransactionSignerRecoveryError,
), ),

5
crates/stages/stages/src/stages/utils.rs
View File

@ -279,5 +279,8 @@ where
let missing_block = Box::new(provider.sealed_header(last_block + 1)?.unwrap_or_default()); let missing_block = Box::new(provider.sealed_header(last_block + 1)?.unwrap_or_default());
Ok(StageError::MissingStaticFileData { block: missing_block, segment }) Ok(StageError::MissingStaticFileData {
block: Box::new(missing_block.block_with_parent()),
segment,
})
} }

15
testing/testing-utils/src/generators.rs
View File

@ -1,7 +1,11 @@
//! Generators for different data structures like block headers, block bodies and ranges of those. //! Generators for different data structures like block headers, block bodies and ranges of those.
use alloy_consensus::{Header, Transaction as _, TxLegacy}; use alloy_consensus::{Header, Transaction as _, TxLegacy};
use alloy_eips::eip4895::{Withdrawal, Withdrawals}; use alloy_eips::{
eip1898::BlockWithParent,
eip4895::{Withdrawal, Withdrawals},
NumHash,
};
use alloy_primitives::{Address, BlockNumber, Bytes, TxKind, B256, U256}; use alloy_primitives::{Address, BlockNumber, Bytes, TxKind, B256, U256};
pub use rand::Rng; pub use rand::Rng;
use rand::{ use rand::{
@ -95,6 +99,15 @@ pub fn random_header_range<R: Rng>(
headers headers
} }
/// Generate a random [`BlockWithParent`].
pub fn random_block_with_parent<R: Rng>(
rng: &mut R,
number: u64,
parent: Option<B256>,
) -> BlockWithParent {
BlockWithParent { parent: parent.unwrap_or_default(), block: NumHash::new(number, rng.gen()) }
}
/// Generate a random [`SealedHeader`]. /// Generate a random [`SealedHeader`].
/// ///
/// The header is assumed to not be correct if validated. /// The header is assumed to not be correct if validated.