blockchain: removed pending transactions order tree and related fns

bin/darkfid/src/rpc/tx.rs

@@ -198,8 +198,8 @@ impl DarkfiNode {
     }
 
     // RPCAPI:
-    // Queries the node pending transactions store to reset all
-    // transactions. Unproposed transactions are removed.
+    // Queries the node pending transactions store to remove all
+    // transactions.
     // Returns `true` if the operation was successful, otherwise, a
     // corresponding error.
     //
@@ -213,18 +213,15 @@ impl DarkfiNode {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        let mut validator = self.validator.write().await;
+        let validator = self.validator.write().await;
         if !validator.synced {
             error!(target: "darkfid::rpc::tx_clean_pending", "Blockchain is not synced");
             return server_error(RpcError::NotSynced, id, None)
         }
 
-        // Retrieve registry transactions
-        let registry_txs = self.registry.state.read().await.proposed_transactions();
-
-        // Purge all unproposed pending transactions from the database
-        if let Err(e) = validator.consensus.purge_unproposed_pending_txs(registry_txs).await {
-            error!(target: "darkfid::rpc::tx_clean_pending", "Failed removing pending txs: {e}");
+        // Purge all pending transactions from the database
+        if let Err(e) = validator.blockchain.transactions.pending.clear() {
+            error!(target: "darkfid::rpc::tx_clean_pending", "Failed cleaning pending txs: {e}");
             return JsonError::new(InternalError, None, id).into()
         };
 

bin/darkfid/src/task/garbage_collect.rs

@@ -17,7 +17,7 @@
  */
 
 use darkfi::{error::TxVerifyFailed, validator::verification::verify_transactions, Error, Result};
-use darkfi_sdk::crypto::MerkleTree;
+use darkfi_sdk::{crypto::MerkleTree, tx::TransactionHash};
 use tracing::{debug, error, info};
 
 use crate::DarkfiNodePtr;
@@ -29,17 +29,21 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
     // Grab all current unproposed transactions.  We verify them in batches,
     // to not load them all in memory.
     let validator = node.validator.read().await;
-    let (mut last_checked, mut txs) =
-        match validator.blockchain.transactions.get_after_pending(0, node.txs_batch_size) {
-            Ok(pair) => pair,
-            Err(e) => {
-                error!(
-                    target: "darkfid::task::garbage_collect_task",
-                    "Uproposed transactions retrieval failed: {e}"
-                );
-                return Ok(())
-            }
-        };
+    let mut last_checked = TransactionHash::none();
+    let mut txs = match validator
+        .blockchain
+        .transactions
+        .get_after_pending(&last_checked, node.txs_batch_size)
+    {
+        Ok(txs) => txs,
+        Err(e) => {
+            error!(
+                target: "darkfid::task::garbage_collect_task",
+                "Uproposed transactions retrieval failed: {e}"
+            );
+            return Ok(())
+        }
+    };
 
     // Check if we have transactions to process
     if txs.is_empty() {
@@ -54,7 +58,7 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
     while !txs.is_empty() {
         // Verify each one against current forks
         for tx in txs {
-            let tx_hash = tx.hash();
+            last_checked = tx.hash();
             let tx_vec = [tx.clone()];
             let mut valid = false;
 
@@ -89,7 +93,7 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
                     };
 
                 // If the hash is contained in the proposals transactions vec, skip it
-                if proposals_txs.contains(&tx_hash) {
+                if proposals_txs.contains(&last_checked) {
                     continue
                 }
 
@@ -121,12 +125,12 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
                     Ok(_) => valid = true,
                     Err(Error::TxVerifyFailed(TxVerifyFailed::ErroneousTxs(_))) => {
                         // Remove transaction from fork's mempool
-                        fork.mempool.retain(|tx| *tx != tx_hash);
+                        fork.mempool.retain(|tx| *tx != last_checked);
                     }
                     Err(e) => {
                         error!(
                             target: "darkfid::task::garbage_collect_task",
-                            "Verifying transaction {tx_hash} failed: {e}"
+                            "Verifying transaction {last_checked} failed: {e}"
                         );
                         return Err(e)
                     }
@@ -135,26 +139,26 @@ pub async fn garbage_collect_task(node: DarkfiNodePtr) -> Result<()> {
 
             // Remove transaction if its invalid for all the forks
             if !valid {
-                debug!(target: "darkfid::task::garbage_collect_task", "Removing invalid transaction: {tx_hash}");
-                if let Err(e) = validator.blockchain.remove_pending_txs_hashes(&[tx_hash]) {
+                debug!(target: "darkfid::task::garbage_collect_task", "Removing invalid transaction: {last_checked}");
+                if let Err(e) = validator.blockchain.remove_pending_txs_hashes(&[last_checked]) {
                     error!(
                         target: "darkfid::task::garbage_collect_task",
-                        "Removing invalid transaction {tx_hash} failed: {e}"
+                        "Removing invalid transaction {last_checked} failed: {e}"
                     );
                 };
             }
         }
 
         // Grab next batch
-        (last_checked, txs) = match node
+        txs = match node
             .validator
             .read()
             .await
             .blockchain
             .transactions
-            .get_after_pending(last_checked + node.txs_batch_size as u64, node.txs_batch_size)
+            .get_after_pending(&last_checked, node.txs_batch_size)
         {
-            Ok(pair) => pair,
+            Ok(txs) => txs,
             Err(e) => {
                 error!(
                     target: "darkfid::task::garbage_collect_task",

bin/darkfid/src/tests/harness.rs

@@ -315,7 +315,5 @@ pub async fn generate_node(
         node.validator.write().await.synced = true;
     }
 
-    node.validator.write().await.purge_pending_txs().await?;
-
     Ok(node)
 }

src/blockchain/mod.rs

@@ -50,8 +50,7 @@ pub use header_store::{
 /// Transactions related storage implementations
 pub mod tx_store;
 pub use tx_store::{
-    TxStore, TxStoreOverlay, SLED_PENDING_TX_ORDER_TREE, SLED_PENDING_TX_TREE,
-    SLED_TX_LOCATION_TREE, SLED_TX_TREE,
+    TxStore, TxStoreOverlay, SLED_PENDING_TX_TREE, SLED_TX_LOCATION_TREE, SLED_TX_TREE,
 };
 
 /// Contracts and Wasm storage implementations
@@ -289,29 +288,17 @@ impl Blockchain {
     /// On success, the function returns the transaction hashes in the same order
     /// as the input transactions.
     pub fn add_pending_txs(&self, txs: &[Transaction]) -> Result<Vec<TransactionHash>> {
-        let (txs_batch, txs_hashes) = self.transactions.insert_batch_pending(txs);
-        let txs_order_batch = self.transactions.insert_batch_pending_order(&txs_hashes)?;
-
-        // 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];
-        self.atomic_write(&trees, &batches)?;
-
-        Ok(txs_hashes)
+        self.transactions.insert_pending(txs)
     }
 
     /// Retrieve all transactions from the pending tx store.
     /// Be careful as this will try to load everything in memory.
     pub fn get_pending_txs(&self) -> Result<Vec<Transaction>> {
         let txs = self.transactions.get_all_pending()?;
-        let indexes = self.transactions.get_all_pending_order()?;
-        if txs.len() != indexes.len() {
-            return Err(Error::InvalidInputLengths)
-        }
 
         let mut ret = Vec::with_capacity(txs.len());
-        for index in indexes {
-            ret.push(txs.get(&index.1).unwrap().clone());
+        for (_, tx) in txs {
+            ret.push(tx);
         }
 
         Ok(ret)
@@ -325,56 +312,7 @@ impl Blockchain {
 
     /// Remove a given slice of pending transactions hashes from the blockchain database.
     pub fn remove_pending_txs_hashes(&self, txs: &[TransactionHash]) -> Result<()> {
-        let indexes = self.transactions.get_all_pending_order()?;
-        // We could do indexes.iter().map(|x| txs.contains(x.1)).collect.map(|x| x.0).collect
-        // but this is faster since we don't do the second iteration
-        let mut removed_indexes = vec![];
-        for index in indexes {
-            if txs.contains(&index.1) {
-                removed_indexes.push(index.0);
-            }
-        }
-
-        let txs_batch = self.transactions.remove_batch_pending(txs);
-        let txs_order_batch = self.transactions.remove_batch_pending_order(&removed_indexes);
-
-        // 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];
-        self.atomic_write(&trees, &batches)?;
-
-        Ok(())
-    }
-
-    /// 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(&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(),
-            self.transactions.pending_order.clone(),
-        ];
-        let batches = [txs_batch, txs_order_batch, txs_new_order_batch];
-        self.atomic_write(&trees, &batches)?;
-
-        Ok(())
+        self.transactions.remove_pending(txs)
     }
 
     /// Auxiliary function to write to multiple trees completely atomic.
@@ -559,7 +497,6 @@ impl BlockchainOverlay {
             SLED_TX_TREE,
             SLED_TX_LOCATION_TREE,
             SLED_PENDING_TX_TREE,
-            SLED_PENDING_TX_ORDER_TREE,
             SLED_CONTRACTS_TREE,
             SLED_CONTRACTS_TREES_TREE,
             SLED_BINCODE_TREE,

src/blockchain/tx_store.rs

@@ -24,12 +24,11 @@ use sled_overlay::sled;
 
 use crate::{tx::Transaction, Error, Result};
 
-use super::{parse_record, parse_u64_key_record, SledDbOverlayPtr};
+use super::{parse_record, SledDbOverlayPtr};
 
 pub const SLED_TX_TREE: &[u8] = b"_transactions";
 pub const SLED_TX_LOCATION_TREE: &[u8] = b"_transaction_location";
 pub const SLED_PENDING_TX_TREE: &[u8] = b"_pending_transactions";
-pub const SLED_PENDING_TX_ORDER_TREE: &[u8] = b"_pending_transactions_order";
 
 /// The `TxStore` is a structure representing all `sled` trees related
 /// to storing the blockchain's transactions information.
@@ -48,10 +47,6 @@ pub struct TxStore {
     /// the key is the transaction hash, and the value is the serialized
     /// transaction.
     pub pending: sled::Tree,
-    /// The `sled` tree storing the order of all the node pending transactions,
-    /// where the key is an incremental value, and the value is the serialized
-    /// transaction.
-    pub pending_order: sled::Tree,
 }
 
 impl TxStore {
@@ -60,8 +55,7 @@ impl TxStore {
         let main = db.open_tree(SLED_TX_TREE)?;
         let location = db.open_tree(SLED_TX_LOCATION_TREE)?;
         let pending = db.open_tree(SLED_PENDING_TX_TREE)?;
-        let pending_order = db.open_tree(SLED_PENDING_TX_ORDER_TREE)?;
-        Ok(Self { main, location, pending, pending_order })
+        Ok(Self { main, location, pending })
     }
 
     /// Insert a slice of [`Transaction`] into the store's main tree.
@@ -85,13 +79,6 @@ impl TxStore {
         Ok(ret)
     }
 
-    /// Insert a slice of [`TransactionHash`] into the store's pending txs order tree.
-    pub fn insert_pending_order(&self, txs_hashes: &[TransactionHash]) -> Result<()> {
-        let batch = self.insert_batch_pending_order(txs_hashes)?;
-        self.pending_order.apply_batch(batch)?;
-        Ok(())
-    }
-
     /// Generate the sled batch corresponding to an insert to the main tree,
     /// so caller can handle the write operation.
     /// The transactions are hashed with BLAKE3 and this hash is used as
@@ -157,27 +144,6 @@ impl TxStore {
         (batch, ret)
     }
 
-    /// Generate the sled batch corresponding to an insert to the pending txs
-    /// order tree, so caller can handle the write operation.
-    pub fn insert_batch_pending_order(&self, tx_hashes: &[TransactionHash]) -> Result<sled::Batch> {
-        let mut batch = sled::Batch::default();
-
-        let next_index = match self.pending_order.last()? {
-            Some(n) => {
-                let prev_bytes: [u8; 8] = n.0.as_ref().try_into().unwrap();
-                let prev = u64::from_be_bytes(prev_bytes);
-                prev + 1
-            }
-            None => 0,
-        };
-
-        for (next_index, tx_hash) in (next_index..).zip(tx_hashes.iter()) {
-            batch.insert(&next_index.to_be_bytes(), tx_hash.inner());
-        }
-
-        Ok(batch)
-    }
-
     /// Check if the store's main tree contains a given transaction hash.
     pub fn contains(&self, tx_hash: &TransactionHash) -> Result<bool> {
         Ok(self.main.contains_key(tx_hash.inner())?)
@@ -310,52 +276,26 @@ impl TxStore {
         Ok(txs)
     }
 
-    /// Retrieve all transactions from the store's pending txs order tree in
-    /// the form of a tuple (`u64`, `TransactionHash`).
-    /// Be careful as this will try to load everything in memory.
-    pub fn get_all_pending_order(&self) -> Result<Vec<(u64, TransactionHash)>> {
+    /// Fetch n transactions after given transaction hash.
+    pub fn get_after_pending(
+        &self,
+        tx_hash: &TransactionHash,
+        n: usize,
+    ) -> Result<Vec<Transaction>> {
         let mut txs = vec![];
-
-        for tx in self.pending_order.iter() {
-            txs.push(parse_u64_key_record(tx.unwrap())?);
-        }
-
-        Ok(txs)
-    }
-
-    /// Fetch n transactions after given order([order..order+n)). 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![];
-
-        // First we grab the order itself
-        if let Some(found) = self.pending_order.get(order.to_be_bytes())? {
-            let hash = deserialize(&found)?;
-            hashes.push(hash);
-        }
-
-        // Then whatever comes after it
-        let mut key = order;
+        let mut key = tx_hash.inner().into();
         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);
+            if let Some((tx_hash, tx)) = self.pending.get_gt(key)? {
+                key = tx_hash;
+                txs.push(deserialize(&tx)?);
                 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))
+        Ok(txs)
     }
 
     /// Retrieve records count of the store's main tree.
@@ -375,13 +315,6 @@ impl TxStore {
         Ok(())
     }
 
-    /// Remove a slice of [`u64`] from the store's pending txs order tree.
-    pub fn remove_pending_order(&self, indexes: &[u64]) -> Result<()> {
-        let batch = self.remove_batch_pending_order(indexes);
-        self.pending_order.apply_batch(batch)?;
-        Ok(())
-    }
-
     /// Generate the sled batch corresponding to a remove from the store's pending
     /// txs tree, so caller can handle the write operation.
     pub fn remove_batch_pending(&self, txs_hashes: &[TransactionHash]) -> sled::Batch {

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, HashSet};
+use std::collections::{BTreeSet, HashMap};
 
 use darkfi_sdk::{crypto::MerkleTree, tx::TransactionHash};
 use darkfi_serial::{async_trait, deserialize, SerialDecodable, SerialEncodable};
@@ -621,35 +621,6 @@ impl Consensus {
 
         Ok(())
     }
-
-    /// Auxiliary function to purge all unproposed pending
-    /// transactions from the database.
-    pub async fn purge_unproposed_pending_txs(
-        &mut self,
-        mut proposed_txs: HashSet<TransactionHash>,
-    ) -> Result<()> {
-        // Iterate over all forks to find proposed txs
-        for fork in &self.forks {
-            // 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 self.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.

src/validator/mod.rs

@@ -237,78 +237,6 @@ impl Validator {
         Ok(())
     }
 
-    /// The node removes invalid transactions from the pending txs
-    /// store.
-    ///
-    /// Note: Always remember to purge new trees from the database if
-    /// not needed.
-    pub async fn purge_pending_txs(&mut self) -> Result<()> {
-        info!(target: "validator::purge_pending_txs", "Removing invalid transactions from pending transactions store...");
-
-        // Check if any pending transactions exist
-        let pending_txs = self.blockchain.get_pending_txs()?;
-        if pending_txs.is_empty() {
-            info!(target: "validator::purge_pending_txs", "No pending transactions found");
-            return Ok(())
-        }
-
-        let mut removed_txs = vec![];
-        for tx in pending_txs {
-            let tx_hash = tx.hash();
-            let tx_vec = [tx.clone()];
-            let mut valid = false;
-
-            // Iterate over node forks to verify transaction validity
-            // in their overlays.
-            for fork in self.consensus.forks.iter_mut() {
-                // Clone fork state
-                let fork_clone = fork.full_clone()?;
-
-                // Grab forks' next block height
-                let next_block_height = fork_clone.get_next_block_height()?;
-
-                // Verify transaction
-                let verify_result = verify_transactions(
-                    &fork_clone.overlay,
-                    next_block_height,
-                    self.consensus.module.target,
-                    &tx_vec,
-                    &mut MerkleTree::new(1),
-                    self.verify_fees,
-                )
-                .await;
-
-                // Handle response
-                match verify_result {
-                    Ok(_) => {
-                        valid = true;
-                        continue
-                    }
-                    Err(Error::TxVerifyFailed(TxVerifyFailed::ErroneousTxs(_))) => {}
-                    Err(e) => return Err(e),
-                }
-
-                // Remove erroneous transaction from forks' mempool
-                fork.mempool.retain(|x| *x != tx_hash);
-            }
-
-            // Remove pending transaction if it's not valid for
-            // canonical or any fork.
-            if !valid {
-                removed_txs.push(tx)
-            }
-        }
-
-        if removed_txs.is_empty() {
-            info!(target: "validator::purge_pending_txs", "No erroneous transactions found");
-            return Ok(())
-        }
-        info!(target: "validator::purge_pending_txs", "Removing {} erroneous transactions...", removed_txs.len());
-        self.blockchain.remove_pending_txs(&removed_txs)?;
-
-        Ok(())
-    }
-
     /// The node tries to append provided proposal to its consensus
     /// state.
     pub async fn append_proposal(&mut self, proposal: &Proposal) -> Result<()> {
@@ -593,10 +521,8 @@ impl Validator {
         // Store the block diffs
         self.blockchain.blocks.insert_state_inverse_diff(&diffs_heights, &inverse_diffs)?;
 
-        // Purge pending erroneous txs since canonical state has been
-        // changed.
+        // Remove blocks transactions from pending txs store
         self.blockchain.remove_pending_txs(&removed_txs)?;
-        self.purge_pending_txs().await?;
 
         // Update PoW module
         self.consensus.module = module;