feat: block executor provider and ethereum + op impl (#7594)

Co-authored-by: Oliver Nordbjerg <onbjerg@users.noreply.github.com>
Co-authored-by: Georgios Konstantopoulos <me@gakonst.com>

Cargo.lock

@@ -6601,6 +6601,7 @@ dependencies = [
 name = "reth-evm"
 version = "0.2.0-beta.5"
 dependencies = [
+ "reth-interfaces",
  "reth-primitives",
  "revm",
  "revm-primitives",
@@ -6611,7 +6612,12 @@ name = "reth-evm-ethereum"
 version = "0.2.0-beta.5"
 dependencies = [
  "reth-evm",
+ "reth-interfaces",
  "reth-primitives",
+ "reth-provider",
+ "reth-revm",
+ "revm-primitives",
+ "tracing",
 ]
 
 [[package]]
@@ -7008,6 +7014,8 @@ dependencies = [
  "reqwest 0.11.27",
  "reth-basic-payload-builder",
  "reth-db",
+ "reth-evm",
+ "reth-interfaces",
  "reth-network",
  "reth-node-api",
  "reth-node-builder",
@@ -7022,9 +7030,11 @@ dependencies = [
  "reth-tracing",
  "reth-transaction-pool",
  "revm",
+ "revm-primitives",
  "serde",
  "serde_json",
  "thiserror",
+ "tracing",
 ]
 
 [[package]]

crates/evm-ethereum/Cargo.toml

@@ -11,6 +11,18 @@ repository.workspace = true
 workspace = true
 
 [dependencies]
+# Reth
 reth-evm.workspace = true
 reth-primitives.workspace = true
+reth-revm.workspace = true
+reth-interfaces.workspace = true
+reth-provider.workspace = true
 
+# Ethereum
+revm-primitives.workspace = true
+
+# misc
+tracing.workspace = true
+
+[dev-dependencies]
+reth-revm = { workspace = true, features = ["test-utils"] }

crates/evm-ethereum/src/execute.rs

@@ -0,0 +1,826 @@
+//! Ethereum block executor.
+
+use crate::EthEvmConfig;
+use reth_evm::{
+    execute::{
+        BatchBlockOutput, BatchExecutor, EthBlockExecutionInput, EthBlockOutput, Executor,
+        ExecutorProvider,
+    },
+    ConfigureEvm, ConfigureEvmEnv,
+};
+use reth_interfaces::{
+    executor::{BlockExecutionError, BlockValidationError},
+    provider::ProviderError,
+};
+use reth_primitives::{
+    BlockWithSenders, ChainSpec, GotExpected, Hardfork, Header, PruneModes, Receipt, Receipts,
+    Withdrawals, U256,
+};
+use reth_provider::BundleStateWithReceipts;
+use reth_revm::{
+    batch::{BlockBatchRecord, BlockExecutorStats},
+    db::states::bundle_state::BundleRetention,
+    eth_dao_fork::{DAO_HARDFORK_BENEFICIARY, DAO_HARDKFORK_ACCOUNTS},
+    processor::verify_receipt,
+    stack::InspectorStack,
+    state_change::{apply_beacon_root_contract_call, post_block_balance_increments},
+    Evm, State,
+};
+use revm_primitives::{
+    db::{Database, DatabaseCommit},
+    BlockEnv, CfgEnvWithHandlerCfg, EnvWithHandlerCfg, ResultAndState,
+};
+use std::sync::Arc;
+use tracing::debug;
+
+/// Provides executors to execute regular ethereum blocks
+#[derive(Debug, Clone)]
+pub struct EthExecutorProvider<EvmConfig> {
+    chain_spec: Arc<ChainSpec>,
+    evm_config: EvmConfig,
+    inspector: Option<InspectorStack>,
+    prune_modes: PruneModes,
+}
+
+impl EthExecutorProvider<EthEvmConfig> {
+    /// Creates a new default ethereum executor provider.
+    pub fn ethereum(chain_spec: Arc<ChainSpec>) -> Self {
+        Self::new(chain_spec, Default::default())
+    }
+}
+
+impl<EvmConfig> EthExecutorProvider<EvmConfig> {
+    /// Creates a new executor provider.
+    pub fn new(chain_spec: Arc<ChainSpec>, evm_config: EvmConfig) -> Self {
+        Self { chain_spec, evm_config, inspector: None, prune_modes: PruneModes::none() }
+    }
+
+    /// Configures an optional inspector stack for debugging.
+    pub fn with_inspector(mut self, inspector: InspectorStack) -> Self {
+        self.inspector = Some(inspector);
+        self
+    }
+
+    /// Configures the prune modes for the executor.
+    pub fn with_prune_modes(mut self, prune_modes: PruneModes) -> Self {
+        self.prune_modes = prune_modes;
+        self
+    }
+}
+
+impl<EvmConfig> EthExecutorProvider<EvmConfig>
+where
+    EvmConfig: ConfigureEvm,
+    EvmConfig: ConfigureEvmEnv<TxMeta = ()>,
+{
+    fn eth_executor<DB>(&self, db: DB) -> EthBlockExecutor<EvmConfig, DB>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        EthBlockExecutor::new(
+            self.chain_spec.clone(),
+            self.evm_config.clone(),
+            State::builder().with_database(db).with_bundle_update().without_state_clear().build(),
+        )
+        .with_inspector(self.inspector.clone())
+    }
+}
+
+impl<EvmConfig> ExecutorProvider for EthExecutorProvider<EvmConfig>
+where
+    EvmConfig: ConfigureEvm,
+    EvmConfig: ConfigureEvmEnv<TxMeta = ()>,
+{
+    type Executor<DB: Database<Error = ProviderError>> = EthBlockExecutor<EvmConfig, DB>;
+
+    type BatchExecutor<DB: Database<Error = ProviderError>> = EthBatchExecutor<EvmConfig, DB>;
+
+    fn executor<DB>(&self, db: DB) -> Self::Executor<DB>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        self.eth_executor(db)
+    }
+
+    fn batch_executor<DB>(&self, db: DB) -> Self::BatchExecutor<DB>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        let executor = self.eth_executor(db);
+        EthBatchExecutor {
+            executor,
+            batch_record: BlockBatchRecord::new(self.prune_modes.clone()),
+            stats: BlockExecutorStats::default(),
+        }
+    }
+}
+
+/// Helper container type for EVM with chain spec.
+#[derive(Debug, Clone)]
+struct EthEvmExecutor<EvmConfig> {
+    /// The chainspec
+    chain_spec: Arc<ChainSpec>,
+    /// How to create an EVM.
+    evm_config: EvmConfig,
+}
+
+impl<EvmConfig> EthEvmExecutor<EvmConfig>
+where
+    EvmConfig: ConfigureEvm,
+    EvmConfig: ConfigureEvmEnv<TxMeta = ()>,
+{
+    /// Executes the transactions in the block and returns the receipts.
+    ///
+    /// This applies the pre-execution changes, and executes the transactions.
+    ///
+    /// # Note
+    ///
+    /// It does __not__ apply post-execution changes.
+    fn execute_pre_and_transactions<Ext, DB>(
+        &mut self,
+        block: &BlockWithSenders,
+        mut evm: Evm<'_, Ext, &mut State<DB>>,
+    ) -> Result<(Vec<Receipt>, u64), BlockExecutionError>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        // apply pre execution changes
+        apply_beacon_root_contract_call(
+            &self.chain_spec,
+            block.timestamp,
+            block.number,
+            block.parent_beacon_block_root,
+            &mut evm,
+        )?;
+
+        // execute transactions
+        let mut cumulative_gas_used = 0;
+        let mut receipts = Vec::with_capacity(block.body.len());
+        for (sender, transaction) in block.transactions_with_sender() {
+            // The sum of the transaction’s gas limit, Tg, and the gas utilized in this block prior,
+            // must be no greater than the block’s gasLimit.
+            let block_available_gas = block.header.gas_limit - cumulative_gas_used;
+            if transaction.gas_limit() > block_available_gas {
+                return Err(BlockValidationError::TransactionGasLimitMoreThanAvailableBlockGas {
+                    transaction_gas_limit: transaction.gas_limit(),
+                    block_available_gas,
+                }
+                .into())
+            }
+
+            EvmConfig::fill_tx_env(evm.tx_mut(), transaction, *sender, ());
+
+            // Execute transaction.
+            let ResultAndState { result, state } = evm.transact().map_err(move |err| {
+                // Ensure hash is calculated for error log, if not already done
+                BlockValidationError::EVM {
+                    hash: transaction.recalculate_hash(),
+                    error: err.into(),
+                }
+            })?;
+            evm.db_mut().commit(state);
+
+            // append gas used
+            cumulative_gas_used += result.gas_used();
+
+            // Push transaction changeset and calculate header bloom filter for receipt.
+            receipts.push(
+                #[allow(clippy::needless_update)] // side-effect of optimism fields
+                Receipt {
+                    tx_type: transaction.tx_type(),
+                    // Success flag was added in `EIP-658: Embedding transaction status code in
+                    // receipts`.
+                    success: result.is_success(),
+                    cumulative_gas_used,
+                    // convert to reth log
+                    logs: result.into_logs(),
+                    ..Default::default()
+                },
+            );
+        }
+        drop(evm);
+
+        // Check if gas used matches the value set in header.
+        if block.gas_used != cumulative_gas_used {
+            let receipts = Receipts::from_block_receipt(receipts);
+            return Err(BlockValidationError::BlockGasUsed {
+                gas: GotExpected { got: cumulative_gas_used, expected: block.gas_used },
+                gas_spent_by_tx: receipts.gas_spent_by_tx()?,
+            }
+            .into())
+        }
+
+        Ok((receipts, cumulative_gas_used))
+    }
+}
+
+/// A basic Ethereum block executor.
+///
+/// Expected usage:
+/// - Create a new instance of the executor.
+/// - Execute the block.
+#[derive(Debug)]
+pub struct EthBlockExecutor<EvmConfig, DB> {
+    /// Chain specific evm config that's used to execute a block.
+    executor: EthEvmExecutor<EvmConfig>,
+    /// The state to use for execution
+    state: State<DB>,
+    /// Optional inspector stack for debugging
+    inspector: Option<InspectorStack>,
+}
+
+impl<EvmConfig, DB> EthBlockExecutor<EvmConfig, DB> {
+    /// Creates a new Ethereum block executor.
+    pub fn new(chain_spec: Arc<ChainSpec>, evm_config: EvmConfig, state: State<DB>) -> Self {
+        Self { executor: EthEvmExecutor { chain_spec, evm_config }, state, inspector: None }
+    }
+
+    /// Sets the inspector stack for debugging.
+    pub fn with_inspector(mut self, inspector: Option<InspectorStack>) -> Self {
+        self.inspector = inspector;
+        self
+    }
+
+    #[inline]
+    fn chain_spec(&self) -> &ChainSpec {
+        &self.executor.chain_spec
+    }
+
+    /// Returns mutable reference to the state that wraps the underlying database.
+    #[allow(unused)]
+    fn state_mut(&mut self) -> &mut State<DB> {
+        &mut self.state
+    }
+}
+
+impl<EvmConfig, DB> EthBlockExecutor<EvmConfig, DB>
+where
+    EvmConfig: ConfigureEvm,
+    // TODO(mattsse): get rid of this
+    EvmConfig: ConfigureEvmEnv<TxMeta = ()>,
+    DB: Database<Error = ProviderError>,
+{
+    /// Configures a new evm configuration and block environment for the given block.
+    ///
+    /// # Caution
+    ///
+    /// This does not initialize the tx environment.
+    fn evm_env_for_block(&self, header: &Header, total_difficulty: U256) -> EnvWithHandlerCfg {
+        let mut cfg = CfgEnvWithHandlerCfg::new(Default::default(), Default::default());
+        let mut block_env = BlockEnv::default();
+        EvmConfig::fill_cfg_and_block_env(
+            &mut cfg,
+            &mut block_env,
+            self.chain_spec(),
+            header,
+            total_difficulty,
+        );
+
+        EnvWithHandlerCfg::new_with_cfg_env(cfg, block_env, Default::default())
+    }
+
+    /// Execute a single block and apply the state changes to the internal state.
+    ///
+    /// Returns the receipts of the transactions in the block and the total gas used.
+    ///
+    /// Returns an error if execution fails or receipt verification fails.
+    fn execute_and_verify(
+        &mut self,
+        block: &BlockWithSenders,
+        total_difficulty: U256,
+    ) -> Result<(Vec<Receipt>, u64), BlockExecutionError> {
+        // 1. prepare state on new block
+        self.on_new_block(&block.header);
+
+        // 2. configure the evm and execute
+        let env = self.evm_env_for_block(&block.header, total_difficulty);
+
+        let (receipts, gas_used) = {
+            if let Some(inspector) = self.inspector.as_mut() {
+                let evm = self.executor.evm_config.evm_with_env_and_inspector(
+                    &mut self.state,
+                    env,
+                    inspector,
+                );
+                self.executor.execute_pre_and_transactions(block, evm)?
+            } else {
+                let evm = self.executor.evm_config.evm_with_env(&mut self.state, env);
+
+                self.executor.execute_pre_and_transactions(block, evm)?
+            }
+        };
+
+        // 3. apply post execution changes
+        self.post_execution(block, total_difficulty)?;
+
+        // Before Byzantium, receipts contained state root that would mean that expensive
+        // operation as hashing that is required for state root got calculated in every
+        // transaction This was replaced with is_success flag.
+        // See more about EIP here: https://eips.ethereum.org/EIPS/eip-658
+        if self.chain_spec().is_byzantium_active_at_block(block.header.number) {
+            if let Err(error) =
+                verify_receipt(block.header.receipts_root, block.header.logs_bloom, receipts.iter())
+            {
+                debug!(target: "evm", %error, ?receipts, "receipts verification failed");
+                return Err(error)
+            };
+        }
+
+        Ok((receipts, gas_used))
+    }
+
+    /// Apply settings before a new block is executed.
+    pub(crate) fn on_new_block(&mut self, header: &Header) {
+        // Set state clear flag if the block is after the Spurious Dragon hardfork.
+        let state_clear_flag = self.chain_spec().is_spurious_dragon_active_at_block(header.number);
+        self.state.set_state_clear_flag(state_clear_flag);
+    }
+
+    /// Apply post execution state changes, including block rewards, withdrawals, and irregular DAO
+    /// hardfork state change.
+    pub fn post_execution(
+        &mut self,
+        block: &BlockWithSenders,
+        total_difficulty: U256,
+    ) -> Result<(), BlockExecutionError> {
+        let mut balance_increments = post_block_balance_increments(
+            self.chain_spec(),
+            block.number,
+            block.difficulty,
+            block.beneficiary,
+            block.timestamp,
+            total_difficulty,
+            &block.ommers,
+            block.withdrawals.as_ref().map(Withdrawals::as_ref),
+        );
+
+        // Irregular state change at Ethereum DAO hardfork
+        if self.chain_spec().fork(Hardfork::Dao).transitions_at_block(block.number) {
+            // drain balances from hardcoded addresses.
+            let drained_balance: u128 = self
+                .state
+                .drain_balances(DAO_HARDKFORK_ACCOUNTS)
+                .map_err(|_| BlockValidationError::IncrementBalanceFailed)?
+                .into_iter()
+                .sum();
+
+            // return balance to DAO beneficiary.
+            *balance_increments.entry(DAO_HARDFORK_BENEFICIARY).or_default() += drained_balance;
+        }
+        // increment balances
+        self.state
+            .increment_balances(balance_increments)
+            .map_err(|_| BlockValidationError::IncrementBalanceFailed)?;
+
+        Ok(())
+    }
+}
+
+impl<EvmConfig, DB> Executor<DB> for EthBlockExecutor<EvmConfig, DB>
+where
+    EvmConfig: ConfigureEvm,
+    EvmConfig: ConfigureEvmEnv<TxMeta = ()>,
+    DB: Database<Error = ProviderError>,
+{
+    type Input<'a> = EthBlockExecutionInput<'a, BlockWithSenders>;
+    type Output = EthBlockOutput<Receipt>;
+    type Error = BlockExecutionError;
+
+    /// Executes the block and commits the state changes.
+    ///
+    /// Returns the receipts of the transactions in the block.
+    ///
+    /// Returns an error if the block could not be executed or failed verification.
+    ///
+    /// State changes are committed to the database.
+    fn execute(mut self, input: Self::Input<'_>) -> Result<Self::Output, Self::Error> {
+        let EthBlockExecutionInput { block, total_difficulty } = input;
+        let (receipts, gas_used) = self.execute_and_verify(block, total_difficulty)?;
+
+        // prepare the state for extraction
+        self.state.merge_transitions(BundleRetention::PlainState);
+
+        Ok(EthBlockOutput { state: self.state.take_bundle(), receipts, gas_used })
+    }
+}
+
+/// An executor for a batch of blocks.
+///
+/// State changes are tracked until the executor is finalized.
+#[derive(Debug)]
+pub struct EthBatchExecutor<EvmConfig, DB> {
+    /// The executor used to execute single blocks
+    ///
+    /// All state changes are committed to the [State].
+    executor: EthBlockExecutor<EvmConfig, DB>,
+    /// Keeps track of the batch and records receipts based on the configured prune mode
+    batch_record: BlockBatchRecord,
+    stats: BlockExecutorStats,
+}
+
+impl<EvmConfig, DB> EthBatchExecutor<EvmConfig, DB> {
+    /// Returns mutable reference to the state that wraps the underlying database.
+    #[allow(unused)]
+    fn state_mut(&mut self) -> &mut State<DB> {
+        self.executor.state_mut()
+    }
+}
+
+impl<EvmConfig, DB> BatchExecutor<DB> for EthBatchExecutor<EvmConfig, DB>
+where
+    EvmConfig: ConfigureEvm,
+    // TODO(mattsse): get rid of this
+    EvmConfig: ConfigureEvmEnv<TxMeta = ()>,
+    DB: Database<Error = ProviderError>,
+{
+    type Input<'a> = EthBlockExecutionInput<'a, BlockWithSenders>;
+    type Output = BundleStateWithReceipts;
+    type Error = BlockExecutionError;
+
+    fn execute_one(&mut self, input: Self::Input<'_>) -> Result<BatchBlockOutput, Self::Error> {
+        let EthBlockExecutionInput { block, total_difficulty } = input;
+        let (receipts, _gas_used) = self.executor.execute_and_verify(block, total_difficulty)?;
+
+        // prepare the state according to the prune mode
+        let retention = self.batch_record.bundle_retention(block.number);
+        self.executor.state.merge_transitions(retention);
+
+        // store receipts in the set
+        self.batch_record.save_receipts(receipts)?;
+
+        Ok(BatchBlockOutput { size_hint: Some(self.executor.state.bundle_size_hint()) })
+    }
+
+    fn finalize(mut self) -> Self::Output {
+        self.stats.log_debug();
+
+        BundleStateWithReceipts::new(
+            self.executor.state.take_bundle(),
+            self.batch_record.take_receipts(),
+            self.batch_record.first_block().unwrap_or_default(),
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::EthEvmConfig;
+    use reth_primitives::{
+        bytes,
+        constants::{BEACON_ROOTS_ADDRESS, SYSTEM_ADDRESS},
+        keccak256, Account, Block, Bytes, ChainSpecBuilder, ForkCondition, B256, MAINNET,
+    };
+    use reth_revm::{
+        database::StateProviderDatabase, test_utils::StateProviderTest, TransitionState,
+    };
+    use std::collections::HashMap;
+
+    static BEACON_ROOT_CONTRACT_CODE: Bytes = bytes!("3373fffffffffffffffffffffffffffffffffffffffe14604d57602036146024575f5ffd5b5f35801560495762001fff810690815414603c575f5ffd5b62001fff01545f5260205ff35b5f5ffd5b62001fff42064281555f359062001fff015500");
+
+    fn create_state_provider_with_beacon_root_contract() -> StateProviderTest {
+        let mut db = StateProviderTest::default();
+
+        let beacon_root_contract_account = Account {
+            balance: U256::ZERO,
+            bytecode_hash: Some(keccak256(BEACON_ROOT_CONTRACT_CODE.clone())),
+            nonce: 1,
+        };
+
+        db.insert_account(
+            BEACON_ROOTS_ADDRESS,
+            beacon_root_contract_account,
+            Some(BEACON_ROOT_CONTRACT_CODE.clone()),
+            HashMap::new(),
+        );
+
+        db
+    }
+
+    fn executor_provider(chain_spec: Arc<ChainSpec>) -> EthExecutorProvider<EthEvmConfig> {
+        EthExecutorProvider {
+            chain_spec,
+            evm_config: Default::default(),
+            inspector: None,
+            prune_modes: Default::default(),
+        }
+    }
+
+    #[test]
+    fn eip_4788_non_genesis_call() {
+        let mut header =
+            Header { timestamp: 1, number: 1, excess_blob_gas: Some(0), ..Header::default() };
+
+        let db = create_state_provider_with_beacon_root_contract();
+
+        let chain_spec = Arc::new(
+            ChainSpecBuilder::from(&*MAINNET)
+                .shanghai_activated()
+                .with_fork(Hardfork::Cancun, ForkCondition::Timestamp(1))
+                .build(),
+        );
+
+        let provider = executor_provider(chain_spec);
+
+        // attempt to execute a block without parent beacon block root, expect err
+        let err = provider
+            .executor(StateProviderDatabase::new(&db))
+            .execute(
+                (
+                    &BlockWithSenders {
+                        block: Block {
+                            header: header.clone(),
+                            body: vec![],
+                            ommers: vec![],
+                            withdrawals: None,
+                        },
+                        senders: vec![],
+                    },
+                    U256::ZERO,
+                )
+                    .into(),
+            )
+            .expect_err(
+                "Executing cancun block without parent beacon block root field should fail",
+            );
+        assert_eq!(
+            err,
+            BlockExecutionError::Validation(BlockValidationError::MissingParentBeaconBlockRoot)
+        );
+
+        // fix header, set a gas limit
+        header.parent_beacon_block_root = Some(B256::with_last_byte(0x69));
+
+        let mut executor = provider.executor(StateProviderDatabase::new(&db));
+
+        // Now execute a block with the fixed header, ensure that it does not fail
+        executor
+            .execute_and_verify(
+                &BlockWithSenders {
+                    block: Block {
+                        header: header.clone(),
+                        body: vec![],
+                        ommers: vec![],
+                        withdrawals: None,
+                    },
+                    senders: vec![],
+                },
+                U256::ZERO,
+            )
+            .unwrap();
+
+        // check the actual storage of the contract - it should be:
+        // * The storage value at header.timestamp % HISTORY_BUFFER_LENGTH should be
+        // header.timestamp
+        // * The storage value at header.timestamp % HISTORY_BUFFER_LENGTH + HISTORY_BUFFER_LENGTH
+        //   // should be parent_beacon_block_root
+        let history_buffer_length = 8191u64;
+        let timestamp_index = header.timestamp % history_buffer_length;
+        let parent_beacon_block_root_index =
+            timestamp_index % history_buffer_length + history_buffer_length;
+
+        // get timestamp storage and compare
+        let timestamp_storage =
+            executor.state.storage(BEACON_ROOTS_ADDRESS, U256::from(timestamp_index)).unwrap();
+        assert_eq!(timestamp_storage, U256::from(header.timestamp));
+
+        // get parent beacon block root storage and compare
+        let parent_beacon_block_root_storage = executor
+            .state
+            .storage(BEACON_ROOTS_ADDRESS, U256::from(parent_beacon_block_root_index))
+            .expect("storage value should exist");
+        assert_eq!(parent_beacon_block_root_storage, U256::from(0x69));
+    }
+
+    #[test]
+    fn eip_4788_no_code_cancun() {
+        // This test ensures that we "silently fail" when cancun is active and there is no code at
+        // // BEACON_ROOTS_ADDRESS
+        let header = Header {
+            timestamp: 1,
+            number: 1,
+            parent_beacon_block_root: Some(B256::with_last_byte(0x69)),
+            excess_blob_gas: Some(0),
+            ..Header::default()
+        };
+
+        let db = StateProviderTest::default();
+
+        // DON'T deploy the contract at genesis
+        let chain_spec = Arc::new(
+            ChainSpecBuilder::from(&*MAINNET)
+                .shanghai_activated()
+                .with_fork(Hardfork::Cancun, ForkCondition::Timestamp(1))
+                .build(),
+        );
+
+        let provider = executor_provider(chain_spec);
+
+        // attempt to execute an empty block with parent beacon block root, this should not fail
+        provider
+            .executor(StateProviderDatabase::new(&db))
+            .execute(
+                (
+                    &BlockWithSenders {
+                        block: Block { header, body: vec![], ommers: vec![], withdrawals: None },
+                        senders: vec![],
+                    },
+                    U256::ZERO,
+                )
+                    .into(),
+            )
+            .expect(
+                "Executing a block with no transactions while cancun is active should not fail",
+            );
+    }
+
+    #[test]
+    fn eip_4788_empty_account_call() {
+        // This test ensures that we do not increment the nonce of an empty SYSTEM_ADDRESS account
+        // // during the pre-block call
+
+        let mut db = create_state_provider_with_beacon_root_contract();
+
+        // insert an empty SYSTEM_ADDRESS
+        db.insert_account(SYSTEM_ADDRESS, Account::default(), None, HashMap::new());
+
+        let chain_spec = Arc::new(
+            ChainSpecBuilder::from(&*MAINNET)
+                .shanghai_activated()
+                .with_fork(Hardfork::Cancun, ForkCondition::Timestamp(1))
+                .build(),
+        );
+
+        let provider = executor_provider(chain_spec);
+
+        // construct the header for block one
+        let header = Header {
+            timestamp: 1,
+            number: 1,
+            parent_beacon_block_root: Some(B256::with_last_byte(0x69)),
+            excess_blob_gas: Some(0),
+            ..Header::default()
+        };
+
+        let mut executor = provider.executor(StateProviderDatabase::new(&db));
+
+        // attempt to execute an empty block with parent beacon block root, this should not fail
+        executor
+            .execute_and_verify(
+                &BlockWithSenders {
+                    block: Block { header, body: vec![], ommers: vec![], withdrawals: None },
+                    senders: vec![],
+                },
+                U256::ZERO,
+            )
+            .expect(
+                "Executing a block with no transactions while cancun is active should not fail",
+            );
+
+        // ensure that the nonce of the system address account has not changed
+        let nonce = executor.state_mut().basic(SYSTEM_ADDRESS).unwrap().unwrap().nonce;
+        assert_eq!(nonce, 0);
+    }
+
+    #[test]
+    fn eip_4788_genesis_call() {
+        let db = create_state_provider_with_beacon_root_contract();
+
+        // activate cancun at genesis
+        let chain_spec = Arc::new(
+            ChainSpecBuilder::from(&*MAINNET)
+                .shanghai_activated()
+                .with_fork(Hardfork::Cancun, ForkCondition::Timestamp(0))
+                .build(),
+        );
+
+        let mut header = chain_spec.genesis_header();
+
+        let provider = executor_provider(chain_spec);
+
+        let mut executor = provider.batch_executor(StateProviderDatabase::new(&db));
+
+        // attempt to execute the genesis block with non-zero parent beacon block root, expect err
+        header.parent_beacon_block_root = Some(B256::with_last_byte(0x69));
+        let _err = executor
+            .execute_one(
+                (
+                    &BlockWithSenders {
+                        block: Block {
+                            header: header.clone(),
+                            body: vec![],
+                            ommers: vec![],
+                            withdrawals: None,
+                        },
+                        senders: vec![],
+                    },
+                    U256::ZERO,
+                )
+                    .into(),
+            )
+            .expect_err(
+                "Executing genesis cancun block with non-zero parent beacon block root field
+    should fail",
+            );
+
+        // fix header
+        header.parent_beacon_block_root = Some(B256::ZERO);
+
+        // now try to process the genesis block again, this time ensuring that a system contract
+        // call does not occur
+        executor
+            .execute_one(
+                (
+                    &BlockWithSenders {
+                        block: Block { header, body: vec![], ommers: vec![], withdrawals: None },
+                        senders: vec![],
+                    },
+                    U256::ZERO,
+                )
+                    .into(),
+            )
+            .unwrap();
+
+        // there is no system contract call so there should be NO STORAGE CHANGES
+        // this means we'll check the transition state
+        let transition_state = executor
+            .state_mut()
+            .transition_state
+            .take()
+            .expect("the evm should be initialized with bundle updates");
+
+        // assert that it is the default (empty) transition state
+        assert_eq!(transition_state, TransitionState::default());
+    }
+
+    #[test]
+    fn eip_4788_high_base_fee() {
+        // This test ensures that if we have a base fee, then we don't return an error when the
+        // system contract is called, due to the gas price being less than the base fee.
+        let header = Header {
+            timestamp: 1,
+            number: 1,
+            parent_beacon_block_root: Some(B256::with_last_byte(0x69)),
+            base_fee_per_gas: Some(u64::MAX),
+            excess_blob_gas: Some(0),
+            ..Header::default()
+        };
+
+        let db = create_state_provider_with_beacon_root_contract();
+
+        let chain_spec = Arc::new(
+            ChainSpecBuilder::from(&*MAINNET)
+                .shanghai_activated()
+                .with_fork(Hardfork::Cancun, ForkCondition::Timestamp(1))
+                .build(),
+        );
+
+        let provider = executor_provider(chain_spec);
+
+        // execute header
+        let mut executor = provider.batch_executor(StateProviderDatabase::new(&db));
+
+        // Now execute a block with the fixed header, ensure that it does not fail
+        executor
+            .execute_one(
+                (
+                    &BlockWithSenders {
+                        block: Block {
+                            header: header.clone(),
+                            body: vec![],
+                            ommers: vec![],
+                            withdrawals: None,
+                        },
+                        senders: vec![],
+                    },
+                    U256::ZERO,
+                )
+                    .into(),
+            )
+            .unwrap();
+
+        // check the actual storage of the contract - it should be:
+        // * The storage value at header.timestamp % HISTORY_BUFFER_LENGTH should be
+        // header.timestamp
+        // * The storage value at header.timestamp % HISTORY_BUFFER_LENGTH + HISTORY_BUFFER_LENGTH
+        //   // should be parent_beacon_block_root
+        let history_buffer_length = 8191u64;
+        let timestamp_index = header.timestamp % history_buffer_length;
+        let parent_beacon_block_root_index =
+            timestamp_index % history_buffer_length + history_buffer_length;
+
+        // get timestamp storage and compare
+        let timestamp_storage = executor
+            .state_mut()
+            .storage(BEACON_ROOTS_ADDRESS, U256::from(timestamp_index))
+            .unwrap();
+        assert_eq!(timestamp_storage, U256::from(header.timestamp));
+
+        // get parent beacon block root storage and compare
+        let parent_beacon_block_root_storage = executor
+            .state_mut()
+            .storage(BEACON_ROOTS_ADDRESS, U256::from(parent_beacon_block_root_index))
+            .unwrap();
+        assert_eq!(parent_beacon_block_root_storage, U256::from(0x69));
+    }
+}

crates/evm-ethereum/src/lib.rs

@@ -14,6 +14,7 @@ use reth_primitives::{
     revm_primitives::{AnalysisKind, CfgEnvWithHandlerCfg, TxEnv},
     Address, ChainSpec, Head, Header, Transaction, U256,
 };
+pub mod execute;
 
 /// Ethereum-related EVM configuration.
 #[derive(Debug, Clone, Copy, Default)]

crates/evm/Cargo.toml

@@ -15,4 +15,5 @@ workspace = true
 reth-primitives.workspace = true
 revm-primitives.workspace = true
 revm.workspace = true
+reth-interfaces.workspace = true
 

crates/evm/src/execute.rs

@@ -0,0 +1,165 @@
+//! Traits for execution.
+
+use reth_interfaces::provider::ProviderError;
+use reth_primitives::U256;
+use revm::db::BundleState;
+use revm_primitives::db::Database;
+
+/// A general purpose executor trait that executes on an input (e.g. blocks) and produces an output
+/// (e.g. state changes and receipts).
+pub trait Executor<DB> {
+    /// The input type for the executor.
+    type Input<'a>;
+    /// The output type for the executor.
+    type Output;
+    /// The error type returned by the executor.
+    type Error;
+
+    /// Consumes the type and executes the block.
+    ///
+    /// Returns the output of the block execution.
+    fn execute(self, input: Self::Input<'_>) -> Result<Self::Output, Self::Error>;
+}
+
+/// An executor that can execute multiple blocks in a row and keep track of the state over the
+/// entire batch.
+pub trait BatchExecutor<DB> {
+    /// The input type for the executor.
+    type Input<'a>;
+    /// The output type for the executor.
+    type Output;
+    /// The error type returned by the executor.
+    type Error;
+
+    /// Executes the next block in the batch and update the state internally.
+    fn execute_one(&mut self, input: Self::Input<'_>) -> Result<BatchBlockOutput, Self::Error>;
+
+    /// Finishes the batch and return the final state.
+    fn finalize(self) -> Self::Output;
+}
+
+/// The output of an executed block in a batch.
+#[derive(Debug, Clone, Copy)]
+pub struct BatchBlockOutput {
+    /// The size hint of the batch's tracked state.
+    pub size_hint: Option<usize>,
+}
+
+/// The output of an ethereum block.
+///
+/// Contains the state changes, transaction receipts, and total gas used in the block.
+///
+/// TODO(mattsse): combine with BundleStateWithReceipts
+#[derive(Debug)]
+pub struct EthBlockOutput<T> {
+    /// The changed state of the block after execution.
+    pub state: BundleState,
+    /// All the receipts of the transactions in the block.
+    pub receipts: Vec<T>,
+    /// The total gas used by the block.
+    pub gas_used: u64,
+}
+
+/// A helper type for ethereum block inputs that consists of a block and the total difficulty.
+#[derive(Debug)]
+pub struct EthBlockExecutionInput<'a, Block> {
+    /// The block to execute.
+    pub block: &'a Block,
+    /// The total difficulty of the block.
+    pub total_difficulty: U256,
+}
+
+impl<'a, Block> EthBlockExecutionInput<'a, Block> {
+    /// Creates a new input.
+    pub fn new(block: &'a Block, total_difficulty: U256) -> Self {
+        Self { block, total_difficulty }
+    }
+}
+
+impl<'a, Block> From<(&'a Block, U256)> for EthBlockExecutionInput<'a, Block> {
+    fn from((block, total_difficulty): (&'a Block, U256)) -> Self {
+        Self::new(block, total_difficulty)
+    }
+}
+
+/// A type that can create a new executor.
+pub trait ExecutorProvider: Send + Sync + Clone {
+    /// An executor that can execute a single block given a database.
+    type Executor<DB: Database<Error = ProviderError>>: Executor<DB>;
+    /// An executor that can execute a batch of blocks given a database.
+
+    type BatchExecutor<DB: Database<Error = ProviderError>>: BatchExecutor<DB>;
+    /// Creates a new executor for single block execution.
+    fn executor<DB>(&self, db: DB) -> Self::Executor<DB>
+    where
+        DB: Database<Error = ProviderError>;
+
+    /// Creates a new batch executor
+    fn batch_executor<DB>(&self, db: DB) -> Self::BatchExecutor<DB>
+    where
+        DB: Database<Error = ProviderError>;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use revm::db::{CacheDB, EmptyDBTyped};
+    use std::marker::PhantomData;
+
+    #[derive(Clone, Default)]
+    struct TestExecutorProvider;
+
+    impl ExecutorProvider for TestExecutorProvider {
+        type Executor<DB: Database<Error = ProviderError>> = TestExecutor<DB>;
+        type BatchExecutor<DB: Database<Error = ProviderError>> = TestExecutor<DB>;
+
+        fn executor<DB>(&self, _db: DB) -> Self::Executor<DB>
+        where
+            DB: Database<Error = ProviderError>,
+        {
+            TestExecutor(PhantomData)
+        }
+
+        fn batch_executor<DB>(&self, _db: DB) -> Self::BatchExecutor<DB>
+        where
+            DB: Database<Error = ProviderError>,
+        {
+            TestExecutor(PhantomData)
+        }
+    }
+
+    struct TestExecutor<DB>(PhantomData<DB>);
+
+    impl<DB> Executor<DB> for TestExecutor<DB> {
+        type Input<'a> = &'static str;
+        type Output = ();
+        type Error = String;
+
+        fn execute(self, _input: Self::Input<'_>) -> Result<Self::Output, Self::Error> {
+            Ok(())
+        }
+    }
+
+    impl<DB> BatchExecutor<DB> for TestExecutor<DB> {
+        type Input<'a> = &'static str;
+        type Output = ();
+        type Error = String;
+
+        fn execute_one(
+            &mut self,
+            _input: Self::Input<'_>,
+        ) -> Result<BatchBlockOutput, Self::Error> {
+            Ok(BatchBlockOutput { size_hint: None })
+        }
+
+        fn finalize(self) -> Self::Output {}
+    }
+
+    #[test]
+    fn test_provider() {
+        let provider = TestExecutorProvider;
+        let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
+        let executor = provider.executor(db);
+        executor.execute("test").unwrap();
+    }
+}

crates/evm/src/lib.rs

@@ -12,6 +12,8 @@ use reth_primitives::{revm::env::fill_block_env, Address, ChainSpec, Header, Tra
 use revm::{inspector_handle_register, Database, Evm, EvmBuilder, GetInspector};
 use revm_primitives::{BlockEnv, CfgEnvWithHandlerCfg, EnvWithHandlerCfg, SpecId, TxEnv};
 
+pub mod execute;
+
 /// Trait for configuring the EVM for executing full blocks.
 pub trait ConfigureEvm: ConfigureEvmEnv {
     /// Returns new EVM with the given database

crates/node-optimism/Cargo.toml

@@ -25,8 +25,12 @@ reth-tracing.workspace = true
 reth-provider.workspace = true
 reth-transaction-pool.workspace = true
 reth-network.workspace = true
+reth-interfaces.workspace = true
+reth-evm.workspace = true
 reth-revm.workspace = true
+
 revm.workspace = true
+revm-primitives.workspace = true
 
 # async
 async-trait.workspace = true
@@ -36,6 +40,7 @@ http-body = "0.4.5"
 reqwest = { version = "0.11", default-features = false, features = [
     "rustls-tls",
 ]}
+tracing.workspace = true
 
 # misc
 clap.workspace = true
@@ -48,6 +53,7 @@ jsonrpsee.workspace = true
 
 [dev-dependencies]
 reth-db.workspace = true
+reth-revm = { workspace = true, features = ["test-utils"]}
 
 [features]
 optimism = [

crates/node-optimism/src/evm/execute.rs

@@ -0,0 +1,745 @@
+//! Optimism block executor.
+
+use crate::OptimismEvmConfig;
+use reth_evm::{
+    execute::{
+        BatchBlockOutput, BatchExecutor, EthBlockExecutionInput, EthBlockOutput, Executor,
+        ExecutorProvider,
+    },
+    ConfigureEvm, ConfigureEvmEnv,
+};
+use reth_interfaces::{
+    executor::{BlockExecutionError, BlockValidationError, OptimismBlockExecutionError},
+    provider::ProviderError,
+};
+use reth_primitives::{
+    proofs::calculate_receipt_root_optimism, BlockWithSenders, Bloom, Bytes, ChainSpec,
+    GotExpected, Hardfork, Header, PruneModes, Receipt, ReceiptWithBloom, Receipts, TxType,
+    Withdrawals, B256, U256,
+};
+use reth_provider::BundleStateWithReceipts;
+use reth_revm::{
+    batch::{BlockBatchRecord, BlockExecutorStats},
+    db::states::bundle_state::BundleRetention,
+    optimism::ensure_create2_deployer,
+    processor::compare_receipts_root_and_logs_bloom,
+    stack::InspectorStack,
+    state_change::{apply_beacon_root_contract_call, post_block_balance_increments},
+    Evm, State,
+};
+use revm_primitives::{
+    db::{Database, DatabaseCommit},
+    BlockEnv, CfgEnvWithHandlerCfg, EnvWithHandlerCfg, ResultAndState,
+};
+use std::sync::Arc;
+use tracing::{debug, trace};
+
+/// Provides executors to execute regular ethereum blocks
+#[derive(Debug, Clone)]
+pub struct OpExecutorProvider<EvmConfig> {
+    chain_spec: Arc<ChainSpec>,
+    evm_config: EvmConfig,
+    inspector: Option<InspectorStack>,
+    prune_modes: PruneModes,
+}
+
+impl OpExecutorProvider<OptimismEvmConfig> {
+    /// Creates a new default optimism executor provider.
+    pub fn optimism(chain_spec: Arc<ChainSpec>) -> Self {
+        Self::new(chain_spec, Default::default())
+    }
+}
+
+impl<EvmConfig> OpExecutorProvider<EvmConfig> {
+    /// Creates a new executor provider.
+    pub fn new(chain_spec: Arc<ChainSpec>, evm_config: EvmConfig) -> Self {
+        Self { chain_spec, evm_config, inspector: None, prune_modes: PruneModes::none() }
+    }
+
+    /// Configures an optional inspector stack for debugging.
+    pub fn with_inspector(mut self, inspector: Option<InspectorStack>) -> Self {
+        self.inspector = inspector;
+        self
+    }
+
+    /// Configures the prune modes for the executor.
+    pub fn with_prune_modes(mut self, prune_modes: PruneModes) -> Self {
+        self.prune_modes = prune_modes;
+        self
+    }
+}
+
+impl<EvmConfig> OpExecutorProvider<EvmConfig>
+where
+    EvmConfig: ConfigureEvm,
+    EvmConfig: ConfigureEvmEnv<TxMeta = Bytes>,
+{
+    fn op_executor<DB>(&self, db: DB) -> OpBlockExecutor<EvmConfig, DB>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        OpBlockExecutor::new(
+            self.chain_spec.clone(),
+            self.evm_config.clone(),
+            State::builder().with_database(db).with_bundle_update().without_state_clear().build(),
+        )
+        .with_inspector(self.inspector.clone())
+    }
+}
+
+impl<EvmConfig> ExecutorProvider for OpExecutorProvider<EvmConfig>
+where
+    EvmConfig: ConfigureEvm,
+    EvmConfig: ConfigureEvmEnv<TxMeta = Bytes>,
+{
+    type Executor<DB: Database<Error = ProviderError>> = OpBlockExecutor<EvmConfig, DB>;
+
+    type BatchExecutor<DB: Database<Error = ProviderError>> = OpBatchExecutor<EvmConfig, DB>;
+    fn executor<DB>(&self, db: DB) -> Self::Executor<DB>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        self.op_executor(db)
+    }
+
+    fn batch_executor<DB>(&self, db: DB) -> Self::BatchExecutor<DB>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        let executor = self.op_executor(db);
+        OpBatchExecutor {
+            executor,
+            batch_record: BlockBatchRecord::new(self.prune_modes.clone()),
+            stats: BlockExecutorStats::default(),
+        }
+    }
+}
+
+/// Helper container type for EVM with chain spec.
+#[derive(Debug, Clone)]
+struct OpEvmExecutor<EvmConfig> {
+    /// The chainspec
+    chain_spec: Arc<ChainSpec>,
+    /// How to create an EVM.
+    evm_config: EvmConfig,
+}
+
+impl<EvmConfig> OpEvmExecutor<EvmConfig>
+where
+    EvmConfig: ConfigureEvm,
+    EvmConfig: ConfigureEvmEnv<TxMeta = Bytes>,
+{
+    /// Executes the transactions in the block and returns the receipts.
+    ///
+    /// This applies the pre-execution changes, and executes the transactions.
+    ///
+    /// # Note
+    ///
+    /// It does __not__ apply post-execution changes.
+    fn execute_pre_and_transactions<Ext, DB>(
+        &mut self,
+        block: &BlockWithSenders,
+        mut evm: Evm<'_, Ext, &mut State<DB>>,
+    ) -> Result<(Vec<Receipt>, u64), BlockExecutionError>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        // apply pre execution changes
+        apply_beacon_root_contract_call(
+            &self.chain_spec,
+            block.timestamp,
+            block.number,
+            block.parent_beacon_block_root,
+            &mut evm,
+        )?;
+
+        // execute transactions
+        let is_regolith =
+            self.chain_spec.fork(Hardfork::Regolith).active_at_timestamp(block.timestamp);
+
+        // Ensure that the create2deployer is force-deployed at the canyon transition. Optimism
+        // blocks will always have at least a single transaction in them (the L1 info transaction),
+        // so we can safely assume that this will always be triggered upon the transition and that
+        // the above check for empty blocks will never be hit on OP chains.
+        ensure_create2_deployer(self.chain_spec.clone(), block.timestamp, evm.db_mut()).map_err(
+            |_| {
+                BlockExecutionError::OptimismBlockExecution(
+                    OptimismBlockExecutionError::ForceCreate2DeployerFail,
+                )
+            },
+        )?;
+
+        let mut cumulative_gas_used = 0;
+        let mut receipts = Vec::with_capacity(block.body.len());
+        for (sender, transaction) in block.transactions_with_sender() {
+            // The sum of the transaction’s gas limit, Tg, and the gas utilized in this block prior,
+            // must be no greater than the block’s gasLimit.
+            let block_available_gas = block.header.gas_limit - cumulative_gas_used;
+            if transaction.gas_limit() > block_available_gas &&
+                (is_regolith || !transaction.is_system_transaction())
+            {
+                return Err(BlockValidationError::TransactionGasLimitMoreThanAvailableBlockGas {
+                    transaction_gas_limit: transaction.gas_limit(),
+                    block_available_gas,
+                }
+                .into())
+            }
+
+            // An optimism block should never contain blob transactions.
+            if matches!(transaction.tx_type(), TxType::Eip4844) {
+                return Err(BlockExecutionError::OptimismBlockExecution(
+                    OptimismBlockExecutionError::BlobTransactionRejected,
+                ))
+            }
+
+            // Cache the depositor account prior to the state transition for the deposit nonce.
+            //
+            // Note that this *only* needs to be done post-regolith hardfork, as deposit nonces
+            // were not introduced in Bedrock. In addition, regular transactions don't have deposit
+            // nonces, so we don't need to touch the DB for those.
+            let depositor = (is_regolith && transaction.is_deposit())
+                .then(|| {
+                    evm.db_mut()
+                        .load_cache_account(*sender)
+                        .map(|acc| acc.account_info().unwrap_or_default())
+                })
+                .transpose()
+                .map_err(|_| {
+                    BlockExecutionError::OptimismBlockExecution(
+                        OptimismBlockExecutionError::AccountLoadFailed(*sender),
+                    )
+                })?;
+
+            let mut buf = Vec::with_capacity(transaction.length_without_header());
+            transaction.encode_enveloped(&mut buf);
+            EvmConfig::fill_tx_env(evm.tx_mut(), transaction, *sender, buf.into());
+
+            // Execute transaction.
+            let ResultAndState { result, state } = evm.transact().map_err(move |err| {
+                // Ensure hash is calculated for error log, if not already done
+                BlockValidationError::EVM {
+                    hash: transaction.recalculate_hash(),
+                    error: err.into(),
+                }
+            })?;
+
+            trace!(
+                target: "evm",
+                ?transaction,
+                "Executed transaction"
+            );
+
+            evm.db_mut().commit(state);
+
+            // append gas used
+            cumulative_gas_used += result.gas_used();
+
+            // Push transaction changeset and calculate header bloom filter for receipt.
+            receipts.push(Receipt {
+                tx_type: transaction.tx_type(),
+                // Success flag was added in `EIP-658: Embedding transaction status code in
+                // receipts`.
+                success: result.is_success(),
+                cumulative_gas_used,
+                logs: result.into_logs(),
+                deposit_nonce: depositor.map(|account| account.nonce),
+                // The deposit receipt version was introduced in Canyon to indicate an update to how
+                // receipt hashes should be computed when set. The state transition process ensures
+                // this is only set for post-Canyon deposit transactions.
+                deposit_receipt_version: (transaction.is_deposit() &&
+                    self.chain_spec
+                        .is_fork_active_at_timestamp(Hardfork::Canyon, block.timestamp))
+                .then_some(1),
+            });
+        }
+        drop(evm);
+
+        // Check if gas used matches the value set in header.
+        if block.gas_used != cumulative_gas_used {
+            let receipts = Receipts::from_block_receipt(receipts);
+            return Err(BlockValidationError::BlockGasUsed {
+                gas: GotExpected { got: cumulative_gas_used, expected: block.gas_used },
+                gas_spent_by_tx: receipts.gas_spent_by_tx()?,
+            }
+            .into())
+        }
+
+        Ok((receipts, cumulative_gas_used))
+    }
+}
+
+/// A basic Ethereum block executor.
+///
+/// Expected usage:
+/// - Create a new instance of the executor.
+/// - Execute the block.
+#[derive(Debug)]
+pub struct OpBlockExecutor<EvmConfig, DB> {
+    /// Chain specific evm config that's used to execute a block.
+    executor: OpEvmExecutor<EvmConfig>,
+    /// The state to use for execution
+    state: State<DB>,
+    /// Optional inspector stack for debugging
+    inspector: Option<InspectorStack>,
+}
+
+impl<EvmConfig, DB> OpBlockExecutor<EvmConfig, DB> {
+    /// Creates a new Ethereum block executor.
+    pub fn new(chain_spec: Arc<ChainSpec>, evm_config: EvmConfig, state: State<DB>) -> Self {
+        Self { executor: OpEvmExecutor { chain_spec, evm_config }, state, inspector: None }
+    }
+
+    /// Sets the inspector stack for debugging.
+    pub fn with_inspector(mut self, inspector: Option<InspectorStack>) -> Self {
+        self.inspector = inspector;
+        self
+    }
+
+    #[inline]
+    fn chain_spec(&self) -> &ChainSpec {
+        &self.executor.chain_spec
+    }
+
+    /// Returns mutable reference to the state that wraps the underlying database.
+    #[allow(unused)]
+    fn state_mut(&mut self) -> &mut State<DB> {
+        &mut self.state
+    }
+}
+
+impl<EvmConfig, DB> OpBlockExecutor<EvmConfig, DB>
+where
+    EvmConfig: ConfigureEvm,
+    // TODO(mattsse): get rid of this
+    EvmConfig: ConfigureEvmEnv<TxMeta = Bytes>,
+    DB: Database<Error = ProviderError>,
+{
+    /// Configures a new evm configuration and block environment for the given block.
+    ///
+    /// Caution: this does not initialize the tx environment.
+    fn evm_env_for_block(&self, header: &Header, total_difficulty: U256) -> EnvWithHandlerCfg {
+        let mut cfg = CfgEnvWithHandlerCfg::new(Default::default(), Default::default());
+        let mut block_env = BlockEnv::default();
+        EvmConfig::fill_cfg_and_block_env(
+            &mut cfg,
+            &mut block_env,
+            self.chain_spec(),
+            header,
+            total_difficulty,
+        );
+
+        EnvWithHandlerCfg::new_with_cfg_env(cfg, block_env, Default::default())
+    }
+
+    /// Execute a single block and apply the state changes to the internal state.
+    ///
+    /// Returns the receipts of the transactions in the block and the total gas used.
+    ///
+    /// Returns an error if execution fails or receipt verification fails.
+    fn execute_and_verify(
+        &mut self,
+        block: &BlockWithSenders,
+        total_difficulty: U256,
+    ) -> Result<(Vec<Receipt>, u64), BlockExecutionError> {
+        // 1. prepare state on new block
+        self.on_new_block(&block.header);
+
+        // 2. configure the evm and execute
+        let env = self.evm_env_for_block(&block.header, total_difficulty);
+
+        let (receipts, gas_used) = {
+            if let Some(inspector) = self.inspector.as_mut() {
+                let evm = self.executor.evm_config.evm_with_env_and_inspector(
+                    &mut self.state,
+                    env,
+                    inspector,
+                );
+                self.executor.execute_pre_and_transactions(block, evm)?
+            } else {
+                let evm = self.executor.evm_config.evm_with_env(&mut self.state, env);
+
+                self.executor.execute_pre_and_transactions(block, evm)?
+            }
+        };
+
+        // 3. apply post execution changes
+        self.post_execution(block, total_difficulty)?;
+
+        // Before Byzantium, receipts contained state root that would mean that expensive
+        // operation as hashing that is required for state root got calculated in every
+        // transaction This was replaced with is_success flag.
+        // See more about EIP here: https://eips.ethereum.org/EIPS/eip-658
+        if self.chain_spec().is_byzantium_active_at_block(block.header.number) {
+            if let Err(error) = verify_receipt_optimism(
+                block.header.receipts_root,
+                block.header.logs_bloom,
+                receipts.iter(),
+                self.chain_spec(),
+                block.timestamp,
+            ) {
+                debug!(target: "evm", %error, ?receipts, "receipts verification failed");
+                return Err(error)
+            };
+        }
+
+        Ok((receipts, gas_used))
+    }
+
+    /// Apply settings before a new block is executed.
+    pub(crate) fn on_new_block(&mut self, header: &Header) {
+        // Set state clear flag if the block is after the Spurious Dragon hardfork.
+        let state_clear_flag = self.chain_spec().is_spurious_dragon_active_at_block(header.number);
+        self.state.set_state_clear_flag(state_clear_flag);
+    }
+
+    /// Apply post execution state changes, including block rewards, withdrawals, and irregular DAO
+    /// hardfork state change.
+    pub fn post_execution(
+        &mut self,
+        block: &BlockWithSenders,
+        total_difficulty: U256,
+    ) -> Result<(), BlockExecutionError> {
+        let balance_increments = post_block_balance_increments(
+            self.chain_spec(),
+            block.number,
+            block.difficulty,
+            block.beneficiary,
+            block.timestamp,
+            total_difficulty,
+            &block.ommers,
+            block.withdrawals.as_ref().map(Withdrawals::as_ref),
+        );
+        // increment balances
+        self.state
+            .increment_balances(balance_increments)
+            .map_err(|_| BlockValidationError::IncrementBalanceFailed)?;
+
+        Ok(())
+    }
+}
+
+impl<EvmConfig, DB> Executor<DB> for OpBlockExecutor<EvmConfig, DB>
+where
+    EvmConfig: ConfigureEvm,
+    EvmConfig: ConfigureEvmEnv<TxMeta = Bytes>,
+    DB: Database<Error = ProviderError>,
+{
+    type Input<'a> = EthBlockExecutionInput<'a, BlockWithSenders>;
+    type Output = EthBlockOutput<Receipt>;
+    type Error = BlockExecutionError;
+
+    /// Executes the block and commits the state changes.
+    ///
+    /// Returns the receipts of the transactions in the block.
+    ///
+    /// Returns an error if the block could not be executed or failed verification.
+    ///
+    /// State changes are committed to the database.
+    fn execute(mut self, input: Self::Input<'_>) -> Result<Self::Output, Self::Error> {
+        let EthBlockExecutionInput { block, total_difficulty } = input;
+        let (receipts, gas_used) = self.execute_and_verify(block, total_difficulty)?;
+
+        // prepare the state for extraction
+        self.state.merge_transitions(BundleRetention::PlainState);
+
+        Ok(EthBlockOutput { state: self.state.take_bundle(), receipts, gas_used })
+    }
+}
+
+/// An executor for a batch of blocks.
+///
+/// State changes are tracked until the executor is finalized.
+#[derive(Debug)]
+pub struct OpBatchExecutor<EvmConfig, DB> {
+    /// The executor used to execute blocks.
+    executor: OpBlockExecutor<EvmConfig, DB>,
+    /// Keeps track of the batch and record receipts based on the configured prune mode
+    batch_record: BlockBatchRecord,
+    stats: BlockExecutorStats,
+}
+
+impl<EvmConfig, DB> OpBatchExecutor<EvmConfig, DB> {
+    /// Returns the receipts of the executed blocks.
+    pub fn receipts(&self) -> &Receipts {
+        self.batch_record.receipts()
+    }
+
+    /// Returns mutable reference to the state that wraps the underlying database.
+    #[allow(unused)]
+    fn state_mut(&mut self) -> &mut State<DB> {
+        self.executor.state_mut()
+    }
+}
+
+impl<EvmConfig, DB> BatchExecutor<DB> for OpBatchExecutor<EvmConfig, DB>
+where
+    EvmConfig: ConfigureEvm,
+    // TODO: get rid of this
+    EvmConfig: ConfigureEvmEnv<TxMeta = Bytes>,
+    DB: Database<Error = ProviderError>,
+{
+    type Input<'a> = EthBlockExecutionInput<'a, BlockWithSenders>;
+    type Output = BundleStateWithReceipts;
+    type Error = BlockExecutionError;
+
+    fn execute_one(&mut self, input: Self::Input<'_>) -> Result<BatchBlockOutput, Self::Error> {
+        let EthBlockExecutionInput { block, total_difficulty } = input;
+        let (receipts, _gas_used) = self.executor.execute_and_verify(block, total_difficulty)?;
+
+        // prepare the state according to the prune mode
+        let retention = self.batch_record.bundle_retention(block.number);
+        self.executor.state.merge_transitions(retention);
+
+        // store receipts in the set
+        self.batch_record.save_receipts(receipts)?;
+
+        Ok(BatchBlockOutput { size_hint: Some(self.executor.state.bundle_size_hint()) })
+    }
+
+    fn finalize(mut self) -> Self::Output {
+        // TODO: track stats
+        self.stats.log_debug();
+
+        BundleStateWithReceipts::new(
+            self.executor.state.take_bundle(),
+            self.batch_record.take_receipts(),
+            self.batch_record.first_block().unwrap_or_default(),
+        )
+    }
+}
+
+/// Verify the calculated receipts root against the expected receipts root.
+pub fn verify_receipt_optimism<'a>(
+    expected_receipts_root: B256,
+    expected_logs_bloom: Bloom,
+    receipts: impl Iterator<Item = &'a Receipt> + Clone,
+    chain_spec: &ChainSpec,
+    timestamp: u64,
+) -> Result<(), BlockExecutionError> {
+    // Calculate receipts root.
+    let receipts_with_bloom = receipts.map(|r| r.clone().into()).collect::<Vec<ReceiptWithBloom>>();
+    let receipts_root =
+        calculate_receipt_root_optimism(&receipts_with_bloom, chain_spec, timestamp);
+
+    // Create header log bloom.
+    let logs_bloom = receipts_with_bloom.iter().fold(Bloom::ZERO, |bloom, r| bloom | r.bloom);
+
+    compare_receipts_root_and_logs_bloom(
+        receipts_root,
+        logs_bloom,
+        expected_receipts_root,
+        expected_logs_bloom,
+    )?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use reth_primitives::{
+        b256, Account, Address, Block, ChainSpecBuilder, Signature, StorageKey, StorageValue,
+        Transaction, TransactionKind, TransactionSigned, TxEip1559, BASE_MAINNET,
+    };
+    use reth_revm::database::StateProviderDatabase;
+    use revm::L1_BLOCK_CONTRACT;
+    use std::{collections::HashMap, str::FromStr};
+
+    use crate::OptimismEvmConfig;
+    use reth_revm::test_utils::StateProviderTest;
+
+    fn create_op_state_provider() -> StateProviderTest {
+        let mut db = StateProviderTest::default();
+
+        let l1_block_contract_account =
+            Account { balance: U256::ZERO, bytecode_hash: None, nonce: 1 };
+
+        let mut l1_block_storage = HashMap::new();
+        // base fee
+        l1_block_storage.insert(StorageKey::with_last_byte(1), StorageValue::from(1000000000));
+        // l1 fee overhead
+        l1_block_storage.insert(StorageKey::with_last_byte(5), StorageValue::from(188));
+        // l1 fee scalar
+        l1_block_storage.insert(StorageKey::with_last_byte(6), StorageValue::from(684000));
+        // l1 free scalars post ecotone
+        l1_block_storage.insert(
+            StorageKey::with_last_byte(3),
+            StorageValue::from_str(
+                "0x0000000000000000000000000000000000001db0000d27300000000000000005",
+            )
+            .unwrap(),
+        );
+
+        db.insert_account(L1_BLOCK_CONTRACT, l1_block_contract_account, None, l1_block_storage);
+
+        db
+    }
+
+    fn executor_provider(chain_spec: Arc<ChainSpec>) -> OpExecutorProvider<OptimismEvmConfig> {
+        OpExecutorProvider {
+            chain_spec,
+            evm_config: Default::default(),
+            inspector: None,
+            prune_modes: Default::default(),
+        }
+    }
+
+    #[test]
+    fn op_deposit_fields_pre_canyon() {
+        let header = Header {
+            timestamp: 1,
+            number: 1,
+            gas_limit: 1_000_000,
+            gas_used: 42_000,
+            receipts_root: b256!(
+                "83465d1e7d01578c0d609be33570f91242f013e9e295b0879905346abbd63731"
+            ),
+            ..Default::default()
+        };
+
+        let mut db = create_op_state_provider();
+
+        let addr = Address::ZERO;
+        let account = Account { balance: U256::MAX, ..Account::default() };
+        db.insert_account(addr, account, None, HashMap::new());
+
+        let chain_spec =
+            Arc::new(ChainSpecBuilder::from(&*BASE_MAINNET).regolith_activated().build());
+
+        let tx = TransactionSigned::from_transaction_and_signature(
+            Transaction::Eip1559(TxEip1559 {
+                chain_id: chain_spec.chain.id(),
+                nonce: 0,
+                gas_limit: 21_000,
+                to: TransactionKind::Call(addr),
+                ..Default::default()
+            }),
+            Signature::default(),
+        );
+
+        let tx_deposit = TransactionSigned::from_transaction_and_signature(
+            Transaction::Deposit(reth_primitives::TxDeposit {
+                from: addr,
+                to: TransactionKind::Call(addr),
+                gas_limit: 21_000,
+                ..Default::default()
+            }),
+            Signature::default(),
+        );
+
+        let provider = executor_provider(chain_spec);
+        let mut executor = provider.batch_executor(StateProviderDatabase::new(&db));
+
+        executor.state_mut().load_cache_account(L1_BLOCK_CONTRACT).unwrap();
+
+        // Attempt to execute a block with one deposit and one non-deposit transaction
+        executor
+            .execute_one(
+                (
+                    &BlockWithSenders {
+                        block: Block {
+                            header,
+                            body: vec![tx, tx_deposit],
+                            ommers: vec![],
+                            withdrawals: None,
+                        },
+                        senders: vec![addr, addr],
+                    },
+                    U256::ZERO,
+                )
+                    .into(),
+            )
+            .unwrap();
+
+        let tx_receipt = executor.receipts()[0][0].as_ref().unwrap();
+        let deposit_receipt = executor.receipts()[0][1].as_ref().unwrap();
+
+        // deposit_receipt_version is not present in pre canyon transactions
+        assert!(deposit_receipt.deposit_receipt_version.is_none());
+        assert!(tx_receipt.deposit_receipt_version.is_none());
+
+        // deposit_nonce is present only in deposit transactions
+        assert!(deposit_receipt.deposit_nonce.is_some());
+        assert!(tx_receipt.deposit_nonce.is_none());
+    }
+
+    #[test]
+    fn op_deposit_fields_post_canyon() {
+        // ensure_create2_deployer will fail if timestamp is set to less then 2
+        let header = Header {
+            timestamp: 2,
+            number: 1,
+            gas_limit: 1_000_000,
+            gas_used: 42_000,
+            receipts_root: b256!(
+                "fffc85c4004fd03c7bfbe5491fae98a7473126c099ac11e8286fd0013f15f908"
+            ),
+            ..Default::default()
+        };
+
+        let mut db = create_op_state_provider();
+        let addr = Address::ZERO;
+        let account = Account { balance: U256::MAX, ..Account::default() };
+
+        db.insert_account(addr, account, None, HashMap::new());
+
+        let chain_spec =
+            Arc::new(ChainSpecBuilder::from(&*BASE_MAINNET).canyon_activated().build());
+
+        let tx = TransactionSigned::from_transaction_and_signature(
+            Transaction::Eip1559(TxEip1559 {
+                chain_id: chain_spec.chain.id(),
+                nonce: 0,
+                gas_limit: 21_000,
+                to: TransactionKind::Call(addr),
+                ..Default::default()
+            }),
+            Signature::default(),
+        );
+
+        let tx_deposit = TransactionSigned::from_transaction_and_signature(
+            Transaction::Deposit(reth_primitives::TxDeposit {
+                from: addr,
+                to: TransactionKind::Call(addr),
+                gas_limit: 21_000,
+                ..Default::default()
+            }),
+            Signature::optimism_deposit_tx_signature(),
+        );
+
+        let provider = executor_provider(chain_spec);
+        let mut executor = provider.batch_executor(StateProviderDatabase::new(&db));
+
+        executor.state_mut().load_cache_account(L1_BLOCK_CONTRACT).unwrap();
+
+        // attempt to execute an empty block with parent beacon block root, this should not fail
+        executor
+            .execute_one(
+                (
+                    &BlockWithSenders {
+                        block: Block {
+                            header,
+                            body: vec![tx, tx_deposit],
+                            ommers: vec![],
+                            withdrawals: None,
+                        },
+                        senders: vec![addr, addr],
+                    },
+                    U256::ZERO,
+                )
+                    .into(),
+            )
+            .expect("Executing a block while canyon is active should not fail");
+
+        let tx_receipt = executor.receipts()[0][0].as_ref().unwrap();
+        let deposit_receipt = executor.receipts()[0][1].as_ref().unwrap();
+
+        // deposit_receipt_version is set to 1 for post canyon deposit transactions
+        assert_eq!(deposit_receipt.deposit_receipt_version, Some(1));
+        assert!(tx_receipt.deposit_receipt_version.is_none());
+
+        // deposit_nonce is present only in deposit transactions
+        assert!(deposit_receipt.deposit_nonce.is_some());
+        assert!(tx_receipt.deposit_nonce.is_none());
+    }
+}

