darkfid: gracefully handle everything in live loops

bin/darkfid/src/proto/protocol_proposal.rs

@@ -19,7 +19,7 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use log::debug;
+use log::{debug, error};
 use smol::Executor;
 use tinyjson::JsonValue;
 
@@ -91,8 +91,7 @@ impl ProtocolProposal {
                 Err(e) => {
                     debug!(
                         target: "darkfid::proto::protocol_proposal::handle_receive_proposal",
-                        "recv fail: {}",
-                        e
+                        "recv fail: {e}"
                     );
                     continue
                 }
@@ -120,8 +119,7 @@ impl ProtocolProposal {
                 Err(e) => {
                     debug!(
                         target: "darkfid::proto::protocol_proposal::handle_receive_proposal",
-                        "append_proposal fail: {}",
-                        e
+                        "append_proposal fail: {e}",
                     );
 
                     match e {
@@ -135,12 +133,27 @@ impl ProtocolProposal {
             debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Asking peer for fork sequence");
 
             // Cleanup subscriber
-            self.proposals_response_sub.clean().await?;
+            if let Err(e) = self.proposals_response_sub.clean().await {
+                error!(
+                    target: "darkfid::proto::protocol_proposal::handle_receive_proposal",
+                    "Error during proposals response subscriber cleanup: {e}"
+                );
+                continue
+            };
 
             // Grab last known block to create the request and execute it
-            let last = self.validator.blockchain.last()?;
+            let last = match self.validator.blockchain.last() {
+                Ok(l) => l,
+                Err(e) => {
+                    debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Blockchain last retriaval failed: {e}");
+                    continue
+                }
+            };
             let request = ForkSyncRequest { tip: last.1, fork_tip: Some(proposal_copy.0.hash) };
-            self.channel.send(&request).await?;
+            if let Err(e) = self.channel.send(&request).await {
+                debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Channel send failed: {e}");
+                continue
+            };
 
             // Node waits for response
             let response = match self
@@ -150,7 +163,7 @@ impl ProtocolProposal {
             {
                 Ok(r) => r,
                 Err(e) => {
-                    debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Asking peer for fork sequence failed: {}", e);
+                    debug!(target: "darkfid::proto::protocol_proposal::handle_receive_proposal", "Asking peer for fork sequence failed: {e}");
                     continue
                 }
             };
@@ -184,7 +197,13 @@ impl ProtocolProposal {
             }
 
             for proposal in &response.proposals {
-                self.validator.append_proposal(proposal).await?;
+                if let Err(e) = self.validator.append_proposal(proposal).await {
+                    error!(
+                        target: "darkfid::proto::protocol_proposal::handle_receive_proposal",
+                        "Error while appending response proposal: {e}"
+                    );
+                    break
+                };
                 let message = ProposalMessage(proposal.clone());
                 self.p2p.broadcast_with_exclude(&message, &exclude_list).await;
                 // Notify subscriber

bin/darkfid/src/proto/protocol_sync.rs

@@ -178,8 +178,7 @@ impl ProtocolSync {
                 Err(e) => {
                     debug!(
                         target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
-                        "recv fail: {}",
-                        e
+                        "recv fail: {e}"
                     );
                     continue
                 }
@@ -207,8 +206,7 @@ impl ProtocolSync {
                     Err(e) => {
                         error!(
                             target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
-                            "block_store.contains fail: {}",
-                            e
+                            "block_store.contains fail: {e}"
                         );
                         continue
                     }
@@ -220,8 +218,7 @@ impl ProtocolSync {
                     Err(e) => {
                         error!(
                             target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
-                            "blockchain.last fail: {}",
-                            e
+                            "blockchain.last fail: {e}"
                         );
                         continue
                     }
@@ -233,8 +230,7 @@ impl ProtocolSync {
             if let Err(e) = self.channel.send(&response).await {
                 error!(
                     target: "darkfid::proto::protocol_sync::handle_receive_tip_request",
-                    "channel send fail: {}",
-                    e
+                    "Channel send fail: {e}"
                 )
             };
         }

bin/darkfid/src/proto/protocol_tx.rs

@@ -84,8 +84,7 @@ impl ProtocolTx {
                 Err(e) => {
                     debug!(
                         target: "darkfid::proto::protocol_tx::handle_receive_tx",
-                        "recv fail: {}",
-                        e
+                        "recv fail: {e}"
                     );
                     continue
                 }
@@ -113,8 +112,7 @@ impl ProtocolTx {
                 Err(e) => {
                     debug!(
                         target: "darkfid::proto::protocol_tx::handle_receive_tx",
-                        "append_tx fail: {}",
-                        e
+                        "append_tx fail: {e}"
                     );
                 }
             }

bin/darkfid/src/task/consensus.rs

@@ -24,8 +24,6 @@ use log::{error, info};
 
 use crate::{task::garbage_collect_task, Darkfid};
 
-// TODO: handle all ? so the task don't stop on errors
-
 /// 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<()> {
     info!(target: "darkfid::task::consensus_task", "Starting consensus task...");
@@ -50,7 +48,16 @@ pub async fn consensus_task(node: Arc<Darkfid>, ex: Arc<smol::Executor<'static>>
         subscription.receive().await;
 
         // Check if we can finalize anything and broadcast them
-        let finalized = node.validator.finalization().await?;
+        let finalized = match node.validator.finalization().await {
+            Ok(f) => f,
+            Err(e) => {
+                error!(
+                    target: "darkfid::task::consensus_task",
+                    "Finalization failed: {e}"
+                );
+                continue
+            }
+        };
         if !finalized.is_empty() {
             let mut notif_blocks = Vec::with_capacity(finalized.len());
             for block in finalized {

bin/darkfid/src/task/garbage_collect.rs

@@ -24,12 +24,10 @@ use darkfi::{
     Error, Result,
 };
 use darkfi_sdk::crypto::MerkleTree;
-use log::info;
+use log::{error, info};
 
 use crate::Darkfid;
 
-// TODO: handle all ? so the task don't stop on errors
-
 /// Async task used for purging erroneous pending transactions from the nodes mempool.
 pub async fn garbage_collect_task(node: Arc<Darkfid>) -> Result<()> {
     info!(target: "darkfid::task::garbage_collect_task", "Starting garbage collection task...");
@@ -37,7 +35,16 @@ pub async fn garbage_collect_task(node: Arc<Darkfid>) -> Result<()> {
     // Grab all current unproposed transactions.  We verify them in batches,
     // to not load them all in memory.
     let (mut last_checked, mut txs) =
-        node.validator.blockchain.transactions.get_after_pending(0, TXS_CAP)?;
+        match node.validator.blockchain.transactions.get_after_pending(0, TXS_CAP) {
+            Ok(pair) => pair,
+            Err(e) => {
+                error!(
+                    target: "darkfid::task::garbage_collect_task",
+                    "Uproposed transactions retrieval failed: {e}"
+                );
+                return Ok(())
+            }
+        };
     while !txs.is_empty() {
         // Verify each one against current forks
         for tx in txs {
@@ -50,11 +57,29 @@ pub async fn garbage_collect_task(node: Arc<Darkfid>) -> Result<()> {
             // Iterate over them to verify transaction validity in their overlays
             for fork in forks.iter_mut() {
                 // Clone forks' overlay
-                let overlay = fork.overlay.lock().unwrap().full_clone()?;
+                let overlay = match fork.overlay.lock().unwrap().full_clone() {
+                    Ok(o) => o,
+                    Err(e) => {
+                        error!(
+                            target: "darkfid::task::garbage_collect_task",
+                            "Overlay full clone creation failed: {e}"
+                        );
+                        break
+                    }
+                };
 
                 // Grab all current proposals transactions hashes
                 let proposals_txs =
-                    overlay.lock().unwrap().get_blocks_txs_hashes(&fork.proposals)?;
+                    match overlay.lock().unwrap().get_blocks_txs_hashes(&fork.proposals) {
+                        Ok(txs) => txs,
+                        Err(e) => {
+                            error!(
+                                target: "darkfid::task::garbage_collect_task",
+                                "Proposal transactions retrieval failed: {e}"
+                            );
+                            break
+                        }
+                    };
 
                 // If the hash is contained in the proposals transactions vec, skip it
                 if proposals_txs.contains(&tx_hash) {
@@ -62,7 +87,16 @@ pub async fn garbage_collect_task(node: Arc<Darkfid>) -> Result<()> {
                 }
 
                 // Grab forks' next block height
-                let next_block_height = fork.get_next_block_height()?;
+                let next_block_height = match fork.get_next_block_height() {
+                    Ok(h) => h,
+                    Err(e) => {
+                        error!(
+                            target: "darkfid::task::garbage_collect_task",
+                            "Next fork block height retrieval failed: {e}"
+                        );
+                        break
+                    }
+                };
 
                 // Verify transaction
                 match verify_transactions(
@@ -87,7 +121,16 @@ pub async fn garbage_collect_task(node: Arc<Darkfid>) -> Result<()> {
             drop(forks);
         }
         (last_checked, txs) =
-            node.validator.blockchain.transactions.get_after_pending(last_checked, TXS_CAP)?;
+            match node.validator.blockchain.transactions.get_after_pending(last_checked, TXS_CAP) {
+                Ok(pair) => pair,
+                Err(e) => {
+                    error!(
+                        target: "darkfid::task::garbage_collect_task",
+                        "Uproposed transactions next batch retrieval failed: {e}"
+                    );
+                    break
+                }
+            };
     }
     info!(target: "darkfid::task::garbage_collect_task", "Garbage collection finished successfully!");
     Ok(())

bin/darkfid/src/task/miner.rs

@@ -135,18 +135,53 @@ pub async fn miner_task(
     loop {
         // Grab best current fork
         let forks = node.validator.consensus.forks.read().await;
-        let extended_fork = forks[best_fork_index(&forks)?].full_clone()?;
+        let index = match best_fork_index(&forks) {
+            Ok(i) => i,
+            Err(e) => {
+                error!(
+                    target: "darkfid::task::miner_task",
+                    "Finding best fork index failed: {e}"
+                );
+                continue
+            }
+        };
+        let extended_fork = match forks[index].full_clone() {
+            Ok(f) => f,
+            Err(e) => {
+                error!(
+                    target: "darkfid::task::miner_task",
+                    "Fork full clone creation failed: {e}"
+                );
+                continue
+            }
+        };
         drop(forks);
 
         // Start listenning for network proposals and mining next block for best fork.
-        smol::future::or(
+        if let Err(e) = smol::future::or(
             listen_to_network(&node, &extended_fork, &subscription, &sender),
             mine(&node, &extended_fork, &mut secret, &recipient, &zkbin, &pk, &stop_signal),
         )
-        .await?;
+        .await
+        {
+            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 = node.validator.finalization().await?;
+        let finalized = match node.validator.finalization().await {
+            Ok(f) => f,
+            Err(e) => {
+                error!(
+                    target: "darkfid::task::miner_task",
+                    "Finalization failed: {e}"
+                );
+                continue
+            }
+        };
         if !finalized.is_empty() {
             let mut notif_blocks = Vec::with_capacity(finalized.len());
             for block in finalized {

bin/darkfid/src/task/mod.rs

@@ -16,8 +16,6 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-// TODO: Handle ? with matches in these files. They should be robust.
-
 pub mod consensus;
 pub use consensus::consensus_task;