fix(l1): cache block hashes when adding blocks in batch (#3034)

**Motivation**
When adding blocks in batches, transactions in blocks may need to read
the hash of a previous block in the batch. This poses an issue, as we
add all blocks after the full batch has been executed ad we don't have
those hashes stored.
Previously, this issue was tackled by storing the block headers before
executing and storing the full block batch. This also required us to
make the fixes on #2831 (as we might have the headers of blocks we have
not yet executed during full sync).
This stopped being a suitable solution once #3022 was merged, as block
hashes are now searched using the block the state is based on's
ancestors. As the state is based on the first block in the batch's
parents, later blocks are not able to access the hashes of previous
blocks in the batch.
This PR aims to fix both issues by adding a `block_hash_cache` to the
`StoreVmDatabase` were we would add the hashes of all blocks in the
batch, and use it when fetching block hashes before looking at the
Store. This way we can access block hashes for previously executed
blocks in the batch without prematurely adding data from blocks yet to
be executed to the state. Allowing us to rollback fixes from #2831
<!-- Why does this pull request exist? What are its goals? -->

**Description**
* Add `block_hash_cache` to `VmStoreDatabase`
* Cache block hashes of the full batch before executing in
`Blockchain::add_block_batch`
* No longer store block headers before execution during full sync
* Rollback fixes from #2831 (no longer needed due to the above change)
<!-- A clear and concise general description of the changes this PR
introduces -->

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

Closes #issue_number

Author: fmoletta

crates/blockchain/blockchain.rs

