always dial mode

Author: erhant

Cargo.lock

@@ -9,7 +9,7 @@ default-members = ["compute"]
 
 [workspace.package]
 edition = "2021"
-version = "0.3.3"
+version = "0.3.4"
 license = "Apache-2.0"
 readme = "README.md"
 

Cargo.toml

@@ -11,7 +11,7 @@ impl DriaComputeNode {
         /// Number of seconds between refreshing for diagnostic prints.
         const DIAGNOSTIC_REFRESH_INTERVAL_SECS: u64 = 30;
         /// Number of seconds between refreshing the available nodes.
-        const AVAILABLE_NODES_REFRESH_INTERVAL_SECS: u64 = 10 * 60; // 30 minutes
+        const AVAILABLE_NODES_REFRESH_INTERVAL_SECS: u64 = 10 * 60;
 
         // prepare durations for sleeps
         let mut diagnostic_refresh_interval =

compute/src/node/core.rs

@@ -1,3 +1,4 @@
+use dkn_p2p::libp2p::multiaddr::Protocol;
 use std::time::Duration;
 use tokio::time::Instant;
 
@@ -86,23 +87,31 @@ impl DriaComputeNode {
         };
 
         // dial all rpc nodes
-        for rpc_addr in self.dria_nodes.rpc_nodes.iter() {
-            log::info!("Dialling RPC node: {}", rpc_addr);
-
-            let fut = self.p2p.dial(rpc_addr.clone());
-            match tokio::time::timeout(Duration::from_secs(10), fut).await {
-                Err(timeout) => {
-                    log::error!("Timeout dialling RPC node: {:?}", timeout);
-                }
-                Ok(res) => match res {
-                    Err(e) => {
-                        log::warn!("Error dialling RPC node: {:?}", e);
+        for addr in self.dria_nodes.rpc_nodes.iter() {
+            log::info!("Dialling RPC node: {}", addr);
+
+            // get peer id from rpc address
+            if let Some(peer_id) = addr.iter().find_map(|p| match p {
+                Protocol::P2p(peer_id) => Some(peer_id),
+                _ => None,
+            }) {
+                let fut = self.p2p.dial(peer_id, addr.clone());
+                match tokio::time::timeout(Duration::from_secs(10), fut).await {
+                    Err(timeout) => {
+                        log::error!("Timeout dialling RPC node: {:?}", timeout);
                     }
-                    Ok(_) => {
-                        log::info!("Successfully dialled RPC node: {}", rpc_addr);
-                    }
-                },
-            };
+                    Ok(res) => match res {
+                        Err(e) => {
+                            log::warn!("Error dialling RPC node: {:?}", e);
+                        }
+                        Ok(_) => {
+                            log::info!("Successfully dialled RPC node: {}", addr);
+                        }
+                    },
+                };
+            } else {
+                log::warn!("Missing peerID in address: {}", addr);
+            }
         }
 
         log::info!("Finished refreshing!");

compute/src/node/diagnostic.rs

@@ -3,6 +3,7 @@ use libp2p::futures::StreamExt;
 use libp2p::gossipsub::{Message, MessageId};
 use libp2p::kad::{GetClosestPeersError, GetClosestPeersOk, QueryResult};
 use libp2p::request_response::{self, ResponseChannel};
+use libp2p::swarm::dial_opts::{DialOpts, PeerCondition};
 use libp2p::swarm::SwarmEvent;
 use libp2p::{autonat, gossipsub, identify, kad, multiaddr::Protocol, noise, tcp, yamux};
 use libp2p::{Multiaddr, PeerId, Swarm, SwarmBuilder};
@@ -182,8 +183,16 @@ impl DriaP2PClient {
     /// Handles a single command, which originates from `DriaP2PCommander`.
     pub async fn handle_command(&mut self, command: DriaP2PCommand) {
         match command {
-            DriaP2PCommand::Dial { peer_id, sender } => {
-                let _ = sender.send(self.swarm.dial(peer_id));
+            DriaP2PCommand::Dial {
+                peer_id,
+                address,
+                sender,
+            } => {
+                let opts = DialOpts::peer_id(peer_id)
+                    .addresses(vec![address])
+                    .condition(PeerCondition::Always)
+                    .build();
+                let _ = sender.send(self.swarm.dial(opts));
             }
             DriaP2PCommand::NetworkInfo { sender } => {
                 let _ = sender.send(self.swarm.network_info());

p2p/src/client.rs

@@ -22,9 +22,10 @@ pub enum DriaP2PCommand {
     PeerCounts {
         sender: oneshot::Sender<(usize, usize)>,
     },
-    /// Dial a peer.
+    /// Dial a known peer.
     Dial {
-        peer_id: Multiaddr,
+        peer_id: PeerId,
+        address: Multiaddr,
         sender: oneshot::Sender<Result<(), swarm::DialError>>,
     },
     /// Subscribe to a topic.
@@ -206,11 +207,15 @@ impl DriaP2PCommander {
     }
 
     /// Dials a given peer.
-    pub async fn dial(&mut self, peer_id: Multiaddr) -> Result<()> {
+    pub async fn dial(&mut self, peer_id: PeerId, address: Multiaddr) -> Result<()> {
         let (sender, receiver) = oneshot::channel();
 
         self.sender
-            .send(DriaP2PCommand::Dial { peer_id, sender })
+            .send(DriaP2PCommand::Dial {
+                peer_id,
+                address,
+                sender,
+            })
             .await
             .wrap_err("could not send")?;