darkfid: include miners templates txs when purging pending txs and remove erroneous ones from fork mempool when retrieving its unproposed ones

bin/darkfid/src/registry/mod.rs

@@ -38,7 +38,10 @@ use darkfi::{
     validator::{consensus::Proposal, ValidatorPtr},
     Error, Result,
 };
-use darkfi_sdk::crypto::{keypair::Network, pasta_prelude::PrimeField};
+use darkfi_sdk::{
+    crypto::{keypair::Network, pasta_prelude::PrimeField},
+    tx::TransactionHash,
+};
 use darkfi_serial::serialize_async;
 
 use crate::{
@@ -500,4 +503,19 @@ impl DarkfiMinersRegistry {
         }
         new_trees
     }
+
+    /// Auxilliary function to retrieve all current block templates
+    /// transactions hashes.
+    pub fn proposed_transactions(
+        &self,
+        block_templates: &HashMap<String, BlockTemplate>,
+    ) -> HashSet<TransactionHash> {
+        let mut proposed_txs = HashSet::new();
+        for block_template in block_templates.values() {
+            for tx in &block_template.block.txs {
+                proposed_txs.insert(tx.hash());
+            }
+        }
+        proposed_txs
+    }
 }

bin/darkfid/src/rpc/tx.rs

@@ -193,7 +193,27 @@ impl DarkfiNode {
             return server_error(RpcError::NotSynced, id, None)
         }
 
-        if let Err(e) = self.validator.consensus.purge_unproposed_pending_txs().await {
+        // Grab node registry locks
+        let submit_lock = self.registry.submit_lock.write().await;
+        let block_templates = self.registry.block_templates.write().await;
+        let jobs = self.registry.jobs.write().await;
+        let mm_jobs = self.registry.mm_jobs.write().await;
+
+        // Purge all unproposed pending transactions from the database
+        let result = self
+            .validator
+            .consensus
+            .purge_unproposed_pending_txs(self.registry.proposed_transactions(&block_templates))
+            .await;
+
+        // Release registry locks
+        drop(block_templates);
+        drop(jobs);
+        drop(mm_jobs);
+        drop(submit_lock);
+
+        // Check result
+        if let Err(e) = result {
             error!(target: "darkfid::rpc::tx_clean_pending", "Failed removing pending txs: {e}");
             return JsonError::new(InternalError, None, id).into()
         };

bin/darkfid/src/tests/unproposed_txs.rs

@@ -87,8 +87,9 @@ async fn simulate_unproposed_txs(
     }
 
     // Obtain fork
-    let forks = validator.consensus.forks.read().await;
-    let best_fork = &forks[best_fork_index(&forks)?];
+    let mut forks = validator.consensus.forks.write().await;
+    let index = best_fork_index(&forks)?;
+    let best_fork = &mut forks[index];
 
     // Retrieve unproposed transactions
     let (tx, total_gas_used, _) = best_fork.unproposed_txs(current_block_height, false).await?;

src/validator/consensus.rs

@@ -668,13 +668,13 @@ impl Consensus {
 
     /// Auxiliary function to purge all unproposed pending
     /// transactions from the database.
-    pub async fn purge_unproposed_pending_txs(&self) -> Result<()> {
+    pub async fn purge_unproposed_pending_txs(
+        &self,
+        mut proposed_txs: HashSet<TransactionHash>,
+    ) -> Result<()> {
         // Grab a lock over current forks
         let mut forks = self.forks.write().await;
 
-        // Keep track of proposed txs
-        let mut proposed_txs = HashSet::new();
-
         // Iterate over all forks to find proposed txs
         for fork in forks.iter() {
             // Grab all current proposals transactions hashes
@@ -829,7 +829,7 @@ impl Fork {
     /// Note: Always remember to purge new trees from the database if
     /// not needed.
     pub async fn unproposed_txs(
-        &self,
+        &mut self,
         verifying_block_height: u32,
         verify_fees: bool,
     ) -> Result<(Vec<Transaction>, u64, u64)> {
@@ -853,6 +853,7 @@ impl Fork {
 
         // Iterate through all pending transactions in the forks' mempool
         let mut unproposed_txs = vec![];
+        let mut erroneous_txs = vec![];
         for tx in &self.mempool {
             // If the hash is contained in the proposals transactions vec, skip it
             if proposals_txs.contains(tx) {
@@ -860,8 +861,14 @@ impl Fork {
             }
 
             // Retrieve the actual unproposed transaction
-            let unproposed_tx =
-                self.blockchain.transactions.get_pending(&[*tx], true)?[0].clone().unwrap();
+            let unproposed_tx = match self.blockchain.transactions.get_pending(&[*tx], true) {
+                Ok(txs) => txs[0].clone().unwrap(),
+                Err(e) => {
+                    debug!(target: "validator::consensus::unproposed_txs", "Transaction retrieval failed: {e}");
+                    erroneous_txs.push(*tx);
+                    continue
+                }
+            };
 
             // Update the verifying keys map
             for call in &unproposed_tx.calls {
@@ -885,6 +892,7 @@ impl Fork {
                 Err(e) => {
                     debug!(target: "validator::consensus::unproposed_txs", "Transaction verification failed: {e}");
                     self.overlay.lock().unwrap().revert_to_checkpoint();
+                    erroneous_txs.push(*tx);
                     continue
                 }
             };
@@ -913,6 +921,9 @@ impl Fork {
             unproposed_txs.push(unproposed_tx);
         }
 
+        // Remove erroneous transactions txs from fork's mempool
+        self.mempool.retain(|tx| !erroneous_txs.contains(tx));
+
         Ok((unproposed_txs, total_gas_used, total_gas_paid))
     }