feat(l1): decouple size of execution batch from header request size during full-sync (#3074)

**Motivation**
Allow us to configure the amount of blocks to execute in a single batch
during full sync. Currently, the only way to do this is by changing the
amount of block headers we ask for in each request.
In order to achieve this, this PR proposes adding the enum
`BlockSyncState` with variants for Full and Snap sync so we can separate
behaviors between each mode and also allow each mode to keep its
separate state. This is key as we will need to persist headers and
bodies through various fetch requests so we can build custom-sized
execution batches.
It also replaces the previous configurable env var `BlOCK_HEADER_LIMIT`
with `EXECUTE_BLOCK_BATCH`
<!-- Why does this pull request exist? What are its goals? -->

**Description**
* Add `BlockSyncState` enum as a way to differentiate between each sync
mode's state during block syncing phase.
* Refactor `request_and_validate_block_bodies`: it now receives a slice
of headers and returns the requested block bodies instead of the full
blocks. This allowed us to completely get rid of header cloning.
* `validate_block_body` now receives a reference to the head & body
instead of the full block (as a result of refactoring its only user)
* `Store::add_block_headers` now only receives the headers (This lets us
simplify caller code)
* Removed `search_head` variable as having both current & search head
serves no purpose.
* Abtract current_head selection into `BlockSyncState::get_current_head`
* Fix bug in condition used to decide wether to switch from snap to full
sync
* `start_sync` no longer receives `current_head`
<!-- A clear and concise general description of the changes this PR
introduces -->

<!-- Link to issues: Resolves #111, Resolves #222 -->

Closes #2894

---------

Co-authored-by: SDartayet <[email protected]>
Co-authored-by: Martin Paulucci <[email protected]>

Author: 3 people

cmd/ethrex/cli.rs

@@ -394,6 +394,10 @@ pub async fn import_blocks(
 
     for blocks in chains {
         let size = blocks.len();
+        let numbers_and_hashes = blocks
+            .iter()
+            .map(|b| (b.header.number, b.hash()))
+            .collect::<Vec<_>>();
         // Execute block by block
         for block in &blocks {
             let hash = block.hash();
@@ -418,7 +422,7 @@ pub async fn import_blocks(
         }
 
         _ = store
-            .mark_chain_as_canonical(&blocks)
+            .mark_chain_as_canonical(&numbers_and_hashes)
             .await
             .inspect_err(|error| warn!("Failed to apply fork choice: {}", error));
 

crates/common/base64.rs

@@ -33,7 +33,7 @@ pub fn encode(bytes: &[u8]) -> Vec<u8> {
 
     let mut bytes_iter = bytes.iter();
     while bytes_iter.len() > 0 {
-        // each block is made of as much as 24 bits (3 bytes)
+        // each block is made up of as many as 24 bits (3 bytes)
         let mut block: Vec<u8> = vec![];
 
         while block.len() < 3 {

crates/common/types/block.rs

@@ -580,21 +580,24 @@ pub fn validate_block_header(
 }
 
 /// Validates that the body matches with the header
-pub fn validate_block_body(block: &Block) -> Result<(), InvalidBlockBodyError> {
+pub fn validate_block_body(
+    block_header: &BlockHeader,
+    block_body: &BlockBody,
+) -> Result<(), InvalidBlockBodyError> {
     // Validates that:
     //  - Transactions root and withdrawals root matches with the header
     //  - Ommers is empty -> https://eips.ethereum.org/EIPS/eip-3675
-    let computed_tx_root = compute_transactions_root(&block.body.transactions);
+    let computed_tx_root = compute_transactions_root(&block_body.transactions);
 
-    if block.header.transactions_root != computed_tx_root {
+    if block_header.transactions_root != computed_tx_root {
         return Err(InvalidBlockBodyError::TransactionsRootNotMatch);
     }
 
-    if !block.body.ommers.is_empty() {
+    if !block_body.ommers.is_empty() {
         return Err(InvalidBlockBodyError::OmmersIsNotEmpty);
     }
 
-    match (block.header.withdrawals_root, &block.body.withdrawals) {
+    match (block_header.withdrawals_root, &block_body.withdrawals) {
         (Some(withdrawals_root), Some(withdrawals)) => {
             let computed_withdrawals_root = compute_withdrawals_root(withdrawals);
             if withdrawals_root != computed_withdrawals_root {

crates/networking/p2p/discv4/lookup.rs

@@ -144,7 +144,7 @@ impl Discv4LookupHandler {
         asked_peers: &mut HashSet<H512>,
         nodes_to_ask: &Vec<Node>,
     ) -> (Vec<Node>, u32) {
-        // send FIND_NODE as much as three times
+        // send FIND_NODE as many as three times
         let alpha = 3;
         let mut queries = 0;
         let mut nodes = vec![];

crates/networking/p2p/peer_handler.rs

@@ -7,7 +7,7 @@ use std::{
 use bytes::Bytes;
 use ethrex_common::{
     H256, U256,
-    types::{AccountState, Block, BlockBody, BlockHeader, Receipt, validate_block_body},
+    types::{AccountState, BlockBody, BlockHeader, Receipt, validate_block_body},
 };
 use ethrex_rlp::encode::RLPEncode;
 use ethrex_trie::Nibbles;
@@ -40,7 +40,7 @@ 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 per request
+// Request as many 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
 //
@@ -268,56 +268,40 @@ impl PeerHandler {
         None
     }
 
-    /// Requests block bodies from any suitable peer given their block hashes and validates them
-    /// Returns the full block or None if:
+    /// Requests block bodies from any suitable peer given their block headers and validates them
+    /// Returns the requested block bodies or None if:
     /// - There are no available peers (the node just started up or was rejected by all other nodes)
     /// - No peer returned a valid response in the given time and retry limits
     /// - The block bodies are invalid given the block headers
-    pub async fn request_and_validate_block_bodies<'a>(
+    pub async fn request_and_validate_block_bodies(
         &self,
-        block_hashes: &mut Vec<H256>,
-        headers_iter: &mut impl Iterator<Item = &BlockHeader>,
-    ) -> Option<Vec<Block>> {
-        let original_hashes = block_hashes.clone();
-        let headers_vec: Vec<&BlockHeader> = headers_iter.collect();
+        block_headers: &[BlockHeader],
+    ) -> Option<Vec<BlockBody>> {
+        let block_hashes: Vec<H256> = block_headers.iter().map(|h| h.hash()).collect();
 
         for _ in 0..REQUEST_RETRY_ATTEMPTS {
-            *block_hashes = original_hashes.clone();
-            let mut headers_iter = headers_vec.iter().copied();
-
             let Some((block_bodies, peer_id)) =
                 self.request_block_bodies_inner(block_hashes.clone()).await
             else {
                 continue; // Retry on empty response
             };
-
-            let mut blocks: Vec<Block> = vec![];
-            let block_bodies_len = block_bodies.len();
-
-            // Push blocks
-            for (_, body) in block_hashes.drain(..block_bodies_len).zip(block_bodies) {
-                let Some(header) = headers_iter.next() else {
-                    debug!("[SYNCING] Header not found for the block bodies received, skipping...");
-                    break; // Break out of block creation and retry with different peer
-                };
-
-                let block = Block::new(header.clone(), body);
-                blocks.push(block);
+            let mut res = Vec::new();
+            let mut validation_success = true;
+            for (header, body) in block_headers[..block_bodies.len()].iter().zip(block_bodies) {
+                if let Err(e) = validate_block_body(header, &body) {
+                    warn!(
+                        "Invalid block body error {e}, discarding peer {peer_id} and retrying..."
+                    );
+                    validation_success = false;
+                    self.record_peer_critical_failure(peer_id).await;
+                    break;
+                }
+                res.push(body);
             }
-
-            // Validate blocks
-            if let Some(e) = blocks
-                .iter()
-                .find_map(|block| validate_block_body(block).err())
-            {
-                warn!(
-                    "[SYNCING] Invalid block body error {e}, discarding peer {peer_id} and retrying..."
-                );
-                self.record_peer_critical_failure(peer_id).await;
-                continue; // Retry on validation failure
+            // Retry on validation failure
+            if validation_success {
+                return Some(res);
             }
-
-            return Some(blocks);
         }
         None
     }

crates/networking/p2p/rlpx/eth/blocks.rs

@@ -11,8 +11,6 @@ use ethrex_rlp::{
     structs::{Decoder, Encoder},
 };
 use ethrex_storage::Store;
-#[cfg(feature = "sync-test")]
-use std::env;
 use tracing::error;
 
 pub const HASH_FIRST_BYTE_DECODER: u8 = 160;
@@ -114,18 +112,6 @@ impl GetBlockHeaders {
             (self.skip + 1) as i64
         };
 
-        #[cfg(feature = "sync-test")]
-        let limit = if let Ok(env_var_block_limit) = env::var("BLOCK_HEADER_LIMIT") {
-            env_var_block_limit
-                .parse()
-                .expect("Block header limit environmental variable is not a number")
-        } else if self.limit > BLOCK_HEADER_LIMIT {
-            BLOCK_HEADER_LIMIT
-        } else {
-            self.limit
-        };
-
-        #[cfg(not(feature = "sync-test"))]
         let limit = if self.limit > BLOCK_HEADER_LIMIT {
             BLOCK_HEADER_LIMIT
         } else {