some p2p refactors on versioning

Author: erhant

Cargo.lock

@@ -10,18 +10,25 @@ license = "Apache-2.0"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [workspace.dependencies]
+# async stuff
 tokio-util = { version = "0.7.10", features = ["rt"] }
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }
-parking_lot = "0.12.2"
+async-trait = "0.1.81"
+
+# serialize & deserialize
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-async-trait = "0.1.81"
+
+# http client
 reqwest = "0.12.5"
 
+# env reading
 dotenvy = "0.15.7"
 
+# randomization
 rand = "0.8.5"
 
+# logging & errors
 env_logger = "0.11.3"
 log = "0.4.21"
 eyre = "0.6.12"

Cargo.toml

@@ -18,7 +18,6 @@ profiling = []
 [dependencies]
 tokio-util = { version = "0.7.10", features = ["rt"] }
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }
-parking_lot = "0.12.2"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 async-trait = "0.1.81"

compute/Cargo.toml

@@ -1,4 +1,4 @@
-use dkn_p2p::{libp2p::gossipsub, DriaP2P};
+use dkn_p2p::{libp2p::gossipsub, DriaP2PClient};
 use eyre::{eyre, Result};
 use std::time::Duration;
 use tokio_util::sync::CancellationToken;
@@ -27,7 +27,7 @@ const RPC_PEER_ID_REFRESH_INTERVAL_SECS: u64 = 30;
 /// ```
 pub struct DriaComputeNode {
     pub config: DriaComputeNodeConfig,
-    pub p2p: DriaP2P,
+    pub p2p: DriaP2PClient,
     pub available_nodes: AvailableNodes,
     pub available_nodes_last_refreshed: tokio::time::Instant,
     pub cancellation: CancellationToken,
@@ -51,11 +51,21 @@ impl DriaComputeNode {
             )
             .sort_dedup();
 
-        let p2p = DriaP2P::new(
+        // we are using the major.minor version as the P2P version
+        // so that patch versions do not interfere with the protocol
+        const P2P_VERSION: &str = concat!(
+            env!("CARGO_PKG_VERSION_MAJOR"),
+            ".",
+            env!("CARGO_PKG_VERSION_MINOR")
+        );
+
+        // create p2p client
+        let p2p = DriaP2PClient::new(
             keypair,
             config.p2p_listen_addr.clone(),
             &available_nodes.bootstrap_nodes,
             &available_nodes.relay_nodes,
+            P2P_VERSION,
         )?;
 
         Ok(DriaComputeNode {

compute/src/node.rs

@@ -1,9 +1,8 @@
 use dkn_p2p::libp2p::{Multiaddr, PeerId};
+use dkn_workflows::split_csv_line;
 use eyre::Result;
 use std::{env, fmt::Debug, str::FromStr};
 
-use dkn_workflows::split_csv_line;
-
 /// Static bootstrap nodes for the Kademlia DHT bootstrap step.
 const STATIC_BOOTSTRAP_NODES: [&str; 4] = [
     "/ip4/44.206.245.139/tcp/4001/p2p/16Uiu2HAm4q3LZU2T9kgjKK4ysy6KZYKLq8KiXQyae4RHdF7uqSt4",

compute/src/utils/available_nodes.rs

@@ -1,4 +1,4 @@
-use dkn_p2p::libp2p_identity::Keypair;
+use dkn_p2p::libp2p_identity;
 use ecies::PublicKey;
 use eyre::{Context, Result};
 use libsecp256k1::{Message, SecretKey};
@@ -55,12 +55,12 @@ pub fn encrypt_bytes(data: impl AsRef<[u8]>, public_key: &PublicKey) -> Result<S
 /// Converts a `libsecp256k1::SecretKey` to a `libp2p_identity::secp256k1::Keypair`.
 /// To do this, we serialize the secret key and create a new keypair from it.
 #[inline]
-pub fn secret_to_keypair(secret_key: &SecretKey) -> Keypair {
+pub fn secret_to_keypair(secret_key: &SecretKey) -> libp2p_identity::Keypair {
     let bytes = secret_key.serialize();
 
     let secret_key = dkn_p2p::libp2p_identity::secp256k1::SecretKey::try_from_bytes(bytes)
         .expect("Failed to create secret key");
-    dkn_p2p::libp2p_identity::secp256k1::Keypair::from(secret_key).into()
+    libp2p_identity::secp256k1::Keypair::from(secret_key).into()
 }
 
 #[cfg(test)]

compute/src/utils/crypto.rs

@@ -33,3 +33,4 @@ eyre.workspace = true
 
 [dev-dependencies]
 env_logger.workspace = true
+tokio.workspace = true

p2p/Cargo.toml

@@ -0,0 +1,13 @@
+# DKN Peer-to-Peer Client
+
+## Installation
+
+Add the package via `git` within your Cargo dependencies:
+
+```toml
+dkn-p2p = { git = "https://github.com/firstbatchxyz/dkn-compute-node" }
+```
+
+## Usage
+
+TODO: !!!

p2p/README.md

@@ -4,9 +4,10 @@ use std::time::Duration;
 
 use libp2p::identity::{Keypair, PeerId, PublicKey};
 use libp2p::kad::store::MemoryStore;
+use libp2p::StreamProtocol;
 use libp2p::{autonat, dcutr, gossipsub, identify, kad, relay, swarm::NetworkBehaviour};
 
-use crate::versioning::{P2P_KADEMLIA_PROTOCOL, P2P_PROTOCOL_STRING};
+// use crate::versioning::{P2P_KADEMLIA_PROTOCOL, P2P_PROTOCOL_STRING};
 
 #[derive(NetworkBehaviour)]
 pub struct DriaBehaviour {
@@ -19,29 +20,37 @@ pub struct DriaBehaviour {
 }
 
 impl DriaBehaviour {
-    pub fn new(key: &Keypair, relay_behavior: relay::client::Behaviour) -> Self {
+    pub fn new(
+        key: &Keypair,
+        relay_behavior: relay::client::Behaviour,
+        identity_protocol: String,
+        kademlia_protocol: StreamProtocol,
+    ) -> Self {
         let public_key = key.public();
         let peer_id = public_key.to_peer_id();
         Self {
             relay: relay_behavior,
             gossipsub: create_gossipsub_behavior(peer_id),
-            kademlia: create_kademlia_behavior(peer_id),
+            kademlia: create_kademlia_behavior(peer_id, kademlia_protocol),
             autonat: create_autonat_behavior(peer_id),
             dcutr: create_dcutr_behavior(peer_id),
-            identify: create_identify_behavior(public_key),
+            identify: create_identify_behavior(public_key, identity_protocol),
         }
     }
 }
 
 /// Configures the Kademlia DHT behavior for the node.
 #[inline]
-fn create_kademlia_behavior(local_peer_id: PeerId) -> kad::Behaviour<MemoryStore> {
+fn create_kademlia_behavior(
+    local_peer_id: PeerId,
+    protocol_name: StreamProtocol,
+) -> kad::Behaviour<MemoryStore> {
     use kad::{Behaviour, Config};
 
     const QUERY_TIMEOUT_SECS: u64 = 5 * 60;
     const RECORD_TTL_SECS: u64 = 30;
 
-    let mut cfg = Config::new(P2P_KADEMLIA_PROTOCOL);
+    let mut cfg = Config::new(protocol_name);
     cfg.set_query_timeout(Duration::from_secs(QUERY_TIMEOUT_SECS))
         .set_record_ttl(Some(Duration::from_secs(RECORD_TTL_SECS)));
 
@@ -50,10 +59,13 @@ fn create_kademlia_behavior(local_peer_id: PeerId) -> kad::Behaviour<MemoryStore
 
 /// Configures the Identify behavior to allow nodes to exchange information like supported protocols.
 #[inline]
-fn create_identify_behavior(local_public_key: PublicKey) -> identify::Behaviour {
+fn create_identify_behavior(
+    local_public_key: PublicKey,
+    protocol_version: String,
+) -> identify::Behaviour {
     use identify::{Behaviour, Config};
 
-    let cfg = Config::new(P2P_PROTOCOL_STRING.to_string(), local_public_key);
+    let cfg = Config::new(protocol_version, local_public_key);
 
     Behaviour::new(cfg)
 }

p2p/src/behaviour.rs

@@ -8,14 +8,12 @@ use libp2p::kad::{GetClosestPeersError, GetClosestPeersOk, QueryResult};
 use libp2p::{
     autonat, gossipsub, identify, kad, multiaddr::Protocol, noise, swarm::SwarmEvent, tcp, yamux,
 };
-use libp2p::{Multiaddr, PeerId, Swarm, SwarmBuilder};
+use libp2p::{Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder};
 use libp2p_identity::Keypair;
-use std::time::Duration;
-use std::time::Instant;
-use versioning::{P2P_KADEMLIA_PREFIX, P2P_KADEMLIA_PROTOCOL, P2P_PROTOCOL_STRING};
+use std::time::{Duration, Instant};
 
 /// P2P client, exposes a simple interface to handle P2P communication.
-pub struct DriaP2P {
+pub struct DriaP2PClient {
     /// `Swarm` instance, everything is accesses through this one.
     swarm: Swarm<DriaBehaviour>,
     /// Peer count for All and Mesh peers.
@@ -25,6 +23,10 @@ pub struct DriaP2P {
     peer_count: (usize, usize),
     /// Last time the peer count was refreshed.
     peer_last_refreshed: Instant,
+    /// Identity protocol string to be used for the Identity behaviour.
+    identity_protocol: String,
+    /// Kademlia protocol, must match with other peers in the network.
+    kademlia_protocol: StreamProtocol,
 }
 
 /// Number of seconds before an idle connection is closed.
@@ -33,14 +35,24 @@ const IDLE_CONNECTION_TIMEOUT_SECS: u64 = 60;
 /// Number of seconds between refreshing the Kademlia DHT.
 const PEER_REFRESH_INTERVAL_SECS: u64 = 30;
 
-impl DriaP2P {
+impl DriaP2PClient {
     /// Creates a new P2P client with the given keypair and listen address.
+    ///
+    /// Can provide a list of bootstrap and relay nodes to connect to as well at the start.
+    ///
+    /// The `version` is used to create the protocol strings for the client, and its very important that
+    /// they match with the clients existing within the network.
     pub fn new(
         keypair: Keypair,
         listen_addr: Multiaddr,
         bootstraps: &[Multiaddr],
         relays: &[Multiaddr],
+        version: &str,
     ) -> Result<Self> {
+        let identity_protocol = format!("{}{}", P2P_IDENTITY_PREFIX, version);
+        let kademlia_protocol =
+            StreamProtocol::try_from_owned(format!("{}{}", P2P_KADEMLIA_PREFIX, version))?;
+
         // this is our peerId
         let node_peerid = keypair.public().to_peer_id();
         log::info!("Compute node peer address: {}", node_peerid);
@@ -54,7 +66,14 @@ impl DriaP2P {
             )?
             .with_quic()
             .with_relay_client(noise::Config::new, yamux::Config::default)?
-            .with_behaviour(|key, relay_behavior| Ok(DriaBehaviour::new(key, relay_behavior)))?
+            .with_behaviour(|key, relay_behavior| {
+                Ok(DriaBehaviour::new(
+                    key,
+                    relay_behavior,
+                    identity_protocol.clone(),
+                    kademlia_protocol.clone(),
+                ))
+            })?
             .with_swarm_config(|c| {
                 c.with_idle_connection_timeout(Duration::from_secs(IDLE_CONNECTION_TIMEOUT_SECS))
             })
@@ -107,6 +126,8 @@ impl DriaP2P {
             swarm,
             peer_count: (0, 0),
             peer_last_refreshed: Instant::now(),
+            identity_protocol,
+            kademlia_protocol,
         })
     }
 
@@ -231,12 +252,12 @@ impl DriaP2P {
     /// - For Kademlia, we check the kademlia protocol and then add the address to the Kademlia routing table.
     fn handle_identify_event(&mut self, peer_id: PeerId, info: identify::Info) {
         // check identify protocol string
-        if info.protocol_version != P2P_PROTOCOL_STRING {
+        if info.protocol_version != self.identity_protocol {
             log::warn!(
                 "Identify: Peer {} has different Identify protocol: (them {}, you {})",
                 peer_id,
                 info.protocol_version,
-                P2P_PROTOCOL_STRING
+                self.identity_protocol
             );
             return;
         }
@@ -248,7 +269,7 @@ impl DriaP2P {
             .find(|p| p.to_string().starts_with(P2P_KADEMLIA_PREFIX))
         {
             // if it matches our protocol, add it to the Kademlia routing table
-            if *kad_protocol == P2P_KADEMLIA_PROTOCOL {
+            if *kad_protocol == self.kademlia_protocol {
                 // filter listen addresses
                 let addrs = info.listen_addrs.into_iter().filter(|listen_addr| {
                     if let Some(Protocol::Ip4(ipv4_addr)) = listen_addr.iter().next() {
@@ -279,7 +300,7 @@ impl DriaP2P {
                     "Identify: Peer {} has different Kademlia version: (them {}, you {})",
                     peer_id,
                     kad_protocol,
-                    P2P_KADEMLIA_PROTOCOL
+                    self.kademlia_protocol
                 );
             }
         }