lints, small rfks

Author: erhant

compute/src/node/core.rs

@@ -59,8 +59,8 @@ impl DriaComputeNode {
                 _ = heartbeat_interval.tick() => {
                   if let Err(e) = self.send_heartbeat().await {
                     log::error!("Error making heartbeat: {:?}", e);
-                }
-              },
+                  }
+                },
 
                 // check if the cancellation token is cancelled
                 // this is expected to be cancelled by the main thread with signal handling

compute/src/node/mod.rs

@@ -1,6 +1,5 @@
 use dkn_p2p::{
-    libp2p::{request_response, PeerId},
-    DriaP2PClient, DriaP2PCommander, DriaP2PProtocol,
+    libp2p::PeerId, DriaP2PClient, DriaP2PCommander, DriaP2PProtocol, DriaReqResMessage,
 };
 use eyre::Result;
 use std::collections::HashMap;
@@ -34,7 +33,7 @@ pub struct DriaComputeNode {
     /// This is used to track the heartbeats, and their acknowledgements.
     pub(crate) heartbeats: HashMap<uuid::Uuid, chrono::DateTime<chrono::Utc>>,
     /// Request-response message receiver, can have both a request or a response.
-    reqres_rx: mpsc::Receiver<(PeerId, request_response::Message<Vec<u8>, Vec<u8>>)>,
+    reqres_rx: mpsc::Receiver<(PeerId, DriaReqResMessage)>,
     /// Task response receiver, will respond to the request-response channel with the given result.
     task_output_rx: mpsc::Receiver<TaskWorkerOutput>,
     /// Task worker transmitter to send batchable tasks.

compute/src/node/reqres.rs

@@ -1,8 +1,9 @@
 use colored::Colorize;
 use dkn_p2p::libp2p::{
-    request_response::{self, OutboundRequestId, ResponseChannel},
+    request_response::{OutboundRequestId, ResponseChannel},
     PeerId,
 };
+use dkn_p2p::DriaReqResMessage;
 use eyre::{eyre, Result};
 
 use crate::{reqres::*, workers::task::TaskWorkerOutput};
@@ -16,13 +17,9 @@ impl DriaComputeNode {
     /// - Response is forwarded to [`handle_response`](DriaComputeNode::handle_response) method.
     ///
     /// Does not return an error, but simply logs it to [`log::error`].
-    pub(crate) async fn handle_reqres(
-        &mut self,
-        peer_id: PeerId,
-        message: request_response::Message<Vec<u8>, Vec<u8>>,
-    ) {
+    pub(crate) async fn handle_reqres(&mut self, peer_id: PeerId, message: DriaReqResMessage) {
         match message {
-            request_response::Message::Request {
+            DriaReqResMessage::Request {
                 request,
                 request_id,
                 channel,
@@ -38,7 +35,7 @@ impl DriaComputeNode {
                 }
             }
 
-            request_response::Message::Response {
+            DriaReqResMessage::Response {
                 response,
                 request_id,
             } => {

compute/src/payloads/response.rs

@@ -5,10 +5,6 @@ use serde::{Deserialize, Serialize};
 use super::TaskStats;
 
 /// A computation task is the task of computing a result from a given input. The result is encrypted with the public key of the requester.
-/// Plain result is signed by the compute node's private key, and a commitment is computed from the signature and plain result.
-///
-/// To check the commitment, one must decrypt the ciphertext and parse plaintext from it,
-/// and compute the digest using SHA256. That digest will then be used for the signature check.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct TaskResponsePayload {
@@ -23,10 +19,7 @@ pub struct TaskResponsePayload {
 }
 
 impl TaskResponsePayload {
-    /// Creates the payload of a computation result.
-    ///
-    /// - Sign `task_id || payload` with node `self.secret_key`
-    /// - Encrypt `result` with `task_public_key`
+    /// Creates the payload of a computation with its (encrypted) result.
     pub fn new(
         result: impl AsRef<[u8]>,
         task_id: impl ToString,

compute/src/payloads/stats.rs

@@ -1,6 +1,7 @@
 use serde::{Deserialize, Serialize};
 
 /// Task stats for diagnostics.
+///
 /// Returning this as the payload helps to debug the errors received at client side, and latencies.
 #[derive(Default, Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]

p2p/src/client.rs

@@ -16,6 +16,18 @@ use crate::DriaP2PProtocol;
 use super::commands::DriaP2PCommand;
 use super::DriaP2PCommander;
 
+/// Number of seconds before an idle connection is closed.
+const IDLE_CONNECTION_TIMEOUT_SECS: u64 = 240;
+/// Buffer size for command channel.
+const COMMAND_CHANNEL_BUFSIZE: usize = 1024;
+/// Buffer size for events channel.
+const MSG_CHANNEL_BUFSIZE: usize = 1024;
+
+/// Request-response message type for Dria protocol, accepts bytes as both request and response.
+///
+/// The additional parsing must be done by the application itself (for now).
+pub type DriaReqResMessage = request_response::Message<Vec<u8>, Vec<u8>>;
+
 /// Peer-to-peer client for Dria Knowledge Network.
 pub struct DriaP2PClient {
     pub peer_id: PeerId,
@@ -24,25 +36,19 @@ pub struct DriaP2PClient {
     /// Dria protocol, used for identifying the client.
     protocol: DriaP2PProtocol,
     /// Request-response protocol messages.
-    reqres_tx: mpsc::Sender<(PeerId, request_response::Message<Vec<u8>, Vec<u8>>)>,
+    reqres_tx: mpsc::Sender<(PeerId, DriaReqResMessage)>,
     /// Command receiver.
     cmd_rx: mpsc::Receiver<DriaP2PCommand>,
 }
 
-/// Number of seconds before an idle connection is closed.
-const IDLE_CONNECTION_TIMEOUT_SECS: u64 = 240;
-/// Buffer size for command channel.
-const COMMAND_CHANNEL_BUFSIZE: usize = 1024;
-/// Buffer size for events channel.
-const MSG_CHANNEL_BUFSIZE: usize = 1024;
-
 impl DriaP2PClient {
     /// Creates a new P2P client with the given keypair and listen address.
     ///
     /// The `version` is used to create the protocol strings for the client, and its very important that
     /// they match with the clients existing within the network.
     ///
     /// If for any reason the given `listen_addr` is not available, it will try to listen on a random port on `localhost`.
+    #[allow(clippy::type_complexity)]
     pub fn new(
         keypair: Keypair,
         listen_addr: Multiaddr,
@@ -51,7 +57,7 @@ impl DriaP2PClient {
     ) -> Result<(
         DriaP2PClient,
         DriaP2PCommander,
-        mpsc::Receiver<(PeerId, request_response::Message<Vec<u8>, Vec<u8>>)>,
+        mpsc::Receiver<(PeerId, DriaReqResMessage)>,
     )> {
         let peer_id = keypair.public().to_peer_id();
 

p2p/src/lib.rs

@@ -1,7 +1,7 @@
 mod behaviour;
 
 mod client;
-pub use client::DriaP2PClient;
+pub use client::{DriaP2PClient, DriaReqResMessage};
 
 mod commands;
 pub use commands::{DriaP2PCommand, DriaP2PCommander};