darkfid: properly handle the garbage collection task

bin/darkfid/src/task/consensus.rs

@@ -37,8 +37,14 @@ pub async fn consensus_task(node: Arc<Darkfid>, ex: Arc<smol::Executor<'static>>
     let proposals_sub = node.subscribers.get("proposals").unwrap();
     let subscription = proposals_sub.sub.clone().subscribe().await;
 
-    // Create channels so threads can signal each other
-    let (gc_sender, gc_stop_signal) = smol::channel::bounded(1);
+    // Create the garbage collection task using a dummy task
+    let gc_task = StoppableTask::new();
+    gc_task.clone().start(
+        async { Ok(()) },
+        |_| async { /* Do nothing */ },
+        Error::GarbageCollectionTaskStopped,
+        ex.clone(),
+    );
 
     loop {
         subscription.receive().await;
@@ -53,17 +59,17 @@ pub async fn consensus_task(node: Arc<Darkfid>, ex: Arc<smol::Executor<'static>>
             }
             block_sub.notify(JsonValue::Array(notif_blocks)).await;
 
-            // Invoke detached garbage collection task
-            gc_sender.send(()).await?;
-            StoppableTask::new().start(
-                garbage_collect_task(node.clone(), gc_stop_signal.clone()),
+            // Invoke the detached garbage collection task
+            gc_task.clone().stop().await;
+            gc_task.clone().start(
+                garbage_collect_task(node.clone()),
                 |res| async {
                     match res {
-                        Ok(()) => { /* Do nothing */ }
+                        Ok(()) | Err(Error::GarbageCollectionTaskStopped) => { /* Do nothing */ }
                         Err(e) => error!(target: "darkfid", "Failed starting garbage collection task: {}", e),
                     }
                 },
-                Error::MinerTaskStopped,
+                Error::GarbageCollectionTaskStopped,
                 ex.clone(),
             );
         }

bin/darkfid/src/task/garbage_collect.rs

@@ -25,21 +25,15 @@ use darkfi::{
 };
 use darkfi_sdk::crypto::MerkleTree;
 use log::info;
-use smol::channel::Receiver;
 
-use crate::{task::miner::wait_stop_signal, Darkfid};
+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>, stop_signal: Receiver<()>) -> Result<()> {
-    // Start mempool puring and wait for stop signal
-    smol::future::or(wait_stop_signal(&stop_signal), purge(&node)).await
-}
+pub async fn garbage_collect_task(node: Arc<Darkfid>) -> Result<()> {
+    info!(target: "darkfid::task::garbage_collect_task", "Starting garbage collection task...");
 
-/// Async task to purge erroneous pending transactions from the nodes mempool.
-async fn purge(node: &Darkfid) -> Result<()> {
-    info!(target: "darkfid::task::garbage_collect::purge", "Starting garbage collection task...");
     // Grab all current unproposed transactions.  We verify them in batches,
     // to not load them all in memory.
     let (mut last_checked, mut txs) =
@@ -95,6 +89,6 @@ async fn purge(node: &Darkfid) -> Result<()> {
         (last_checked, txs) =
             node.validator.blockchain.transactions.get_after_pending(last_checked, TXS_CAP)?;
     }
-    info!(target: "darkfid::task::garbage_collect::purge", "Garbage collection finished successfully!");
+    info!(target: "darkfid::task::garbage_collect_task", "Garbage collection finished successfully!");
     Ok(())
 }

bin/darkfid/src/task/miner.rs

@@ -119,7 +119,15 @@ pub async fn miner_task(
 
     // Create channels so threads can signal each other
     let (sender, stop_signal) = smol::channel::bounded(1);
-    let (gc_sender, gc_stop_signal) = smol::channel::bounded(1);
+
+    // Create the garbage collection task using a dummy task
+    let gc_task = StoppableTask::new();
+    gc_task.clone().start(
+        async { Ok(()) },
+        |_| async { /* Do nothing */ },
+        Error::GarbageCollectionTaskStopped,
+        ex.clone(),
+    );
 
     info!(target: "darkfid::task::miner_task", "Miner initialized successfully!");
 
@@ -147,17 +155,17 @@ pub async fn miner_task(
             }
             block_sub.notify(JsonValue::Array(notif_blocks)).await;
 
-            // Invoke detached garbage collection task
-            gc_sender.send(()).await?;
-            StoppableTask::new().start(
-                garbage_collect_task(node.clone(), gc_stop_signal.clone()),
+            // Invoke the detached garbage collection task
+            gc_task.clone().stop().await;
+            gc_task.clone().start(
+                garbage_collect_task(node.clone()),
                 |res| async {
                     match res {
-                        Ok(()) => { /* Do nothing */ }
+                        Ok(()) | Err(Error::GarbageCollectionTaskStopped) => { /* Do nothing */ }
                         Err(e) => error!(target: "darkfid", "Failed starting garbage collection task: {}", e),
                     }
                 },
-                Error::MinerTaskStopped,
+                Error::GarbageCollectionTaskStopped,
                 ex.clone(),
             );
         }

src/error.rs

@@ -325,6 +325,9 @@ pub enum Error {
     #[error("Miner task stopped")]
     MinerTaskStopped,
 
+    #[error("Garbage collection task stopped")]
+    GarbageCollectionTaskStopped,
+
     #[error("Calculated total work is zero")]
     PoWTotalWorkIsZero,