feat(l1): snap sync overhaul

crates/networking/p2p/net.rs

@@ -596,7 +596,7 @@ async fn peers_revalidation(
             table.update_peer_ping_with_revalidation(peer.node.node_id, ping_hash);
             previously_pinged_peers.insert(peer.node.node_id);
 
-            debug!("Pinging peer {:?} to re-validate!", peer.node.node_id);
+            debug!("Pinging peer {} to re-validate!", peer.node.node_id);
         }
 
         debug!("Peer revalidation finished");
@@ -1029,7 +1029,7 @@ pub fn node_id_from_signing_key(signer: &SigningKey) -> H512 {
 
 /// Shows the amount of connected peers, active peers, and peers suitable for snap sync on a set interval
 pub async fn periodically_show_peer_stats(peer_table: Arc<Mutex<KademliaTable>>) {
-    const INTERVAL_DURATION: tokio::time::Duration = tokio::time::Duration::from_secs(60);
+    const INTERVAL_DURATION: tokio::time::Duration = tokio::time::Duration::from_secs(120);
     let mut interval = tokio::time::interval(INTERVAL_DURATION);
     loop {
         peer_table.lock().await.show_peer_stats();

crates/networking/p2p/peer_channels.rs

@@ -29,7 +29,7 @@ use crate::{
 
 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 MAX_RESPONSE_BYTES: u64 = 2 * 1024 * 1024;
 pub const HASH_MAX: H256 = H256([0xFF; 32]);
 
 #[derive(Debug, Clone)]
@@ -260,7 +260,7 @@ impl PeerChannels {
     /// 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 true if the last account's 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
@@ -447,4 +447,68 @@ impl PeerChannels {
             })
             .flatten()
     }
+
+    /// Requests a single storage range for an accouns given its hashed address and storage root, and the root of its state trie
+    /// This is a simplified version of `request_storage_range` meant to be used for large tries that require their own single requests
+    /// account_hashes & storage_roots must have the same length
+    /// storage_root must not be an empty trie hash, we will treat empty ranges as invalid responses
+    /// Returns true if the account's storage was not completely fetched by the request
+    /// Returns the list of hashed storage keys and values for the 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_range(
+        &self,
+        state_root: H256,
+        storage_root: H256,
+        account_hash: H256,
+        start: H256,
+    ) -> Option<(Vec<H256>, Vec<U256>, bool)> {
+        let request_id = rand::random();
+        let request = RLPxMessage::GetStorageRanges(GetStorageRanges {
+            id: request_id,
+            root_hash: state_root,
+            account_hashes: vec![account_hash],
+            starting_hash: start,
+            limit_hash: HASH_MAX,
+            response_bytes: MAX_RESPONSE_BYTES,
+        });
+        let mut receiver = self.receiver.lock().await;
+        self.sender.send(request).await.ok()?;
+        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() != 1 {
+            return None;
+        }
+        // Unzip & validate response
+        let proof = encodable_to_proof(&proof);
+        let (storage_keys, storage_values): (Vec<H256>, Vec<U256>) = slots
+            .remove(0)
+            .into_iter()
+            .map(|slot| (slot.hash, slot.data))
+            .unzip();
+        let encoded_values = storage_values
+            .iter()
+            .map(|val| val.encode_to_vec())
+            .collect::<Vec<_>>();
+        // Verify storage range
+        let should_continue =
+            verify_range(storage_root, &start, &storage_keys, &encoded_values, &proof).ok()?;
+        Some((storage_keys, storage_values, should_continue))
+    }
 }

crates/networking/p2p/rlpx/connection.rs

@@ -51,7 +51,7 @@ use tokio::{
 };
 use tokio_stream::StreamExt;
 use tokio_util::codec::Framed;
-use tracing::{debug, error};
+use tracing::{debug, error, warn};
 const CAP_P2P: (Capability, u8) = (Capability::P2p, 5);
 const CAP_ETH: (Capability, u8) = (Capability::Eth, 68);
 const CAP_SNAP: (Capability, u8) = (Capability::Snap, 1);
@@ -614,6 +614,10 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
         let msg_size = u16::from_be_bytes(ack_data) as usize;
 
         // Read the rest of the message
+        // Guard unwrap
+        if buf.len() < msg_size + 2 {
+            return Err(RLPxError::CryptographyError(String::from("bad buf size")));
+        }
         self.framed
             .get_mut()
             .read_exact(&mut buf[2..msg_size + 2])

crates/networking/p2p/rlpx/handshake.rs

@@ -116,7 +116,9 @@ fn decrypt_message(
 
     // Verify the MAC.
     let expected_d = sha256_hmac(&mac_key, &[iv, c], size_data);
-    assert_eq!(d, expected_d);
+    if d != expected_d {
+        return Err(RLPxError::HandshakeError(String::from("Invalid MAC")));
+    }
 
     // Decrypt the message with the AES key.
     let mut stream_cipher = Aes128Ctr64BE::new_from_slices(aes_key, iv)?;