rm autonat, add re-dial, increase timeout

Author: erhant

Cargo.lock

@@ -20,7 +20,7 @@ impl DriaComputeNode {
     }
 
     /// Peer refresh simply reports the peer count to the user.
-    pub(crate) async fn handle_diagnostic_refresh(&self) {
+    pub(crate) async fn handle_diagnostic_refresh(&mut self) {
         let mut diagnostics = vec![format!("Diagnostics (v{}):", DRIA_COMPUTE_NODE_VERSION)];
 
         // print steps
@@ -38,6 +38,21 @@ impl DriaComputeNode {
                 "Completed Tasks (single/batch): {} / {}",
                 self.completed_tasks_single, self.completed_tasks_batch
             ));
+
+            diagnostics.push(format!(
+                "RPC {}: {}",
+                self.dria_rpc.peer_id,
+                if self
+                    .p2p
+                    .is_connected(self.dria_rpc.peer_id)
+                    .await
+                    .unwrap_or(false)
+                {
+                    "Connected".green()
+                } else {
+                    "Disconnected".red()
+                }
+            ));
         }
 
         // print peer id and address

compute/src/node/diagnostic.rs

@@ -1,6 +1,6 @@
 use colored::Colorize;
 use dkn_p2p::libp2p::{
-    request_response::{self, ResponseChannel},
+    request_response::{self, OutboundRequestId, ResponseChannel},
     PeerId,
 };
 use eyre::{eyre, Result};
