darkfid: gracefully handle miner network dc

bin/darkfid/src/main.rs

@@ -18,7 +18,6 @@
 
 use std::{
     collections::{HashMap, HashSet},
-    str::FromStr,
     sync::Arc,
 };
 
@@ -29,7 +28,7 @@ use url::Url;
 
 use darkfi::{
     async_daemonize,
-    blockchain::{BlockInfo, HeaderHash},
+    blockchain::BlockInfo,
     cli_desc,
     net::{settings::SettingsOpt, P2pPtr},
     rpc::{
@@ -42,7 +41,6 @@ use darkfi::{
     validator::{Validator, ValidatorConfig, ValidatorPtr},
     Error, Result,
 };
-use darkfi_sdk::crypto::PublicKey;
 use darkfi_serial::deserialize_async;
 
 #[cfg(test)]
@@ -58,7 +56,7 @@ mod rpc_tx;
 
 /// Validator async tasks
 mod task;
-use task::{consensus_task, miner_task, sync_task};
+use task::consensus_init_task;
 
 /// P2P net protocols
 mod proto;
@@ -313,66 +311,27 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     info!(target: "darkfid", "Starting P2P network");
     p2p.clone().start().await?;
 
-    // Sync blockchain
-    if !blockchain_config.skip_sync {
-        // Parse configured checkpoint
-        if blockchain_config.checkpoint_height.is_some() && blockchain_config.checkpoint.is_none() {
-            return Err(Error::ParseFailed("Blockchain configured checkpoint hash missing"))
-        }
-
-        let checkpoint = if let Some(height) = blockchain_config.checkpoint_height {
-            Some((height, HeaderHash::from_str(&blockchain_config.checkpoint.unwrap())?))
-        } else {
-            None
-        };
-
-        sync_task(&darkfid, checkpoint).await?;
-    } else {
-        *darkfid.validator.synced.write().await = true;
-    }
-
     // Consensus protocol
     info!(target: "darkfid", "Starting consensus protocol task");
-    let consensus_task = if blockchain_config.miner {
-        // Grab rewards recipient public key(address)
-        if blockchain_config.recipient.is_none() {
-            return Err(Error::ParseFailed("Recipient address missing"))
-        }
-        let recipient = match PublicKey::from_str(&blockchain_config.recipient.unwrap()) {
-            Ok(address) => address,
-            Err(_) => return Err(Error::InvalidAddress),
-        };
-
-        let task = StoppableTask::new();
-        task.clone().start(
-            miner_task(darkfid.clone(), recipient, blockchain_config.skip_sync, ex.clone()),
-            |res| async move {
-                match res {
-                    Ok(()) | Err(Error::MinerTaskStopped) => { /* Do nothing */ }
-                    Err(e) => error!(target: "darkfid", "Failed starting miner task: {}", e),
-                }
-            },
-            Error::MinerTaskStopped,
+    let consensus_task = StoppableTask::new();
+    consensus_task.clone().start(
+        consensus_init_task(
+            darkfid.clone(),
+            blockchain_config.skip_sync,
+            blockchain_config.checkpoint_height,
+            blockchain_config.checkpoint, blockchain_config.miner,
+            blockchain_config.recipient,
             ex.clone(),
-        );
-
-        task
-    } else {
-        let task = StoppableTask::new();
-        task.clone().start(
-            consensus_task(darkfid.clone(), ex.clone()),
-            |res| async move {
-                match res {
-                    Ok(()) | Err(Error::ConsensusTaskStopped) => { /* Do nothing */ }
-                    Err(e) => error!(target: "darkfid", "Failed starting consensus task: {}", e),
-                }
-            },
-            Error::ConsensusTaskStopped,
-            ex.clone(),
-        );
-
-        task
-    };
+        ),
+        |res| async move {
+            match res {
+                Ok(()) | Err(Error::ConsensusTaskStopped) | Err(Error::MinerTaskStopped) => { /* Do nothing */ }
+                Err(e) => error!(target: "darkfid", "Failed starting consensus initialization task: {}", e),
+            }
+        },
+        Error::ConsensusTaskStopped,
+        ex.clone(),
+    );
 
     // Signal handling for graceful termination.
     let (signals_handler, signals_task) = SignalHandler::new(ex)?;

bin/darkfid/src/task/consensus.rs

@@ -16,16 +16,90 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::Arc;
+use std::{str::FromStr, sync::Arc};
 
-use darkfi::{rpc::util::JsonValue, system::StoppableTask, util::encoding::base64, Error, Result};
+use darkfi::{
+    blockchain::HeaderHash, rpc::util::JsonValue, system::StoppableTask, util::encoding::base64,
+    Error, Result,
+};
+use darkfi_sdk::crypto::PublicKey;
 use darkfi_serial::serialize_async;
 use log::{error, info};
 
-use crate::{task::garbage_collect_task, Darkfid};
+use crate::{
+    task::{garbage_collect_task, miner_task, sync_task},
+    Darkfid,
+};
+
+/// Sync the node consensus state and start the corresponding task, based on node type.
+pub async fn consensus_init_task(
+    node: Arc<Darkfid>,
+    skip_sync: bool,
+    checkpoint_height: Option<u32>,
+    checkpoint: Option<String>,
+    miner: bool,
+    recipient: Option<String>,
+    ex: Arc<smol::Executor<'static>>,
+) -> Result<()> {
+    // Sync blockchain
+    let checkpoint = if !skip_sync {
+        // Parse configured checkpoint
+        if checkpoint_height.is_some() && checkpoint.is_none() {
+            return Err(Error::ParseFailed("Blockchain configured checkpoint hash missing"))
+        }
+
+        let checkpoint = if let Some(height) = checkpoint_height {
+            Some((height, HeaderHash::from_str(checkpoint.as_ref().unwrap())?))
+        } else {
+            None
+        };
+
+        sync_task(&node, checkpoint).await?;
+        checkpoint
+    } else {
+        *node.validator.synced.write().await = true;
+        None
+    };
+
+    // Grab rewards recipient public key(address) if node is a miner
+    let recipient = if miner {
+        if recipient.is_none() {
+            return Err(Error::ParseFailed("Recipient address missing"))
+        }
+        match PublicKey::from_str(recipient.as_ref().unwrap()) {
+            Ok(address) => Some(address),
+            Err(_) => return Err(Error::InvalidAddress),
+        }
+    } else {
+        None
+    };
+
+    // Gracefully handle network disconnections
+    loop {
+        let result = if miner {
+            miner_task(node.clone(), recipient.unwrap(), skip_sync, ex.clone()).await
+        } else {
+            replicator_task(node.clone(), ex.clone()).await
+        };
+
+        match result {
+            Ok(_) => return Ok(()),
+            Err(Error::NetworkOperationFailed) => {
+                // Sync node again
+                *node.validator.synced.write().await = false;
+                if !skip_sync {
+                    sync_task(&node, checkpoint).await?;
+                } else {
+                    *node.validator.synced.write().await = true;
+                }
+            }
+            Err(e) => return Err(e),
+        }
+    }
+}
 
 /// async task used for listening for new blocks and perform consensus.
-pub async fn consensus_task(node: Arc<Darkfid>, ex: Arc<smol::Executor<'static>>) -> Result<()> {
+pub async fn replicator_task(node: Arc<Darkfid>, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     info!(target: "darkfid::task::consensus_task", "Starting consensus task...");
 
     // Grab blocks subscriber

bin/darkfid/src/task/miner.rs

@@ -48,8 +48,6 @@ use smol::channel::{Receiver, Sender};
 
 use crate::{proto::ProposalMessage, task::garbage_collect_task, Darkfid};
 
-// TODO: handle all ? so the task don't stop on errors
-
 /// Async task used for participating in the PoW block production.
 /// Miner initializes their setup and waits for next finalization,
 /// by listenning for new proposals from the network, for optimal
@@ -158,18 +156,35 @@ pub async fn miner_task(
         drop(forks);
 
         // Start listenning for network proposals and mining next block for best fork.
-        if let Err(e) = smol::future::or(
+        match smol::future::or(
             listen_to_network(&node, &extended_fork, &subscription, &sender),
-            mine(&node, &extended_fork, &mut secret, &recipient, &zkbin, &pk, &stop_signal),
+            mine(
+                &node,
+                &extended_fork,
+                &mut secret,
+                &recipient,
+                &zkbin,
+                &pk,
+                &stop_signal,
+                skip_sync,
+            ),
         )
         .await
         {
-            error!(
-                target: "darkfid::task::miner_task",
-                "Error during listen_to_network() or mine(): {e}"
-            );
-            continue
-        };
+            Ok(_) => { /* Do nothing */ }
+            Err(Error::NetworkOperationFailed) => {
+                error!(target: "darkfid::task::miner_task", "Node disconnected from the network");
+                subscription.unsubscribe().await;
+                return Err(Error::NetworkOperationFailed)
+            }
+            Err(e) => {
+                error!(
+                    target: "darkfid::task::miner_task",
+                    "Error during listen_to_network() or mine(): {e}"
+                );
+                continue
+            }
+        }
 
         // Check if we can finalize anything and broadcast them
         let finalized = match node.validator.finalization().await {
@@ -238,7 +253,7 @@ async fn listen_to_network(
     // Signal miner to abort mining
     sender.send(()).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);
+        error!(target: "darkfid::task::miner::listen_to_network", "Failed to execute miner daemon abort request: {}", e);
     }
 
     Ok(())
@@ -246,6 +261,7 @@ async fn listen_to_network(
 
 /// Async task to generate and mine provided fork index next block,
 /// while listening for a stop signal.
+#[allow(clippy::too_many_arguments)]
 async fn mine(
     node: &Darkfid,
     extended_fork: &Fork,
@@ -254,10 +270,11 @@ async fn mine(
     zkbin: &ZkBinary,
     pk: &ProvingKey,
     stop_signal: &Receiver<()>,
+    skip_sync: bool,
 ) -> Result<()> {
     smol::future::or(
         wait_stop_signal(stop_signal),
-        mine_next_block(node, extended_fork, secret, recipient, zkbin, pk),
+        mine_next_block(node, extended_fork, secret, recipient, zkbin, pk, skip_sync),
     )
     .await
 }
@@ -283,6 +300,7 @@ async fn mine_next_block(
     recipient: &PublicKey,
     zkbin: &ZkBinary,
     pk: &ProvingKey,
+    skip_sync: bool,
 ) -> Result<()> {
     // Grab next target and block
     let (next_target, mut next_block) = generate_next_block(
@@ -309,6 +327,11 @@ async fn mine_next_block(
     // Verify it
     extended_fork.module.verify_current_block(&next_block)?;
 
+    // Check if we are connected to the network
+    if !skip_sync && node.p2p.hosts().channels().await.is_empty() {
+        return Err(Error::NetworkOperationFailed)
+    }
+
     // Append the mined block as a proposal
     let proposal = Proposal::new(next_block);
     node.validator.append_proposal(&proposal).await?;

bin/darkfid/src/task/mod.rs

@@ -17,7 +17,7 @@
  */
 
 pub mod consensus;
-pub use consensus::consensus_task;
+pub use consensus::consensus_init_task;
 
 pub mod miner;
 pub use miner::miner_task;

bin/darkfid/src/task/sync.rs

@@ -204,7 +204,7 @@ async fn synced_peers(
         }
 
         warn!(target: "darkfid::task::sync::synced_peers", "Node is not connected to other nodes, waiting to retry...");
-        sleep(10).await;
+        sleep(node.p2p.settings().outbound_connect_timeout).await;
     }
 
     Ok(tips)