feat(l1): process blocks in batches when syncing and importing (#2174)

**Motivation**
Accelerate syncing!

**Description**
This PR introduces block batching during full sync:
1. Instead of storing and computing the state root for each block
individually, we now maintain a single state tree for the entire batch,
committing it only at the end. This results in one state trie per `n`
blocks instead of one per block (we'll need less storage also).
2. The new full sync process:
    - Request 1024 headers
    - Request 1024 block bodies and collect them
- Once all blocks are received, process them in batches using a single
state trie, which is attached to the last block.
3. Blocks are now stored in a single transaction.
4. State root, receipts root, and request root validation are only
required for the last block in the batch.
5. The new add_blocks_in_batch function includes a flag,
`should_commit_intermediate_tries`. When set to true, it stores the
tries for each block. This functionality is added to make the hive test
pass. Currently, this is handled by verifying if the block is within the
`STATE_TRIES_TO_KEEP` range. In a real syncing scenario, my intuition is
that it would be better to wait until we are fully synced and then we
would start storing the state of the new blocks and pruning when we
reach `STATE_TRIES_TO_KEEP`.
6. Throughput when syncing is now measured per batches.
7. A new command was added to import blocks in batch

Considerations:
1. ~Optimize account updates: Instead of inserting updates into the
state trie after each block execution, batch them at the end, merging
repeated accounts to reduce insertions and improve performance (see
#2216)~ Closes #2216.
2. Improve transaction handling: Avoid committing storage tries to the
database separately. Instead, create a single transaction for storing
receipts, storage tries, and blocks. This would require additional
abstractions for transaction management (see #2217).
3. This isn't working for `levm` backend we need it to cache the
executions state and persist it between them, as we don't store anything
until the final of the batch (see #2218).
4. In #2210 a new ci is added to run a bench comparing main and `head`
branch using `import-in-batch`

Closes None

---------

Co-authored-by: Martin Paulucci <martin.c.paulucci@gmail.com>
Co-authored-by: fmoletta <99273364+fmoletta@users.noreply.github.com>

Author: 3 people

CHANGELOG.md

@@ -3,10 +3,14 @@
 ## Perf
 
 #### 2025-03-30
-* Faster block import, use a slice instead of copy
-[#2097](https://github.com/lambdaclass/ethrex/pull/2097)
 
+- Faster block import, use a slice instead of copy
+  [#2097](https://github.com/lambdaclass/ethrex/pull/2097)
 
 #### 2025-02-28
 
-* Don't recompute transaction senders when building blocks [#2097](https://github.com/lambdaclass/ethrex/pull/2097)
+- Don't recompute transaction senders when building blocks [#2097](https://github.com/lambdaclass/ethrex/pull/2097)
+
+#### 2025-03-21
+
+- Process blocks in batches when syncing and importing [#2174](https://github.com/lambdaclass/ethrex/pull/2174)

crates/blockchain/blockchain.rs

@@ -18,12 +18,13 @@ use ethrex_common::types::{
     BlockHeader, BlockNumber, ChainConfig, EIP4844Transaction, Receipt, Transaction,
 };
 
-use ethrex_common::{Address, H256};
+use ethrex_common::{Address, H160, H256};
 use mempool::Mempool;
+use std::collections::HashMap;
 use std::{ops::Div, time::Instant};
 
 use ethrex_storage::error::StoreError;
-use ethrex_storage::Store;
+use ethrex_storage::{AccountUpdate, Store};
 use ethrex_vm::{BlockExecutionResult, Evm, EvmEngine};
 use fork_choice::apply_fork_choice;
 use tracing::{error, info, warn};
@@ -38,6 +39,12 @@ pub struct Blockchain {
     pub mempool: Mempool,
 }
 
+#[derive(Debug, Clone)]
+pub struct BatchBlockProcessingFailure {
+    pub last_valid_hash: H256,
+    pub failed_block_hash: H256,
+}
+
 impl Blockchain {
     pub fn new(evm_engine: EvmEngine, store: Store) -> Self {
         Self {
@@ -55,7 +62,8 @@ impl Blockchain {
         }
     }
 
-    pub fn execute_block(&self, block: &Block) -> Result<BlockExecutionResult, ChainError> {
+    /// Executes a block withing a new vm instance and state
+    fn execute_block(&self, block: &Block) -> Result<BlockExecutionResult, ChainError> {
         // Validate if it can be the new head and find the parent
         let Ok(parent_header) = find_parent_header(&block.header, &self.storage) else {
             // If the parent is not present, we store it as pending.
@@ -74,45 +82,55 @@ impl Blockchain {
         );
         let execution_result = vm.execute_block(block)?;
 
+        // Validate execution went alright
+        validate_gas_used(&execution_result.receipts, &block.header)?;
+        validate_receipts_root(&block.header, &execution_result.receipts)?;
+        validate_requests_hash(&block.header, &chain_config, &execution_result.requests)?;
+
         Ok(execution_result)
     }
 
-    pub fn store_block(
+    /// Executes a block from a given vm instance an does not clear its state
+    fn execute_block_from_state(
         &self,
+        parent_header: &BlockHeader,
         block: &Block,
-        execution_result: BlockExecutionResult,
-    ) -> Result<(), ChainError> {
-        // Assumes block is valid
-        let BlockExecutionResult {
-            receipts,
-            requests,
-            account_updates,
-        } = execution_result;
-        let chain_config = self.storage.get_chain_config()?;
+        chain_config: &ChainConfig,
+        vm: &mut Evm,
+    ) -> Result<BlockExecutionResult, ChainError> {
+        // Validate the block pre-execution
+        validate_block(block, parent_header, chain_config)?;
 
-        let block_hash = block.header.compute_block_hash();
+        let execution_result = vm.execute_block_without_clearing_state(block)?;
 
-        validate_gas_used(&receipts, &block.header)?;
+        // Validate execution went alright
+        validate_gas_used(&execution_result.receipts, &block.header)?;
+        validate_receipts_root(&block.header, &execution_result.receipts)?;
+        validate_requests_hash(&block.header, chain_config, &execution_result.requests)?;
 
+        Ok(execution_result)
+    }
+
+    pub fn store_block(
+        &self,
+        block: &Block,
+        execution_result: BlockExecutionResult,
+    ) -> Result<(), ChainError> {
         // Apply the account updates over the last block's state and compute the new state root
         let new_state_root = self
             .storage
-            .apply_account_updates(block.header.parent_hash, &account_updates)?
+            .apply_account_updates(block.header.parent_hash, &execution_result.account_updates)?
             .ok_or(ChainError::ParentStateNotFound)?;
 
         // Check state root matches the one in block header
         validate_state_root(&block.header, new_state_root)?;
 
-        // Check receipts root matches the one in block header
-        validate_receipts_root(&block.header, &receipts)?;
-
-        // Processes requests from receipts, computes the requests_hash and compares it against the header
-        validate_requests_hash(&block.header, &chain_config, &requests)?;
-
-        store_block(&self.storage, block.clone())?;
-        store_receipts(&self.storage, receipts, block_hash)?;
-
-        Ok(())
+        self.storage
+            .add_block(block.clone())
+            .map_err(ChainError::StoreError)?;
+        self.storage
+            .add_receipts(block.hash(), execution_result.receipts)
+            .map_err(ChainError::StoreError)
     }
 
     pub fn add_block(&self, block: &Block) -> Result<(), ChainError> {
@@ -132,6 +150,141 @@ impl Blockchain {
         result
     }
 
+    /// Adds multiple blocks in a batch.
+    ///
+    /// If an error occurs, returns a tuple containing:
+    /// - The error type ([`ChainError`]).
+    /// - [`BatchProcessingFailure`] (if the error was caused by block processing).
+    ///
+    /// Note: only the last block's state trie is stored in the db
+    pub fn add_blocks_in_batch(
+        &self,
+        blocks: &[Block],
+    ) -> Result<(), (ChainError, Option<BatchBlockProcessingFailure>)> {
+        let mut last_valid_hash = H256::default();
+
+        let Some(first_block_header) = blocks.first().map(|e| e.header.clone()) else {
+            return Err((ChainError::Custom("First block not found".into()), None));
+        };
+
+        let chain_config: ChainConfig = self
+            .storage
+            .get_chain_config()
+            .map_err(|e| (e.into(), None))?;
+        let mut vm = Evm::new(
+            self.evm_engine,
+            self.storage.clone(),
+            first_block_header.parent_hash,
+        );
+
+        let blocks_len = blocks.len();
+        let mut all_receipts: HashMap<BlockHash, Vec<Receipt>> = HashMap::new();
+        let mut all_account_updates: HashMap<H160, AccountUpdate> = HashMap::new();
+        let mut total_gas_used = 0;
+        let mut transactions_count = 0;
+
+        let interval = Instant::now();
+        for (i, block) in blocks.iter().enumerate() {
+            // for the first block, we need to query the store
+            let parent_header = if i == 0 {
+                let Ok(parent_header) = find_parent_header(&block.header, &self.storage) else {
+                    return Err((
+                        ChainError::ParentNotFound,
+                        Some(BatchBlockProcessingFailure {
+                            failed_block_hash: block.hash(),
+                            last_valid_hash,
+                        }),
+                    ));
+                };
+                parent_header
+            } else {
+                // for the subsequent ones, the parent is the previous block
+                blocks[i - 1].header.clone()
+            };
+
+            let BlockExecutionResult {
+                receipts,
+                account_updates,
+                ..
+            } = match self.execute_block_from_state(&parent_header, block, &chain_config, &mut vm) {
+                Ok(result) => result,
+                Err(err) => {
+                    return Err((
+                        err,
+                        Some(BatchBlockProcessingFailure {
+                            failed_block_hash: block.hash(),
+                            last_valid_hash,
+                        }),
+                    ))
+                }
+            };
+
+            // Merge account updates
+            for account_update in account_updates {
+                let Some(cache) = all_account_updates.get_mut(&account_update.address) else {
+                    all_account_updates.insert(account_update.address, account_update);
+                    continue;
+                };
+
+                cache.removed = account_update.removed;
+                if let Some(code) = account_update.code {
+                    cache.code = Some(code);
+                };
+
+                if let Some(info) = account_update.info {
+                    cache.info = Some(info);
+                }
+
+                for (k, v) in account_update.added_storage.into_iter() {
+                    cache.added_storage.insert(k, v);
+                }
+            }
+
+            last_valid_hash = block.hash();
+            total_gas_used += block.header.gas_used;
+            transactions_count += block.body.transactions.len();
+            all_receipts.insert(block.hash(), receipts);
+        }
+
+        let Some(last_block) = blocks.last() else {
+            return Err((ChainError::Custom("Last block not found".into()), None));
+        };
+
+        // Apply the account updates over all blocks and compute the new state root
+        let new_state_root = self
+            .storage
+            .apply_account_updates(
+                first_block_header.parent_hash,
+                &all_account_updates.into_values().collect::<Vec<_>>(),
+            )
+            .map_err(|e| (e.into(), None))?
+            .ok_or((ChainError::ParentStateNotFound, None))?;
+
+        // Check state root matches the one in block header
+        validate_state_root(&last_block.header, new_state_root).map_err(|e| (e, None))?;
+
+        self.storage
+            .add_blocks(blocks)
+            .map_err(|e| (e.into(), None))?;
+        self.storage
+            .add_receipts_for_blocks(all_receipts)
+            .map_err(|e| (e.into(), None))?;
+
+        let elapsed_total = interval.elapsed().as_millis();
+        let mut throughput = 0.0;
+        if elapsed_total != 0 && total_gas_used != 0 {
+            let as_gigas = (total_gas_used as f64).div(10_f64.powf(9_f64));
+            throughput = (as_gigas) / (elapsed_total as f64) * 1000_f64;
+        }
+
+        info!(
+            "[METRICS] Executed and stored: Range: {}, Total transactions: {}, Throughput: {} Gigagas/s",
+            blocks_len, transactions_count, throughput
+        );
+
+        Ok(())
+    }
+
     //TODO: Forkchoice Update shouldn't be part of this function
     pub fn import_blocks(&self, blocks: &Vec<Block>) {
         let size = blocks.len();
@@ -364,21 +517,6 @@ pub fn validate_requests_hash(
     Ok(())
 }
 
-/// Stores block and header in the database
-pub fn store_block(storage: &Store, block: Block) -> Result<(), ChainError> {
-    storage.add_block(block)?;
-    Ok(())
-}
-
-pub fn store_receipts(
-    storage: &Store,
-    receipts: Vec<Receipt>,
-    block_hash: BlockHash,
-) -> Result<(), ChainError> {
-    storage.add_receipts(block_hash, receipts)?;
-    Ok(())
-}
-
 /// Performs post-execution checks
 pub fn validate_state_root(
     block_header: &BlockHeader,

crates/common/trie/trie.rs

@@ -125,9 +125,7 @@ impl Trie {
     /// Returns keccak(RLP_NULL) if the trie is empty
     /// Also commits changes to the DB
     pub fn hash(&mut self) -> Result<H256, TrieError> {
-        if let Some(ref root) = self.root {
-            self.state.commit(root)?;
-        }
+        self.commit()?;
         Ok(self
             .root
             .as_ref()
@@ -144,6 +142,13 @@ impl Trie {
             .unwrap_or(*EMPTY_TRIE_HASH)
     }
 
+    pub fn commit(&mut self) -> Result<(), TrieError> {
+        if let Some(ref root) = self.root {
+            self.state.commit(root)?;
+        }
+        Ok(())
+    }
+
     /// Obtain a merkle proof for the given path.
     /// The proof will contain all the encoded nodes traversed until reaching the node where the path is stored (including this last node).
     /// The proof will still be constructed even if the path is not stored in the trie, proving its absence.

crates/networking/p2p/peer_handler.rs

@@ -14,7 +14,9 @@ use crate::{
     kademlia::{KademliaTable, PeerChannels},
     rlpx::{
         eth::{
-            blocks::{BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders},
+            blocks::{
+                BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders, BLOCK_HEADER_LIMIT,
+            },
             receipts::{GetReceipts, Receipts},
         },
         message::Message as RLPxMessage,
@@ -33,14 +35,13 @@ pub const REQUEST_RETRY_ATTEMPTS: usize = 5;
 pub const MAX_RESPONSE_BYTES: u64 = 512 * 1024;
 pub const HASH_MAX: H256 = H256([0xFF; 32]);
 
-// Ask as much as 128 block bodies and 192 block headers per request
-// these magic numbers are not part of the protocol and are taken from geth, see:
+// Ask as much as 128 block bodies per request
+// this magic number is not part of the protocol and is taken from geth, see:
 // https://github.com/ethereum/go-ethereum/blob/2585776aabbd4ae9b00050403b42afb0cee968ec/eth/downloader/downloader.go#L42-L43
 //
 // Note: We noticed that while bigger values are supported
 // increasing them may be the cause of peers disconnection
 pub const MAX_BLOCK_BODIES_TO_REQUEST: usize = 128;
-pub const MAX_BLOCK_HEADERS_TO_REQUEST: usize = 192;
 
 /// An abstraction over the [KademliaTable] containing logic to make requests to peers
 #[derive(Debug, Clone)]
@@ -83,14 +84,13 @@ impl PeerHandler {
         &self,
         start: H256,
         order: BlockRequestOrder,
-        limit: u64,
     ) -> Option<Vec<BlockHeader>> {
         for _ in 0..REQUEST_RETRY_ATTEMPTS {
             let request_id = rand::random();
             let request = RLPxMessage::GetBlockHeaders(GetBlockHeaders {
                 id: request_id,
                 startblock: start.into(),
-                limit,
+                limit: BLOCK_HEADER_LIMIT,
                 skip: 0,
                 reverse: matches!(order, BlockRequestOrder::NewToOld),
             });