fix bad error return in node, some log fixes & version fix

firstbatchxyz · Feb 3, 2025 · 49aebb2 · 49aebb2
1 parent 624a60a
commit 49aebb2
Show file tree

Hide file tree

Showing 7 changed files with 19 additions and 26 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -9,7 +9,7 @@ default-members = ["compute"]
 
 [workspace.package]
 edition = "2021"
-version = "0.3.0"
+version = "0.3.1"
 license = "Apache-2.0"
 readme = "README.md"
 

diff --git a/compute/src/main.rs b/compute/src/main.rs
@@ -15,6 +15,7 @@ async fn main() -> Result<()> {
         .filter_module("dkn_compute", log::LevelFilter::Info)
         .filter_module("dkn_p2p", log::LevelFilter::Info)
         .filter_module("dkn_workflows", log::LevelFilter::Info)
+        .filter_module("libp2p", log::LevelFilter::Error)
         .parse_default_env() // reads RUST_LOG variable
         .init();
 

diff --git a/compute/src/node/core.rs b/compute/src/node/core.rs
@@ -32,7 +32,9 @@ impl DriaComputeNode {
                     let task_response_msg = task_response_msg_opt.ok_or(
                       eyre!("Publish channel closed unexpectedly, we still have {} batch and {} single tasks.", self.pending_tasks_batch.len(), self.pending_tasks_single.len())
                     )?; {
-                        self.handle_task_response(task_response_msg).await?;
+                        if let Err(e) = self.handle_task_response(task_response_msg).await {
+                          log::error!("Error responding to task: {:?}", e);
+                      }
                     }
                 },
 

diff --git a/p2p/src/behaviour.rs b/p2p/src/behaviour.rs
@@ -54,10 +54,13 @@ fn create_request_response_behaviour(
     protocol_name: StreamProtocol,
 ) -> request_response::cbor::Behaviour<Vec<u8>, Vec<u8>> {
     use request_response::{Behaviour, Config, ProtocolSupport};
-    const REQUEST_RESPONSE_TIMEOUT: u64 = 180;
 
-    Behaviour::new([(protocol_name, ProtocolSupport::Full)],
-                   Config::default().with_request_timeout(Duration::from_secs(REQUEST_RESPONSE_TIMEOUT)))
+    const REQUEST_RESPONSE_TIMEOUT_SECS: u64 = 180;
+
+    Behaviour::new(
+        [(protocol_name, ProtocolSupport::Full)],
+        Config::default().with_request_timeout(Duration::from_secs(REQUEST_RESPONSE_TIMEOUT_SECS)),
+    )
 }
 
 /// Configures the connection limits.

diff --git a/p2p/src/client.rs b/p2p/src/client.rs
@@ -263,21 +263,9 @@ impl DriaP2PClient {
                 );
             }
             DriaP2PCommand::Peers { sender } => {
-                let mesh = self
-                    .swarm
-                    .behaviour()
-                    .gossipsub
-                    .all_mesh_peers()
-                    .cloned()
-                    .collect();
-                let all = self
-                    .swarm
-                    .behaviour()
-                    .gossipsub
-                    .all_peers()
-                    .map(|(p, _)| p)
-                    .cloned()
-                    .collect();
+                let gossipsub = &self.swarm.behaviour().gossipsub;
+                let mesh = gossipsub.all_mesh_peers().cloned().collect();
+                let all = gossipsub.all_peers().map(|(p, _)| p).cloned().collect();
                 let _ = sender.send((mesh, all));
             }
             DriaP2PCommand::PeerCounts { sender } => {
@@ -340,7 +328,6 @@ impl DriaP2PClient {
                     response,
                 } => {
                     // while we support the protocol, we dont really make any requests
-                    // TODO: should p2p crate support this?
                     log::warn!(
                         "Unexpected response message with request_id {}: {:?}",
                         request_id,

diff --git a/p2p/src/commands.rs b/p2p/src/commands.rs
@@ -183,7 +183,7 @@ impl DriaP2PCommander {
         receiver
             .await
             .wrap_err("could not receive")?
-            .wrap_err("could not publish")
+            .wrap_err("could not respond")
     }
 
     pub async fn request(