feat: build local pending block (#3811)

crates/rpc/rpc/src/eth/api/block.rs

@@ -7,6 +7,7 @@ use crate::{
     },
     EthApi,
 };
+use reth_network_api::NetworkInfo;
 use reth_primitives::{BlockId, BlockNumberOrTag, TransactionMeta};
 use reth_provider::{BlockReaderIdExt, EvmEnvProvider, StateProviderFactory};
 use reth_rpc_types::{Block, Index, RichBlock, TransactionReceipt};
@@ -16,7 +17,7 @@ impl<Provider, Pool, Network> EthApi<Provider, Pool, Network>
 where
     Provider: BlockReaderIdExt + StateProviderFactory + EvmEnvProvider + 'static,
     Pool: TransactionPool + Clone + 'static,
-    Network: Send + Sync + 'static,
+    Network: NetworkInfo + Send + Sync + 'static,
 {
     /// Returns the uncle headers of the given block
     ///

crates/rpc/rpc/src/eth/api/mod.rs

@@ -4,6 +4,7 @@
 //! files.
 
 use crate::eth::{
+    api::pending_block::{PendingBlock, PendingBlockEnv, PendingBlockEnvOrigin},
     cache::EthStateCache,
     error::{EthApiError, EthResult},
     gas_oracle::GasPriceOracle,
@@ -20,7 +21,11 @@ use reth_rpc_types::{SyncInfo, SyncStatus};
 use reth_tasks::{TaskSpawner, TokioTaskExecutor};
 use reth_transaction_pool::TransactionPool;
 use revm_primitives::{BlockEnv, CfgEnv};
-use std::{future::Future, sync::Arc, time::Instant};
+use std::{
+    future::Future,
+    sync::Arc,
+    time::{Duration, Instant},
+};
 use tokio::sync::{oneshot, Mutex};
 
 mod block;
@@ -32,7 +37,6 @@ mod sign;
 mod state;
 mod transactions;
 
-use crate::eth::api::pending_block::{PendingBlock, PendingBlockEnv, PendingBlockEnvOrigin};
 pub use transactions::{EthTransactions, TransactionSource};
 
 /// `Eth` API trait.
@@ -220,7 +224,7 @@ impl<Provider, Pool, Network> EthApi<Provider, Pool, Network>
 where
     Provider: BlockReaderIdExt + StateProviderFactory + EvmEnvProvider + 'static,
     Pool: TransactionPool + Clone + 'static,
-    Network: Send + Sync + 'static,
+    Network: NetworkInfo + Send + Sync + 'static,
 {
     /// Configures the [CfgEnv] and [BlockEnv] for the pending block
     ///
@@ -261,24 +265,41 @@ where
 
         // no pending block from the CL yet, so we need to build it ourselves via txpool
         self.on_blocking_task(|this| async move {
-            let PendingBlockEnv { cfg: _, block_env, origin } = pending;
-            let lock = this.inner.pending_block.lock().await;
+            let mut lock = this.inner.pending_block.lock().await;
             let now = Instant::now();
-            // this is guaranteed to be the `latest` header
-            let parent_header = origin.into_header();
 
             // check if the block is still good
-            if let Some(pending) = lock.as_ref() {
-                if block_env.number.to::<u64>() == pending.block.number &&
-                    pending.block.parent_hash == parent_header.parent_hash &&
-                    now <= pending.expires_at
+            if let Some(pending_block) = lock.as_ref() {
+                // this is guaranteed to be the `latest` header
+                if pending.block_env.number.to::<u64>() == pending_block.block.number &&
+                    pending.origin.header().hash == pending_block.block.parent_hash &&
+                    now <= pending_block.expires_at
                 {
-                    return Ok(Some(pending.block.clone()))
+                    return Ok(Some(pending_block.block.clone()))
                 }
             }
 
-            // TODO(mattsse): actually build the pending block
-            Ok(None)
+            // if we're currently syncing, we're unable to build a pending block
+            if this.network().is_syncing() {
+                return Ok(None)
+            }
+
+            // we rebuild the block
+            let pending_block = match pending.build_block(this.provider(), this.pool()) {
+                Ok(block) => block,
+                Err(err) => {
+                    tracing::debug!(target = "rpc", "Failed to build pending block: {:?}", err);
+                    return Ok(None)
+                }
+            };
+
+            let now = Instant::now();
+            *lock = Some(PendingBlock {
+                block: pending_block.clone(),
+                expires_at: now + Duration::from_secs(3),
+            });
+
+            Ok(Some(pending_block))
         })
         .await
     }

crates/rpc/rpc/src/eth/api/pending_block.rs

@@ -1,7 +1,18 @@
 //! Support for building a pending block via local txpool.
 
-use reth_primitives::{SealedBlock, SealedHeader};
-use revm_primitives::{BlockEnv, CfgEnv};
+use crate::eth::error::EthResult;
+use reth_primitives::{
+    constants::{BEACON_NONCE, EMPTY_WITHDRAWALS},
+    proofs, Block, Header, IntoRecoveredTransaction, Receipt, SealedBlock, SealedHeader,
+    EMPTY_OMMER_ROOT, H256, U256,
+};
+use reth_provider::{PostState, StateProviderFactory};
+use reth_revm::{
+    database::State, env::tx_env_with_recovered, executor::commit_state_changes, into_reth_log,
+};
+use reth_transaction_pool::TransactionPool;
+use revm::db::CacheDB;
+use revm_primitives::{BlockEnv, CfgEnv, EVMError, Env, InvalidTransaction, ResultAndState};
 use std::time::Instant;
 
 /// Configured [BlockEnv] and [CfgEnv] for a pending block
@@ -15,6 +26,133 @@ pub(crate) struct PendingBlockEnv {
     pub(crate) origin: PendingBlockEnvOrigin,
 }
 
+impl PendingBlockEnv {
+    /// Builds a pending block from the given client and pool.
+    pub(crate) fn build_block<Client, Pool>(
+        self,
+        client: &Client,
+        pool: &Pool,
+    ) -> EthResult<SealedBlock>
+    where
+        Client: StateProviderFactory,
+        Pool: TransactionPool,
+    {
+        let Self { cfg, block_env, origin } = self;
+
+        let parent_hash = origin.build_target_hash();
+        let state = State::new(client.history_by_block_hash(parent_hash)?);
+        let mut db = CacheDB::new(state);
+        let mut post_state = PostState::default();
+
+        let mut cumulative_gas_used = 0;
+        let block_gas_limit: u64 = block_env.gas_limit.try_into().unwrap_or(u64::MAX);
+        let base_fee = block_env.basefee.to::<u64>();
+        let block_number = block_env.number.to::<u64>();
+
+        let mut executed_txs = Vec::new();
+        let mut best_txs = pool.best_transactions_with_base_fee(base_fee as u128);
+
+        while let Some(pool_tx) = best_txs.next() {
+            // ensure we still have capacity for this transaction
+            if cumulative_gas_used + pool_tx.gas_limit() > block_gas_limit {
+                // we can't fit this transaction into the block, so we need to mark it as invalid
+                // which also removes all dependent transaction from the iterator before we can
+                // continue
+                best_txs.mark_invalid(&pool_tx);
+                continue
+            }
+
+            // convert tx to a signed transaction
+            let tx = pool_tx.to_recovered_transaction();
+
+            // Configure the environment for the block.
+            let env =
+                Env { cfg: cfg.clone(), block: block_env.clone(), tx: tx_env_with_recovered(&tx) };
+
+            let mut evm = revm::EVM::with_env(env);
+            evm.database(&mut db);
+
+            let ResultAndState { result, state } = match evm.transact() {
+                Ok(res) => res,
+                Err(err) => {
+                    match err {
+                        EVMError::Transaction(err) => {
+                            if matches!(err, InvalidTransaction::NonceTooLow { .. }) {
+                                // if the nonce is too low, we can skip this transaction
+                            } else {
+                                // if the transaction is invalid, we can skip it and all of its
+                                // descendants
+                                best_txs.mark_invalid(&pool_tx);
+                            }
+                            continue
+                        }
+                        err => {
+                            // this is an error that we should treat as fatal for this attempt
+                            return Err(err.into())
+                        }
+                    }
+                }
+            };
+
+            let gas_used = result.gas_used();
+
+            // commit changes
+            commit_state_changes(&mut db, &mut post_state, block_number, state, true);
+
+            // add gas used by the transaction to cumulative gas used, before creating the receipt
+            cumulative_gas_used += gas_used;
+
+            // Push transaction changeset and calculate header bloom filter for receipt.
+            post_state.add_receipt(
+                block_number,
+                Receipt {
+                    tx_type: tx.tx_type(),
+                    success: result.is_success(),
+                    cumulative_gas_used,
+                    logs: result.logs().into_iter().map(into_reth_log).collect(),
+                },
+            );
+            // append transaction to the list of executed transactions
+            executed_txs.push(tx.into_signed());
+        }
+
+        let receipts_root = post_state.receipts_root(block_number);
+        let logs_bloom = post_state.logs_bloom(block_number);
+
+        // calculate the state root
+        let state_root = db.db.state().state_root(post_state)?;
+
+        // create the block header
+        let transactions_root = proofs::calculate_transaction_root(&executed_txs);
+
+        let header = Header {
+            parent_hash,
+            ommers_hash: EMPTY_OMMER_ROOT,
+            beneficiary: block_env.coinbase,
+            state_root,
+            transactions_root,
+            receipts_root,
+            withdrawals_root: Some(EMPTY_WITHDRAWALS),
+            logs_bloom,
+            timestamp: block_env.timestamp.to::<u64>(),
+            mix_hash: block_env.prevrandao.unwrap_or_default(),
+            nonce: BEACON_NONCE,
+            base_fee_per_gas: Some(base_fee),
+            number: block_number,
+            gas_limit: block_gas_limit,
+            difficulty: U256::ZERO,
+            gas_used: cumulative_gas_used,
+            extra_data: Default::default(),
+        };
+
+        // seal the block
+        let block = Block { header, body: executed_txs, ommers: vec![], withdrawals: Some(vec![]) };
+        let sealed_block = block.seal_slow();
+
+        Ok(sealed_block)
+    }
+}
+
 /// The origin for a configured [PendingBlockEnv]
 #[derive(Clone, Debug)]
 pub(crate) enum PendingBlockEnvOrigin {
@@ -38,6 +176,14 @@ impl PendingBlockEnvOrigin {
         }
     }
 
+    /// Returns the hash of the pending block should be built on
+    fn build_target_hash(&self) -> H256 {
+        match self {
+            PendingBlockEnvOrigin::ActualPending(block) => block.parent_hash,
+            PendingBlockEnvOrigin::DerivedFromLatest(header) => header.hash,
+        }
+    }
+
     /// Returns the header this pending block is based on.
     pub(crate) fn header(&self) -> &SealedHeader {
         match self {
@@ -45,14 +191,6 @@ impl PendingBlockEnvOrigin {
             PendingBlockEnvOrigin::DerivedFromLatest(header) => header,
         }
     }
-
-    /// Consumes the type and returns the header this pending block is based on.
-    pub(crate) fn into_header(self) -> SealedHeader {
-        match self {
-            PendingBlockEnvOrigin::ActualPending(block) => block.header,
-            PendingBlockEnvOrigin::DerivedFromLatest(header) => header,
-        }
-    }
 }
 
 /// In memory pending block for `pending` tag
@@ -63,5 +201,3 @@ pub(crate) struct PendingBlock {
     /// Timestamp when the pending block is considered outdated
     pub(crate) expires_at: Instant,
 }
-
-impl PendingBlock {}

crates/rpc/rpc/src/eth/api/transactions.rs

@@ -663,7 +663,7 @@ impl<Provider, Pool, Network> EthApi<Provider, Pool, Network>
 where
     Pool: TransactionPool + 'static,
     Provider: BlockReaderIdExt + StateProviderFactory + EvmEnvProvider + 'static,
-    Network: Send + Sync + 'static,
+    Network: NetworkInfo + Send + Sync + 'static,
 {
     pub(crate) fn sign_request(
         &self,