@@ -43,7 +43,7 @@ impl DriaComputeNode {
                 request_id,
             } => {
                 log::debug!("Received a response ({}) from {}", request_id, peer_id);
-                if let Err(e) = self.handle_response(peer_id, response).await {
+                if let Err(e) = self.handle_response(peer_id, request_id, response).await {
                     log::error!("Error handling response: {:?}", e);
                 }
             }
@@ -55,12 +55,16 @@ impl DriaComputeNode {
     /// - Internally, the data is expected to be some JSON serialized data that is expected to be parsed and handled.
     /// - Can be inlined because it is only called by [`DriaComputeNode::handle_reqres`].
     #[inline]
-    async fn handle_response(&mut self, peer_id: PeerId, data: Vec<u8>) -> Result<()> {
+    async fn handle_response(
+        &mut self,
+        peer_id: PeerId,
+        request_id: OutboundRequestId,
+        data: Vec<u8>,
+    ) -> Result<()> {
         if let Ok(heartbeat_response) = HeartbeatRequester::try_parse_response(&data) {
             log::info!(
-                "Received a {} response from {}",
+                "Received a {} response ({request_id}) from {peer_id}",
                 "heartbeat".blue(),
-                peer_id
             );
             HeartbeatRequester::handle_ack(self, heartbeat_response).await
         } else {
@@ -97,19 +101,17 @@ impl DriaComputeNode {
         channel: ResponseChannel<Vec<u8>>,
     ) -> Result<()> {
         log::info!(
-            "Got a {} request from peer {} with id {}",
+            "Got a {} request from peer {peer_id} with id {}",
             "spec".green(),
-            peer_id,
             spec_request.request_id
         );
 
         let response = SpecResponder::respond(spec_request, self.spec_collector.collect().await);
         let response_data = serde_json::to_vec(&response)?;
 
         log::info!(
-            "Responding to {} request from peer {} with id {}",
+            "Responding to {} request from peer {peer_id} with id {}",
             "spec".green(),
-            peer_id,
             response.request_id
         );
         self.p2p.respond(response_data, channel).await?;

compute/src/node/reqres.rs

@@ -52,27 +52,14 @@ mod tests {
     #[test]
     fn test_stats() {
         let mut stats = TaskStats::default();
+        let unix_epoch = chrono::DateTime::<chrono::Utc>::default();
 
-        assert_eq!(
-            stats.received_at,
-            chrono::DateTime::<chrono::Utc>::default()
-        );
+        assert_eq!(stats.received_at, unix_epoch);
         stats = stats.record_received_at();
-        assert_ne!(
-            stats.received_at,
-            chrono::DateTime::<chrono::Utc>::default()
-        );
+        assert_ne!(stats.received_at, unix_epoch);
 
-        assert_eq!(
-            stats.published_at,
-            chrono::DateTime::<chrono::Utc>::default()
-        );
+        assert_eq!(stats.published_at, unix_epoch);
         stats = stats.record_published_at();
-        assert_ne!(
-            stats.published_at,
-            chrono::DateTime::<chrono::Utc>::default()
-        );
-
-        println!("{:?}", stats);
+        assert_ne!(stats.published_at, unix_epoch);
     }
 }

compute/src/payloads/stats.rs

@@ -56,6 +56,6 @@ mod tests {
         let steps = get_points("0xa43536a6032a3907ccf60e8109429ee1047b207c")
             .await
             .unwrap();
-        println!("{:?}", steps);
+        assert!(steps.score != 0);
     }
 }

compute/src/utils/points.rs

@@ -67,6 +67,6 @@ mod tests {
     #[tokio::test]
     async fn test_dria_nodes() {
         let node = DriaRPC::new(DriaNetworkType::Community).await;
-        println!("{:?}", node);
+        assert!(node.is_ok());
     }
 }

compute/src/utils/rpc.rs

@@ -94,8 +94,5 @@ mod tests {
         assert!(!specs.os.is_empty());
         assert!(!specs.arch.is_empty());
         assert!(specs.lookup.is_some());
-
-        // print optionally:
-        // println!("{}", serde_json::to_string_pretty(&specs).unwrap());
     }
 }

compute/src/utils/specs.rs

@@ -324,7 +324,7 @@ mod tests {
             let result = publish_rx.recv().await.unwrap();
             log::info!("Got result {}", i + 1,);
             if result.result.is_err() {
-                println!("Error: {:?}", result.result);
+                log::error!("Error: {:?}", result.result);
             }
             results.push(result);
         }

compute/src/workers/task.rs

@@ -15,7 +15,6 @@ libp2p = { git = "https://github.com/anilaltuner/rust-libp2p.git", rev = "7ce9f9
   "identify",
   "tokio",
   "noise",
-  "autonat",
   "macros",
   "request-response",
   "cbor",

p2p/Cargo.toml

@@ -1,24 +1,21 @@
 use eyre::Result;
 use libp2p::identity::{Keypair, PublicKey};
-use libp2p::{autonat, identify, request_response, PeerId, StreamProtocol};
+use libp2p::{identify, request_response, PeerId, StreamProtocol};
 use std::time::Duration;
 
 #[derive(libp2p::swarm::NetworkBehaviour)]
 pub struct DriaBehaviour {
     pub identify: identify::Behaviour,
-    pub autonat: autonat::Behaviour,
     pub request_response: request_response::cbor::Behaviour<Vec<u8>, Vec<u8>>,
 }
 
 impl DriaBehaviour {
     pub fn new(key: &Keypair, identity_protocol: String, reqres_protocol: StreamProtocol) -> Self {
         let public_key = key.public();
-        let peer_id = public_key.to_peer_id();
 
         Self {
             identify: create_identify_behaviour(public_key, identity_protocol),
             request_response: create_request_response_behaviour(reqres_protocol),
-            autonat: create_autonat_behaviour(peer_id),
         }
     }
 }
@@ -50,17 +47,3 @@ fn create_identify_behaviour(
 
     Behaviour::new(Config::new(protocol_version, local_public_key))
 }
-
-/// Configures the Autonat behavior to assist in network address translation detection.
-#[inline]
-fn create_autonat_behaviour(local_peer_id: PeerId) -> autonat::Behaviour {
-    use autonat::{Behaviour, Config};
-
-    Behaviour::new(
-        local_peer_id,
-        Config {
-            // only_global_ips: false,
-            ..Default::default()
-        },
-    )
-}