msg refactors, fix few clones

Author: erhant

Cargo.lock

@@ -7,7 +7,7 @@ default-members = ["compute"]
 
 [workspace.package]
 edition = "2021"
-version = "0.2.24"
+version = "0.2.25"
 license = "Apache-2.0"
 readme = "README.md"
 

Cargo.toml

@@ -34,8 +34,6 @@ rand.workspace = true
 env_logger.workspace = true
 log.workspace = true
 eyre.workspace = true
-# tracing = { version = "0.1.40" }
-# tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 
 # encryption (ecies) & signatures (ecdsa) & hashing & bloom-filters
 ecies = { version = "0.2", default-features = false, features = ["pure"] }
@@ -48,7 +46,7 @@ fastbloom-rs = "0.5.9"
 dkn-p2p = { path = "../p2p" }
 dkn-workflows = { path = "../workflows" }
 
-# Vendor OpenSSL so that its easier to build cross-platform packages
+# vendor OpenSSL so that its easier to build cross-platform packages
 [dependencies.openssl]
 version = "*"
 features = ["vendored"]

compute/Cargo.toml

@@ -131,7 +131,7 @@ impl DriaComputeNodeConfig {
             .map(|addr| addr.trim_matches('"').to_string())
             .unwrap_or(DEFAULT_P2P_LISTEN_ADDR.to_string());
         let p2p_listen_addr = Multiaddr::from_str(&p2p_listen_addr_str)
-            .expect("Could not parse the given P2P listen address.");
+            .expect("could not parse the given P2P listen address.");
 
         // parse network type
         let network_type = env::var("DKN_NETWORK")

compute/src/config.rs

@@ -11,7 +11,9 @@ pub struct PingpongHandler;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 struct PingpongPayload {
+    /// UUID of the ping request, prevents replay attacks.
     uuid: String,
+    /// Deadline for the ping request.
     deadline: u128,
 }
 
@@ -37,11 +39,11 @@ impl PingpongHandler {
     /// 7. Returns `MessageAcceptance::Accept` so that ping is propagated to others as well.
     pub(crate) async fn handle_ping(
         node: &mut DriaComputeNode,
-        message: DKNMessage,
+        ping_message: &DKNMessage,
     ) -> Result<MessageAcceptance> {
-        let pingpong = message
+        let pingpong = ping_message
             .parse_payload::<PingpongPayload>(true)
-            .wrap_err("Could not parse ping request")?;
+            .wrap_err("could not parse ping request")?;
 
         // check deadline
         let current_time = get_current_time_nanos();

compute/src/handlers/pingpong.rs

@@ -31,14 +31,12 @@ impl WorkflowHandler {
 
     pub(crate) async fn handle_compute(
         node: &mut DriaComputeNode,
-        message: DKNMessage,
+        compute_message: &DKNMessage,
     ) -> Result<Either<MessageAcceptance, WorkflowsWorkerInput>> {
-        let task = message
+        let stats = TaskStats::new().record_received_at();
+        let task = compute_message
             .parse_payload::<TaskRequestPayload<WorkflowPayload>>(true)
-            .wrap_err("Could not parse workflow task")?;
-
-        // TODO: !!!
-        let task_stats = TaskStats::default().record_received_at();
+            .wrap_err("could not parse workflow task")?;
 
         // check if deadline is past or not
         let current_time = get_current_time_nanos();
@@ -106,15 +104,15 @@ impl WorkflowHandler {
             model_name,
             task_id: task.task_id,
             public_key: task_public_key,
-            stats: task_stats,
+            stats,
         }))
     }
 
     pub(crate) async fn handle_publish(
         node: &mut DriaComputeNode,
         task: WorkflowsWorkerOutput,
-    ) -> Result<MessageAcceptance> {
-        let (message, acceptance) = match task.result {
+    ) -> Result<()> {
+        let message = match task.result {
             Ok(result) => {
                 // prepare signed and encrypted payload
                 let payload = TaskResponsePayload::new(
@@ -126,17 +124,16 @@ impl WorkflowHandler {
                     task.stats.record_published_at(),
                 )?;
                 let payload_str = serde_json::to_string(&payload)
-                    .wrap_err("Could not serialize response payload")?;
+                    .wrap_err("could not serialize response payload")?;
 
                 // prepare signed message
                 log::debug!(
                     "Publishing result for task {}\n{}",
                     task.task_id,
                     payload_str
                 );
-                let message = DKNMessage::new(payload_str, Self::RESPONSE_TOPIC);
-                // accept so that if there are others included in filter they can do the task
-                (message, MessageAcceptance::Accept)
+
+                DKNMessage::new(payload_str, Self::RESPONSE_TOPIC)
             }
             Err(err) => {
                 // use pretty display string for error logging with causes
@@ -151,22 +148,20 @@ impl WorkflowHandler {
                     stats: task.stats.record_published_at(),
                 };
                 let error_payload_str = serde_json::to_string(&error_payload)
-                    .wrap_err("Could not serialize error payload")?;
+                    .wrap_err("could not serialize error payload")?;
 
                 // prepare signed message
-                let message = DKNMessage::new_signed(
+                DKNMessage::new_signed(
                     error_payload_str,
                     Self::RESPONSE_TOPIC,
                     &node.config.secret_key,
-                );
-                // ignore just in case, workflow may be bugged
-                (message, MessageAcceptance::Ignore)
+                )
             }
         };
 
         // try publishing the result
         if let Err(publish_err) = node.publish(message).await {
-            let err_msg = format!("Could not publish result: {:?}", publish_err);
+            let err_msg = format!("could not publish result: {:?}", publish_err);
             log::error!("{}", err_msg);
 
             let payload = serde_json::json!({
@@ -181,6 +176,6 @@ impl WorkflowHandler {
             node.publish(message).await?;
         };
 
-        Ok(acceptance)
+        Ok(())
     }
 }

compute/src/handlers/workflow.rs

@@ -1,6 +1,3 @@
-// #![doc = include_str!("../README.md")]
-// TODO: this line breaks docker, find a way to ignore during compose?
-
 pub(crate) mod config;
 pub(crate) mod handlers;
 pub(crate) mod node;

compute/src/lib.rs

@@ -13,7 +13,7 @@ async fn main() -> Result<()> {
         .format_timestamp(Some(env_logger::TimestampPrecision::Millis))
         .init();
     if let Err(e) = dotenv_result {
-        log::warn!("Could not load .env file: {}", e);
+        log::warn!("could not load .env file: {}", e);
     }
 
     log::info!(