blockchain/tx_store: use a new tree to keep pending transactions order

src/blockchain/mod.rs

@@ -20,8 +20,11 @@ use std::sync::{Arc, Mutex};
 
 use log::debug;
 
+use darkfi_serial::serialize;
+
 use crate::{
     consensus::{Block, BlockInfo, SlotCheckpoint},
+    tx::Transaction,
     util::time::Timestamp,
     Result,
 };
@@ -33,7 +36,7 @@ pub mod slot_checkpoint_store;
 pub use slot_checkpoint_store::SlotCheckpointStore;
 
 pub mod tx_store;
-pub use tx_store::{PendingTxStore, TxStore};
+pub use tx_store::{PendingTxOrderStore, PendingTxStore, TxStore};
 
 pub mod contract_store;
 pub use contract_store::{
@@ -57,6 +60,8 @@ pub struct Blockchain {
     pub transactions: TxStore,
     /// Pending transactions sled tree
     pub pending_txs: PendingTxStore,
+    /// Pending transactions order sled tree
+    pub pending_txs_order: PendingTxOrderStore,
     /// Contract states
     pub contracts: ContractStateStore,
     /// Wasm bincodes
@@ -72,6 +77,7 @@ impl Blockchain {
         let slot_checkpoints = SlotCheckpointStore::new(db)?;
         let transactions = TxStore::new(db)?;
         let pending_txs = PendingTxStore::new(db)?;
+        let pending_txs_order = PendingTxOrderStore::new(db)?;
         let contracts = ContractStateStore::new(db)?;
         let wasm_bincode = WasmStore::new(db)?;
 
@@ -83,6 +89,7 @@ impl Blockchain {
             slot_checkpoints,
             transactions,
             pending_txs,
+            pending_txs_order,
             contracts,
             wasm_bincode,
         })
@@ -224,6 +231,55 @@ impl Blockchain {
         };
         Ok(!vec.is_empty())
     }
+
+    /// Insert a given slice of pending transactions into the blockchain database.
+    /// 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<blake3::Hash>> {
+        // TODO: Make db writes here completely atomic
+        let txs_hashes = self.pending_txs.insert(&txs)?;
+        self.pending_txs_order.insert(&txs_hashes)?;
+
+        Ok(txs_hashes)
+    }
+
+    /// 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.pending_txs.get_all()?;
+        let indexes = self.pending_txs_order.get_all()?;
+        assert_eq!(txs.len(), indexes.len());
+
+        let mut ret = Vec::with_capacity(txs.len());
+        for index in indexes {
+            ret.push(txs.get(&index.1).unwrap().clone());
+        }
+
+        Ok(ret)
+    }
+
+    /// Remove a given slice of pending transactions from the blockchain database.
+    pub fn remove_pending_txs(&self, txs: &[Transaction]) -> Result<()> {
+        let mut txs_hashes = Vec::with_capacity(txs.len());
+        for tx in txs {
+            let tx_hash = blake3::hash(&serialize(tx));
+            txs_hashes.push(tx_hash);
+        }
+
+        let indexes = self.pending_txs_order.get_all()?;
+        let mut removed_indexes = vec![];
+        for index in indexes {
+            if txs_hashes.contains(&index.1) {
+                removed_indexes.push(index.0);
+            }
+        }
+
+        // TODO: Make db writes here completely atomic
+        self.pending_txs.remove(&txs_hashes)?;
+        self.pending_txs_order.remove(&removed_indexes)?;
+
+        Ok(())
+    }
 }
 
 /// Atomic pointer to sled db overlay.

src/blockchain/tx_store.rs

@@ -16,12 +16,15 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use std::collections::HashMap;
+
 use darkfi_serial::{deserialize, serialize};
 
 use crate::{tx::Transaction, Error, Result};
 
 const SLED_TX_TREE: &[u8] = b"_transactions";
 const SLED_PENDING_TX_TREE: &[u8] = b"_pending_transactions";
+const SLED_PENDING_TX_ORDER_TREE: &[u8] = b"_pending_transactions_order";
 
 /// The `TxStore` is a `sled` tree storing all the blockchain's
 /// transactions where the key is the transaction hash, and the value is
