chore(execution): Refactor unused executor (#1096)

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

crates/executor/src/eth_dao_fork.rs

@@ -3,10 +3,10 @@
 use reth_primitives::{hex_literal::hex, H160};
 
 /// Dao hardfork beneficiary that received ether from accounts from DAO and DAO creator children.
-pub const DAO_HARDFORK_BENEFICIARY: H160 = H160(hex!("bf4ed7b27f1d666546e30d74d50d173d20bca754"));
+pub static DAO_HARDFORK_BENEFICIARY: H160 = H160(hex!("bf4ed7b27f1d666546e30d74d50d173d20bca754"));
 
 /// DAO hardfork account that ether was taken and added to beneficiry
-pub const DAO_HARDKFORK_ACCOUNTS: [H160; 116] = [
+pub static DAO_HARDKFORK_ACCOUNTS: [H160; 116] = [
     H160(hex!("d4fe7bc31cedb7bfb8a345f31e668033056b2728")),
     H160(hex!("b3fb0e5aba0e20e5c49d252dfd30e102b171a425")),
     H160(hex!("2c19c7f9ae8b751e37aeb2d93a699722395ae18f")),

crates/executor/src/execution_result.rs

@@ -0,0 +1,171 @@
+use reth_db::{models::AccountBeforeTx, tables, transaction::DbTxMut, Error as DbError};
+use reth_primitives::{Account, Address, Receipt, H256, U256};
+use revm::Bytecode;
+use std::collections::BTreeMap;
+
+/// Execution Result containing vector of transaction changesets
+/// and block reward if present
+#[derive(Debug)]
+pub struct ExecutionResult {
+    /// Transaction changeset containing [Receipt], changed [Accounts][Account] and Storages.
+    pub changesets: Vec<TransactionChangeSet>,
+    /// Block reward if present. It represent changeset for block reward slot in
+    /// [tables::AccountChangeSet] .
+    pub block_reward: Option<BTreeMap<Address, AccountInfoChangeSet>>,
+}
+
+/// After transaction is executed this structure contain
+/// transaction [Receipt] every change to state ([Account], Storage, [Bytecode])
+/// that this transaction made and its old values
+/// so that history account table can be updated.
+#[derive(Debug, Clone)]
+pub struct TransactionChangeSet {
+    /// Transaction receipt
+    pub receipt: Receipt,
+    /// State change that this transaction made on state.
+    pub changeset: BTreeMap<Address, AccountChangeSet>,
+    /// new bytecode created as result of transaction execution.
+    pub new_bytecodes: BTreeMap<H256, Bytecode>,
+}
+
+/// Contains old/new account changes
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub enum AccountInfoChangeSet {
+    /// The account is newly created.
+    Created {
+        /// The newly created account.
+        new: Account,
+    },
+    /// An account was deleted (selfdestructed) or we have touched
+    /// an empty account and we need to remove/destroy it.
+    /// (Look at state clearing [EIP-158](https://eips.ethereum.org/EIPS/eip-158))
+    Destroyed {
+        /// The account that was destroyed.
+        old: Account,
+    },
+    /// The account was changed.
+    Changed {
+        /// The account after the change.
+        new: Account,
+        /// The account prior to the change.
+        old: Account,
+    },
+    /// Nothing was changed for the account (nonce/balance).
+    NoChange,
+}
+
+impl AccountInfoChangeSet {
+    /// Apply the changes from the changeset to a database transaction.
+    pub fn apply_to_db<'a, TX: DbTxMut<'a>>(
+        self,
+        tx: &TX,
+        address: Address,
+        tx_index: u64,
+        has_state_clear_eip: bool,
+    ) -> Result<(), DbError> {
+        match self {
+            AccountInfoChangeSet::Changed { old, new } => {
+                // insert old account in AccountChangeSet
+                // check for old != new was already done
+                tx.put::<tables::AccountChangeSet>(
+                    tx_index,
+                    AccountBeforeTx { address, info: Some(old) },
+                )?;
+                tx.put::<tables::PlainAccountState>(address, new)?;
+            }
+            AccountInfoChangeSet::Created { new } => {
+                // Ignore account that are created empty and state clear (SpuriousDragon) hardfork
+                // is activated.
+                if has_state_clear_eip && new.is_empty() {
+                    return Ok(())
+                }
+                tx.put::<tables::AccountChangeSet>(
+                    tx_index,
+                    AccountBeforeTx { address, info: None },
+                )?;
+                tx.put::<tables::PlainAccountState>(address, new)?;
+            }
+            AccountInfoChangeSet::Destroyed { old } => {
+                tx.delete::<tables::PlainAccountState>(address, None)?;
+                tx.put::<tables::AccountChangeSet>(
+                    tx_index,
+                    AccountBeforeTx { address, info: Some(old) },
+                )?;
+            }
+            AccountInfoChangeSet::NoChange => {
+                // do nothing storage account didn't change
+            }
+        }
+        Ok(())
+    }
+}
+
+/// Diff change set that is needed for creating history index and updating current world state.
+#[derive(Debug, Clone)]
+pub struct AccountChangeSet {
+    /// Old and New account account change.
+    pub account: AccountInfoChangeSet,
+    /// Storage containing key -> (OldValue,NewValue). in case that old value is not existing
+    /// we can expect to have U256::ZERO, same with new value.
+    pub storage: BTreeMap<U256, (U256, U256)>,
+    /// Just to make sure that we are taking selfdestruct cleaning we have this field that wipes
+    /// storage. There are instances where storage is changed but account is not touched, so we
+    /// can't take into account that if new account is None that it is selfdestruct.
+    pub wipe_storage: bool,
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use reth_db::{
+        database::Database,
+        mdbx::{test_utils, Env, EnvKind, WriteMap},
+        transaction::DbTx,
+    };
+    use reth_primitives::H160;
+
+    use super::*;
+
+    #[test]
+    fn apply_account_info_changeset() {
+        let db: Arc<Env<WriteMap>> = test_utils::create_test_db(EnvKind::RW);
+        let address = H160::zero();
+        let tx_num = 0;
+        let acc1 = Account { balance: U256::from(1), nonce: 2, bytecode_hash: Some(H256::zero()) };
+        let acc2 = Account { balance: U256::from(3), nonce: 4, bytecode_hash: Some(H256::zero()) };
+
+        let tx = db.tx_mut().unwrap();
+
+        // check Changed changeset
+        AccountInfoChangeSet::Changed { new: acc1, old: acc2 }
+            .apply_to_db(&tx, address, tx_num, true)
+            .unwrap();
+        assert_eq!(
+            tx.get::<tables::AccountChangeSet>(tx_num),
+            Ok(Some(AccountBeforeTx { address, info: Some(acc2) }))
+        );
+        assert_eq!(tx.get::<tables::PlainAccountState>(address), Ok(Some(acc1)));
+
+        AccountInfoChangeSet::Created { new: acc1 }
+            .apply_to_db(&tx, address, tx_num, true)
+            .unwrap();
+        assert_eq!(
+            tx.get::<tables::AccountChangeSet>(tx_num),
+            Ok(Some(AccountBeforeTx { address, info: None }))
+        );
+        assert_eq!(tx.get::<tables::PlainAccountState>(address), Ok(Some(acc1)));
+
+        // delete old value, as it is dupsorted
+        tx.delete::<tables::AccountChangeSet>(tx_num, None).unwrap();
+
+        AccountInfoChangeSet::Destroyed { old: acc2 }
+            .apply_to_db(&tx, address, tx_num, true)
+            .unwrap();
+        assert_eq!(tx.get::<tables::PlainAccountState>(address), Ok(None));
+        assert_eq!(
+            tx.get::<tables::AccountChangeSet>(tx_num),
+            Ok(Some(AccountBeforeTx { address, info: Some(acc2) }))
+        );
+    }
+}

crates/executor/src/lib.rs

@@ -9,6 +9,9 @@
 
 pub mod config;
 pub mod eth_dao_fork;
+
+/// Execution result types
+pub mod execution_result;
 /// Executor
 pub mod executor;
 /// Wrapper around revm database and types

crates/executor/src/revm_wrap.rs

@@ -1,6 +1,6 @@
 use reth_interfaces::Error;
 use reth_primitives::{
-    Account, Header, Transaction, TransactionKind, TransactionSignedEcRecovered, TxEip1559,
+    Account, Address, Header, Log, Transaction, TransactionKind, TransactionSigned, TxEip1559,
     TxEip2930, TxLegacy, H160, H256, KECCAK_EMPTY, U256,
 };
 use reth_provider::StateProvider;
@@ -82,9 +82,9 @@ pub fn fill_block_env(block_env: &mut BlockEnv, header: &Header, after_merge: bo
 }
 
 /// Fill transaction environment from Transaction.
-pub fn fill_tx_env(tx_env: &mut TxEnv, transaction: &TransactionSignedEcRecovered) {
-    tx_env.caller = transaction.signer();
-    match transaction.as_ref().as_ref() {
+pub fn fill_tx_env(tx_env: &mut TxEnv, transaction: &TransactionSigned, sender: Address) {
+    tx_env.caller = sender;
+    match transaction.as_ref() {
         Transaction::Legacy(TxLegacy {
             nonce,
             chain_id,
@@ -192,6 +192,15 @@ pub fn is_log_equal(revm_log: &revm::Log, reth_log: &reth_primitives::Log) -> bo
             .any(|(revm_topic, reth_topic)| revm_topic.0 != reth_topic.0)
 }
 
+/// Into reth primitive [Log] from [revm::Log].
+pub fn into_reth_log(log: revm::Log) -> Log {
+    Log {
+        address: H160(log.address.0),
+        topics: log.topics.into_iter().map(|h| H256(h.0)).collect(),
+        data: log.data.into(),
+    }
+}
+
 /// Create reth primitive [Account] from [revm::AccountInfo].
 /// Check if revm bytecode hash is [KECCAK_EMPTY] and put None to reth [Account]
 pub fn to_reth_acc(revm_acc: &revm::AccountInfo) -> Account {

crates/interfaces/src/executor.rs

@@ -1,12 +1,19 @@
 use async_trait::async_trait;
-use reth_primitives::{Block, Bloom, H256};
+use reth_primitives::{Address, Block, Bloom, H256};
 use thiserror::Error;
 
-/// Takes block and executes it, returns error
+/// An executor capable of executing a block.
 #[async_trait]
-pub trait BlockExecutor: Send + Sync {
-    /// Execute block
-    async fn execute(&self, _block: Block) -> Error;
+pub trait BlockExecutor<T> {
+    /// Execute a block.
+    ///
+    /// The number of `senders` should be equal to the number of transactions in the block.
+    ///
+    /// If no senders are specified, the `execute` function MUST recover the senders for the
+    /// provided block's transactions internally. We use this to allow for calculating senders in
+    /// parallel in e.g. staged sync, so that execution can happen without paying for sender
+    /// recovery costs.
+    fn execute(&mut self, block: &Block, senders: Option<Vec<Address>>) -> Result<T, Error>;
 }
 
 /// BlockExecutor Errors
@@ -17,6 +24,8 @@ pub enum Error {
     VerificationFailed,
     #[error("Fatal internal error")]
     ExecutionFatalError,
+    #[error("Failed to recover sender for transaction")]
+    SenderRecoveryError,
     #[error("Receipt cumulative gas used {got:?} is different from expected {expected:?}")]
     ReceiptCumulativeGasUsedDiff { got: u64, expected: u64 },
     #[error("Receipt log count {got:?} is different from expected {expected:?}.")]

crates/rpc/rpc-engine-api/src/engine_api.rs

@@ -138,18 +138,20 @@ impl<Client: HeaderProvider + BlockProvider + StateProvider> EngineApi<Client> {
                 }))
             }
         };
+        let block_hash = block.header.hash();
+        let parent_hash = block.parent_hash;
 
         // The block already exists in our database
-        if self.client.is_known(&block.hash())? {
-            return Ok(PayloadStatus::new(PayloadStatusEnum::Valid, block.hash()))
+        if self.client.is_known(&block_hash)? {
+            return Ok(PayloadStatus::new(PayloadStatusEnum::Valid, block_hash))
         }
 
-        let Some(parent) = self.client.block(BlockId::Hash(EthersH256(block.parent_hash.0)))? else {
+        let Some(parent) = self.client.block(BlockId::Hash(EthersH256(parent_hash.0)))? else {
              // TODO: cache block for storing later
              return Ok(PayloadStatus::from_status(PayloadStatusEnum::Syncing))
         };
 
-        if let Some(parent_td) = self.client.header_td(&block.parent_hash)? {
+        if let Some(parent_td) = self.client.header_td(&parent_hash)? {
             if Some(parent_td) <= self.chain_spec.paris_status().terminal_total_difficulty() {
                 return Ok(PayloadStatus::from_status(PayloadStatusEnum::Invalid {
                     validation_error: EngineApiError::PayloadPreMerge.to_string(),
@@ -167,26 +169,17 @@ impl<Client: HeaderProvider + BlockProvider + StateProvider> EngineApi<Client> {
             }))
         }
 
-        let (header, body, _) = block.split();
-        let transactions = body
-            .into_iter()
-            .map(|tx| {
-                let tx_hash = tx.hash;
-                tx.into_ecrecovered().ok_or(EngineApiError::PayloadSignerRecovery { hash: tx_hash })
-            })
-            .collect::<Result<Vec<_>, EngineApiError>>()?;
         let mut state_provider = SubState::new(State::new(&*self.client));
         match executor::execute_and_verify_receipt(
-            &header,
-            &transactions,
-            &[],
+            &block.unseal(),
+            None,
             &self.chain_spec,
             &mut state_provider,
         ) {
-            Ok(_) => Ok(PayloadStatus::new(PayloadStatusEnum::Valid, header.hash())),
+            Ok(_) => Ok(PayloadStatus::new(PayloadStatusEnum::Valid, block_hash)),
             Err(err) => Ok(PayloadStatus::new(
                 PayloadStatusEnum::Invalid { validation_error: err.to_string() },
-                header.parent_hash, // The parent hash is already in our database hence it is valid
+                parent_hash, // The parent hash is already in our database hence it is valid
             )),
         }
     }

crates/stages/src/stages/execution.rs

@@ -10,12 +10,11 @@ use reth_db::{
     transaction::{DbTx, DbTxMut},
 };
 use reth_executor::{
-    executor::AccountChangeSet,
+    execution_result::AccountChangeSet,
     revm_wrap::{State, SubState},
 };
 use reth_primitives::{
-    Address, ChainSpec, Hardfork, Header, StorageEntry, TransactionSignedEcRecovered, H256,
-    MAINNET, U256,
+    Address, Block, ChainSpec, Hardfork, Header, StorageEntry, H256, MAINNET, U256,
 };
 use reth_provider::LatestStateProviderRef;
 use std::fmt::Debug;
@@ -134,9 +133,9 @@ impl<DB: Database> Stage<DB> for ExecutionStage {
 
         // Fetch transactions, execute them and generate results
         let mut block_change_patches = Vec::with_capacity(canonical_batch.len());
-        for (header, body, ommers) in block_batch.iter() {
-            let num = header.number;
-            tracing::trace!(target: "sync::stages::execution", ?num, "Execute block.");
+        for (header, body, ommers) in block_batch.into_iter() {
+            let block_number = header.number;
+            tracing::trace!(target: "sync::stages::execution", ?block_number, "Execute block.");
             // iterate over all transactions
             let mut tx_walker = tx_cursor.walk(body.start_tx_id)?;
             let mut transactions = Vec::with_capacity(body.tx_count as usize);
@@ -145,7 +144,7 @@ impl<DB: Database> Stage<DB> for ExecutionStage {
                 let (tx_index, tx) =
                     tx_walker.next().ok_or(DatabaseIntegrityError::EndOfTransactionTable)??;
                 if tx_index != index {
-                    error!(target: "sync::stages::execution", block = header.number, expected = index, found = tx_index, ?body, "Transaction gap");
+                    error!(target: "sync::stages::execution", block = block_number, expected = index, found = tx_index, ?body, "Transaction gap");
                     return Err(DatabaseIntegrityError::TransactionsGap { missing: tx_index }.into())
                 }
                 transactions.push(tx);
@@ -159,23 +158,15 @@ impl<DB: Database> Stage<DB> for ExecutionStage {
                     .next()
                     .ok_or(DatabaseIntegrityError::EndOfTransactionSenderTable)??;
                 if tx_index != index {
-                    error!(target: "sync::stages::execution", block = header.number, expected = index, found = tx_index, ?body, "Signer gap");
+                    error!(target: "sync::stages::execution", block = block_number, expected = index, found = tx_index, ?body, "Signer gap");
                     return Err(
                         DatabaseIntegrityError::TransactionsSignerGap { missing: tx_index }.into()
                     )
                 }
                 signers.push(tx);
             }
-            // create ecRecovered transaction by matching tx and its signer
-            let recovered_transactions: Vec<_> = transactions
-                .into_iter()
-                .zip(signers.into_iter())
-                .map(|(tx, signer)| {
-                    TransactionSignedEcRecovered::from_signed_transaction(tx, signer)
-                })
-                .collect();
 
-            trace!(target: "sync::stages::execution", number = header.number, txs = recovered_transactions.len(), "Executing block");
+            trace!(target: "sync::stages::execution", number = block_number, txs = transactions.len(), "Executing block");
 
             // For ethereum tests that has MAX gas that calls contract until max depth (1024 calls)
             // revm can take more then default allocated stack space. For this case we are using
@@ -188,9 +179,8 @@ impl<DB: Database> Stage<DB> for ExecutionStage {
                     .spawn_scoped(scope, || {
                         // execute and store output to results
                         reth_executor::executor::execute_and_verify_receipt(
-                            header,
-                            &recovered_transactions,
-                            ommers,
+                            &Block { header, body: transactions, ommers },
+                            Some(signers),
                             &self.chain_spec,
                             &mut state_provider,
                         )
@@ -198,8 +188,8 @@ impl<DB: Database> Stage<DB> for ExecutionStage {
                     .expect("Expects that thread name is not null");
                 handle.join().expect("Expects for thread to not panic")
             })
-            .map_err(|error| StageError::ExecutionError { block: header.number, error })?;
-            block_change_patches.push((changeset, num));
+            .map_err(|error| StageError::ExecutionError { block: block_number, error })?;
+            block_change_patches.push((changeset, block_number));
         }
 
         // Get last tx count so that we can know amount of transaction in the block.