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

feat: add Consensus to ExecutionStage (#14447)

Browse Source
This commit is contained in:
Arsenii Kulikov
2025-02-12 18:48:14 +04:00
committed by GitHub
parent 3e07d65751
commit 172369afd5
31 changed files with 355 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/stages/stages/Cargo.toml
View File

@ -66,6 +66,7 @@ reth-chainspec.workspace = true
reth-primitives = { workspace = true, features = ["test-utils", "arbitrary"] }
reth-db = { workspace = true, features = ["test-utils", "mdbx"] }
reth-evm-ethereum.workspace = true
reth-ethereum-consensus.workspace = true
reth-execution-errors.workspace = true
reth-consensus = { workspace = true, features = ["test-utils"] }
reth-network-p2p = { workspace = true, features = ["test-utils"] }

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

@ -32,9 +32,10 @@
//! # use reth_config::config::StageConfig;
//! # use reth_consensus::{Consensus, ConsensusError};
//! # use reth_consensus::test_utils::TestConsensus;
//! # use reth_consensus::FullConsensus;
//! #
//! # let chain_spec = MAINNET.clone();
//! # let consensus: Arc<dyn Consensus<reth_primitives::Block, Error = ConsensusError>> = Arc::new(TestConsensus::default());
//! # let consensus: Arc<dyn FullConsensus<reth_primitives::EthPrimitives, Error = ConsensusError>> = Arc::new(TestConsensus::default());
//! # let headers_downloader = ReverseHeadersDownloaderBuilder::default().build(
//! # Arc::new(TestHeadersClient::default()),
//! # consensus.clone().as_header_validator()
@ -42,7 +43,7 @@
//! # let provider_factory = create_test_provider_factory();
//! # let bodies_downloader = BodiesDownloaderBuilder::default().build(
//! # Arc::new(TestBodiesClient { responder: |_| Ok((PeerId::ZERO, vec![]).into()) }),
//! # consensus.clone(),
//! # consensus.clone().as_consensus(),
//! # provider_factory.clone()
//! # );
//! # let (tip_tx, tip_rx) = watch::channel(B256::default());

75
crates/stages/stages/src/sets.rs
View File

@ -21,15 +21,17 @@
//! # use reth_config::config::StageConfig;
//! # use reth_evm::execute::BlockExecutorProvider;
//! # use reth_primitives::EthPrimitives;
//! # use std::sync::Arc;
//! # use reth_consensus::{FullConsensus, ConsensusError};
//!
//! # fn create(exec: impl BlockExecutorProvider<Primitives = EthPrimitives>) {
//! # fn create(exec: impl BlockExecutorProvider<Primitives = EthPrimitives>, consensus: impl FullConsensus<EthPrimitives, Error = ConsensusError> + 'static) {
//!
//! let provider_factory = create_test_provider_factory();
//! let static_file_producer =
//! StaticFileProducer::new(provider_factory.clone(), PruneModes::default());
//! // Build a pipeline with all offline stages.
//! let pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
//! .add_stages(OfflineStages::new(exec, StageConfig::default(), PruneModes::default()))
//! .add_stages(OfflineStages::new(exec, Arc::new(consensus), StageConfig::default(), PruneModes::default()))
//! .build(provider_factory, static_file_producer);
//!
//! # }
@ -44,9 +46,10 @@ use crate::{
};
use alloy_primitives::B256;
use reth_config::config::StageConfig;
use reth_consensus::{Consensus, ConsensusError};
use reth_consensus::{Consensus, ConsensusError, FullConsensus};
use reth_evm::execute::BlockExecutorProvider;
use reth_network_p2p::{bodies::downloader::BodyDownloader, headers::downloader::HeaderDownloader};
use reth_primitives::NodePrimitives;
use reth_primitives_traits::Block;
use reth_provider::HeaderSyncGapProvider;
use reth_prune_types::PruneModes;
@ -78,15 +81,18 @@ use tokio::sync::watch;
/// - [`PruneStage`] (execute)
/// - [`FinishStage`]
#[derive(Debug)]
pub struct DefaultStages<Provider, H, B, EF>
pub struct DefaultStages<Provider, H, B, E>
where
H: HeaderDownloader,
B: BodyDownloader,
E: BlockExecutorProvider,
{
/// Configuration for the online stages
online: OnlineStages<Provider, H, B>,
/// Executor factory needs for execution stage
executor_factory: EF,
executor_provider: E,
/// Consensus instance
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
/// Configuration for each stage in the pipeline
stages_config: StageConfig,
/// Prune configuration for every segment that can be pruned
@ -97,32 +103,31 @@ impl<Provider, H, B, E> DefaultStages<Provider, H, B, E>
where
H: HeaderDownloader,
B: BodyDownloader,
E: BlockExecutorProvider<Primitives: NodePrimitives<BlockHeader = H::Header, Block = B::Block>>,
{
/// Create a new set of default stages with default values.
#[allow(clippy::too_many_arguments)]
pub fn new(
provider: Provider,
tip: watch::Receiver<B256>,
consensus: Arc<dyn Consensus<B::Block, Error = ConsensusError>>,
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
header_downloader: H,
body_downloader: B,
executor_factory: E,
executor_provider: E,
stages_config: StageConfig,
prune_modes: PruneModes,
) -> Self
where
E: BlockExecutorProvider,
{
) -> Self {
Self {
online: OnlineStages::new(
provider,
tip,
consensus,
consensus.clone().as_consensus(),
header_downloader,
body_downloader,
stages_config.clone(),
),
executor_factory,
executor_provider,
consensus,
stages_config,
prune_modes,
}
@ -138,7 +143,8 @@ where
/// Appends the default offline stages and default finish stage to the given builder.
pub fn add_offline_stages<Provider>(
default_offline: StageSetBuilder<Provider>,
executor_factory: E,
executor_provider: E,
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
stages_config: StageConfig,
prune_modes: PruneModes,
) -> StageSetBuilder<Provider>
@ -147,7 +153,7 @@ where
{
StageSetBuilder::default()
.add_set(default_offline)
.add_set(OfflineStages::new(executor_factory, stages_config, prune_modes))
.add_set(OfflineStages::new(executor_provider, consensus, stages_config, prune_modes))
.add_stage(FinishStage)
}
}
@ -164,7 +170,8 @@ where
fn builder(self) -> StageSetBuilder<Provider> {
Self::add_offline_stages(
self.online.builder(),
self.executor_factory,
self.executor_provider,
self.consensus,
self.stages_config.clone(),
self.prune_modes,
)
@ -286,25 +293,28 @@ where
/// - [`HashingStages`]
/// - [`HistoryIndexingStages`]
/// - [`PruneStage`]
#[derive(Debug, Default)]
#[derive(Debug)]
#[non_exhaustive]
pub struct OfflineStages<EF> {
pub struct OfflineStages<E: BlockExecutorProvider> {
/// Executor factory needs for execution stage
executor_factory: EF,
executor_provider: E,
/// Consensus instance for validating blocks.
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
/// Configuration for each stage in the pipeline
stages_config: StageConfig,
/// Prune configuration for every segment that can be pruned
prune_modes: PruneModes,
}
impl<EF> OfflineStages<EF> {
impl<E: BlockExecutorProvider> OfflineStages<E> {
/// Create a new set of offline stages with default values.
pub const fn new(
executor_factory: EF,
executor_provider: E,
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
stages_config: StageConfig,
prune_modes: PruneModes,
) -> Self {
Self { executor_factory, stages_config, prune_modes }
Self { executor_provider, consensus, stages_config, prune_modes }
}
}
@ -318,7 +328,7 @@ where
PruneStage: Stage<Provider>,
{
fn builder(self) -> StageSetBuilder<Provider> {
ExecutionStages::new(self.executor_factory, self.stages_config.clone())
ExecutionStages::new(self.executor_provider, self.consensus, self.stages_config.clone())
.builder()
// If sender recovery prune mode is set, add the prune sender recovery stage.
.add_stage_opt(self.prune_modes.sender_recovery.map(|prune_mode| {
@ -341,17 +351,23 @@ where
/// A set containing all stages that are required to execute pre-existing block data.
#[derive(Debug)]
#[non_exhaustive]
pub struct ExecutionStages<E> {
pub struct ExecutionStages<E: BlockExecutorProvider> {
/// Executor factory that will create executors.
executor_factory: E,
executor_provider: E,
/// Consensus instance for validating blocks.
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
/// Configuration for each stage in the pipeline
stages_config: StageConfig,
}
impl<E> ExecutionStages<E> {
impl<E: BlockExecutorProvider> ExecutionStages<E> {
/// Create a new set of execution stages with default values.
pub const fn new(executor_factory: E, stages_config: StageConfig) -> Self {
Self { executor_factory, stages_config }
pub const fn new(
executor_provider: E,
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
stages_config: StageConfig,
) -> Self {
Self { executor_provider, consensus, stages_config }
}
}
@ -365,7 +381,8 @@ where
StageSetBuilder::default()
.add_stage(SenderRecoveryStage::new(self.stages_config.sender_recovery))
.add_stage(ExecutionStage::from_config(
self.executor_factory,
self.executor_provider,
self.consensus,
self.stages_config.execution,
self.stages_config.execution_external_clean_threshold(),
))

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

@ -4,9 +4,10 @@ use alloy_eips::{eip1898::BlockWithParent, NumHash};
use alloy_primitives::BlockNumber;
use num_traits::Zero;
use reth_config::config::ExecutionConfig;
use reth_consensus::{ConsensusError, FullConsensus, PostExecutionInput};
use reth_db::{static_file::HeaderMask, tables};
use reth_evm::{
execute::{BatchExecutor, BlockExecutorProvider},
execute::{BlockExecutorProvider, Executor},
metrics::ExecutorMetrics,
};
use reth_execution_types::Chain;
@ -15,9 +16,9 @@ use reth_primitives::StaticFileSegment;
use reth_primitives_traits::{format_gas_throughput, Block, BlockBody, NodePrimitives};
use reth_provider::{
providers::{StaticFileProvider, StaticFileWriter},
BlockHashReader, BlockReader, DBProvider, HeaderProvider, LatestStateProviderRef,
OriginalValuesKnown, ProviderError, StateCommitmentProvider, StateWriter,
StaticFileProviderFactory, StatsReader, StorageLocation, TransactionVariant,
BlockHashReader, BlockReader, DBProvider, ExecutionOutcome, HeaderProvider,
LatestStateProviderRef, OriginalValuesKnown, ProviderError, StateCommitmentProvider,
StateWriter, StaticFileProviderFactory, StatsReader, StorageLocation, TransactionVariant,
};
use reth_revm::database::StateProviderDatabase;
use reth_stages_api::{
@ -72,6 +73,9 @@ where
{
/// The stage's internal block executor
executor_provider: E,
/// The consensus instance for validating blocks.
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
/// The consensu
/// The commit thresholds of the execution stage.
thresholds: ExecutionStageThresholds,
/// The highest threshold (in number of blocks) for switching between incremental
@ -100,6 +104,7 @@ where
/// Create new execution stage with specified config.
pub fn new(
executor_provider: E,
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
thresholds: ExecutionStageThresholds,
external_clean_threshold: u64,
exex_manager_handle: ExExManagerHandle<E::Primitives>,
@ -107,6 +112,7 @@ where
Self {
external_clean_threshold,
executor_provider,
consensus,
thresholds,
post_execute_commit_input: None,
post_unwind_commit_input: None,
@ -118,9 +124,13 @@ where
/// Create an execution stage with the provided executor.
///
/// The commit threshold will be set to [`MERKLE_STAGE_DEFAULT_CLEAN_THRESHOLD`].
pub fn new_with_executor(executor_provider: E) -> Self {
pub fn new_with_executor(
executor_provider: E,
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
) -> Self {
Self::new(
executor_provider,
consensus,
ExecutionStageThresholds::default(),
MERKLE_STAGE_DEFAULT_CLEAN_THRESHOLD,
ExExManagerHandle::empty(),
@ -130,11 +140,13 @@ where
/// Create new instance of [`ExecutionStage`] from configuration.
pub fn from_config(
executor_provider: E,
consensus: Arc<dyn FullConsensus<E::Primitives, Error = ConsensusError>>,
config: ExecutionConfig,
external_clean_threshold: u64,
) -> Self {
Self::new(
executor_provider,
consensus,
config.into(),
external_clean_threshold,
ExExManagerHandle::empty(),
@ -283,7 +295,7 @@ where
self.ensure_consistency(provider, input.checkpoint().block_number, None)?;
let db = StateProviderDatabase(LatestStateProviderRef::new(provider));
let mut executor = self.executor_provider.batch_executor(db);
let mut executor = self.executor_provider.executor(db);
// Progress tracking
let mut stage_progress = start_block;
@ -310,6 +322,7 @@ where
let batch_start = Instant::now();
let mut blocks = Vec::new();
let mut results = Vec::new();
for block_number in start_block..=max_block {
// Fetch the block
let fetch_block_start = Instant::now();
@ -329,8 +342,8 @@ where
// Execute the block
let execute_start = Instant::now();
self.metrics.metered_one(&block, |input| {
executor.execute_and_verify_one(input).map_err(|error| {
let result = self.metrics.metered_one(&block, |input| {
executor.execute_one(input).map_err(|error| {
let header = block.header();
StageError::Block {
block: Box::new(BlockWithParent::new(
@ -342,6 +355,20 @@ where
})
})?;
if let Err(err) = self.consensus.validate_block_post_execution(
&block,
PostExecutionInput::new(&result.receipts, &result.requests),
) {
return Err(StageError::Block {
block: Box::new(BlockWithParent::new(
block.header().parent_hash(),
NumHash::new(block.header().number(), block.hash_slow()),
)),
error: BlockErrorKind::Validation(err),
})
}
results.push(result);
execution_duration += execute_start.elapsed();
// Log execution throughput
@ -369,10 +396,9 @@ where
}
// Check if we should commit now
let bundle_size_hint = executor.size_hint().unwrap_or_default() as u64;
if self.thresholds.is_end_of_batch(
block_number - start_block,
bundle_size_hint,
executor.size_hint() as u64,
cumulative_gas,
batch_start.elapsed(),
) {
@ -382,7 +408,11 @@ where
// prepare execution output for writing
let time = Instant::now();
let mut state = executor.finalize();
let mut state = ExecutionOutcome::from_blocks(
start_block,
executor.into_state().take_bundle(),
results,
);
let write_preparation_duration = time.elapsed();
// log the gas per second for the range we just executed
@ -649,6 +679,7 @@ mod tests {
use reth_chainspec::ChainSpecBuilder;
use reth_db::transaction::DbTx;
use reth_db_api::{models::AccountBeforeTx, transaction::DbTxMut};
use reth_ethereum_consensus::EthBeaconConsensus;
use reth_evm::execute::BasicBlockExecutorProvider;
use reth_evm_ethereum::execute::EthExecutionStrategyFactory;
use reth_primitives::{Account, Bytecode, SealedBlock, StorageEntry};
@ -666,8 +697,12 @@ mod tests {
ChainSpecBuilder::mainnet().berlin_activated().build(),
));
let executor_provider = BasicBlockExecutorProvider::new(strategy_factory);
let consensus = Arc::new(EthBeaconConsensus::new(Arc::new(
ChainSpecBuilder::mainnet().berlin_activated().build(),
)));
ExecutionStage::new(
executor_provider,
consensus,
ExecutionStageThresholds {
max_blocks: Some(100),
max_changes: None,

4
crates/stages/stages/src/stages/mod.rs
View File

@ -57,6 +57,7 @@ mod tests {
table::Table,
transaction::{DbTx, DbTxMut},
};
use reth_ethereum_consensus::EthBeaconConsensus;
use reth_evm_ethereum::execute::EthExecutorProvider;
use reth_exex::ExExManagerHandle;
use reth_primitives::{Account, Bytecode, SealedBlock, StaticFileSegment};
@ -152,6 +153,9 @@ mod tests {
EthExecutorProvider::ethereum(Arc::new(
ChainSpecBuilder::mainnet().berlin_activated().build(),
)),
Arc::new(EthBeaconConsensus::new(Arc::new(
ChainSpecBuilder::mainnet().berlin_activated().build(),
))),
ExecutionStageThresholds {
max_blocks: Some(100),
max_changes: None,