darkfid/rpc/tx: properly cleanup pending txs

bin/darkfid/src/rpc/tx.rs

@@ -173,8 +173,8 @@ impl DarkfiNode {
     }
 
     // RPCAPI:
-    // Queries the node pending transactions store to remove all
-    // transactions.
+    // Queries the node pending transactions store to reset all
+    // transactions. Unproposed transactions are removed.
     // Returns `true` if the operation was successful, otherwise, a
     // corresponding error.
     //
@@ -193,7 +193,7 @@ impl DarkfiNode {
             return server_error(RpcError::NotSynced, id, None)
         }
 
-        if let Err(e) = self.validator.blockchain.remove_all_pending_txs() {
+        if let Err(e) = self.validator.consensus.purge_unproposed_pending_txs().await {
             error!(target: "darkfid::rpc::tx_clean_pending", "Failed removing pending txs: {e}");
             return JsonError::new(InternalError, None, id).into()
         };

src/blockchain/mod.rs

@@ -346,19 +346,32 @@ impl Blockchain {
         Ok(())
     }
 
-    /// Remove all transactions from the pending tx store.
-    pub fn remove_all_pending_txs(&self) -> Result<()> {
-        let txs: Vec<TransactionHash> =
-            self.transactions.get_all_pending()?.keys().copied().collect();
+    /// Remove all transactions from the pending tx store not in the
+    /// provided vector and rebuild the remaining ones order.
+    pub fn reset_pending_txs(&self, exclude_txs: &[TransactionHash]) -> Result<()> {
+        let mut txs = vec![];
+        let mut removed_txs = vec![];
+        for tx in self.transactions.get_all_pending()?.keys() {
+            if exclude_txs.contains(tx) {
+                txs.push(*tx);
+                continue
+            }
+            removed_txs.push(*tx);
+        }
         let indexes: Vec<u64> =
             self.transactions.get_all_pending_order()?.iter().map(|(k, _)| *k).collect();
 
-        let txs_batch = self.transactions.remove_batch_pending(&txs);
+        let txs_batch = self.transactions.remove_batch_pending(&removed_txs);
         let txs_order_batch = self.transactions.remove_batch_pending_order(&indexes);
+        let txs_new_order_batch = self.transactions.insert_batch_pending_order(&txs)?;
 
         // Perform an atomic transaction over the trees and apply the batches.
-        let trees = [self.transactions.pending.clone(), self.transactions.pending_order.clone()];
-        let batches = [txs_batch, txs_order_batch];
+        let trees = [
+            self.transactions.pending.clone(),
+            self.transactions.pending_order.clone(),
+            self.transactions.pending_order.clone(),
+        ];
+        let batches = [txs_batch, txs_order_batch, txs_new_order_batch];
         self.atomic_write(&trees, &batches)?;
 
         Ok(())

src/validator/consensus.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::collections::{BTreeSet, HashMap};
+use std::collections::{BTreeSet, HashMap, HashSet};
 
 use darkfi_sdk::{crypto::MerkleTree, tx::TransactionHash};
 use darkfi_serial::{async_trait, deserialize, SerialDecodable, SerialEncodable};
@@ -665,6 +665,38 @@ impl Consensus {
 
         Ok(())
     }
+
+    /// Auxiliary function to purge all unproposed pending
+    /// transactions from the database.
+    pub async fn purge_unproposed_pending_txs(&self) -> 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
+            let proposals_txs =
+                fork.overlay.lock().unwrap().get_blocks_txs_hashes(&fork.proposals)?;
+            for tx in proposals_txs {
+                proposed_txs.insert(tx);
+            }
+        }
+
+        // Iterate over all forks again to remove unproposed txs from
+        // their mempools.
+        for fork in forks.iter_mut() {
+            fork.mempool.retain(|tx| proposed_txs.contains(tx));
+        }
+
+        // Remove unproposed txs from the pending store
+        let proposed_txs: Vec<TransactionHash> = proposed_txs.into_iter().collect();
+        self.blockchain.reset_pending_txs(&proposed_txs)?;
+
+        Ok(())
+    }
 }
 
 /// This struct represents a block proposal, used for consensus.