feat(l1): snap sync

cmd/ethrex/ethrex.rs

@@ -6,7 +6,10 @@ use ethrex_core::{
     H256,
 };
 use ethrex_net::{
-    bootnode::BootNode, node_id_from_signing_key, peer_table, sync::SyncManager, types::Node,
+    bootnode::BootNode,
+    node_id_from_signing_key, peer_table,
+    sync::{SyncManager, SyncMode},
+    types::Node,
 };
 use ethrex_rlp::decode::RLPDecode;
 use ethrex_storage::{EngineType, Store};
@@ -115,10 +118,7 @@ async fn main() {
         .get_one::<String>("datadir")
         .map_or(set_datadir(DEFAULT_DATADIR), |datadir| set_datadir(datadir));
 
-    let snap_sync = is_snap_sync(&matches);
-    if snap_sync {
-        info!("snap-sync not available, defaulting to full-sync");
-    }
+    let sync_mode = sync_mode(&matches);
 
     cfg_if::cfg_if! {
         if #[cfg(feature = "redb")] {
@@ -187,7 +187,7 @@ async fn main() {
     // Create Kademlia Table here so we can access it from rpc server (for syncing)
     let peer_table = peer_table(signer.clone());
     // Create SyncManager
-    let syncer = SyncManager::new(peer_table.clone(), snap_sync);
+    let syncer = SyncManager::new(peer_table.clone(), sync_mode);
 
     // TODO: Check every module starts properly.
     let tracker = TaskTracker::new();
@@ -299,16 +299,16 @@ fn parse_socket_addr(addr: &str, port: &str) -> io::Result<SocketAddr> {
         ))
 }
 
-fn is_snap_sync(matches: &clap::ArgMatches) -> bool {
+fn sync_mode(matches: &clap::ArgMatches) -> SyncMode {
     let syncmode = matches.get_one::<String>("syncmode");
     if let Some(syncmode) = syncmode {
         match &**syncmode {
-            "full" => false,
-            "snap" => true,
+            "full" => SyncMode::Full,
+            "snap" => SyncMode::Snap,
             other => panic!("Invalid syncmode {other} expected either snap or full"),
         }
     } else {
-        true
+        SyncMode::Snap
     }
 }
 

crates/common/types/blobs_bundle.rs

@@ -226,7 +226,6 @@ pub enum BlobsBundleError {
 }
 
 #[cfg(test)]
-
 mod tests {
     use super::*;
     use crate::{

crates/networking/docs/Sync.md

@@ -0,0 +1,24 @@
+# Syncing
+
+## Snap Sync
+
+A snap sync cycle begins by fetching all the block headers (via p2p) between the current head (latest canonical block) and the sync head (block hash sent by a forkChoiceUpdate).
+The next two steps are performed in parallel:
+On one side, blocks and receipts for all fetched headers are fetched via p2p and stored.
+
+On the other side, the state is reconstructed via p2p snap requests. Our current implementation of this works as follows:
+We will spawn two processes, the `bytecode_fetcher` which will remain active and process bytecode requests in batches by requesting bytecode batches from peers and storing them, and the `fetch_snap_state` process, which will iterate over the fetched headers and rebuild the block's state via `rebuild_state_trie`.
+
+`rebuild_state_trie` will spawn a `storage_fetcher` process (which works similarly to the `bytecode_fetcher` and is kept alive for the duration of the rebuild process), it will open a new state trie and will fetch the block's accounts in batches and for each account it will: send the account's code hash to the `bytecode_fetcher` (if not empty), send the account's address and storage root to the `storage_fetcher` (if not empty), and add the account to the state trie. Once all accounts are processed, the final state root will be checked and committed.
+
+(Not implemented yet) When `fetch_snap_state` runs out of available state (aka, the state we need to fetch is older than 128 blocks and peers don't provide it), it will begin the `state_healing` process.
+This diagram illustrates the process described above:
+
+![snap_sync](/crates/networking/docs/diagrams/snap_sync.jpg)
+
+The `bytecode_fetcher` has its own channel where it receives code hashes from active `rebuild_state_trie` processes. Once a code hash is received, it is added to a pending queue. When the queue has enough messages for a full batch it will request a batch of bytecodes via snap p2p and store them. If a bytecode could not be fetched by the request (aka, we reached the response limit) it is added back to the pending queue. After the whole state is synced `fetch_snap_state` will send an empty list to the `bytecode_fetcher` to signal the end of the requests so it can request the last (incomplete) bytecode batch and end gracefully.
+This diagram illustrates the process described above:
+
+![snap_sync](/crates/networking/docs/diagrams/bytecode_fetcher.jpg)
+
+The `storage_fetcher` works almost alike, but one will be spawned for each `rebuild_state_trie` process as we can't fetch storages from different blocks in the same request.

crates/networking/p2p/peer_channels.rs

@@ -1,20 +1,31 @@
 use std::{sync::Arc, time::Duration};
 
+use bytes::Bytes;
 use ethrex_core::{
-    types::{BlockBody, BlockHeader},
-    H256,
+    types::{AccountState, BlockBody, BlockHeader},
+    H256, U256,
 };
+use ethrex_rlp::encode::RLPEncode;
+use ethrex_trie::verify_range;
 use tokio::sync::{mpsc, Mutex};
 
 use crate::{
-    rlpx::eth::blocks::{
-        BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders, BLOCK_HEADER_LIMIT,
+    rlpx::{
+        eth::blocks::{
+            BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders, BLOCK_HEADER_LIMIT,
+        },
+        snap::{
+            AccountRange, ByteCodes, GetAccountRange, GetByteCodes, GetStorageRanges, StorageRanges,
+        },
     },
+    snap::encodable_to_proof,
     RLPxMessage,
 };
 
 pub const PEER_REPLY_TIMOUT: Duration = Duration::from_secs(45);
 pub const MAX_MESSAGES_IN_PEER_CHANNEL: usize = 25;
+pub const MAX_RESPONSE_BYTES: u64 = 512 * 1024;
+pub const HASH_MAX: H256 = H256([0xFF; 32]);
 
 #[derive(Debug, Clone)]
 /// Holds the respective sender and receiver ends of the communication channels bewteen the peer data and its active connection
@@ -41,8 +52,8 @@ impl PeerChannels {
         )
     }
 
-    /// Requests block headers from the peer
-    /// Returns the response message or None if:
+    /// Requests block headers from the peer, starting from the `start` block hash towards newer blocks
+    /// Returns the block headers or None if:
     /// - There are no available peers (the node just started up or was rejected by all other nodes)
     /// - The response timed out
     /// - The response was empty or not valid
@@ -76,8 +87,8 @@ impl PeerChannels {
         (!block_headers.is_empty()).then_some(block_headers)
     }
 
-    /// Requests block headers from the peer
-    /// Returns the response message or None if:
+    /// Requests block bodies from the peer given their block hashes
+    /// Returns the block bodies or None if:
     /// - There are no available peers (the node just started up or was rejected by all other nodes)
     /// - The response timed out
     /// - The response was empty or not valid
@@ -109,4 +120,202 @@ impl PeerChannels {
         // Check that the response is not empty and does not contain more bodies than the ones requested
         (!block_bodies.is_empty() && block_bodies.len() <= block_hashes_len).then_some(block_bodies)
     }
+
+    /// Requests an account range from the peer given the state trie's root and the starting hash (the limit hash will be the maximum value of H256)
+    /// Will also return a boolean indicating if there is more state to be fetched towards the right of the trie
+    /// Returns the response message or None if:
+    /// - There are no available peers (the node just started up or was rejected by all other nodes)
+    /// - The response timed out
+    /// - The response was not valid
+    pub async fn request_account_range(
+        &self,
+        state_root: H256,
+        start: H256,
+    ) -> Option<(Vec<H256>, Vec<AccountState>, bool)> {
+        let request_id = rand::random();
+        let request = RLPxMessage::GetAccountRange(GetAccountRange {
+            id: request_id,
+            root_hash: state_root,
+            starting_hash: start,
+            limit_hash: HASH_MAX,
+            response_bytes: MAX_RESPONSE_BYTES,
+        });
+        self.sender.send(request).await.ok()?;
+        let mut receiver = self.receiver.lock().await;
+        let (accounts, proof) = tokio::time::timeout(PEER_REPLY_TIMOUT, async move {
+            loop {
+                match receiver.recv().await {
+                    Some(RLPxMessage::AccountRange(AccountRange {
+                        id,
+                        accounts,
+                        proof,
+                    })) if id == request_id => return Some((accounts, proof)),
+                    // Ignore replies that don't match the expected id (such as late responses)
+                    Some(_) => continue,
+                    None => return None,
+                }
+            }
+        })
+        .await
+        .ok()??;
+        // Unzip & validate response
+        let proof = encodable_to_proof(&proof);
+        let (account_hashes, accounts): (Vec<_>, Vec<_>) = accounts
+            .into_iter()
+            .map(|unit| (unit.hash, AccountState::from(unit.account)))
+            .unzip();
+        let encoded_accounts = accounts
+            .iter()
+            .map(|acc| acc.encode_to_vec())
+            .collect::<Vec<_>>();
+        let should_continue = verify_range(
+            state_root,
+            &start,
+            &account_hashes,
+            &encoded_accounts,
+            &proof,
+        )
+        .ok()?;
+        Some((account_hashes, accounts, should_continue))
+    }
+
+    /// Requests bytecodes for the given code hashes
+    /// Returns the bytecodes or None if:
+    /// - There are no available peers (the node just started up or was rejected by all other nodes)
+    /// - The response timed out
+    /// - The response was empty or not valid
+    pub async fn request_bytecodes(&self, hashes: Vec<H256>) -> Option<Vec<Bytes>> {
+        let request_id = rand::random();
+        let hashes_len = hashes.len();
+        let request = RLPxMessage::GetByteCodes(GetByteCodes {
+            id: request_id,
+            hashes,
+            bytes: MAX_RESPONSE_BYTES,
+        });
+        self.sender.send(request).await.ok()?;
+        let mut receiver = self.receiver.lock().await;
+        let codes = tokio::time::timeout(PEER_REPLY_TIMOUT, async move {
+            loop {
+                match receiver.recv().await {
+                    Some(RLPxMessage::ByteCodes(ByteCodes { id, codes })) if id == request_id => {
+                        return Some(codes)
+                    }
+                    // Ignore replies that don't match the expected id (such as late responses)
+                    Some(_) => continue,
+                    None => return None,
+                }
+            }
+        })
+        .await
+        .ok()??;
+        (!codes.is_empty() && codes.len() <= hashes_len).then_some(codes)
+    }
+
+    /// Requests storage ranges for accounts given their hashed address and storage roots, and the root of their state trie
+    /// account_hashes & storage_roots must have the same length
+    /// storage_roots must not contain empty trie hashes, we will treat empty ranges as invalid responses
+    /// Returns true if the last accoun't storage was not completely fetched by the request
+    /// Returns the list of hashed storage keys and values for each account's storage or None if:
+    /// - There are no available peers (the node just started up or was rejected by all other nodes)
+    /// - The response timed out
+    /// - The response was empty or not valid
+    pub async fn request_storage_ranges(
+        &self,
+        state_root: H256,
+        mut storage_roots: Vec<H256>,
+        account_hashes: Vec<H256>,
+        start: H256,
+    ) -> Option<(Vec<Vec<H256>>, Vec<Vec<U256>>, bool)> {
+        let request_id = rand::random();
+        let request = RLPxMessage::GetStorageRanges(GetStorageRanges {
+            id: request_id,
+            root_hash: state_root,
+            account_hashes,
+            starting_hash: start,
+            limit_hash: HASH_MAX,
+            response_bytes: MAX_RESPONSE_BYTES,
+        });
+        self.sender.send(request).await.ok()?;
+        let mut receiver = self.receiver.lock().await;
+        let (mut slots, proof) = tokio::time::timeout(PEER_REPLY_TIMOUT, async move {
+            loop {
+                match receiver.recv().await {
+                    Some(RLPxMessage::StorageRanges(StorageRanges { id, slots, proof }))
+                        if id == request_id =>
+                    {
+                        return Some((slots, proof))
+                    }
+                    // Ignore replies that don't match the expected id (such as late responses)
+                    Some(_) => continue,
+                    None => return None,
+                }
+            }
+        })
+        .await
+        .ok()??;
+        // Check we got a reasonable amount of storage ranges
+        if slots.len() > storage_roots.len() || slots.is_empty() {
+            return None;
+        }
+        // Unzip & validate response
+        let mut proof = encodable_to_proof(&proof);
+        let mut storage_keys = vec![];
+        let mut storage_values = vec![];
+        let mut should_continue = false;
+        // Validate each storage range
+        while !slots.is_empty() {
+            let (hahsed_keys, values): (Vec<_>, Vec<_>) = slots
+                .remove(0)
+                .into_iter()
+                .map(|slot| (slot.hash, slot.data))
+                .unzip();
+            // We won't accept empty storage ranges
+            if hahsed_keys.is_empty() {
+                return None;
+            }
+            let encoded_values = values
+                .iter()
+                .map(|val| val.encode_to_vec())
+                .collect::<Vec<_>>();
+            let storage_root = storage_roots.remove(0);
+
+            // We have 3 cases (as we won't accept empty storage ranges):
+            // - The range has only 1 element (with key matching the start): We expect one edge proof
+            // - The range has the full storage: We expect no proofs
+            // - The range is not the full storage (last range): We expect 2 edge proofs
+            if hahsed_keys.len() == 1 && hahsed_keys[0] == start {
+                if proof.is_empty() {
+                    return None;
+                };
+                let first_proof = vec![proof.remove(0)];
+                verify_range(
+                    storage_root,
+                    &start,
+                    &hahsed_keys,
+                    &encoded_values,
+                    &first_proof,
+                )
+                .ok()?;
+            }
+            // Last element with two edge proofs
+            if slots.is_empty() && proof.len() >= 2 {
+                let last_proof = vec![proof.remove(0), proof.remove(0)];
+                should_continue = verify_range(
+                    storage_root,
+                    &start,
+                    &hahsed_keys,
+                    &encoded_values,
+                    &last_proof,
+                )
+                .ok()?;
+            } else {
+                // Full range (no proofs)
+                verify_range(storage_root, &start, &hahsed_keys, &encoded_values, &[]).ok()?;
+            }
+
+            storage_keys.push(hahsed_keys);
+            storage_values.push(values);
+        }
+        Some((storage_keys, storage_values, should_continue))
+    }
 }

crates/networking/p2p/rlpx/error.rs

@@ -41,12 +41,20 @@ pub(crate) enum RLPxError {
     BroadcastError(String),
     #[error(transparent)]
     RecvError(#[from] RecvError),
-    #[error(transparent)]
-    Send(#[from] tokio::sync::mpsc::error::SendError<Message>),
+    #[error("Failed to send msg: {0}")]
+    SendMessage(String),
     #[error("Error when inserting transaction in the mempool: {0}")]
     MempoolError(#[from] MempoolError),
 }
 
+// tokio::sync::mpsc::error::SendError<Message> is too large to be part of the RLPxError enum directly
+// so we will instead save the error's display message
+impl From<tokio::sync::mpsc::error::SendError<Message>> for RLPxError {
+    fn from(value: tokio::sync::mpsc::error::SendError<Message>) -> Self {
+        Self::SendMessage(value.to_string())
+    }
+}
+
 // Grouping all cryptographic related errors in a single CryptographicError variant
 // We can improve this to individual errors if required
 impl From<k256::ecdsa::Error> for RLPxError {

crates/networking/p2p/rlpx/snap.rs

@@ -169,8 +169,14 @@ impl RLPxMessage for GetStorageRanges {
         let (id, decoder) = decoder.decode_field("request-id")?;
         let (root_hash, decoder) = decoder.decode_field("rootHash")?;
         let (account_hashes, decoder) = decoder.decode_field("accountHashes")?;
-        let (starting_hash, decoder) = decoder.decode_field("startingHash")?;
-        let (limit_hash, decoder) = decoder.decode_field("limitHash")?;
+        let (starting_hash, decoder): (Bytes, _) = decoder.decode_field("startingHash")?;
+        let starting_hash = (!starting_hash.is_empty())
+            .then(|| H256::from_slice(&starting_hash))
+            .unwrap_or_default();
+        let (limit_hash, decoder): (Bytes, _) = decoder.decode_field("limitHash")?;
+        let limit_hash = (!limit_hash.is_empty())
+            .then(|| H256::from_slice(&limit_hash))
+            .unwrap_or(H256([0xFF; 32]));
         let (response_bytes, decoder) = decoder.decode_field("responseBytes")?;
         decoder.finish()?;