darkfid: pending txs garbage collection added

bin/darkfid/src/main.rs

@@ -332,9 +332,6 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
         *darkfid.validator.synced.write().await = true;
     }
 
-    // Clean node pending transactions
-    darkfid.validator.purge_pending_txs().await?;
-
     // Consensus protocol
     info!(target: "darkfid", "Starting consensus protocol task");
     let consensus_task = if blockchain_config.miner {
@@ -349,9 +346,8 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
 
         let task = StoppableTask::new();
         task.clone().start(
-            // Weird hack to prevent lifetimes hell
-            async move { miner_task(&darkfid, &recipient, blockchain_config.skip_sync).await },
-            |res| async {
+            miner_task(darkfid, 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),
@@ -365,9 +361,8 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     } else {
         let task = StoppableTask::new();
         task.clone().start(
-            // Weird hack to prevent lifetimes hell
-            async move { consensus_task(&darkfid).await },
-            |res| async {
+            consensus_task(darkfid, ex.clone()),
+            |res| async move {
                 match res {
                     Ok(()) | Err(Error::ConsensusTaskStopped) => { /* Do nothing */ }
                     Err(e) => error!(target: "darkfid", "Failed starting consensus task: {}", e),

bin/darkfid/src/task/consensus.rs

@@ -16,16 +16,18 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use darkfi::{rpc::util::JsonValue, util::encoding::base64, Result};
+use std::sync::Arc;
+
+use darkfi::{rpc::util::JsonValue, system::StoppableTask, util::encoding::base64, Error, Result};
 use darkfi_serial::serialize_async;
-use log::info;
+use log::{error, info};
 
-use crate::Darkfid;
+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: &Darkfid) -> Result<()> {
+/// 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...");
 
     // Grab blocks subscriber
@@ -35,6 +37,9 @@ pub async fn consensus_task(node: &Darkfid) -> Result<()> {
     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);
+
     loop {
         subscription.receive().await;
 
@@ -47,6 +52,20 @@ pub async fn consensus_task(node: &Darkfid) -> Result<()> {
                     .push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
             }
             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()),
+                |res| async {
+                    match res {
+                        Ok(()) => { /* Do nothing */ }
+                        Err(e) => error!(target: "darkfid", "Failed starting garbage collection task: {}", e),
+                    }
+                },
+                Error::MinerTaskStopped,
+                ex.clone(),
+            );
         }
     }
 }

bin/darkfid/src/task/garbage_collect.rs

@@ -0,0 +1,100 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2024 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::sync::Arc;
+
+use darkfi::{
+    error::TxVerifyFailed,
+    validator::{consensus::TXS_CAP, verification::verify_transactions},
+    Error, Result,
+};
+use darkfi_sdk::crypto::MerkleTree;
+use log::info;
+use smol::channel::Receiver;
+
+use crate::{task::miner::wait_stop_signal, 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
+}
+
+/// 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) =
+        node.validator.blockchain.transactions.get_after_pending(0, TXS_CAP)?;
+    while !txs.is_empty() {
+        // Verify each one against current forks
+        for tx in txs {
+            let tx_hash = tx.hash();
+            let tx_vec = [tx.clone()];
+
+            // Grab a lock over current consensus forks state
+            let mut forks = node.validator.consensus.forks.write().await;
+
+            // 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()?;
+
+                // Grab all current proposals transactions hashes
+                let proposals_txs =
+                    overlay.lock().unwrap().get_blocks_txs_hashes(&fork.proposals)?;
+
+                // If the hash is contained in the proposals transactions vec, skip it
+                if proposals_txs.contains(&tx_hash) {
+                    continue
+                }
+
+                // Grab forks' next block height
+                let next_block_height = fork.get_next_block_height()?;
+
+                // Verify transaction
+                match verify_transactions(
+                    &overlay,
+                    next_block_height,
+                    &tx_vec,
+                    &mut MerkleTree::new(1),
+                    false,
+                )
+                .await
+                {
+                    Ok(_) => {}
+                    Err(Error::TxVerifyFailed(TxVerifyFailed::ErroneousTxs(_))) => {
+                        // Remove transaction from fork's mempool
+                        fork.mempool.retain(|tx| *tx != tx_hash);
+                    }
+                    Err(e) => return Err(e),
+                }
+            }
+
+            // Drop forks lock
+            drop(forks);
+        }
+        (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!");
+    Ok(())
+}

bin/darkfid/src/task/miner.rs

@@ -16,10 +16,12 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use std::sync::Arc;
+
 use darkfi::{
     blockchain::BlockInfo,
     rpc::{jsonrpc::JsonNotification, util::JsonValue},
-    system::Subscription,
+    system::{StoppableTask, Subscription},
     tx::{ContractCallLeaf, Transaction, TransactionBuilder},
     util::encoding::base64,
     validator::{
@@ -28,7 +30,7 @@ use darkfi::{
     },
     zk::{empty_witnesses, ProvingKey, ZkCircuit},
     zkas::ZkBinary,
-    Result,
+    Error, Result,
 };
 use darkfi_money_contract::{
     client::pow_reward_v1::PoWRewardCallBuilder, MoneyFunction, MONEY_CONTRACT_ZKAS_MINT_NS_V1,
@@ -44,7 +46,7 @@ use num_bigint::BigUint;
 use rand::rngs::OsRng;
 use smol::channel::{Receiver, Sender};
 
-use crate::{proto::ProposalMessage, Darkfid};
+use crate::{proto::ProposalMessage, task::garbage_collect_task, Darkfid};
 
 // TODO: handle all ? so the task don't stop on errors
 
@@ -58,7 +60,12 @@ use crate::{proto::ProposalMessage, Darkfid};
 /// proposals produce a new best ranking fork. If they do, the stop
 /// mining. These two tasks run in parallel, and after one of them
 /// finishes, node triggers finallization check.
-pub async fn miner_task(node: &Darkfid, recipient: &PublicKey, skip_sync: bool) -> Result<()> {
+pub async fn miner_task(
+    node: Arc<Darkfid>,
+    recipient: PublicKey,
+    skip_sync: bool,
+    ex: Arc<smol::Executor<'static>>,
+) -> Result<()> {
     // Initialize miner configuration
     info!(target: "darkfid::task::miner_task", "Starting miner task...");
 
@@ -112,6 +119,7 @@ pub async fn miner_task(node: &Darkfid, recipient: &PublicKey, skip_sync: bool)
 
     // 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);
 
     info!(target: "darkfid::task::miner_task", "Miner initialized successfully!");
 
@@ -124,8 +132,8 @@ pub async fn miner_task(node: &Darkfid, recipient: &PublicKey, skip_sync: bool)
 
         // Start listenning for network proposals and mining next block for best fork.
         smol::future::or(
-            listen_to_network(node, &extended_fork, &subscription, &sender),
-            mine(node, &extended_fork, &mut secret, recipient, &zkbin, &pk, &stop_signal),
+            listen_to_network(&node, &extended_fork, &subscription, &sender),
+            mine(&node, &extended_fork, &mut secret, &recipient, &zkbin, &pk, &stop_signal),
         )
         .await?;
 
@@ -138,11 +146,25 @@ pub async fn miner_task(node: &Darkfid, recipient: &PublicKey, skip_sync: bool)
                     .push(JsonValue::String(base64::encode(&serialize_async(&block).await)));
             }
             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()),
+                |res| async {
+                    match res {
+                        Ok(()) => { /* Do nothing */ }
+                        Err(e) => error!(target: "darkfid", "Failed starting garbage collection task: {}", e),
+                    }
+                },
+                Error::MinerTaskStopped,
+                ex.clone(),
+            );
         }
     }
 }
 
-/// Async task to listen for incoming proposals and check if the best fork has changed
+/// Async task to listen for incoming proposals and check if the best fork has changed.
 async fn listen_to_network(
     node: &Darkfid,
     extended_fork: &Fork,
@@ -180,7 +202,7 @@ async fn listen_to_network(
 }
 
 /// Async task to generate and mine provided fork index next block,
-/// while listening for a stop signal
+/// while listening for a stop signal.
 async fn mine(
     node: &Darkfid,
     extended_fork: &Fork,
@@ -198,7 +220,7 @@ async fn mine(
 }
 
 /// Async task to wait for listener's stop signal.
-async fn wait_stop_signal(stop_signal: &Receiver<()>) -> Result<()> {
+pub async fn wait_stop_signal(stop_signal: &Receiver<()>) -> Result<()> {
     // Clean stop signal channel
     if stop_signal.is_full() {
         stop_signal.recv().await?;
@@ -210,7 +232,7 @@ async fn wait_stop_signal(stop_signal: &Receiver<()>) -> Result<()> {
     Ok(())
 }
 
-/// Async task to generate and mine provided fork index next block
+/// Async task to generate and mine provided fork index next block.
 async fn mine_next_block(
     node: &Darkfid,
     extended_fork: &Fork,
@@ -247,7 +269,7 @@ async fn mine_next_block(
     Ok(())
 }
 
-/// Auxiliary function to generate next block in an atomic manner
+/// Auxiliary function to generate next block in an atomic manner.
 async fn generate_next_block(
     extended_fork: &Fork,
     secret: &mut SecretKey,
@@ -276,7 +298,7 @@ async fn generate_next_block(
     Ok((target, next_block))
 }
 
-/// Auxiliary function to generate a Money::PoWReward transaction
+/// Auxiliary function to generate a Money::PoWReward transaction.
 fn generate_transaction(
     block_height: u32,
     secret: &SecretKey,

bin/darkfid/src/task/mod.rs

@@ -26,3 +26,6 @@ pub use miner::miner_task;
 
 pub mod sync;
 pub use sync::sync_task;
+
+pub mod garbage_collect;
+pub use garbage_collect::garbage_collect_task;

src/blockchain/tx_store.rs

@@ -323,6 +323,34 @@ impl TxStore {
         Ok(txs)
     }
 
+    /// Fetch n transactions after given order. In the iteration, if a transaction
+    /// order is not found, the iteration stops and the function returns what
+    /// it has found so far in the store's pending order tree.
+    pub fn get_after_pending(&self, order: u64, n: usize) -> Result<(u64, Vec<Transaction>)> {
+        let mut hashes = vec![];
+
+        let mut key = order;
+        let mut counter = 0;
+        while counter < n {
+            if let Some(found) = self.pending_order.get_gt(key.to_be_bytes())? {
+                let (order, hash) = parse_u64_key_record(found)?;
+                key = order;
+                hashes.push(hash);
+                counter += 1;
+                continue
+            }
+            break
+        }
+
+        if hashes.is_empty() {
+            return Ok((key, vec![]))
+        }
+
+        let txs = self.get_pending(&hashes, true)?.iter().map(|tx| tx.clone().unwrap()).collect();
+
+        Ok((key, txs))
+    }
+
     /// Retrieve records count of the store's main tree.
     pub fn len(&self) -> usize {
         self.main.len()

src/validator/mod.rs

@@ -148,8 +148,7 @@ impl Validator {
         // Grab a lock over current consensus forks state
         let mut forks = self.consensus.forks.write().await;
 
-        // If node participates in consensus and holds any forks, iterate over them
-        // to verify transaction validity in their overlays
+        // 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()?;
@@ -180,30 +179,12 @@ impl Validator {
             }
         }
 
-        // Verify transaction against canonical state
-        let overlay = BlockchainOverlay::new(&self.blockchain)?;
-        let next_block_height = self.blockchain.last_block()?.header.height + 1;
-        let mut erroneous_txs = vec![];
-        match verify_transactions(
-            &overlay,
-            next_block_height,
-            &tx_vec,
-            &mut MerkleTree::new(1),
-            false,
-        )
-        .await
-        {
-            Ok(_) => valid = true,
-            Err(Error::TxVerifyFailed(TxVerifyFailed::ErroneousTxs(etx))) => erroneous_txs = etx,
-            Err(e) => return Err(e),
-        }
-
         // Drop forks lock
         drop(forks);
 
-        // Return error if transaction is not valid for canonical or any fork
+        // Return error if transaction is not valid for any fork
         if !valid {
-            return Err(TxVerifyFailed::ErroneousTxs(erroneous_txs).into())
+            return Err(TxVerifyFailed::ErroneousTxs(tx_vec.to_vec()).into())
         }
 
         // Add transaction to pending txs store