feat(l1): add rpc endpoint admin_peers (#2732)

**Motivation**
Support rpc endpoint `admin_peers`
<!-- Why does this pull request exist? What are its goals? -->

**Description**
* Add rpc endpoint `admin_peers`
* Track inbound connections 
* Store peer node version when starting a connection
* Add `peer_handler: PeerHandler` field to `RpcContext` so we can access
peers from the rpc
* (Misc) `Syncer` & `SyncManager` now receive a `PeerHandler` upon
creation instead of a `KademliaTable`
* (Misc) Fix common typo across the project
<!-- A clear and concise general description of the changes this PR
introduces -->

Data missing compared to geth implementation:
* The local address of each connection
* Whether a connection is trusted, static (we have no notion of this
yet)
<!-- Link to issues: Resolves #111, Resolves #222 -->

Closes #2671

Author: fmoletta

cmd/ef_tests/blockchain/types.rs

@@ -53,7 +53,7 @@ pub struct BlobSchedule {
 }
 
 impl TestUnit {
-    /// Checks wether a test has a block where the inner_block is none.
+    /// Checks whether a test has a block where the inner_block is none.
     /// These tests only check for failures in decoding invalid rlp and expect an exception.
     pub fn is_rlp_only_test(&self) -> bool {
         let mut is_rlp_only = false;

cmd/ethrex/initializers.rs

@@ -10,6 +10,7 @@ use ethrex_blockchain::Blockchain;
 use ethrex_p2p::{
     kademlia::KademliaTable,
     network::{public_key_from_signing_key, P2PContext},
+    peer_handler::PeerHandler,
     sync_manager::SyncManager,
     types::{Node, NodeRecord},
 };
@@ -134,9 +135,11 @@ pub async fn init_rpc_api(
     tracker: TaskTracker,
     #[cfg(feature = "l2")] rollup_store: StoreRollup,
 ) {
+    let peer_handler = PeerHandler::new(peer_table);
+
     // Create SyncManager
     let syncer = SyncManager::new(
-        peer_table.clone(),
+        peer_handler.clone(),
         opts.syncmode.clone(),
         cancel_token,
         blockchain.clone(),
@@ -153,6 +156,7 @@ pub async fn init_rpc_api(
         local_p2p_node,
         local_node_record,
         syncer,
+        peer_handler,
         get_client_version(),
         #[cfg(feature = "based")]
         get_gateway_http_client(&l2_opts.based_opts),

crates/blockchain/payload.rs

@@ -401,7 +401,7 @@ impl Blockchain {
             // TODO: maybe fetch hash too when filtering mempool so we don't have to compute it here (we can do this in the same refactor as adding timestamp)
             let tx_hash = head_tx.tx.compute_hash();
 
-            // Check wether the tx is replay-protected
+            // Check whether the tx is replay-protected
             if head_tx.tx.protected() && !chain_config.is_eip155_activated(context.block_number()) {
                 // Ignore replay protected tx & all txs from the sender
                 // Pull transaction from the mempool

crates/common/trie/trie.rs

@@ -715,7 +715,7 @@ mod test {
 
         #[test]
         // The previous test needs to sort the input values in order to get rid of duplicate entries, leading to ordered insertions
-        // This check has a fixed way of determining wether a value should be removed but doesn't require ordered insertions
+        // This check has a fixed way of determining whether a value should be removed but doesn't require ordered insertions
         fn proptest_get_insert_with_removals_unsorted(data in btree_set(vec(any::<u8>(), 5..100), 1..100)) {
             let mut trie = Trie::new_temp();
             // Remove all values that have an odd first value
@@ -787,7 +787,7 @@ mod test {
 
         #[test]
         // The previous test needs to sort the input values in order to get rid of duplicate entries, leading to ordered insertions
-        // This check has a fixed way of determining wether a value should be removed but doesn't require ordered insertions
+        // This check has a fixed way of determining whether a value should be removed but doesn't require ordered insertions
         fn proptest_compare_hash_with_removals_unsorted(data in btree_set(vec(any::<u8>(), 5..100), 1..100)) {
             let mut trie = Trie::new_temp();
             let mut cita_trie = cita_trie();
@@ -876,7 +876,7 @@ mod test {
 
         #[test]
         // The previous test needs to sort the input values in order to get rid of duplicate entries, leading to ordered insertions
-        // This check has a fixed way of determining wether a value should be removed but doesn't require ordered insertions
+        // This check has a fixed way of determining whether a value should be removed but doesn't require ordered insertions
         fn proptest_compare_proof_with_removals_unsorted(data in btree_set(vec(any::<u8>(), 5..100), 1..100)) {
             let mut trie = Trie::new_temp();
             let mut cita_trie = cita_trie();

crates/l2/sequencer/block_producer/payload_builder.rs

@@ -135,7 +135,7 @@ pub async fn fill_transactions(
         // TODO: maybe fetch hash too when filtering mempool so we don't have to compute it here (we can do this in the same refactor as adding timestamp)
         let tx_hash = head_tx.tx.compute_hash();
 
-        // Check wether the tx is replay-protected
+        // Check whether the tx is replay-protected
         if head_tx.tx.protected() && !chain_config.is_eip155_activated(context.block_number()) {
             // Ignore replay protected tx & all txs from the sender
             // Pull transaction from the mempool

crates/networking/p2p/kademlia.rs

@@ -232,7 +232,7 @@ impl KademliaTable {
     }
 
     /// Returns an iterator for all peers in the table that match the filter
-    fn filter_peers<'a>(
+    pub fn filter_peers<'a>(
         &'a self,
         filter: &'a dyn Fn(&'a PeerData) -> bool,
     ) -> impl Iterator<Item = &'a PeerData> {
@@ -297,17 +297,20 @@ impl KademliaTable {
     /// Set the sender end of the channel between the kademlia table and the peer's active connection
     /// Set the peer's supported capabilities
     /// This function should be called each time a connection is established so the backend can send requests to the peers
+    /// Receives a boolean indicating if the connection is inbound (aka if it was started by the peer and not by this node)
     pub(crate) fn init_backend_communication(
         &mut self,
         node_id: H256,
         channels: PeerChannels,
         capabilities: Vec<Capability>,
+        inbound: bool,
     ) {
         let peer = self.get_by_node_id_mut(node_id);
         if let Some(peer) = peer {
             peer.channels = Some(channels);
             peer.supported_capabilities = capabilities;
             peer.is_connected = true;
+            peer.is_connection_inbound = inbound;
         } else {
             debug!(
                 "[PEERS] Peer with node_id {:?} not found in the kademlia table when trying to init backend communication",
@@ -354,9 +357,12 @@ pub struct PeerData {
     pub revalidation: Option<bool>,
     /// communication channels between the peer data and its active connection
     pub channels: Option<PeerChannels>,
-    /// Starts as false when a node is added. Set to true when a connection si active. When a
+    /// Starts as false when a node is added. Set to true when a connection becomes active. When a
     /// connection fails, the peer record is removed, so no need to set it to false.
     pub is_connected: bool,
+    /// Set to true if the connection is inbound (aka the connection was started by the peer and not by this node)
+    /// It is only valid as long as is_connected is true
+    pub is_connection_inbound: bool,
 }
 
 impl PeerData {
@@ -375,6 +381,7 @@ impl PeerData {
             channels: None,
             supported_capabilities: vec![],
             is_connected: false,
+            is_connection_inbound: false,
         }
     }
 

crates/networking/p2p/network.rs

@@ -138,7 +138,7 @@ fn listener(tcp_addr: SocketAddr) -> Result<TcpListener, io::Error> {
 async fn handle_peer_as_receiver(context: P2PContext, peer_addr: SocketAddr, stream: TcpStream) {
     let table = context.table.clone();
     match handshake::as_receiver(context, peer_addr, stream).await {
-        Ok(mut conn) => conn.start(table).await,
+        Ok(mut conn) => conn.start(table, true).await,
         Err(e) => {
             debug!("Error creating tcp connection with peer at {peer_addr}: {e}")
         }
@@ -157,7 +157,7 @@ pub async fn handle_peer_as_initiator(context: P2PContext, node: Node) {
     };
     let table = context.table.clone();
     match handshake::as_initiator(context, node.clone(), stream).await {
-        Ok(mut conn) => conn.start(table).await,
+        Ok(mut conn) => conn.start(table, false).await,
         Err(e) => {
             log_peer_error(&node, &format!("Error creating tcp connection {e}"));
             table.lock().await.replace_peer(node.node_id());

crates/networking/p2p/peer_handler.rs

@@ -11,7 +11,7 @@ use ethrex_trie::{verify_range, Node};
 use tokio::sync::Mutex;
 
 use crate::{
-    kademlia::{KademliaTable, PeerChannels},
+    kademlia::{KademliaTable, PeerChannels, PeerData},
     rlpx::{
         eth::{
             blocks::{
@@ -58,6 +58,13 @@ impl PeerHandler {
     pub fn new(peer_table: Arc<Mutex<KademliaTable>>) -> PeerHandler {
         Self { peer_table }
     }
+
+    /// Creates a dummy PeerHandler for tests where interacting with peers is not needed
+    /// This should only be used in tests as it won't be able to interact with the node's connected peers
+    pub fn dummy() -> PeerHandler {
+        let dummy_peer_table = Arc::new(Mutex::new(KademliaTable::new(Default::default())));
+        PeerHandler::new(dummy_peer_table)
+    }
     /// Returns the channel ends to an active peer connection that supports the given capability
     /// The peer is selected randomly, and doesn't guarantee that the selected peer is not currently busy
     /// If no peer is found, this method will try again after 10 seconds
@@ -637,4 +644,17 @@ impl PeerHandler {
         }
         None
     }
+
+    /// Returns the PeerData for each connected Peer
+    /// Returns None if it fails to aquire the lock on the kademlia table
+    pub fn read_connected_peers(&self) -> Option<Vec<PeerData>> {
+        Some(
+            self.peer_table
+                .try_lock()
+                .ok()?
+                .filter_peers(&|peer| peer.is_connected)
+                .cloned()
+                .collect::<Vec<_>>(),
+        )
+    }
 }

crates/networking/p2p/rlpx/connection.rs

@@ -143,7 +143,11 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
 
     /// Handshake already performed, now it starts a peer connection.
     /// It runs in it's own task and blocks until the connection is dropped
-    pub async fn start(&mut self, table: Arc<Mutex<crate::kademlia::KademliaTable>>) {
+    pub async fn start(
+        &mut self,
+        table: Arc<Mutex<crate::kademlia::KademliaTable>>,
+        inbound: bool,
+    ) {
         log_peer_debug(&self.node, "Starting RLPx connection");
 
         if let Err(reason) = self.post_handshake_checks(table.clone()).await {
@@ -174,6 +178,7 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
                     self.node.node_id(),
                     peer_channels,
                     self.capabilities.clone(),
+                    inbound,
                 );
             }
             if let Err(e) = self.connection_loop(sender, receiver).await {
@@ -295,6 +300,8 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
                     self.negotiated_snap_version = negotiated_snap_cap.version;
                 }
 
+                self.node.version = Some(hello_message.client_id);
+
                 Ok(())
             }
             Message::Disconnect(disconnect) => {

crates/networking/p2p/rlpx/p2p.rs

@@ -12,6 +12,7 @@ use ethrex_rlp::{
     error::{RLPDecodeError, RLPEncodeError},
 };
 use k256::PublicKey;
+use serde::Serialize;
 
 pub const CAP_P2P_5: Capability = Capability::p2p(5);
 pub const CAP_ETH_68: Capability = Capability::eth(68);
@@ -69,6 +70,15 @@ impl RLPDecode for Capability {
     }
 }
 
+impl Serialize for Capability {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        serializer.serialize_str(&format!("{}/{}", self.protocol, self.version))
+    }
+}
+
 #[derive(Debug)]
 pub(crate) struct HelloMessage {
     pub(crate) capabilities: Vec<Capability>,

crates/networking/p2p/sync.rs

@@ -26,17 +26,14 @@ use std::{
 };
 use storage_healing::storage_healer;
 use tokio::{
-    sync::{mpsc::error::SendError, Mutex},
+    sync::mpsc::error::SendError,
     time::{Duration, Instant},
 };
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
 use trie_rebuild::TrieRebuilder;
 
-use crate::{
-    kademlia::KademliaTable,
-    peer_handler::{BlockRequestOrder, PeerHandler, HASH_MAX, MAX_BLOCK_BODIES_TO_REQUEST},
-};
+use crate::peer_handler::{BlockRequestOrder, PeerHandler, HASH_MAX, MAX_BLOCK_BODIES_TO_REQUEST};
 
 /// The minimum amount of blocks from the head that we want to full sync during a snap sync
 const MIN_FULL_BLOCKS: usize = 64;
@@ -95,14 +92,14 @@ pub struct Syncer {
 
 impl Syncer {
     pub fn new(
-        peer_table: Arc<Mutex<KademliaTable>>,
+        peers: PeerHandler,
         snap_enabled: Arc<AtomicBool>,
         cancel_token: CancellationToken,
         blockchain: Arc<Blockchain>,
     ) -> Self {
         Self {
             snap_enabled,
-            peers: PeerHandler::new(peer_table),
+            peers,
             last_snap_pivot: 0,
             trie_rebuilder: None,
             cancel_token,
@@ -113,10 +110,9 @@ impl Syncer {
     /// Creates a dummy Syncer for tests where syncing is not needed
     /// This should only be used in tests as it won't be able to connect to the p2p network
     pub fn dummy() -> Self {
-        let dummy_peer_table = Arc::new(Mutex::new(KademliaTable::new(Default::default())));
         Self {
             snap_enabled: Arc::new(AtomicBool::new(false)),
-            peers: PeerHandler::new(dummy_peer_table),
+            peers: PeerHandler::dummy(),
             last_snap_pivot: 0,
             trie_rebuilder: None,
             // This won't be used

crates/networking/p2p/sync_manager.rs

@@ -14,7 +14,7 @@ use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
 
 use crate::{
-    kademlia::KademliaTable,
+    peer_handler::PeerHandler,
     sync::{SyncMode, Syncer},
 };
 
@@ -31,15 +31,15 @@ pub struct SyncManager {
 
 impl SyncManager {
     pub async fn new(
-        peer_table: Arc<Mutex<KademliaTable>>,
+        peer_handler: PeerHandler,
         sync_mode: SyncMode,
         cancel_token: CancellationToken,
         blockchain: Arc<Blockchain>,
         store: Store,
     ) -> Self {
         let snap_enabled = Arc::new(AtomicBool::new(matches!(sync_mode, SyncMode::Snap)));
         let syncer = Arc::new(Mutex::new(Syncer::new(
-            peer_table,
+            peer_handler,
             snap_enabled.clone(),
             cancel_token,
             blockchain,

crates/networking/p2p/types.rs

@@ -65,6 +65,7 @@ pub struct Node {
     pub udp_port: u16,
     pub tcp_port: u16,
     pub public_key: H512,
+    pub version: Option<String>,
     node_id: OnceLock<H256>,
 }
 
@@ -119,6 +120,7 @@ impl Node {
             udp_port,
             tcp_port,
             public_key,
+            version: None,
             node_id: OnceLock::new(),
         }
     }

crates/networking/rpc/admin/mod.rs

@@ -5,6 +5,8 @@ use serde_json::Value;
 use std::collections::HashMap;
 
 use crate::{rpc::NodeData, utils::RpcErr};
+mod peers;
+pub use peers::peers;
 
 #[derive(Serialize, Debug)]
 struct NodeInfo {