few rfks, todo: rpc choice logic

Author: erhant

compute/src/node/diagnostic.rs

@@ -99,7 +99,7 @@ impl DriaComputeNode {
         };
 
         // dial all rpc nodes
-        for addr in self.dria_nodes.rpc_nodes.iter() {
+        for addr in self.dria_nodes.rpc_addrs.iter() {
             log::info!("Dialling RPC node: {}", addr);
 
             // get peer id from rpc address

compute/src/reqres/heartbeat.rs

@@ -32,8 +32,11 @@ pub struct HeartbeatRequest {
 pub struct HeartbeatResponse {
     /// UUID as given in the request.
     pub(crate) heartbeat_id: Uuid,
-    /// Acknowledgement of the heartbeat.
-    pub(crate) ack: bool,
+    /// An associated error with the response,
+    ///
+    /// - `None` means that the heartbeat was acknowledged.
+    /// - `Some` means that the heartbeat was not acknowledged for the given reason.
+    pub(crate) error: Option<String>,
 }
 
 impl IsResponder for HeartbeatRequester {
@@ -80,14 +83,18 @@ impl HeartbeatRequester {
         res: HeartbeatResponse,
     ) -> Result<()> {
         if let Some(deadline) = node.heartbeats.remove(&res.heartbeat_id) {
-            if !res.ack {
-                Err(eyre!("Heartbeat was not acknowledged."))
-            } else if chrono::Utc::now() > deadline {
-                Err(eyre!("Acknowledged heartbeat was past the deadline."))
+            if let Some(err) = res.error {
+                Err(eyre!("Heartbeat was not acknowledged: {}", err))
             } else {
+                // acknowledge heartbeat
                 node.last_heartbeat_at = chrono::Utc::now();
                 node.num_heartbeats += 1;
 
+                // for diagnostics, we can check if the heartbeat was past its deadline as well
+                if chrono::Utc::now() > deadline {
+                    log::warn!("Acknowledged heartbeat was past its deadline.")
+                }
+
                 Ok(())
             }
         } else {

compute/src/utils/nodes.rs

@@ -1,4 +1,7 @@
-use dkn_p2p::{libp2p::PeerId, DriaNetworkType, DriaNodes};
+use dkn_p2p::{
+    libp2p::{Multiaddr, PeerId},
+    DriaNetworkType, DriaNodes,
+};
 use dkn_utils::parse_vec;
 use eyre::Result;
 
@@ -8,7 +11,7 @@ pub async fn refresh_dria_nodes(nodes: &mut DriaNodes) -> Result<()> {
     struct DriaNodesApiResponse {
         pub rpcs: Vec<String>,
         #[serde(rename = "rpcAddrs")]
-        pub rpc_addrs: Vec<String>,
+        pub rpc_addrs: Vec<Multiaddr>,
     }
 
     // url to be used is determined by the network type
@@ -22,12 +25,7 @@ pub async fn refresh_dria_nodes(nodes: &mut DriaNodes) -> Result<()> {
     let response = reqwest::get(url).await?;
     let response_body = response.json::<DriaNodesApiResponse>().await?;
 
-    nodes
-        .rpc_nodes
-        .extend(parse_vec(response_body.rpc_addrs).unwrap_or_else(|e| {
-            log::error!("Failed to parse rpc nodes: {}", e);
-            vec![]
-        }));
+    nodes.rpc_addrs.extend(response_body.rpc_addrs);
     nodes
         .rpc_peerids
         .extend(parse_vec::<PeerId>(response_body.rpcs).unwrap_or_else(|e| {
@@ -46,12 +44,12 @@ mod tests {
     async fn test_refresh_dria_nodes() {
         let mut nodes = DriaNodes::new(DriaNetworkType::Community);
         refresh_dria_nodes(&mut nodes).await.unwrap();
-        assert!(!nodes.rpc_nodes.is_empty());
+        assert!(!nodes.rpc_addrs.is_empty());
         assert!(!nodes.rpc_peerids.is_empty());
 
         let mut nodes = DriaNodes::new(DriaNetworkType::Pro);
         refresh_dria_nodes(&mut nodes).await.unwrap();
-        assert!(!nodes.rpc_nodes.is_empty());
+        assert!(!nodes.rpc_addrs.is_empty());
         assert!(!nodes.rpc_peerids.is_empty());
     }
 }

p2p/src/client.rs

@@ -83,7 +83,7 @@ impl DriaP2PClient {
 
         // dial rpc nodes
         // this will cause `identify` event to be called on their side
-        for rpc_addr in nodes.rpc_nodes.iter().cloned() {
+        for rpc_addr in nodes.rpc_addrs.iter().cloned() {
             log::info!("Dialing RPC node: {}", rpc_addr);
             if let Err(e) = swarm.dial(rpc_addr) {
                 log::error!("Could not dial RPC node: {:?}", e);

p2p/src/nodes.rs

@@ -3,12 +3,9 @@ use std::{collections::HashSet, fmt::Debug};
 
 use crate::DriaNetworkType;
 
-/// Dria-owned nodes within the hybrid P2P network.
-///
-/// - RPC: used for RPC nodes for task & ping messages.
 #[derive(Debug, Clone)]
 pub struct DriaNodes {
-    pub rpc_nodes: HashSet<Multiaddr>,
+    pub rpc_addrs: HashSet<Multiaddr>,
     pub rpc_peerids: HashSet<PeerId>,
     pub network: DriaNetworkType,
 }
@@ -17,13 +14,13 @@ impl DriaNodes {
     /// Creates a new `AvailableNodes` struct for the given network type.
     pub fn new(network: DriaNetworkType) -> Self {
         Self {
-            rpc_nodes: HashSet::new(),
+            rpc_addrs: HashSet::new(),
             rpc_peerids: HashSet::new(),
             network,
         }
     }
     pub fn with_rpc_nodes(mut self, addresses: impl IntoIterator<Item = Multiaddr>) -> Self {
-        self.rpc_nodes.extend(addresses);
+        self.rpc_addrs.extend(addresses);
         self
     }
 
@@ -34,7 +31,7 @@ impl DriaNodes {
 
     /// Adds the static nodes to the struct, with respect to network type.
     pub fn with_statics(mut self) -> Self {
-        self.rpc_nodes.extend(self.network.get_static_rpc_nodes());
+        self.rpc_addrs.extend(self.network.get_static_rpc_nodes());
         self.rpc_peerids
             .extend(self.network.get_static_rpc_peer_ids());