@@ -146,40 +149,98 @@ impl PendingTxStore {
         Ok(self.0.contains_key(tx_hash.as_bytes())?)
     }
 
-    /// Retrieve all transactions from the pending tx store in the form of a tuple
-    /// (`tx_hash`, `tx`).
+    /// Retrieve all transactions from the pending tx store in the form of
+    /// a HashMap with key the transaction hash and value the transaction
+    /// itself.
     /// Be careful as this will try to load everything in memory.
-    pub fn get_all(&self) -> Result<Vec<(blake3::Hash, Transaction)>> {
-        let mut txs = vec![];
+    pub fn get_all(&self) -> Result<HashMap<blake3::Hash, Transaction>> {
+        let mut txs = HashMap::new();
 
         for tx in self.0.iter() {
             let (key, value) = tx.unwrap();
             let hash_bytes: [u8; 32] = key.as_ref().try_into().unwrap();
             let tx = deserialize(&value)?;
-            txs.push((hash_bytes.into(), tx));
+            txs.insert(hash_bytes.into(), tx);
         }
 
         Ok(txs)
     }
 
-    /// Retrieve all transactions from the pending tx store.
+    /// Remove a slice of [`blake3::Hash`] from the pending tx store.
+    /// With sled, the operation is done as a batch.
+    pub fn remove(&self, txs_hashes: &[blake3::Hash]) -> Result<()> {
+        let mut batch = sled::Batch::default();
+
+        for tx_hash in txs_hashes {
+            batch.remove(tx_hash.as_bytes());
+        }
+
+        self.0.apply_batch(batch)?;
+        Ok(())
+    }
+}
+
+/// The `PendingTxOrderStore` is a `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.
+#[derive(Clone)]
+pub struct PendingTxOrderStore(sled::Tree);
+
+impl PendingTxOrderStore {
+    /// Opens a new or existing `PendingTxOrderStore` on the given sled database.
+    pub fn new(db: &sled::Db) -> Result<Self> {
+        let tree = db.open_tree(SLED_PENDING_TX_ORDER_TREE)?;
+        Ok(Self(tree))
+    }
+
+    /// Insert a slice of [`blake3::Hash`] into the pending tx order store.
+    /// With sled, the operation is done as a batch.
+    pub fn insert(&self, txs_hashes: &[blake3::Hash]) -> Result<()> {
+        let mut batch = sled::Batch::default();
+
+        let mut next_index = match self.0.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 txs_hash in txs_hashes {
+            batch.insert(&next_index.to_be_bytes(), txs_hash.as_bytes());
+            next_index += 1;
+        }
+
+        self.0.apply_batch(batch)?;
+        Ok(())
+    }
+
+    /// Retrieve all transactions from the pending tx order store in the form
+    /// of a tuple (`u64`, `blake3::Hash`).
     /// Be careful as this will try to load everything in memory.
-    pub fn get_all_txs(&self) -> Result<Vec<Transaction>> {
-        let txs = self.get_all()?;
-        Ok(txs.iter().map(|x| x.1.clone()).rev().collect())
+    pub fn get_all(&self) -> Result<Vec<(u64, blake3::Hash)>> {
+        let mut txs = vec![];
+
+        for tx in self.0.iter() {
+            let (key, value) = tx.unwrap();
+            let index_bytes: [u8; 8] = key.as_ref().try_into().unwrap();
+            let hash_bytes: [u8; 32] = value.as_ref().try_into().unwrap();
+            let index = u64::from_be_bytes(index_bytes);
+            let hash = blake3::Hash::from(hash_bytes);
+            txs.push((index, hash));
+        }
+
+        Ok(txs)
     }
 
-    /// Remove a slice of [`Transaction`] from the pending tx store.
+    /// Remove a slice of [`u64`] from the pending tx order store.
     /// With sled, the operation is done as a batch.
-    /// The transactions are hashed with BLAKE3 and this hash is used as
-    /// the key to remove.
-    pub fn remove(&self, transactions: &[Transaction]) -> Result<()> {
+    pub fn remove(&self, indexes: &[u64]) -> Result<()> {
         let mut batch = sled::Batch::default();
 
-        for tx in transactions {
-            let serialized = serialize(tx);
-            let tx_hash = blake3::hash(&serialized);
-            batch.remove(tx_hash.as_bytes());
+        for index in indexes {
+            batch.remove(&index.to_be_bytes());
         }
 
         self.0.apply_batch(batch)?;

src/consensus/proto/protocol_sync_consensus.rs

@@ -99,7 +99,7 @@ impl ProtocolSyncConsensus {
             for fork in &lock.consensus.forks {
                 forks.push(fork.clone().into());
             }
-            let pending_txs = match lock.blockchain.pending_txs.get_all_txs() {
+            let pending_txs = match lock.blockchain.get_pending_txs() {
                 Ok(v) => v,
                 Err(e) => {
                     debug!(

src/consensus/validator.rs

@@ -239,7 +239,7 @@ impl ValidatorState {
             }
         }
 
-        if let Err(e) = self.blockchain.pending_txs.insert(&[tx]) {
+        if let Err(e) = self.blockchain.add_pending_txs(&[tx]) {
             error!(target: "consensus::validator", "append_tx(): Failed to insert transaction to pending txs store: {}", e);
             return false
         }
@@ -291,7 +291,7 @@ impl ValidatorState {
             filtered_txs.retain(|x| !erroneous_txs.contains(&x));
         }
 
-        if let Err(e) = self.blockchain.pending_txs.insert(&filtered_txs) {
+        if let Err(e) = self.blockchain.add_pending_txs(&filtered_txs) {
             error!(target: "consensus::validator", "append_pending_txs(): Failed to insert transactions to pending txs store: {}", e);
             return
         }
@@ -301,7 +301,7 @@ impl ValidatorState {
     /// The node removes erroneous transactions from the pending txs store.
     async fn purge_pending_txs(&self) -> Result<()> {
         info!(target: "consensus::validator", "purge_pending_txs(): Removing erroneous transactions from pending transactions store...");
-        let pending_txs = self.blockchain.pending_txs.get_all_txs()?;
+        let pending_txs = self.blockchain.get_pending_txs()?;
         if pending_txs.is_empty() {
             info!(target: "consensus::validator", "purge_pending_txs(): No pending transactions found");
             return Ok(())
@@ -312,7 +312,7 @@ impl ValidatorState {
             return Ok(())
         }
         info!(target: "consensus::validator", "purge_pending_txs(): Removing {} erroneous transactions...", erroneous_txs.len());
-        self.blockchain.pending_txs.remove(&erroneous_txs)?;
+        self.blockchain.remove_pending_txs(&erroneous_txs)?;
 
         // TODO: Don't hardcode this:
         let err_txs_subscriber = self.subscribers.get("err_txs").unwrap();
@@ -413,11 +413,11 @@ impl ValidatorState {
         let unproposed_txs = if index == -1 {
             // If index is -1 (canonical blockchain) a new fork will be generated,
             // therefore all unproposed transactions can be included in the proposal.
-            self.blockchain.pending_txs.get_all_txs()?
+            self.blockchain.get_pending_txs()?
         } else {
             // We iterate over the fork chain proposals to find already proposed
             // transactions and remove them from the local unproposed_txs vector.
-            let mut filtered_txs = self.blockchain.pending_txs.get_all_txs()?;
+            let mut filtered_txs = self.blockchain.get_pending_txs()?;
             let chain = &self.consensus.forks[index as usize];
             for state_checkpoint in &chain.sequence {
                 for tx in &state_checkpoint.proposal.block.txs {
@@ -778,7 +778,7 @@ impl ValidatorState {
             }
 
             // Remove proposal transactions from pending txs store
-            if let Err(e) = self.blockchain.pending_txs.remove(&proposal.txs) {
+            if let Err(e) = self.blockchain.remove_pending_txs(&proposal.txs) {
                 error!(target: "consensus::validator", "Removing finalized block transactions failed: {}", e);
                 return Err(e)
             }
@@ -900,7 +900,7 @@ impl ValidatorState {
         blocks_subscriber.notify(notif).await;
 
         info!(target: "consensus::validator", "receive_finalized_block(): Removing block transactions from pending txs store");
-        self.blockchain.pending_txs.remove(&block.txs)?;
+        self.blockchain.remove_pending_txs(&block.txs)?;
 
         // Purge pending erroneous txs since canonical state has been changed
         if let Err(e) = self.purge_pending_txs().await {