Skip to content

Commit 49aebb2

Browse files
committed
fix bad error return in node, some log fixes & version fix
1 parent 624a60a commit 49aebb2

File tree

7 files changed

+19
-26
lines changed

7 files changed

+19
-26
lines changed

Cargo.lock

Lines changed: 4 additions & 4 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

Cargo.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ default-members = ["compute"]
99

1010
[workspace.package]
1111
edition = "2021"
12-
version = "0.3.0"
12+
version = "0.3.1"
1313
license = "Apache-2.0"
1414
readme = "README.md"
1515

compute/src/main.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ async fn main() -> Result<()> {
1515
.filter_module("dkn_compute", log::LevelFilter::Info)
1616
.filter_module("dkn_p2p", log::LevelFilter::Info)
1717
.filter_module("dkn_workflows", log::LevelFilter::Info)
18+
.filter_module("libp2p", log::LevelFilter::Error)
1819
.parse_default_env() // reads RUST_LOG variable
1920
.init();
2021

compute/src/node/core.rs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,9 @@ impl DriaComputeNode {
3232
let task_response_msg = task_response_msg_opt.ok_or(
3333
eyre!("Publish channel closed unexpectedly, we still have {} batch and {} single tasks.", self.pending_tasks_batch.len(), self.pending_tasks_single.len())
3434
)?; {
35-
self.handle_task_response(task_response_msg).await?;
35+
if let Err(e) = self.handle_task_response(task_response_msg).await {
36+
log::error!("Error responding to task: {:?}", e);
37+
}
3638
}
3739
},
3840

p2p/src/behaviour.rs

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -54,10 +54,13 @@ fn create_request_response_behaviour(
5454
protocol_name: StreamProtocol,
5555
) -> request_response::cbor::Behaviour<Vec<u8>, Vec<u8>> {
5656
use request_response::{Behaviour, Config, ProtocolSupport};
57-
const REQUEST_RESPONSE_TIMEOUT: u64 = 180;
5857

59-
Behaviour::new([(protocol_name, ProtocolSupport::Full)],
60-
Config::default().with_request_timeout(Duration::from_secs(REQUEST_RESPONSE_TIMEOUT)))
58+
const REQUEST_RESPONSE_TIMEOUT_SECS: u64 = 180;
59+
60+
Behaviour::new(
61+
[(protocol_name, ProtocolSupport::Full)],
62+
Config::default().with_request_timeout(Duration::from_secs(REQUEST_RESPONSE_TIMEOUT_SECS)),
63+
)
6164
}
6265

6366
/// Configures the connection limits.

p2p/src/client.rs

Lines changed: 3 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -263,21 +263,9 @@ impl DriaP2PClient {
263263
);
264264
}
265265
DriaP2PCommand::Peers { sender } => {
266-
let mesh = self
267-
.swarm
268-
.behaviour()
269-
.gossipsub
270-
.all_mesh_peers()
271-
.cloned()
272-
.collect();
273-
let all = self
274-
.swarm
275-
.behaviour()
276-
.gossipsub
277-
.all_peers()
278-
.map(|(p, _)| p)
279-
.cloned()
280-
.collect();
266+
let gossipsub = &self.swarm.behaviour().gossipsub;
267+
let mesh = gossipsub.all_mesh_peers().cloned().collect();
268+
let all = gossipsub.all_peers().map(|(p, _)| p).cloned().collect();
281269
let _ = sender.send((mesh, all));
282270
}
283271
DriaP2PCommand::PeerCounts { sender } => {
@@ -340,7 +328,6 @@ impl DriaP2PClient {
340328
response,
341329
} => {
342330
// while we support the protocol, we dont really make any requests
343-
// TODO: should p2p crate support this?
344331
log::warn!(
345332
"Unexpected response message with request_id {}: {:?}",
346333
request_id,

p2p/src/commands.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -183,7 +183,7 @@ impl DriaP2PCommander {
183183
receiver
184184
.await
185185
.wrap_err("could not receive")?
186-
.wrap_err("could not publish")
186+
.wrap_err("could not respond")
187187
}
188188

189189
pub async fn request(

0 commit comments

Comments
 (0)