crates/node-optimism/src/evm/mod.rs

@@ -6,6 +6,9 @@ use reth_primitives::{
 };
 use revm::{inspector_handle_register, Database, Evm, EvmBuilder, GetInspector};
 
+mod execute;
+pub use execute::*;
+
 /// Optimism-related EVM configuration.
 #[derive(Debug, Default, Clone, Copy)]
 #[non_exhaustive]

crates/node-optimism/src/rpc.rs

@@ -6,7 +6,6 @@ use reth_rpc::eth::{
     error::{EthApiError, EthResult, ToRpcError},
     traits::RawTransactionForwarder,
 };
-use reth_tracing::tracing;
 use std::sync::{atomic::AtomicUsize, Arc};
 
 /// Error type when interacting with the Sequencer

crates/revm/Cargo.toml

@@ -18,6 +18,7 @@ reth-interfaces.workspace = true
 reth-provider.workspace = true
 reth-consensus-common.workspace = true
 reth-evm.workspace = true
+reth-trie = { workspace = true, optional = true }
 
 # revm
 revm.workspace = true
@@ -30,6 +31,7 @@ tracing.workspace = true
 reth-trie.workspace = true
 
 [features]
+test-utils = ["dep:reth-trie"]
 optimism = [
     "revm/optimism",
     "reth-primitives/optimism",

crates/revm/src/lib.rs

@@ -38,7 +38,7 @@ pub mod stack;
 pub mod optimism;
 
 /// Common test helpers
-#[cfg(test)]
+#[cfg(any(test, feature = "test-utils"))]
 pub mod test_utils;
 
 // Convenience re-exports.

crates/revm/src/processor.rs

@@ -460,20 +460,16 @@ pub fn compare_receipts_root_and_logs_bloom(
 
 #[cfg(test)]
 mod tests {
-    use std::collections::HashMap;
+    use super::*;
-
+    use crate::test_utils::{StateProviderTest, TestEvmConfig};
-    use revm::{Database, TransitionState};
-
     use reth_primitives::{
         bytes,
         constants::{BEACON_ROOTS_ADDRESS, EIP1559_INITIAL_BASE_FEE, SYSTEM_ADDRESS},
         keccak256, Account, Bytes, ChainSpecBuilder, ForkCondition, Signature, Transaction,
         TransactionKind, TxEip1559, MAINNET,
     };
-
+    use revm::{Database, TransitionState};
-    use crate::test_utils::{StateProviderTest, TestEvmConfig};
+    use std::collections::HashMap;
-
-    use super::*;
 
     static BEACON_ROOT_CONTRACT_CODE: Bytes = bytes!("3373fffffffffffffffffffffffffffffffffffffffe14604d57602036146024575f5ffd5b5f35801560495762001fff810690815414603c575f5ffd5b62001fff01545f5260205ff35b5f5ffd5b62001fff42064281555f359062001fff015500");
 

crates/revm/src/test_utils.rs

@@ -25,6 +25,7 @@ use {
     },
 };
 
+/// Mock state for testing
 #[derive(Debug, Default, Clone, Eq, PartialEq)]
 pub struct StateProviderTest {
     accounts: HashMap<Address, (HashMap<StorageKey, U256>, Account)>,