change rpc-dial logic

Author: erhant

compute/src/node/core.rs

@@ -10,18 +10,18 @@ impl DriaComputeNode {
     /// This method is not expected to return until cancellation occurs for the given token.
     pub async fn run(&mut self, cancellation: CancellationToken) {
         /// Duration between refreshing for diagnostic prints.
-        const DIAGNOSTIC_REFRESH_INTERVAL_SECS: Duration = Duration::from_secs(30);
+        const DIAGNOSTIC_REFRESH_INTERVAL_SECS: Duration = Duration::from_secs(45);
         /// Duration between refreshing the available nodes.
-        const AVAILABLE_NODES_REFRESH_INTERVAL_SECS: Duration = Duration::from_secs(2 * 60);
+        const RPC_LIVENESS_REFRESH_INTERVAL_SECS: Duration = Duration::from_secs(2 * 60);
         /// Duration between each heartbeat sent to the RPC.
         const HEARTBEAT_INTERVAL_SECS: Duration = Duration::from_secs(60);
 
         let mut diagnostic_refresh_interval =
             tokio::time::interval(DIAGNOSTIC_REFRESH_INTERVAL_SECS);
         diagnostic_refresh_interval.tick().await; // move each one tick
-        let mut available_node_refresh_interval =
-            tokio::time::interval(AVAILABLE_NODES_REFRESH_INTERVAL_SECS);
-        available_node_refresh_interval.tick().await; // move each one tick
+        let mut rpc_liveness_refresh_interval =
+            tokio::time::interval(RPC_LIVENESS_REFRESH_INTERVAL_SECS);
+        rpc_liveness_refresh_interval.tick().await; // move each one tick
 
         let mut heartbeat_interval = tokio::time::interval(HEARTBEAT_INTERVAL_SECS);
 
@@ -52,8 +52,8 @@ impl DriaComputeNode {
                 // check peer count every now and then
                 _ = diagnostic_refresh_interval.tick() => self.handle_diagnostic_refresh().await,
 
-                // available nodes are refreshed every now and then
-                _ = available_node_refresh_interval.tick() => self.handle_available_nodes_refresh().await,
+                // check RPC, and get a new one if we are disconnected
+                _ = rpc_liveness_refresh_interval.tick() => self.handle_rpc_liveness_check().await,
 
                 _ = heartbeat_interval.tick() => {
                   if let Err(e) = self.send_heartbeat().await {

compute/src/node/diagnostic.rs

@@ -98,7 +98,7 @@ impl DriaComputeNode {
     ///
     /// If there is an error while doing that,
     /// it will try to get a new RPC node and dial it.
-    pub(crate) async fn handle_available_nodes_refresh(&mut self) {
+    pub(crate) async fn handle_rpc_liveness_check(&mut self) {
         log::debug!("Checking RPC connections for diagnostics.");
 
         // check if we are connected
@@ -108,19 +108,15 @@ impl DriaComputeNode {
             .await
             .unwrap_or(false);
 
-        // if we are not connected, try to dial it again
+        // if we are not connected, get a new RPC and dial it again
         if !is_connected {
-            log::info!("Dialling RPC at: {}", self.dria_rpc.addr);
-            if let Err(err) = self
-                .dial_with_timeout(self.dria_rpc.peer_id, self.dria_rpc.addr.clone())
-                .await
-            {
-                // if we also cannot dial it, get a new RPC node
-                log::warn!(
-                    "Could not dial to RPC at: {}: {err:?}\nWill get a new RPC node.",
-                    self.dria_rpc.addr,
-                );
-                if let Ok(new_rpc) = DriaRPC::new(self.dria_rpc.network).await {
+            // if we also cannot dial it, get a new RPC node
+            log::warn!(
+                "Connection to RPC {} is lost, geting a new one!",
+                self.dria_rpc.addr,
+            );
+            match DriaRPC::new_for_network(self.dria_rpc.network).await {
+                Ok(new_rpc) => {
                     self.dria_rpc = new_rpc;
 
                     // now dial this new RPC again
@@ -129,14 +125,13 @@ impl DriaComputeNode {
                         .await
                     {
                         // worst-case we cant dial this one too, just leave it for the next diagnostic
-                        log::error!("Could not dial the new RPC: {err:?}\nWill try again in the next diagnostic refresh.");
+                        log::error!("Could not dial the new RPC: {err:?}");
                     }
-                } else {
-                    log::error!("Could not get a new RPC node!\nWill try again in the next diagnostic refresh.");
                 }
-            } else {
-                log::info!("Successfully dialled to RPC at: {}", self.dria_rpc.addr);
-            }
+                Err(err) => {
+                    log::error!("Could not get a new RPC node: {err:?}");
+                }
+            };
         } else {
             log::debug!("Connection with {} is intact.", self.dria_rpc.peer_id);
         }

compute/src/node/mod.rs

@@ -70,7 +70,7 @@ impl DriaComputeNode {
         let keypair = secret_to_keypair(&config.secret_key);
 
         // get available rpc node
-        let dria_nodes = DriaRPC::new(config.network_type)
+        let dria_nodes = DriaRPC::new_for_network(config.network_type)
             .await
             .expect("could not get RPC to connect to");
 

compute/src/utils/rpc.rs

@@ -12,11 +12,8 @@ pub struct DriaRPC {
 }
 
 impl DriaRPC {
-    /// Creates a new `AvailableNodes` struct for the given network type.
-    ///
-    /// Will panic if anything goes wrong.
-    pub async fn new(network: DriaNetworkType) -> Result<Self> {
-        let addr = refresh_rpc_addr(&network).await?;
+    /// Creates a new RPC target at the given type, along with a network type for refreshing the RPC address.
+    pub fn new(addr: Multiaddr, network: DriaNetworkType) -> Result<Self> {
         let peer_id = addr
             .iter()
             .find_map(|p| match p {
@@ -31,12 +28,18 @@ impl DriaRPC {
             network,
         })
     }
+
+    /// Creates a new RPC target for the given network type.
+    pub async fn new_for_network(network: DriaNetworkType) -> Result<Self> {
+        let addr = get_rpc_for_network(&network).await?;
+        Self::new(addr, network)
+    }
 }
 
 /// Calls the DKN API to get an RPC address for the given network type.
 ///
 /// The peer id is expected to be within the multi-address.
-async fn refresh_rpc_addr(network: &DriaNetworkType) -> Result<Multiaddr> {
+async fn get_rpc_for_network(network: &DriaNetworkType) -> Result<Multiaddr> {
     #[derive(serde::Deserialize, Debug)]
     struct DriaNodesApiResponse {
         pub rpc: Multiaddr,
@@ -66,7 +69,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_dria_nodes() {
-        let node = DriaRPC::new(DriaNetworkType::Community).await;
+        let node = DriaRPC::new_for_network(DriaNetworkType::Community).await;
         assert!(node.is_ok());
     }
 }