Merge branch 'snap_sync' into snap_sync_fix_ci

Author: gianbelinche

cmd/ethrex/initializers.rs

@@ -191,7 +191,7 @@ pub async fn init_network(
         signer,
         peer_table.clone(),
         store,
-        blockchain,
+        blockchain.clone(),
         get_client_version(),
         based_context,
     );
@@ -200,7 +200,10 @@ pub async fn init_network(
         .await
         .expect("Network starts");
 
-    tracker.spawn(ethrex_p2p::periodically_show_peer_stats());
+    tracker.spawn(ethrex_p2p::periodically_show_peer_stats(
+        blockchain,
+        peer_table.peers.clone(),
+    ));
 }
 
 #[cfg(feature = "dev")]

crates/networking/p2p/network.rs

@@ -3,21 +3,24 @@ use crate::{
         server::{DiscoveryServer, DiscoveryServerError},
         side_car::{DiscoverySideCar, DiscoverySideCarError},
     },
-    kademlia::Kademlia,
+    kademlia::{Kademlia, PeerData},
     metrics::METRICS,
     rlpx::{
         connection::server::{RLPxConnBroadcastSender, RLPxConnection},
         initiator::{RLPxInitiator, RLPxInitiatorError},
         l2::l2_connection::P2PBasedContext,
         message::Message,
+        p2p::SUPPORTED_SNAP_CAPABILITIES,
     },
     tx_broadcaster::{TxBroadcaster, TxBroadcasterError},
     types::{Node, NodeRecord},
 };
 use ethrex_blockchain::Blockchain;
+use ethrex_common::H256;
 use ethrex_storage::Store;
 use secp256k1::SecretKey;
 use std::{
+    collections::BTreeMap,
     io,
     net::SocketAddr,
     sync::Arc,
@@ -175,9 +178,20 @@ fn listener(tcp_addr: SocketAddr) -> Result<TcpListener, io::Error> {
     tcp_socket.listen(50)
 }
 
-pub async fn periodically_show_peer_stats() {
+pub async fn periodically_show_peer_stats(
+    blockchain: Arc<Blockchain>,
+    peers: Arc<Mutex<BTreeMap<H256, PeerData>>>,
+) {
+    periodically_show_peer_stats_during_syncing(blockchain).await;
+    periodically_show_peer_stats_after_sync(peers).await;
+}
+
+pub async fn periodically_show_peer_stats_during_syncing(blockchain: Arc<Blockchain>) {
     let start = std::time::Instant::now();
     loop {
+        if blockchain.is_synced() {
+            return;
+        }
         let metrics_enabled = *METRICS.enabled.lock().await;
         // Show the metrics only when these are enabled
         if !metrics_enabled {
@@ -424,6 +438,33 @@ bytecodes progress: {bytecodes_download_progress} (total: {bytecodes_to_download
     }
 }
 
+/// 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_after_sync(peers: Arc<Mutex<BTreeMap<H256, PeerData>>>) {
+    const INTERVAL_DURATION: tokio::time::Duration = tokio::time::Duration::from_secs(60);
+    let mut interval = tokio::time::interval(INTERVAL_DURATION);
+    loop {
+        // clone peers to keep the lock short
+        let peers: Vec<PeerData> = peers.lock().await.values().cloned().collect();
+        let active_peers = peers
+            .iter()
+            .filter(|peer| -> bool { peer.channels.as_ref().is_some() })
+            .count();
+        let snap_active_peers = peers
+            .iter()
+            .filter(|peer| -> bool {
+                peer.channels.as_ref().is_some()
+                    && SUPPORTED_SNAP_CAPABILITIES
+                        .iter()
+                        .any(|cap| peer.supported_capabilities.contains(cap))
+            })
+            .count();
+        info!(
+            "Snap Peers: {snap_active_peers} / Total Peers: {active_peers}"
+        );
+        interval.tick().await;
+    }
+}
+
 fn format_duration(duration: Duration) -> String {
     let total_seconds = duration.as_secs();
     let hours = total_seconds / 3600;

crates/networking/p2p/rlpx/connection/server.rs

@@ -10,7 +10,8 @@ use ethrex_common::{
     H256,
     types::{MempoolTransaction, Transaction},
 };
-use ethrex_storage::Store;
+use ethrex_storage::{Store, error::StoreError};
+use ethrex_trie::TrieError;
 use futures::{SinkExt as _, Stream, stream::SplitSink};
 use rand::random;
 use secp256k1::{PublicKey, SecretKey};
@@ -327,6 +328,19 @@ impl GenServer for RLPxConnection {
                         );
                         return CastResponse::Stop;
                     }
+                    RLPxError::StoreError(StoreError::Trie(TrieError::InconsistentTree)) => {
+                        if established_state.blockchain.is_synced() {
+                            log_peer_error(
+                                &established_state.node,
+                                &format!("Error handling cast message: {e}"),
+                            );
+                        } else {
+                            log_peer_debug(
+                                &established_state.node,
+                                &format!("Error handling cast message: {e}"),
+                            );
+                        }
+                    }
                     _ => {
                         log_peer_warn(
                             &established_state.node,