added autonat, some log rfks

Author: erhant

Cargo.lock

@@ -20,7 +20,6 @@ debug = true
 # async stuff
 tokio-util = { version = "0.7.10", features = ["rt"] }
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }
-async-trait = "0.1.81"
 
 # serialize & deserialize
 serde = { version = "1.0", features = ["derive"] }
@@ -31,8 +30,6 @@ reqwest = "0.12.5"
 
 # utilities
 dotenvy = "0.15.7"
-
-# randomization
 rand = "0.8.5"
 
 # logging & errors

Cargo.toml

@@ -10,7 +10,6 @@ authors = ["Erhan Tezcan <[email protected]>"]
 # async stuff
 tokio-util.workspace = true
 tokio.workspace = true
-async-trait.workspace = true
 
 # serialize & deserialize
 serde.workspace = true

compute/Cargo.toml

@@ -19,16 +19,13 @@ impl DriaComputeNode {
         // prepare durations for sleeps
         let mut diagnostic_refresh_interval =
             tokio::time::interval(Duration::from_secs(DIAGNOSTIC_REFRESH_INTERVAL_SECS));
+        diagnostic_refresh_interval.tick().await; // move each one tick
         let mut available_node_refresh_interval =
             tokio::time::interval(Duration::from_secs(AVAILABLE_NODES_REFRESH_INTERVAL_SECS));
+        available_node_refresh_interval.tick().await; // move each one tick
         let mut heartbeat_interval =
             tokio::time::interval(Duration::from_secs(HEARTBEAT_INTERVAL_SECS));
 
-        // move each one tick
-        available_node_refresh_interval.tick().await;
-        diagnostic_refresh_interval.tick().await;
-        heartbeat_interval.tick().await;
-
         loop {
             tokio::select! {
                 // a task is completed by the worker & should be responded to the requesting peer

compute/src/node/core.rs

@@ -27,7 +27,7 @@ impl DriaComputeNode {
                 request_id,
                 channel,
             } => {
-                log::info!("Received a request ({}) from {}", request_id, peer_id);
+                log::debug!("Received a request ({}) from {}", request_id, peer_id);
 
                 // ensure that message is from the known RPCs
                 if self.dria_rpc.peer_id != peer_id {
@@ -42,7 +42,7 @@ impl DriaComputeNode {
                 response,
                 request_id,
             } => {
-                log::info!("Received a response ({}) from {}", request_id, peer_id);
+                log::debug!("Received a response ({}) from {}", request_id, peer_id);
                 if let Err(e) = self.handle_response(peer_id, response).await {
                     log::error!("Error handling response: {:?}", e);
                 }
@@ -57,6 +57,11 @@ impl DriaComputeNode {
     #[inline]
     async fn handle_response(&mut self, peer_id: PeerId, data: Vec<u8>) -> Result<()> {
         if let Ok(heartbeat_response) = HeartbeatRequester::try_parse_response(&data) {
+            log::info!(
+                "Received a {} response from {}",
+                "heartbeat".blue(),
+                peer_id
+            );
             HeartbeatRequester::handle_ack(self, heartbeat_response).await
         } else {
             Err(eyre::eyre!("Received unhandled request from {}", peer_id))
@@ -196,7 +201,12 @@ impl DriaComputeNode {
     pub(crate) async fn send_heartbeat(&mut self) -> Result<()> {
         let peer_id = self.dria_rpc.peer_id;
         let request_id = HeartbeatRequester::send_heartbeat(self, peer_id).await?;
-        log::info!("Sent heartbeat request ({}) to {}", request_id, peer_id);
+        log::info!(
+            "Sent {} request ({}) to {}",
+            "heartbeat".blue(),
+            request_id,
+            peer_id
+        );
 
         Ok(())
     }

compute/src/node/reqres.rs

@@ -38,7 +38,7 @@ pub struct HeartbeatResponse {
 
 impl IsResponder for HeartbeatRequester {
     type Request = DriaMessage; // TODO: HeartbeatRequest;
-    type Response = DriaMessage; // TODO: HeartbeatResponse;
+    type Response = HeartbeatResponse;
 }
 
 /// Any acknowledged heartbeat that is older than this duration is considered dead.
@@ -80,10 +80,8 @@ impl HeartbeatRequester {
     /// Handles the heartbeat request received from the network.
     pub(crate) async fn handle_ack(
         node: &mut DriaComputeNode,
-        ack_message: DriaMessage,
+        res: HeartbeatResponse,
     ) -> Result<()> {
-        let res = ack_message.parse_payload::<HeartbeatResponse>()?;
-
         if let Some(deadline) = node.heartbeats.remove(&res.heartbeat_id) {
             if let Some(err) = res.error {
                 Err(eyre!("Heartbeat was not acknowledged: {}", err))

compute/src/reqres/heartbeat.rs

@@ -10,7 +10,6 @@ authors = ["Erhan Tezcan <[email protected]>"]
 # async stuff
 tokio-util.workspace = true
 tokio.workspace = true
-async-trait.workspace = true
 
 # serialize & deserialize
 serde.workspace = true

monitor/Cargo.toml

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

p2p/Cargo.toml

@@ -1,22 +1,24 @@
 use eyre::Result;
 use libp2p::identity::{Keypair, PublicKey};
-use libp2p::StreamProtocol;
-use libp2p::{identify, request_response};
+use libp2p::{autonat, 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),
         }
     }
 }
@@ -30,11 +32,11 @@ fn create_request_response_behaviour(
 ) -> request_response::cbor::Behaviour<Vec<u8>, Vec<u8>> {
     use request_response::{Behaviour, Config, ProtocolSupport};
 
-    const REQUEST_RESPONSE_TIMEOUT_SECS: u64 = 180;
+    const REQUEST_RESPONSE_TIMEOUT: Duration = Duration::from_secs(180);
 
     Behaviour::new(
         [(protocol_name, ProtocolSupport::Full)],
-        Config::default().with_request_timeout(Duration::from_secs(REQUEST_RESPONSE_TIMEOUT_SECS)),
+        Config::default().with_request_timeout(REQUEST_RESPONSE_TIMEOUT),
     )
 }
 
@@ -48,3 +50,17 @@ 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()
+        },
+    )
+}

p2p/src/behaviour.rs

@@ -4,7 +4,7 @@ use libp2p::swarm::{
     dial_opts::{DialOpts, PeerCondition},
     SwarmEvent,
 };
-use libp2p::{identify, noise, request_response, tcp, yamux};
+use libp2p::{autonat, identify, noise, request_response, tcp, yamux};
 use libp2p::{Multiaddr, PeerId, Swarm, SwarmBuilder};
 use libp2p_identity::Keypair;
 use std::time::Duration;
@@ -90,6 +90,7 @@ impl DriaP2PClient {
 
         // create p2p client itself
         let (reqres_tx, reqres_rx) = mpsc::channel(MSG_CHANNEL_BUFSIZE);
+
         let client = Self {
             peer_id,
             swarm,
@@ -241,17 +242,23 @@ impl DriaP2PClient {
             })) => self.handle_identify_event(peer_id, info),
 
             SwarmEvent::NewListenAddr { address, .. } => {
-                log::warn!("Local node is listening on {}", address);
+                log::warn!("Local node is listening on {address}");
             }
             SwarmEvent::NewExternalAddrOfPeer { peer_id, address } => {
-                log::info!(
-                    "External address of peer {} confirmed: {}",
-                    peer_id,
-                    address
-                );
+                log::info!("External address of peer {peer_id} confirmed: {address}");
             }
             SwarmEvent::ExternalAddrConfirmed { address } => {
-                log::info!("External address confirmed: {}", address);
+                log::info!("External address confirmed: {address}");
+            }
+
+            /*****************************************
+             * AutoNAT stuff                         *
+             *****************************************/
+            SwarmEvent::Behaviour(DriaBehaviourEvent::Autonat(autonat::Event::StatusChanged {
+                old,
+                new,
+            })) => {
+                log::info!("AutoNAT status changed from {old:?} to {new:?}");
             }
 
             /*****************************************
@@ -303,7 +310,18 @@ impl DriaP2PClient {
                 );
             }
 
-            event => log::trace!("Unhandled Swarm Event: {:?}", event),
+            SwarmEvent::ExpiredListenAddr {
+                address,
+                listener_id,
+            } => {
+                log::warn!("Listener {listener_id} expired: {address}");
+            }
+
+            SwarmEvent::ListenerError { listener_id, error } => {
+                log::error!("Listener {listener_id} error: {error}");
+            }
+
+            event => log::debug!("Unhandled Swarm Event: {:?}", event),
         }
     }
 
@@ -313,6 +331,7 @@ impl DriaP2PClient {
     ///
     /// - 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) {
+        println!("{}: {:?}", peer_id, info.protocols);
         // check identify protocol string
         if info.protocol_version != self.protocol.identity {
             log::warn!(