validator: random tx handling fixes

bin/darkfid/src/main.rs

@@ -276,14 +276,9 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     };
 
     // Initialize node
-    let darkfid = Darkfid::new(
-        p2p.clone(),
-        validator.clone(),
-        blockchain_config.miner,
-        subscribers,
-        rpc_client,
-    )
-    .await;
+    let darkfid =
+        Darkfid::new(p2p.clone(), validator, blockchain_config.miner, subscribers, rpc_client)
+            .await;
     let darkfid = Arc::new(darkfid);
     info!(target: "darkfid", "Node initialized successfully!");
 

src/blockchain/mod.rs

@@ -292,6 +292,29 @@ impl Blockchain {
         Ok(())
     }
 
+    /// 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(())
+    }
+
     /// Auxiliary function to write to multiple trees completely atomic.
     fn atomic_write(&self, trees: &[sled::Tree], batches: &[sled::Batch]) -> Result<()> {
         if trees.len() != batches.len() {

src/validator/consensus.rs

@@ -475,10 +475,10 @@ impl Consensus {
         forks.retain(|_| *iter.next().unwrap());
 
         // Remove finalized proposals txs from the unporposed txs sled tree
-        self.blockchain.transactions.remove_pending(&finalized_txs_hashes)?;
+        self.blockchain.remove_pending_txs_hashes(&finalized_txs_hashes)?;
 
         // Remove unreferenced txs from the unporposed txs sled tree
-        self.blockchain.transactions.remove_pending(&Vec::from_iter(dropped_txs))?;
+        self.blockchain.remove_pending_txs_hashes(&Vec::from_iter(dropped_txs))?;
 
         // Drop forks lock
         drop(forks);
@@ -594,11 +594,10 @@ impl Fork {
 
     /// Auxiliary function to retrieve last proposal.
     pub fn last_proposal(&self) -> Result<Proposal> {
-        let block = if self.proposals.is_empty() {
-            self.overlay.lock().unwrap().last_block()?
+        let block = if let Some(last) = self.proposals.last() {
+            self.overlay.lock().unwrap().get_blocks_by_hash(&[*last])?[0].clone()
         } else {
-            self.overlay.lock().unwrap().get_blocks_by_hash(&[*self.proposals.last().unwrap()])?[0]
-                .clone()
+            self.overlay.lock().unwrap().last_block()?
         };
 
         Ok(Proposal::new(block))

src/validator/mod.rs

@@ -135,9 +135,7 @@ impl Validator {
     pub async fn calculate_gas(&self, tx: &Transaction, verify_fee: bool) -> Result<u64> {
         // Grab the best fork to verify against
         let forks = self.consensus.forks.read().await;
-        let fork = &forks[best_fork_index(&forks)?];
-        let overlay = fork.overlay.lock().unwrap().full_clone()?;
-        let next_block_height = fork.get_next_block_height()?;
+        let fork = forks[best_fork_index(&forks)?].full_clone()?;
         drop(forks);
 
         // Map of ZK proof verifying keys for the transaction
@@ -146,9 +144,12 @@ impl Validator {
             vks.insert(call.data.contract_id.to_bytes(), HashMap::new());
         }
 
+        // Grab forks' next block height
+        let next_block_height = fork.get_next_block_height()?;
+
         // Verify transaction to grab the gas used
         let verify_result = verify_transaction(
-            &overlay,
+            &fork.overlay,
             next_block_height,
             tx,
             &mut MerkleTree::new(1),
@@ -158,7 +159,7 @@ impl Validator {
         .await;
 
         // Purge new trees
-        overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
+        fork.overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
 
         Ok(verify_result?.0)
     }
@@ -185,24 +186,29 @@ impl Validator {
         // Grab a lock over current consensus forks state
         let mut forks = self.consensus.forks.write().await;
 
-        // Iterate over them to verify transaction validity in their overlays
+        // Iterate over node forks to verify transaction validity in their overlays
         for fork in forks.iter_mut() {
-            // Clone forks' overlay
-            let overlay = fork.overlay.lock().unwrap().full_clone()?;
+            // Clone fork state
+            let fork_clone = fork.full_clone()?;
 
             // Grab forks' next block height
-            let next_block_height = fork.get_next_block_height()?;
+            let next_block_height = fork_clone.get_next_block_height()?;
 
             // Verify transaction
-            match verify_transactions(
-                &overlay,
+            let verify_result = verify_transactions(
+                &fork_clone.overlay,
                 next_block_height,
                 &tx_vec,
                 &mut MerkleTree::new(1),
                 self.verify_fees,
             )
-            .await
-            {
+            .await;
+
+            // Purge new trees
+            fork_clone.overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
+
+            // Handle response
+            match verify_result {
                 Ok(_) => {}
                 Err(Error::TxVerifyFailed(TxVerifyFailed::ErroneousTxs(_))) => continue,
                 Err(e) => return Err(e),
@@ -253,25 +259,29 @@ impl Validator {
             let tx_vec = [tx.clone()];
             let mut valid = false;
 
-            // If node participates in consensus and holds any forks, iterate over them
-            // to verify transaction validity in their overlays
+            // Iterate over node forks to verify transaction validity in their overlays
             for fork in forks.iter_mut() {
-                // Clone forks' overlay
-                let overlay = fork.overlay.lock().unwrap().full_clone()?;
+                // Clone fork state
+                let fork_clone = fork.full_clone()?;
 
                 // Grab forks' next block height
-                let next_block_height = fork.get_next_block_height()?;
+                let next_block_height = fork_clone.get_next_block_height()?;
 
                 // Verify transaction
-                match verify_transactions(
-                    &overlay,
+                let verify_result = verify_transactions(
+                    &fork_clone.overlay,
                     next_block_height,
                     &tx_vec,
                     &mut MerkleTree::new(1),
                     self.verify_fees,
                 )
-                .await
-                {
+                .await;
+
+                // Purge new trees
+                fork_clone.overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
+
+                // Handle response
+                match verify_result {
                     Ok(_) => {
                         valid = true;
                         continue
@@ -284,23 +294,6 @@ impl Validator {
                 fork.mempool.retain(|x| *x != tx_hash);
             }
 
-            // Verify transaction against canonical state
-            let overlay = BlockchainOverlay::new(&self.blockchain)?;
-            let next_block_height = self.blockchain.last_block()?.header.height + 1;
-            match verify_transactions(
-                &overlay,
-                next_block_height,
-                &tx_vec,
-                &mut MerkleTree::new(1),
-                self.verify_fees,
-            )
-            .await
-            {
-                Ok(_) => valid = true,
-                Err(Error::TxVerifyFailed(TxVerifyFailed::ErroneousTxs(_))) => {}
-                Err(e) => return Err(e),
-            }
-
             // Remove pending transaction if it's not valid for canonical or any fork
             if !valid {
                 removed_txs.push(tx)