feat: add bootnode support and basic discv5 support (#38)

* feat: make discv5 peer discovery work

* fix: resolve PR concerns

Cargo.toml

@@ -30,7 +30,8 @@ anyhow = "1.0"
 blst = "0.3"
 kzg = { git = "https://github.com/grandinetech/rust-kzg" }
 clap = "4"
-discv5 = "0.9"
+discv5 = { version = "0.9.0", features = ["libp2p"] }
+enr = "0.13.0"
 ethereum_hashing = { git = "https://github.com/ReamLabs/ethereum_hashing.git" }
 ethereum_serde_utils = "0.7"
 ethereum_ssz = "0.8"

bin/ream/src/main.rs

@@ -4,7 +4,7 @@ use clap::Parser;
 use ream::cli::{Cli, Commands};
 use ream_discv5::config::NetworkConfig;
 use ream_executor::ReamExecutor;
-use ream_p2p::Network;
+use ream_p2p::network::Network;
 use tracing::info;
 use tracing_subscriber::EnvFilter;
 

crates/networking/discv5/Cargo.toml

@@ -10,9 +10,11 @@ rust-version.workspace = true
 version.workspace = true
 
 [dependencies]
-
-discv5 = { workspace = true }
-futures = { workspace = true }
-libp2p-identity = { workspace = true }
-libp2p-mplex = { workspace = true }
-libp2p = { workspace = true }
+anyhow.workspace = true
+discv5.workspace = true
+futures.workspace = true
+libp2p-identity.workspace = true
+libp2p-mplex.workspace = true
+libp2p.workspace = true
+tokio.workspace = true
+tracing.workspace = true

crates/networking/discv5/src/discovery.rs

@@ -6,8 +6,12 @@ use std::{
     time::Instant,
 };
 
-use discv5::{enr::CombinedKey, Discv5, Enr};
-use futures::{stream::FuturesUnordered, StreamExt, TryFutureExt};
+use anyhow::anyhow;
+use discv5::{
+    enr::{k256::ecdsa::SigningKey, CombinedKey, NodeId},
+    Discv5, Enr,
+};
+use futures::{stream::FuturesUnordered, FutureExt, StreamExt, TryFutureExt};
 use libp2p::{
     core::{transport::PortUse, Endpoint},
     identity::Keypair,
@@ -17,16 +21,22 @@ use libp2p::{
     },
     Multiaddr, PeerId,
 };
+use tokio::sync::mpsc;
+use tracing::{error, info, warn};
 
 use crate::config::NetworkConfig;
 
 #[derive(Debug)]
 pub struct DiscoveredPeers {
-    pub _peers: HashMap<Enr, Option<Instant>>,
+    pub peers: HashMap<Enr, Option<Instant>>,
 }
 
 enum EventStream {
     Inactive,
+    Awaiting(
+        Pin<Box<dyn Future<Output = Result<mpsc::Receiver<discv5::Event>, discv5::Error>> + Send>>,
+    ),
+    Present(mpsc::Receiver<discv5::Event>),
 }
 
 #[derive(Debug, Clone, PartialEq)]
@@ -48,13 +58,13 @@ pub struct Discovery {
 }
 
 impl Discovery {
-    pub async fn new(local_key: Keypair, config: &NetworkConfig) -> Result<Self, String> {
+    pub async fn new(local_key: Keypair, config: &NetworkConfig) -> anyhow::Result<Self> {
         let enr_local = convert_to_enr(local_key)?;
         let enr = Enr::builder().build(&enr_local).unwrap();
         let node_local_id = enr.node_id();
 
         let mut discv5 = Discv5::new(enr, enr_local, config.discv5_config.clone())
-            .map_err(|e| format!("Discv5 service failed. Error: {:?}", e))?;
+            .map_err(|err| anyhow!("Failed to create discv5: {err:?}"))?;
 
         // adding bootnode to DHT
         for bootnode_enr in config.boot_nodes_enr.clone() {
@@ -63,17 +73,19 @@ impl Discovery {
                 continue;
             }
 
-            let _ = discv5.add_enr(bootnode_enr).map_err(|e| {
-                println!("Discv5 service failed. Error: {:?}", e);
+            let _ = discv5.add_enr(bootnode_enr).map_err(|err| {
+                error!("Failed to add bootnode to DHT {err:?}");
             });
         }
 
         // init ports
         let event_stream = if !config.disable_discovery {
-            discv5.start().map_err(|e| e.to_string()).await?;
-            println!("Started discovery");
-            // EventStream::Awaiting(Box::pin(discv5.event_stream()))
-            EventStream::Inactive
+            discv5
+                .start()
+                .map_err(|err| anyhow!("Failed to start discv5 {err:?}"))
+                .await?;
+            info!("Started discovery");
+            EventStream::Awaiting(Box::pin(discv5.event_stream()))
         } else {
             EventStream::Inactive
         };
@@ -90,7 +102,7 @@ impl Discovery {
     pub fn discover_peers(&mut self, target_peers: usize) {
         // If the discv5 service isn't running or we are in the process of a query, don't bother
         // queuing a new one.
-        println!("Discovering peers {:?}", self.discv5.local_enr());
+        info!("Discovering peers {:?}", self.discv5.local_enr());
 
         if !self.started || self.find_peer_active {
             return;
@@ -100,19 +112,46 @@ impl Discovery {
         self.start_query(QueryType::FindPeers, target_peers);
     }
 
-    fn process_queries(&mut self, cx: &mut Context) -> bool {
-        let mut processed = false;
-
+    fn process_queries(&mut self, cx: &mut Context) -> Option<HashMap<Enr, Option<Instant>>> {
         while let Poll::Ready(Some(query)) = self.discovery_queries.poll_next_unpin(cx) {
-            println!("query{:?} {:?}", query.result, query.query_type);
-            processed = true;
-            // TODO: add query types and push them to mesh
+            let result = match query.query_type {
+                QueryType::FindPeers => {
+                    self.find_peer_active = false;
+                    match query.result {
+                        Ok(peers) => {
+                            info!("Found {} peers", peers.len());
+                            let mut peer_map = HashMap::new();
+                            for peer in peers {
+                                peer_map.insert(peer, None);
+                            }
+                            Some(peer_map)
+                        }
+                        Err(e) => {
+                            warn!("Failed to find peers: {:?}", e);
+                            None
+                        }
+                    }
+                }
+            };
+
+            if result.is_some() {
+                return result;
+            }
         }
-        processed
+        None
     }
 
     fn start_query(&mut self, query: QueryType, _total_peers: usize) {
-        println!("Query! queryType={:?}", query);
+        info!("Query! queryType={:?}", query);
+        let query_future = self
+            .discv5
+            .find_node(NodeId::random())
+            .map(|result| QueryResult {
+                query_type: query,
+                result,
+            });
+
+        self.discovery_queries.push(Box::pin(query_future));
     }
 }
 
@@ -139,16 +178,6 @@ impl NetworkBehaviour for Discovery {
         Ok(ConnectionHandler)
     }
 
-    fn handle_pending_outbound_connection(
-        &mut self,
-        _connection_id: ConnectionId,
-        _maybe_peer: Option<PeerId>,
-        _addresses: &[Multiaddr],
-        _effective_role: Endpoint,
-    ) -> Result<Vec<Multiaddr>, ConnectionDenied> {
-        Ok(Vec::new())
-    }
-
     fn handle_established_outbound_connection(
         &mut self,
         _connection_id: ConnectionId,
@@ -161,7 +190,7 @@ impl NetworkBehaviour for Discovery {
     }
 
     fn on_swarm_event(&mut self, event: FromSwarm) {
-        println!("Swarm event: {:?}", event);
+        info!("Discv5 on swarm event gotten: {:?}", event);
     }
 
     fn on_connection_handler_event(
@@ -170,26 +199,47 @@ impl NetworkBehaviour for Discovery {
         _connection_id: ConnectionId,
         _event: THandlerOutEvent<Self>,
     ) {
-        println!("ConnectionHandlerOutEvent");
     }
 
     fn poll(
         &mut self,
         cx: &mut Context<'_>,
     ) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
-        self.process_queries(cx);
+        if !self.started {
+            return Poll::Pending;
+        }
 
-        match self.event_stream {
-            EventStream::Inactive => println!("inactive"),
+        if let Some(peers) = self.process_queries(cx) {
+            return Poll::Ready(ToSwarm::GenerateEvent(DiscoveredPeers { peers }));
+        }
+
+        match &mut self.event_stream {
+            EventStream::Inactive => {}
+            EventStream::Awaiting(fut) => {
+                if let Poll::Ready(event_stream) = fut.poll_unpin(cx) {
+                    match event_stream {
+                        Ok(stream) => {
+                            self.event_stream = EventStream::Present(stream);
+                        }
+                        Err(e) => {
+                            error!("Failed to start discovery event stream: {:?}", e);
+                            self.event_stream = EventStream::Inactive;
+                        }
+                    }
+                }
+            }
+            EventStream::Present(_receiver) => {}
         };
 
         Poll::Pending
     }
 }
 
-fn convert_to_enr(key: Keypair) -> Result<CombinedKey, &'static str> {
-    let key = key.try_into_secp256k1().expect("right key type");
-    let secret = discv5::enr::k256::ecdsa::SigningKey::from_slice(&key.secret().to_bytes())
-        .expect("libp2p key must be valid");
+fn convert_to_enr(key: Keypair) -> anyhow::Result<CombinedKey> {
+    let key = key
+        .try_into_secp256k1()
+        .map_err(|err| anyhow!("Failed to get secp256k1 keypair: {err:?}"))?;
+    let secret = SigningKey::from_slice(&key.secret().to_bytes())
+        .map_err(|err| anyhow!("Failed to convert keypair to SigningKey: {err:?}"))?;
     Ok(CombinedKey::Secp256k1(secret))
 }

crates/networking/p2p/Cargo.toml

@@ -11,15 +11,19 @@ version.workspace = true
 
 [dependencies]
 # ream dependencies
-ream-executor = { workspace = true }
-ream-discv5 = { workspace = true }
+ream-executor.workspace = true
+ream-discv5.workspace = true
 
 # misc
-discv5 = { workspace = true }
-futures = { workspace = true }
-hex = { workspace = true }
-libp2p-identity = { workspace = true }
-libp2p-mplex = { workspace = true }
-libp2p = { workspace = true, default-features = false, features = ["identify", "yamux", "noise", "dns", "tcp", "tokio", "plaintext", "secp256k1", "macros", "ecdsa", "metrics", "quic", "upnp", "gossipsub", "ping"] }
-serde = { workspace = true }
-tokio = { workspace = true }
+anyhow.workspace = true
+enr.workspace = true
+discv5.workspace = true
+futures.workspace = true
+hex.workspace = true
+libp2p-identity.workspace = true
+libp2p-mplex.workspace = true
+libp2p.workspace = true
+serde.workspace = true
+serde_yaml.workspace = true
+tokio.workspace = true
+tracing.workspace = true

crates/networking/p2p/resources/bootstrap_nodes.yaml

@@ -0,0 +1,27 @@
+# Eth mainnet consensus layer bootnodes
+# Taken from https://github.com/eth-clients/mainnet/blob/main/metadata/bootstrap_nodes.yaml
+
+# Teku team's bootnodes
+- enr:-KG4QNTx85fjxABbSq_Rta9wy56nQ1fHK0PewJbGjLm1M4bMGx5-3Qq4ZX2-iFJ0pys_O90sVXNNOxp2E7afBsGsBrgDhGV0aDKQu6TalgMAAAD__________4JpZIJ2NIJpcIQEnfA2iXNlY3AyNTZrMaECGXWQ-rQ2KZKRH1aOW4IlPDBkY4XDphxg9pxKytFCkayDdGNwgiMog3VkcIIjKA # 4.157.240.54 | azure-us-east-virginia
+- enr:-KG4QF4B5WrlFcRhUU6dZETwY5ZzAXnA0vGC__L1Kdw602nDZwXSTs5RFXFIFUnbQJmhNGVU6OIX7KVrCSTODsz1tK4DhGV0aDKQu6TalgMAAAD__________4JpZIJ2NIJpcIQExNYEiXNlY3AyNTZrMaECQmM9vp7KhaXhI-nqL_R0ovULLCFSFTa9CPPSdb1zPX6DdGNwgiMog3VkcIIjKA # 4.196.214.4  | azure-au-east-sydney
+
+# Prylab team's bootnodes
+- enr:-Ku4QImhMc1z8yCiNJ1TyUxdcfNucje3BGwEHzodEZUan8PherEo4sF7pPHPSIB1NNuSg5fZy7qFsjmUKs2ea1Whi0EBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpD1pf1CAAAAAP__________gmlkgnY0gmlwhBLf22SJc2VjcDI1NmsxoQOVphkDqal4QzPMksc5wnpuC3gvSC8AfbFOnZY_On34wIN1ZHCCIyg # 18.223.219.100 | aws-us-east-2-ohio
+- enr:-Ku4QP2xDnEtUXIjzJ_DhlCRN9SN99RYQPJL92TMlSv7U5C1YnYLjwOQHgZIUXw6c-BvRg2Yc2QsZxxoS_pPRVe0yK8Bh2F0dG5ldHOIAAAAAAAAAACEZXRoMpD1pf1CAAAAAP__________gmlkgnY0gmlwhBLf22SJc2VjcDI1NmsxoQMeFF5GrS7UZpAH2Ly84aLK-TyvH-dRo0JM1i8yygH50YN1ZHCCJxA # 18.223.219.100 | aws-us-east-2-ohio
+- enr:-Ku4QPp9z1W4tAO8Ber_NQierYaOStqhDqQdOPY3bB3jDgkjcbk6YrEnVYIiCBbTxuar3CzS528d2iE7TdJsrL-dEKoBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpD1pf1CAAAAAP__________gmlkgnY0gmlwhBLf22SJc2VjcDI1NmsxoQMw5fqqkw2hHC4F5HZZDPsNmPdB1Gi8JPQK7pRc9XHh-oN1ZHCCKvg # 18.223.219.100 | aws-us-east-2-ohio
+
+# Lighthouse team's bootnodes
+- enr:-Le4QPUXJS2BTORXxyx2Ia-9ae4YqA_JWX3ssj4E_J-3z1A-HmFGrU8BpvpqhNabayXeOZ2Nq_sbeDgtzMJpLLnXFgAChGV0aDKQtTA_KgEAAAAAIgEAAAAAAIJpZIJ2NIJpcISsaa0Zg2lwNpAkAIkHAAAAAPA8kv_-awoTiXNlY3AyNTZrMaEDHAD2JKYevx89W0CcFJFiskdcEzkH_Wdv9iW42qLK79ODdWRwgiMohHVkcDaCI4I # 172.105.173.25 | linode-au-sydney
+- enr:-Le4QLHZDSvkLfqgEo8IWGG96h6mxwe_PsggC20CL3neLBjfXLGAQFOPSltZ7oP6ol54OvaNqO02Rnvb8YmDR274uq8ChGV0aDKQtTA_KgEAAAAAIgEAAAAAAIJpZIJ2NIJpcISLosQxg2lwNpAqAX4AAAAAAPA8kv_-ax65iXNlY3AyNTZrMaEDBJj7_dLFACaxBfaI8KZTh_SSJUjhyAyfshimvSqo22WDdWRwgiMohHVkcDaCI4I # 139.162.196.49 | linode-uk-london
+- enr:-Le4QH6LQrusDbAHPjU_HcKOuMeXfdEB5NJyXgHWFadfHgiySqeDyusQMvfphdYWOzuSZO9Uq2AMRJR5O4ip7OvVma8BhGV0aDKQtTA_KgEAAAAAIgEAAAAAAIJpZIJ2NIJpcISLY9ncg2lwNpAkAh8AgQIBAAAAAAAAAAmXiXNlY3AyNTZrMaECDYCZTZEksF-kmgPholqgVt8IXr-8L7Nu7YrZ7HUpgxmDdWRwgiMohHVkcDaCI4I # 139.99.217.220 | ovh-au-sydney
+- enr:-Le4QIqLuWybHNONr933Lk0dcMmAB5WgvGKRyDihy1wHDIVlNuuztX62W51voT4I8qD34GcTEOTmag1bcdZ_8aaT4NUBhGV0aDKQtTA_KgEAAAAAIgEAAAAAAIJpZIJ2NIJpcISLY04ng2lwNpAkAh8AgAIBAAAAAAAAAA-fiXNlY3AyNTZrMaEDscnRV6n1m-D9ID5UsURk0jsoKNXt1TIrj8uKOGW6iluDdWRwgiMohHVkcDaCI4I # 139.99.78.39 | ovh-singapore
+
+# EF bootnodes
+- enr:-Ku4QHqVeJ8PPICcWk1vSn_XcSkjOkNiTg6Fmii5j6vUQgvzMc9L1goFnLKgXqBJspJjIsB91LTOleFmyWWrFVATGngBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhAMRHkWJc2VjcDI1NmsxoQKLVXFOhp2uX6jeT0DvvDpPcU8FWMjQdR4wMuORMhpX24N1ZHCCIyg # 3.17.30.69 | aws-us-east-2-ohio
+- enr:-Ku4QG-2_Md3sZIAUebGYT6g0SMskIml77l6yR-M_JXc-UdNHCmHQeOiMLbylPejyJsdAPsTHJyjJB2sYGDLe0dn8uYBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhBLY-NyJc2VjcDI1NmsxoQORcM6e19T1T9gi7jxEZjk_sjVLGFscUNqAY9obgZaxbIN1ZHCCIyg # 18.216.248.220 | aws-us-east-2-ohio
+- enr:-Ku4QPn5eVhcoF1opaFEvg1b6JNFD2rqVkHQ8HApOKK61OIcIXD127bKWgAtbwI7pnxx6cDyk_nI88TrZKQaGMZj0q0Bh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhDayLMaJc2VjcDI1NmsxoQK2sBOLGcUb4AwuYzFuAVCaNHA-dy24UuEKkeFNgCVCsIN1ZHCCIyg # 54.178.44.198 | aws-ap-northeast-1-tokyo
+- enr:-Ku4QEWzdnVtXc2Q0ZVigfCGggOVB2Vc1ZCPEc6j21NIFLODSJbvNaef1g4PxhPwl_3kax86YPheFUSLXPRs98vvYsoBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhDZBrP2Jc2VjcDI1NmsxoQM6jr8Rb1ktLEsVcKAPa08wCsKUmvoQ8khiOl_SLozf9IN1ZHCCIyg # 54.65.172.253 | aws-ap-northeast-1-tokyo
+
+# Nimbus team's bootnodes
+- enr:-LK4QA8FfhaAjlb_BXsXxSfiysR7R52Nhi9JBt4F8SPssu8hdE1BXQQEtVDC3qStCW60LSO7hEsVHv5zm8_6Vnjhcn0Bh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhAN4aBKJc2VjcDI1NmsxoQJerDhsJ-KxZ8sHySMOCmTO6sHM3iCFQ6VMvLTe948MyYN0Y3CCI4yDdWRwgiOM # 3.120.104.18 | aws-eu-central-1-frankfurt
+- enr:-LK4QKWrXTpV9T78hNG6s8AM6IO4XH9kFT91uZtFg1GcsJ6dKovDOr1jtAAFPnS2lvNltkOGA9k29BUN7lFh_sjuc9QBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpC1MD8qAAAAAP__________gmlkgnY0gmlwhANAdd-Jc2VjcDI1NmsxoQLQa6ai7y9PMN5hpLe5HmiJSlYzMuzP7ZhwRiwHvqNXdoN0Y3CCI4yDdWRwgiOM # 3.64.117.223 | aws-eu-central-1-frankfurt

crates/networking/p2p/src/bootnodes.rs

@@ -0,0 +1,20 @@
+use discv5::Enr;
+
+pub struct Bootnodes {
+    pub bootnodes: Vec<Enr>,
+}
+
+impl Bootnodes {
+    pub fn new() -> Self {
+        let bootnodes: Vec<Enr> =
+            serde_yaml::from_str(include_str!("../resources/bootstrap_nodes.yaml"))
+                .expect("should deserialize bootnodes");
+        Self { bootnodes }
+    }
+}
+
+impl Default for Bootnodes {
+    fn default() -> Self {
+        Self::new()
+    }
+}

crates/networking/p2p/src/lib.rs

@@ -1,3 +1,2 @@
-mod network;
-
-pub use network::Network;
+pub mod bootnodes;
+pub mod network;