added logs for req-res

Author: erhant

compute/src/node.rs

@@ -354,17 +354,28 @@ impl DriaComputeNode {
 
         // respond w.r.t data
         let response_data = if let Ok(req) = SpecResponder::try_parse_request(&data) {
+            log::info!(
+                "Got a spec request from peer {} with id {}",
+                peer_id,
+                req.request_id
+            );
+
             let response = SpecResponder::respond(req, self.spec_collector.collect().await);
-            serde_json::to_vec(&response).unwrap()
+            serde_json::to_vec(&response)?
         } else if let Ok(req) = WorkflowResponder::try_parse_request(&data) {
             log::info!("Received a task request with id: {}", req.task_id);
             return Err(eyre::eyre!(
                 "REQUEST RESPONSE FOR TASKS ARE NOT IMPLEMENTED YET"
             ));
         } else {
-            return Err(eyre::eyre!("Received unknown request: {:?}", data));
+            return Err(eyre::eyre!(
+                "Received unknown request from {}: {:?}",
+                peer_id,
+                data,
+            ));
         };
 
+        log::info!("Responding to peer {}", peer_id);
         self.p2p.respond(response_data, channel).await
     }
 

compute/src/responders/specs.rs

@@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
 #[derive(Serialize, Deserialize)]
 pub struct Request {
     /// UUID of the specs request, prevents replay attacks.
-    request_id: String,
+    pub request_id: String,
 }
 
 #[derive(Serialize, Deserialize)]

p2p/src/behaviour.rs

@@ -53,12 +53,9 @@ impl DriaBehaviour {
 fn create_request_response_behaviour(
     protocol_name: StreamProtocol,
 ) -> request_response::cbor::Behaviour<Vec<u8>, Vec<u8>> {
-    use request_response::{Behaviour, ProtocolSupport};
+    use request_response::{Behaviour, Config, ProtocolSupport};
 
-    Behaviour::new(
-        [(protocol_name, ProtocolSupport::Full)],
-        request_response::Config::default(),
-    )
+    Behaviour::new([(protocol_name, ProtocolSupport::Full)], Config::default())
 }
 
 /// Configures the connection limits.

p2p/src/protocol.rs

@@ -41,17 +41,22 @@ impl Default for DriaP2PProtocol {
 
 impl DriaP2PProtocol {
     /// Creates a new instance of the protocol with the given `name` and `version`.
-    pub fn new(name: &str, version: &str) -> Self {
+    pub fn new(name: impl ToString, version: impl ToString) -> Self {
+        let name = name.to_string();
+        let version = version.to_string();
+
         let identity = format!("{}/{}", name, version);
-        let kademlia = format!("/{}/kad/{}", name, version);
-        let request_response = format!("/{}/rr/{}", name, version);
+        let kademlia =
+            StreamProtocol::try_from_owned(format!("/{}/kad/{}", name, version)).unwrap();
+        let request_response =
+            StreamProtocol::try_from_owned(format!("/{}/rr/{}", name, version)).unwrap();
 
         Self {
-            name: name.to_string(),
-            version: version.to_string(),
+            name,
+            version,
             identity,
-            kademlia: StreamProtocol::try_from_owned(kademlia).unwrap(), // guaranteed to unwrap
-            request_response: StreamProtocol::try_from_owned(request_response).unwrap(), // guaranteed to unwrap
+            kademlia,
+            request_response,
         }
     }
 

p2p/tests/gossipsub_test.rs

@@ -35,19 +35,11 @@ async fn test_gossipsub() -> Result<()> {
         listen_addr,
         &nodes,
         DriaP2PProtocol::default(),
-    )
-    .expect("could not create p2p client");
-
-    // spawn task
+    )?;
     let task_handle = tokio::spawn(async move { client.run().await });
 
-    // subscribe to the given topic
-    commander
-        .subscribe(TOPIC)
-        .await
-        .expect("could not subscribe");
-
     // wait for a single gossipsub message on this topic
+    commander.subscribe(TOPIC).await?;
     log::info!("Waiting for messages...");
     let message = msg_rx.recv().await;
     match message {
@@ -58,20 +50,13 @@ async fn test_gossipsub() -> Result<()> {
             log::warn!("No message received for topic: {}", TOPIC);
         }
     }
+    commander.unsubscribe(TOPIC).await?;
 
-    // unsubscribe to the given topic
-    commander
-        .unsubscribe(TOPIC)
-        .await
-        .expect("could not unsubscribe");
-
-    // close command channel
-    commander.shutdown().await.expect("could not shutdown");
-
-    // close message channel
+    // close everything
+    commander.shutdown().await?;
     msg_rx.close();
 
-    log::info!("Waiting for p2p task to finish...");
+    // wait for handle to return
     task_handle.await?;
 
     log::info!("Done!");

p2p/tests/request_test.rs

@@ -1,5 +1,6 @@
 use std::str::FromStr;
 
+use dkn_p2p::DriaNetworkType::Community;
 use dkn_p2p::{DriaNodes, DriaP2PClient, DriaP2PProtocol};
 use eyre::Result;
 use libp2p::PeerId;
@@ -25,9 +26,9 @@ async fn test_request_message() -> Result<()> {
     let listen_addr = "/ip4/0.0.0.0/tcp/4001".parse()?;
 
     // prepare nodes
-    let nodes = DriaNodes::new(dkn_p2p::DriaNetworkType::Community)
-    .with_bootstrap_nodes(["/ip4/44.206.245.139/tcp/4001/p2p/16Uiu2HAm4q3LZU2T9kgjKK4ysy6KZYKLq8KiXQyae4RHdF7uqSt4".parse()?])
-    .with_relay_nodes(["/ip4/34.201.33.141/tcp/4001/p2p/16Uiu2HAkuXiV2CQkC9eJgU6cMnJ9SMARa85FZ6miTkvn5fuHNufa".parse()?]);
+    let nodes = DriaNodes::new(Community)
+        .with_bootstrap_nodes(Community.get_static_bootstrap_nodes())
+        .with_relay_nodes(Community.get_static_relay_nodes());
 
     // spawn P2P client in another task
     let (client, mut commander, mut msg_rx, mut req_rx) = DriaP2PClient::new(