feat(l1): improve healing

crates/networking/p2p/discv4/server.rs

@@ -117,7 +117,7 @@ impl Discv4Server {
             debug!("Received {read} bytes from {from}");
 
             match Packet::decode(&buf[..read]) {
-                Err(e) => error!("Could not decode packet: {:?}", e),
+                Err(e) => debug!("Could not decode packet: {:?}", e),
                 Ok(packet) => {
                     let msg = packet.get_message();
                     let msg_name = msg.to_string();

crates/networking/p2p/net.rs

@@ -16,7 +16,7 @@ use tokio::{
     sync::Mutex,
 };
 use tokio_util::task::TaskTracker;
-use tracing::{error, info};
+use tracing::{debug, error, info};
 use types::Node;
 
 pub(crate) mod discv4;
@@ -139,7 +139,7 @@ async fn handle_peer_as_receiver(context: P2PContext, peer_addr: SocketAddr, str
             // TODO We should remove the peer from the table if connection failed
             // but currently it will make the tests fail
             // table.lock().await.replace_peer(node.node_id);
-            error!("Error creating tcp connection with peer at {peer_addr}: {e}")
+            debug!("Error creating tcp connection with peer at {peer_addr}: {e}")
         }
     }
 }
@@ -152,7 +152,7 @@ async fn handle_peer_as_initiator(context: P2PContext, node: Node) {
             // TODO We should remove the peer from the table if connection failed
             // but currently it will make the tests fail
             // table.lock().await.replace_peer(node.node_id);
-            error!("Error establishing tcp connection with peer at {addr}: {e}");
+            debug!("Error establishing tcp connection with peer at {addr}: {e}");
             return;
         }
     };
@@ -163,7 +163,7 @@ async fn handle_peer_as_initiator(context: P2PContext, node: Node) {
             // TODO We should remove the peer from the table if connection failed
             // but currently it will make the tests fail
             // table.lock().await.replace_peer(node.node_id);
-            error!("Error creating tcp connection with peer at {addr}: {e}")
+            debug!("Error creating tcp connection with peer at {addr}: {e}")
         }
     };
 }

crates/storage/store/engines/api.rs

