darkfid: use a hot-swapable JSON-RPC client to handle errors while communicating with minerd

bin/darkfid/src/main.rs

@@ -143,6 +143,21 @@ pub struct BlockchainNetwork {
     pub net: SettingsOpt,
 }
 
+/// Structure to hold a JSON-RPC client and its config,
+/// so we can recreate it in case of an error.
+pub struct MinerRpcCLient {
+    endpoint: Url,
+    ex: Arc<smol::Executor<'static>>,
+    client: RpcChadClient,
+}
+
+impl MinerRpcCLient {
+    pub async fn new(endpoint: Url, ex: Arc<smol::Executor<'static>>) -> Result<Self> {
+        let client = RpcChadClient::new(endpoint.clone(), ex.clone()).await?;
+        Ok(Self { endpoint, ex, client })
+    }
+}
+
 /// Daemon structure
 pub struct Darkfid {
     /// P2P network pointer
@@ -156,7 +171,7 @@ pub struct Darkfid {
     /// JSON-RPC connection tracker
     rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
     /// JSON-RPC client to execute requests to the miner daemon
-    rpc_client: Option<RpcChadClient>,
+    rpc_client: Option<Mutex<MinerRpcCLient>>,
 }
 
 impl Darkfid {
@@ -165,7 +180,7 @@ impl Darkfid {
         validator: ValidatorPtr,
         miner: bool,
         subscribers: HashMap<&'static str, JsonSubscriber>,
-        rpc_client: Option<RpcChadClient>,
+        rpc_client: Option<Mutex<MinerRpcCLient>>,
     ) -> Self {
         Self {
             p2p,
@@ -238,12 +253,12 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     // Initialize JSON-RPC client to perform requests to minerd
     let rpc_client = if blockchain_config.miner {
         let Ok(rpc_client) =
-            RpcChadClient::new(blockchain_config.minerd_endpoint, ex.clone()).await
+            MinerRpcCLient::new(blockchain_config.minerd_endpoint, ex.clone()).await
         else {
             error!(target: "darkfid", "Failed to initialize miner daemon rpc client, check if minerd is running");
             return Err(Error::RpcClientStopped)
         };
-        Some(rpc_client)
+        Some(Mutex::new(rpc_client))
     } else {
         None
     };

bin/darkfid/src/rpc.rs

@@ -19,16 +19,17 @@
 use std::{collections::HashSet, time::Instant};
 
 use async_trait::async_trait;
-use log::{debug, error};
+use log::{debug, error, info};
 use smol::lock::MutexGuard;
 use tinyjson::JsonValue;
 
 use darkfi::{
     rpc::{
+        client::RpcChadClient,
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
         server::RequestHandler,
     },
-    system::StoppableTaskPtr,
+    system::{sleep, StoppableTaskPtr},
     util::time::Timestamp,
     Error, Result,
 };
@@ -134,21 +135,65 @@ impl Darkfid {
         JsonResponse::new(JsonValue::Boolean(true), id).into()
     }
 
+    /// Ping configured miner daemon JSON-RPC endpoint.
     pub async fn ping_miner_daemon(&self) -> Result<()> {
         debug!(target: "darkfid::ping_miner_daemon", "Pinging miner daemon...");
-        self.miner_daemon_request("ping", JsonValue::Array(vec![])).await?;
+        self.miner_daemon_request("ping", &JsonValue::Array(vec![])).await?;
         Ok(())
     }
 
-    pub async fn miner_daemon_request(&self, method: &str, params: JsonValue) -> Result<JsonValue> {
+    /// Auxiliary function to execute a request towards the configured miner daemon JSON-RPC endpoint.
+    pub async fn miner_daemon_request(
+        &self,
+        method: &str,
+        params: &JsonValue,
+    ) -> Result<JsonValue> {
         let Some(ref rpc_client) = self.rpc_client else { return Err(Error::RpcClientStopped) };
         debug!(target: "darkfid::rpc::miner_daemon_request", "Executing request {} with params: {:?}", method, params);
         let latency = Instant::now();
-        let req = JsonRequest::new(method, params);
+        let req = JsonRequest::new(method, params.clone());
-        let rep = rpc_client.request(req).await?;
+        let lock = rpc_client.lock().await;
+        let rep = lock.client.request(req).await?;
+        drop(lock);
         let latency = latency.elapsed();
         debug!(target: "darkfid::rpc::miner_daemon_request", "Got reply: {:?}", rep);
         debug!(target: "darkfid::rpc::miner_daemon_request", "Latency: {:?}", latency);
         Ok(rep)
     }
+
+    /// Auxiliary function to execute a request towards the configured miner daemon JSON-RPC endpoint,
+    /// but in case of failure, sleep and retry until connection is re-established.
+    pub async fn miner_daemon_request_with_retry(
+        &self,
+        method: &str,
+        params: &JsonValue,
+    ) -> JsonValue {
+        loop {
+            // Try to execute the request using current client
+            match self.miner_daemon_request(method, params).await {
+                Ok(v) => return v,
+                Err(e) => {
+                    error!(target: "darkfid::rpc::miner_daemon_request_with_retry", "Failed to execute miner daemon request: {}", e);
+                }
+            }
+            loop {
+                // Sleep a bit before retrying
+                info!(target: "darkfid::rpc::miner_daemon_request_with_retry", "Sleeping so we can retry later");
+                sleep(10).await;
+                // Create a new client
+                let mut rpc_client = self.rpc_client.as_ref().unwrap().lock().await;
+                let Ok(client) =
+                    RpcChadClient::new(rpc_client.endpoint.clone(), rpc_client.ex.clone()).await
+                else {
+                    error!(target: "darkfid::rpc::miner_daemon_request_with_retry", "Failed to initialize miner daemon rpc client, check if minerd is running");
+                    drop(rpc_client);
+                    continue
+                };
+                info!(target: "darkfid::rpc::miner_daemon_request_with_retry", "Connection re-established!");
+                // Set the new client as the daemon one
+                rpc_client.client = client;
+                break;
+            }
+        }
+    }
 }

bin/darkfid/src/task/miner.rs

@@ -39,7 +39,7 @@ use darkfi_sdk::{
     ContractCall,
 };
 use darkfi_serial::{serialize_async, Encodable};
-use log::info;
+use log::{error, info};
 use num_bigint::BigUint;
 use rand::rngs::OsRng;
 use smol::channel::{Receiver, Sender};
@@ -172,7 +172,9 @@ async fn listen_to_network(
 
     // Signal miner to abort mining
     sender.send(()).await?;
-    node.miner_daemon_request("abort", JsonValue::Array(vec![])).await?;
+    if let Err(e) = node.miner_daemon_request("abort", &JsonValue::Array(vec![])).await {
+        error!(target: "darkfid::task::miner_task::listen_to_network", "Failed to execute miner daemon abort request: {}", e);
+    }
 
     Ok(())
 }
@@ -224,7 +226,8 @@ async fn mine_next_block(
     // Execute request to minerd and parse response
     let target = JsonValue::String(next_target.to_string());
     let block = JsonValue::String(base64::encode(&serialize_async(&next_block).await));
-    let response = node.miner_daemon_request("mine", JsonValue::Array(vec![target, block])).await?;
+    let response =
+        node.miner_daemon_request_with_retry("mine", &JsonValue::Array(vec![target, block])).await;
     next_block.header.nonce = *response.get::<f64>().unwrap() as u64;
 
     // Sign the mined block