perf(l1, l2): move post-execution trie updates to the background (#4989)

**Motivation**

This PR moves the trie portion of the `apply_updates` function to a
background thread, improving block execution times. This background
thread receives messages through a channel to apply new trie updates and
does two things:

- First, it updates the in-memory diff layers and notifies the process
that sent the message (i.e. the block production thread) so it can
continue with block execution.
- Second, it performs the logic of persisting the bottom-most diff layer
to disk. To do this, we first pause the ongoing thread in charge of
generating the snapshots (i.e. FlatKeyValue) through a message, then
persist the diff layer to disk, then notify again the thread to continue
snapshot generation.

**Description**

<!-- A clear and concise general description of the changes this PR
introduces -->

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

Closes #issue_number

---------

Co-authored-by: Mario Rugiero <[email protected]>
Co-authored-by: Javier Chatruc <[email protected]>
Co-authored-by: Javier Rodríguez Chatruc <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 - Reusing FindNode message per lookup loop instead of randomizing the key for each message. [#5047](https://github.com/lambdaclass/ethrex/pull/5047)
 
+### 2025-10-23
+
+- Move trie updates post block execution to a background thread. [#4989](https://github.com/lambdaclass/ethrex/pull/4989).
+
 ### 2025-10-21
 
 - Instead of lazy computation of blocklist, do greedy computation of allowlist and store the result, fetch it with the DB. [#4961](https://github.com/lambdaclass/ethrex/pull/4961)

crates/networking/p2p/snap.rs

@@ -180,6 +180,7 @@ mod tests {
     use ethrex_common::{BigEndianHash, H256, types::AccountState};
     use ethrex_rlp::{decode::RLPDecode, encode::RLPEncode};
     use ethrex_storage::EngineType;
+    use ethrex_trie::EMPTY_TRIE_HASH;
 
     use crate::rlpx::snap::AccountStateSlim;
 
@@ -996,7 +997,7 @@ mod tests {
 
         // Create a store and load it up with the accounts
         let store = Store::new("null", EngineType::InMemory).unwrap();
-        let mut state_trie = store.new_state_trie_for_test()?;
+        let mut state_trie = store.open_direct_state_trie(*EMPTY_TRIE_HASH)?;
         for (address, account) in accounts {
             let hashed_address = H256::from_str(address).unwrap().as_bytes().to_vec();
             let account = AccountState::from(AccountStateSlim::decode(&account).unwrap());

crates/storage/store.rs

@@ -1538,7 +1538,7 @@ mod tests {
             })
             .collect();
         accounts.sort_by_key(|a| a.0);
-        let mut trie = store.open_state_trie(*EMPTY_TRIE_HASH).unwrap();
+        let mut trie = store.open_direct_state_trie(*EMPTY_TRIE_HASH).unwrap();
         for (address, state) in &accounts {
             trie.insert(address.0.to_vec(), state.encode_to_vec())
                 .unwrap();
@@ -1564,13 +1564,13 @@ mod tests {
             .collect();
         slots.sort_by_key(|a| a.0);
         let mut trie = store
-            .open_storage_trie(address, *EMPTY_TRIE_HASH, *EMPTY_TRIE_HASH)
+            .open_direct_storage_trie(address, *EMPTY_TRIE_HASH)
             .unwrap();
         for (slot, value) in &slots {
             trie.insert(slot.0.to_vec(), value.encode_to_vec()).unwrap();
         }
         let storage_root = trie.hash().unwrap();
-        let mut trie = store.open_state_trie(*EMPTY_TRIE_HASH).unwrap();
+        let mut trie = store.open_direct_state_trie(*EMPTY_TRIE_HASH).unwrap();
         trie.insert(
             address.0.to_vec(),
             AccountState {

crates/storage/store_db/in_memory.rs

@@ -15,7 +15,7 @@ use std::{
     collections::HashMap,
     fmt::Debug,
     path::Path,
-    sync::{Arc, Mutex, MutexGuard, RwLock},
+    sync::{Arc, Mutex, MutexGuard},
 };
 
 // NOTE: we use a different commit threshold than rocksdb since tests
@@ -39,7 +39,7 @@ pub struct StoreInner {
     // Maps transaction hashes to their blocks (height+hash) and index within the blocks.
     transaction_locations: HashMap<H256, Vec<(BlockNumber, BlockHash, Index)>>,
     receipts: HashMap<BlockHash, HashMap<Index, Receipt>>,
-    trie_cache: Arc<RwLock<TrieLayerCache>>,
+    trie_cache: Arc<TrieLayerCache>,
     // Contains account trie nodes
     state_trie_nodes: NodeMap,
     pending_blocks: HashMap<BlockHash, Block>,
@@ -91,31 +91,28 @@ impl StoreEngine for Store {
         let mut store = self.inner()?;
 
         // Store trie updates
-        {
-            let mut trie = store
-                .trie_cache
-                .write()
-                .map_err(|_| StoreError::LockError)?;
-            let parent = update_batch
-                .blocks
-                .first()
-                .ok_or(StoreError::UpdateBatchNoBlocks)?
-                .header
-                .parent_hash;
-
-            let pre_state_root = store
-                .headers
-                .get(&parent)
-                .map(|header| header.state_root)
-                .unwrap_or_default();
-
-            let last_state_root = update_batch
-                .blocks
-                .last()
-                .ok_or(StoreError::UpdateBatchNoBlocks)?
-                .header
-                .state_root;
+        let mut trie = TrieLayerCache::clone(&store.trie_cache);
+        let parent = update_batch
+            .blocks
+            .first()
+            .ok_or(StoreError::UpdateBatchNoBlocks)?
+            .header
+            .parent_hash;
+
+        let pre_state_root = store
+            .headers
+            .get(&parent)
+            .map(|header| header.state_root)
+            .unwrap_or_default();
+
+        let last_state_root = update_batch
+            .blocks
+            .last()
+            .ok_or(StoreError::UpdateBatchNoBlocks)?
+            .header
+            .state_root;
 
+        {
             let mut state_trie = store
                 .state_trie_nodes
                 .lock()
@@ -131,19 +128,21 @@ impl StoreEngine for Store {
                     }
                 }
             }
-            let key_values = update_batch
-                .storage_updates
-                .into_iter()
-                .flat_map(|(account_hash, nodes)| {
-                    nodes
-                        .into_iter()
-                        .map(move |(path, node)| (apply_prefix(Some(account_hash), path), node))
-                })
-                .chain(update_batch.account_updates)
-                .collect();
-            trie.put_batch(pre_state_root, last_state_root, key_values);
         }
 
+        let key_values = update_batch
+            .storage_updates
+            .into_iter()
+            .flat_map(|(account_hash, nodes)| {
+                nodes
+                    .into_iter()
+                    .map(move |(path, node)| (apply_prefix(Some(account_hash), path), node))
+            })
+            .chain(update_batch.account_updates)
+            .collect();
+        trie.put_batch(pre_state_root, last_state_root, key_values);
+        store.trie_cache = Arc::new(trie);
+
         for block in update_batch.blocks {
             // store block
             let number = block.header.number;