@@ -7,7 +7,7 @@ use ethrex_core::types::{
 use std::{fmt::Debug, panic::RefUnwindSafe};
 
 use crate::error::StoreError;
-use ethrex_trie::Trie;
+use ethrex_trie::{Nibbles, Trie};
 
 pub trait StoreEngine: Debug + Send + Sync + RefUnwindSafe {
     /// Add block header
@@ -258,35 +258,34 @@ pub trait StoreEngine: Debug + Send + Sync + RefUnwindSafe {
     /// Gets the hash of the last header downloaded during a snap sync
     fn get_header_download_checkpoint(&self) -> Result<Option<BlockHash>, StoreError>;
 
-    /// Clears the hash of the last header downloaded during a snap sync
-    fn clear_header_download_checkpoint(&self) -> Result<(), StoreError>;
-
     /// Sets the current state root of the state trie being rebuilt during snap sync
     fn set_state_trie_root_checkpoint(&self, current_root: H256) -> Result<(), StoreError>;
 
     /// Gets the current state root of the state trie being rebuilt during snap sync
     fn get_state_trie_root_checkpoint(&self) -> Result<Option<H256>, StoreError>;
 
-    /// Clears the current state root of the state trie being rebuilt during snap sync
-    fn clear_state_trie_root_checkpoint(&self) -> Result<(), StoreError>;
-
     /// Sets the last key fetched from the state trie being fetched during snap sync
     fn set_state_trie_key_checkpoint(&self, last_key: H256) -> Result<(), StoreError>;
 
     /// Gets the last key fetched from the state trie being fetched during snap sync
     fn get_state_trie_key_checkpoint(&self) -> Result<Option<H256>, StoreError>;
 
-    /// Clears the last key fetched from the state trie being fetched during snap sync
-    fn clear_state_trie_key_checkpoint(&self) -> Result<(), StoreError>;
+    /// Sets the storage trie paths in need of healing, grouped by hashed address
+    fn set_storage_heal_paths(&self, accounts: Vec<(H256, Vec<Nibbles>)>)
+        -> Result<(), StoreError>;
+
+    /// Gets the storage trie paths in need of healing, grouped by hashed address
+    #[allow(clippy::type_complexity)]
+    fn get_storage_heal_paths(&self) -> Result<Option<Vec<(H256, Vec<Nibbles>)>>, StoreError>;
 
-    /// Sets the list of account hashes whose storage needs healing
-    fn set_pending_storage_heal_accounts(&self, accounts: Vec<H256>) -> Result<(), StoreError>;
+    /// Sets the state trie paths in need of healing
+    fn set_state_heal_paths(&self, paths: Vec<Nibbles>) -> Result<(), StoreError>;
 
-    /// Gets the list of account hashes whos storage needs healing
-    fn get_pending_storage_heal_accounts(&self) -> Result<Option<Vec<H256>>, StoreError>;
+    /// Gets the state trie paths in need of healing
+    fn get_state_heal_paths(&self) -> Result<Option<Vec<Nibbles>>, StoreError>;
 
-    /// Clears the list of account hashes whose storage needs healing
-    fn clear_pending_storage_heal_accounts(&self) -> Result<(), StoreError>;
+    /// Clears all checkpoint data created during the last snap sync
+    fn clear_snap_state(&self) -> Result<(), StoreError>;
 
     fn is_synced(&self) -> Result<bool, StoreError>;
 

crates/storage/store/engines/in_memory.rs

@@ -4,7 +4,7 @@ use ethereum_types::{H256, U256};
 use ethrex_core::types::{
     BlobsBundle, Block, BlockBody, BlockHash, BlockHeader, BlockNumber, ChainConfig, Index, Receipt,
 };
-use ethrex_trie::{InMemoryTrieDB, Trie};
+use ethrex_trie::{InMemoryTrieDB, Nibbles, Trie};
 use std::{
     collections::HashMap,
     fmt::Debug,
@@ -65,7 +65,9 @@ pub struct SnapState {
     /// Last downloaded key of the latest State Trie
     state_trie_key_checkpoint: Option<H256>,
     /// Accounts which storage needs healing
-    pending_storage_heal_accounts: Option<Vec<H256>>,
+    storage_heal_paths: Option<Vec<(H256, Vec<Nibbles>)>>,
+    /// State trie Paths in need of healing
+    state_heal_paths: Option<Vec<Nibbles>>,
 }
 
 impl Store {
@@ -449,11 +451,6 @@ impl StoreEngine for Store {
         Ok(self.inner().snap_state.header_download_checkpoint)
     }
 
-    fn clear_header_download_checkpoint(&self) -> Result<(), StoreError> {
-        self.inner().snap_state.header_download_checkpoint = None;
-        Ok(())
-    }
-
     fn set_state_trie_root_checkpoint(&self, current_root: H256) -> Result<(), StoreError> {
         self.inner().snap_state.state_trie_root_checkpoint = Some(current_root);
         Ok(())
@@ -463,11 +460,6 @@ impl StoreEngine for Store {
         Ok(self.inner().snap_state.state_trie_root_checkpoint)
     }
 
-    fn clear_state_trie_root_checkpoint(&self) -> Result<(), StoreError> {
-        self.inner().snap_state.state_trie_root_checkpoint = None;
-        Ok(())
-    }
-
     fn set_state_trie_key_checkpoint(&self, last_key: H256) -> Result<(), StoreError> {
         self.inner().snap_state.state_trie_key_checkpoint = Some(last_key);
         Ok(())
@@ -477,26 +469,20 @@ impl StoreEngine for Store {
         Ok(self.inner().snap_state.state_trie_key_checkpoint)
     }
 
-    fn clear_state_trie_key_checkpoint(&self) -> Result<(), StoreError> {
-        self.inner().snap_state.state_trie_key_checkpoint = None;
-        Ok(())
-    }
-
-    fn set_pending_storage_heal_accounts(&self, accounts: Vec<H256>) -> Result<(), StoreError> {
-        self.inner().snap_state.pending_storage_heal_accounts = Some(accounts);
+    fn set_storage_heal_paths(
+        &self,
+        accounts: Vec<(H256, Vec<Nibbles>)>,
+    ) -> Result<(), StoreError> {
+        self.inner().snap_state.storage_heal_paths = Some(accounts);
         Ok(())
     }
 
-    fn get_pending_storage_heal_accounts(&self) -> Result<Option<Vec<H256>>, StoreError> {
-        Ok(self
-            .inner()
-            .snap_state
-            .pending_storage_heal_accounts
-            .clone())
+    fn get_storage_heal_paths(&self) -> Result<Option<Vec<(H256, Vec<Nibbles>)>>, StoreError> {
+        Ok(self.inner().snap_state.storage_heal_paths.clone())
     }
 
-    fn clear_pending_storage_heal_accounts(&self) -> Result<(), StoreError> {
-        self.inner().snap_state.pending_storage_heal_accounts = None;
+    fn clear_snap_state(&self) -> Result<(), StoreError> {
+        self.inner().snap_state = Default::default();
         Ok(())
     }
 
@@ -508,6 +494,15 @@ impl StoreEngine for Store {
         self.inner().chain_data.is_synced = status;
         Ok(())
     }
+
+    fn set_state_heal_paths(&self, paths: Vec<Nibbles>) -> Result<(), StoreError> {
+        self.inner().snap_state.state_heal_paths = Some(paths);
+        Ok(())
+    }
+
+    fn get_state_heal_paths(&self) -> Result<Option<Vec<Nibbles>>, StoreError> {
+        Ok(self.inner().snap_state.state_heal_paths.clone())
+    }
 }
 
 impl Debug for Store {

crates/storage/store/engines/libmdbx.rs

@@ -14,7 +14,7 @@ use ethrex_core::types::{
 };
 use ethrex_rlp::decode::RLPDecode;
 use ethrex_rlp::encode::RLPEncode;
-use ethrex_trie::{LibmdbxDupsortTrieDB, LibmdbxTrieDB, Trie};
+use ethrex_trie::{LibmdbxDupsortTrieDB, LibmdbxTrieDB, Nibbles, Trie};
 use libmdbx::orm::{Decodable, Encodable, Table};
 use libmdbx::{
     dupsort,
@@ -72,17 +72,6 @@ impl Store {
         txn.get::<T>(key).map_err(StoreError::LibmdbxError)
     }
 
-    // Helper method to remove a value from a libmdbx table
-    fn delete<T: Table>(&self, key: T::Key) -> Result<(), StoreError> {
-        let txn = self
-            .db
-            .begin_readwrite()
-            .map_err(StoreError::LibmdbxError)?;
-        txn.delete::<T>(key, None)
-            .map_err(StoreError::LibmdbxError)?;
-        txn.commit().map_err(StoreError::LibmdbxError)
-    }
-
     fn get_block_hash_by_block_number(
         &self,
         number: BlockNumber,
@@ -540,10 +529,6 @@ impl StoreEngine for Store {
             .map_err(StoreError::RLPDecode)
     }
 
-    fn clear_header_download_checkpoint(&self) -> Result<(), StoreError> {
-        self.delete::<SnapState>(SnapStateIndex::HeaderDownloadCheckpoint)
-    }
-
     fn set_state_trie_root_checkpoint(&self, current_root: H256) -> Result<(), StoreError> {
         self.write::<SnapState>(
             SnapStateIndex::StateTrieRootCheckpoint,
@@ -558,10 +543,6 @@ impl StoreEngine for Store {
             .map_err(StoreError::RLPDecode)
     }
 
-    fn clear_state_trie_root_checkpoint(&self) -> Result<(), StoreError> {
-        self.delete::<SnapState>(SnapStateIndex::StateTrieRootCheckpoint)
-    }
-
     fn set_state_trie_key_checkpoint(&self, last_key: H256) -> Result<(), StoreError> {
         self.write::<SnapState>(
             SnapStateIndex::StateTrieRootCheckpoint,
@@ -576,37 +557,51 @@ impl StoreEngine for Store {
             .map_err(StoreError::RLPDecode)
     }
 
-    fn clear_state_trie_key_checkpoint(&self) -> Result<(), StoreError> {
-        self.delete::<SnapState>(SnapStateIndex::StateTrieRootCheckpoint)
-    }
-
-    fn set_pending_storage_heal_accounts(&self, accounts: Vec<H256>) -> Result<(), StoreError> {
-        self.write::<SnapState>(
-            SnapStateIndex::PendingStorageHealAccounts,
-            accounts.encode_to_vec(),
-        )
+    fn set_storage_heal_paths(
+        &self,
+        accounts: Vec<(H256, Vec<Nibbles>)>,
+    ) -> Result<(), StoreError> {
+        self.write::<SnapState>(SnapStateIndex::StorageHealPaths, accounts.encode_to_vec())
     }
 
-    fn get_pending_storage_heal_accounts(&self) -> Result<Option<Vec<H256>>, StoreError> {
-        self.read::<SnapState>(SnapStateIndex::PendingStorageHealAccounts)?
-            .map(|ref h| <Vec<H256>>::decode(h))
+    fn get_storage_heal_paths(&self) -> Result<Option<Vec<(H256, Vec<Nibbles>)>>, StoreError> {
+        self.read::<SnapState>(SnapStateIndex::StorageHealPaths)?
+            .map(|ref h| <Vec<(H256, Vec<Nibbles>)>>::decode(h))
             .transpose()
             .map_err(StoreError::RLPDecode)
     }
 
-    fn clear_pending_storage_heal_accounts(&self) -> Result<(), StoreError> {
-        self.delete::<SnapState>(SnapStateIndex::PendingStorageHealAccounts)
-    }
-
     fn is_synced(&self) -> Result<bool, StoreError> {
         match self.read::<ChainData>(ChainDataIndex::IsSynced)? {
             None => Err(StoreError::Custom("Sync status not found".to_string())),
             Some(ref rlp) => RLPDecode::decode(rlp).map_err(|_| StoreError::DecodeError),
         }
     }
+
     fn update_sync_status(&self, status: bool) -> Result<(), StoreError> {
         self.write::<ChainData>(ChainDataIndex::IsSynced, status.encode_to_vec())
     }
+
+    fn set_state_heal_paths(&self, paths: Vec<Nibbles>) -> Result<(), StoreError> {
+        self.write::<SnapState>(SnapStateIndex::StateHealPaths, paths.encode_to_vec())
+    }
+
+    fn get_state_heal_paths(&self) -> Result<Option<Vec<Nibbles>>, StoreError> {
+        self.read::<SnapState>(SnapStateIndex::StateHealPaths)?
+            .map(|ref h| <Vec<Nibbles>>::decode(h))
+            .transpose()
+            .map_err(StoreError::RLPDecode)
+    }
+
+    fn clear_snap_state(&self) -> Result<(), StoreError> {
+        let txn = self
+            .db
+            .begin_readwrite()
+            .map_err(StoreError::LibmdbxError)?;
+        txn.clear_table::<SnapState>()
+            .map_err(StoreError::LibmdbxError)?;
+        txn.commit().map_err(StoreError::LibmdbxError)
+    }
 }
 
 impl Debug for Store {