Add basic peer manager

Cargo.toml

@@ -86,6 +86,7 @@ parking_lot = "0.12"
 reqwest = "0.12.12"
 rusqlite = "0.28.0"
 serde = { version = "1.0.208", features = ["derive"] }
+smallvec = "1.13.2"
 strum = { version = "0.24", features = ["derive"] }
 tokio = { version = "1.39.2", features = [
     "rt",

anchor/network/Cargo.toml

@@ -5,13 +5,16 @@ edition = { workspace = true }
 authors = ["Sigma Prime <[email protected]>"]
 
 [dependencies]
+arbitrary = "1.4.1"
+delay_map = "0.4"
 dirs = { workspace = true }
 discv5 = { workspace = true }
 ethereum_serde_utils = "1.0.0-beta.0"
 ethereum_ssz = "0.8.1"
 ethereum_ssz_derive = "0.8.1"
 futures = { workspace = true }
 hex = "0.4.3"
+itertools = "0.10.5"
 libp2p = { version = "0.54", default-features = false, features = [
   "identify",
   "yamux",
@@ -25,8 +28,11 @@ libp2p = { version = "0.54", default-features = false, features = [
   "ping",
 ] }
 lighthouse_network = { workspace = true }
+parking_lot = { workspace = true }
 serde = { workspace = true }
+smallvec = { workspace = true }
 ssz_types = "0.10"
+strum = "0.24.1"
 task_executor = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

anchor/network/src/behaviour.rs

@@ -1,4 +1,5 @@
 use crate::discovery::Discovery;
+use crate::peer_manager::PeerManager;
 use libp2p::swarm::NetworkBehaviour;
 use libp2p::{gossipsub, identify, ping};
 
@@ -12,4 +13,6 @@ pub struct AnchorBehaviour {
     pub gossipsub: gossipsub::Behaviour,
     /// Discv5 Discovery protocol.
     pub discovery: Discovery,
+    /// The peer manager that keeps track of peer's reputation and status.
+    pub peer_manager: PeerManager,
 }

anchor/network/src/lib.rs

@@ -5,6 +5,7 @@ mod config;
 mod discovery;
 mod keypair_utils;
 mod network;
+mod peer_manager;
 mod transport;
 pub mod types;
 

anchor/network/src/network.rs

@@ -13,7 +13,7 @@ use libp2p::{futures, gossipsub, identify, ping, PeerId, Swarm, SwarmBuilder};
 use lighthouse_network::discovery::DiscoveredPeers;
 use lighthouse_network::discv5::enr::k256::sha2::{Digest, Sha256};
 use task_executor::TaskExecutor;
-use tracing::{debug, error, info, trace, warn};
+use tracing::{debug, error, info, warn};
 
 use crate::behaviour::AnchorBehaviour;
 use crate::behaviour::AnchorBehaviourEvent;
@@ -22,8 +22,8 @@ use crate::keypair_utils::load_private_key;
 use crate::transport::build_transport;
 use crate::Config;
 
+use crate::peer_manager::{PeerManager, PeerManagerEvent};
 use crate::types::ssv_message::SignedSSVMessage;
-use lighthouse_network::EnrExt;
 use ssz::Decode;
 
 pub struct Network {
@@ -99,6 +99,7 @@ impl Network {
         loop {
             tokio::select! {
                 swarm_message = self.swarm.select_next_some() => {
+                    debug!(connected_peers = self.swarm.connected_peers().count(), "Connected peers");
                     match swarm_message {
                         SwarmEvent::Behaviour(behaviour_event) => match behaviour_event {
                             AnchorBehaviourEvent::Gossipsub(ge) => {
@@ -134,16 +135,13 @@ impl Network {
                             // The peer manager will subsequently decide which peers need to be dialed and then dial
                             // them.
                             AnchorBehaviourEvent::Discovery(DiscoveredPeers { peers }) => {
-                                //self.peer_manager_mut().peers_discovered(peers);
+                                let peer_manager = &mut self.swarm.behaviour_mut().peer_manager;
+                                peer_manager.peers_discovered(peers.clone());
                                 debug!(peers =  ?peers, "Peers discovered");
-                                for (enr, _) in peers {
-                                    for tcp in enr.multiaddr_tcp() {
-                                        trace!(address = ?tcp, "Dialing peer");
-                                        if let Err(e) = self.swarm.dial(tcp.clone()) {
-                                            error!(address = ?tcp, error = ?e, "Error dialing peer");
-                                        }
-                                    }
-                                }
+                            }
+                            AnchorBehaviourEvent::PeerManager(PeerManagerEvent::DiscoverPeers(peers_to_find)) => {
+                                // Peer manager has requested a discovery query for more peers.
+                                self.discovery_mut().discover_peers(peers_to_find);
                             }
                             // TODO handle other behaviour events
                             _ => {
@@ -160,6 +158,11 @@ impl Network {
             }
         }
     }
+
+    /// Discv5 Discovery protocol.
+    pub fn discovery_mut(&mut self) -> &mut Discovery {
+        &mut self.swarm.behaviour_mut().discovery
+    }
 }
 
 async fn build_anchor_behaviour(
@@ -217,11 +220,18 @@ async fn build_anchor_behaviour(
         discovery
     };
 
+    let peer_manager = PeerManager::new(
+        lighthouse_network::peer_manager::config::Config::default(),
+        vec![],
+        false,
+    );
+
     AnchorBehaviour {
         identify,
         ping: ping::Behaviour::default(),
         gossipsub,
         discovery,
+        peer_manager,
     }
 }