@@ -74,19 +74,10 @@ impl Blockchain {
         block: &Block,
     ) -> Result<(BlockExecutionResult, Vec<AccountUpdate>), ChainError> {
         // Validate if it can be the new head and find the parent
-        // Feda (#2831): We search for the entire block because during full/batch sync
-        // we can have the header without the body indicating it's still syncing.
-        let parent = self
-            .storage
-            .get_block_by_hash(block.header.parent_hash)
-            .await?;
-        let parent_header = match parent {
-            Some(parent_block) => parent_block.header,
-            None => {
-                // If the parent is not present, we store it as pending.
-                self.storage.add_pending_block(block.clone()).await?;
-                return Err(ChainError::ParentNotFound);
-            }
+        let Ok(parent_header) = find_parent_header(&block.header, &self.storage) else {
+            // If the parent is not present, we store it as pending.
+            self.storage.add_pending_block(block.clone()).await?;
+            return Err(ChainError::ParentNotFound);
         };
 
         let chain_config = self.storage.get_chain_config()?;
@@ -217,7 +208,14 @@ impl Blockchain {
             .get_chain_config()
             .map_err(|e| (e.into(), None))?;
 
-        let vm_db = StoreVmDatabase::new(self.storage.clone(), first_block_header.parent_hash);
+        // Cache block hashes for the full batch so we can access them during execution without having to store the blocks beforehand
+        let block_hash_cache = blocks.iter().map(|b| (b.header.number, b.hash())).collect();
+
+        let vm_db = StoreVmDatabase::new_with_block_hash_cache(
+            self.storage.clone(),
+            first_block_header.parent_hash,
+            block_hash_cache,
+        );
         let mut vm = Evm::new(self.evm_engine, vm_db);
 
         let blocks_len = blocks.len();

crates/blockchain/fork_choice.rs

@@ -40,23 +40,19 @@ pub async fn apply_fork_choice(
         None
     };
 
-    // Feda (#2831): We search for the entire block because during full/batch sync
-    // we can have the header without the body indicating it's still syncing.
-    let head_block_res = store.get_block_by_hash(head_hash).await?;
-
-    let Some(head_block) = head_block_res else {
-        return Err(InvalidForkChoice::Syncing);
-    };
+    let head_res = store.get_block_header_by_hash(head_hash)?;
 
     if !safe_hash.is_zero() {
-        check_order(safe_res.as_ref(), Some(&head_block.header))?;
+        check_order(&safe_res, &head_res)?;
     }
 
     if !finalized_hash.is_zero() && !safe_hash.is_zero() {
-        check_order(finalized_res.as_ref(), safe_res.as_ref())?;
+        check_order(&finalized_res, &safe_res)?;
     }
 
-    let head = head_block.header;
+    let Some(head) = head_res else {
+        return Err(InvalidForkChoice::Syncing);
+    };
 
     let latest = store.get_latest_block_number().await?;
 
@@ -131,8 +127,8 @@ pub async fn apply_fork_choice(
 
 // Checks that block 1 is prior to block 2 and that if the second is present, the first one is too.
 fn check_order(
-    block_1: Option<&BlockHeader>,
-    block_2: Option<&BlockHeader>,
+    block_1: &Option<BlockHeader>,
+    block_2: &Option<BlockHeader>,
 ) -> Result<(), InvalidForkChoice> {
     // We don't need to perform the check if the hashes are null
     match (block_1, block_2) {

crates/blockchain/vm.rs

@@ -1,21 +1,41 @@
 use bytes::Bytes;
 use ethrex_common::{
-    types::{AccountInfo, BlockHash, ChainConfig, EMPTY_KECCACK_HASH},
+    types::{AccountInfo, BlockHash, BlockNumber, ChainConfig, EMPTY_KECCACK_HASH},
     Address, H256, U256,
 };
 use ethrex_storage::Store;
 use ethrex_vm::{EvmError, VmDatabase};
-use std::cmp::Ordering;
+use std::{cmp::Ordering, collections::HashMap};
 
 #[derive(Clone)]
 pub struct StoreVmDatabase {
     pub store: Store,
     pub block_hash: BlockHash,
+    // Used to store known block hashes
+    // We use this when executing blocks in batches, as we will only add the blocks at the end
+    // And may need to access hashes of blocks previously executed in the batch
+    pub block_hash_cache: HashMap<BlockNumber, BlockHash>,
 }
 
 impl StoreVmDatabase {
     pub fn new(store: Store, block_hash: BlockHash) -> Self {
-        StoreVmDatabase { store, block_hash }
+        StoreVmDatabase {
+            store,
+            block_hash,
+            block_hash_cache: HashMap::new(),
+        }
+    }
+
+    pub fn new_with_block_hash_cache(
+        store: Store,
+        block_hash: BlockHash,
+        block_hash_cache: HashMap<BlockNumber, BlockHash>,
+    ) -> Self {
+        StoreVmDatabase {
+            store,
+            block_hash,
+            block_hash_cache,
+        }
     }
 }
 
@@ -33,6 +53,10 @@ impl VmDatabase for StoreVmDatabase {
     }
 
     fn get_block_hash(&self, block_number: u64) -> Result<H256, EvmError> {
+        // Check if we have it cached
+        if let Some(block_hash) = self.block_hash_cache.get(&block_number) {
+            return Ok(*block_hash);
+        }
         // First check if our block is canonical, if it is then it's ancestor will also be canonical and we can look it up directly
         if self
             .store

crates/networking/p2p/sync.rs

@@ -274,26 +274,29 @@ impl Syncer {
             // Discard the first header as we already have it
             block_hashes.remove(0);
             block_headers.remove(0);
-            // Store headers and save hashes for full block retrieval
-            all_block_hashes.extend_from_slice(&block_hashes[..]);
-            // This step is necessary for full sync because some opcodes depend on previous blocks during execution.
-            store
-                .add_block_headers(block_hashes.clone(), block_headers.clone())
-                .await?;
-
-            if sync_mode == SyncMode::Full {
-                let last_block_hash = self
-                    .download_and_run_blocks(
-                        &block_hashes,
-                        &block_headers,
-                        sync_head,
-                        sync_head_found,
-                        store.clone(),
-                    )
-                    .await?;
-                if let Some(last_block_hash) = last_block_hash {
-                    current_head = last_block_hash;
-                    search_head = current_head;
+
+            match sync_mode {
+                SyncMode::Snap => {
+                    // Store headers and save hashes for full block retrieval
+                    all_block_hashes.extend_from_slice(&block_hashes[..]);
+                    store
+                        .add_block_headers(block_hashes.clone(), block_headers.clone())
+                        .await?;
+                }
+                SyncMode::Full => {
+                    let last_block_hash = self
+                        .download_and_run_blocks(
+                            &block_hashes,
+                            &block_headers,
+                            sync_head,
+                            sync_head_found,
+                            store.clone(),
+                        )
+                        .await?;
+                    if let Some(last_block_hash) = last_block_hash {
+                        current_head = last_block_hash;
+                        search_head = current_head;
+                    }
                 }
             }