consensus/validator.rs: verify_transactions refactor and erroneous txs handling

Tests pass, still need to be tested in live conditions of node running

src/consensus/task/proposal.rs

@@ -209,7 +209,7 @@ async fn propose_period(consensus_p2p: P2pPtr, state: ValidatorStatePtr) -> bool
     let (won, fork_index, coin_index) =
         state.write().await.consensus.is_slot_leader(sigma1, sigma2);
     let result = if won {
-        state.write().await.propose(processing_slot, fork_index, coin_index, sigma1, sigma2)
+        state.write().await.propose(processing_slot, fork_index, coin_index, sigma1, sigma2).await
     } else {
         Ok(None)
     };

src/consensus/validator.rs

@@ -43,7 +43,7 @@ use super::{
 };
 
 use crate::{
-    blockchain::{Blockchain, BlockchainOverlay},
+    blockchain::{Blockchain, BlockchainOverlay, BlockchainOverlayPtr},
     rpc::jsonrpc::JsonNotification,
     runtime::vm_runtime::Runtime,
     system::{Subscriber, SubscriberPtr},
@@ -224,10 +224,18 @@ impl ValidatorState {
         }
 
         info!(target: "consensus::validator", "append_tx(): Starting state transition validation");
-        if let Err(e) = self.verify_transactions(&[tx.clone()], false).await {
-            error!(target: "consensus::validator", "append_tx(): Failed to verify transaction: {}", e);
-            return false
-        };
+        match self.verify_transactions(&[tx.clone()], false).await {
+            Ok(erroneous_txs) => {
+                if !erroneous_txs.is_empty() {
+                    error!(target: "consensus::validator", "append_tx(): Erroneous transaction detected");
+                    return false
+                }
+            }
+            Err(e) => {
+                error!(target: "consensus::validator", "append_tx(): Failed to verify transaction: {}", e);
+                return false
+            }
+        }
 
         if let Err(e) = self.blockchain.pending_txs.insert(&[tx]) {
             error!(target: "consensus::validator", "append_tx(): Failed to insert transaction to pending txs store: {}", e);
@@ -241,6 +249,7 @@ impl ValidatorState {
     /// and appends successfull ones to the pending txs store.
     pub async fn append_pending_txs(&mut self, txs: &[Transaction]) {
         let mut filtered_txs = vec![];
+        // Filter already seen transactions
         for tx in txs {
             let tx_hash = blake3::hash(&serialize(tx));
             let tx_in_txstore = match self.blockchain.transactions.contains(&tx_hash) {
@@ -267,12 +276,18 @@ impl ValidatorState {
             filtered_txs.push(tx.clone());
         }
 
+        // Verify transactions and filter erroneous ones
         info!(target: "consensus::validator", "append_pending_txs(): Starting state transition validation");
-        // TODO: verify_transactions should return erroneous txs of the set to ignore them
-        if let Err(e) = self.verify_transactions(&filtered_txs[..], false).await {
-            error!(target: "consensus::validator", "append_pending_txs(): Failed to verify transaction: {}", e);
-            return
+        let erroneous_txs = match self.verify_transactions(&filtered_txs[..], false).await {
+            Ok(erroneous_txs) => erroneous_txs,
+            Err(e) => {
+                error!(target: "consensus::validator", "append_pending_txs(): Failed to verify transactions: {}", e);
+                return
+            }
         };
+        if !erroneous_txs.is_empty() {
+            filtered_txs.retain(|x| !erroneous_txs.contains(&x));
+        }
 
         if let Err(e) = self.blockchain.pending_txs.insert(&filtered_txs) {
             error!(target: "consensus::validator", "append_pending_txs(): Failed to insert transactions to pending txs store: {}", e);
@@ -281,10 +296,28 @@ impl ValidatorState {
         info!(target: "consensus::validator", "append_pending_txs(): Appended tx to pending txs store");
     }
 
+    /// 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()?;
+        if pending_txs.is_empty() {
+            info!(target: "consensus::validator", "purge_pending_txs(): No pending transactions found");
+            return Ok(())
+        }
+        let erroneous_txs = self.verify_transactions(&pending_txs[..], false).await?;
+        if erroneous_txs.is_empty() {
+            info!(target: "consensus::validator", "purge_pending_txs(): No erroneous transactions found");
+            return Ok(())
+        }
+        info!(target: "consensus::validator", "purge_pending_txs(): Removing {} erroneous transactions...", erroneous_txs.len());
+        self.blockchain.pending_txs.remove(&erroneous_txs)?;
+        Ok(())
+    }
+
     /// Generate a block proposal for the current slot, containing all
     /// pending transactions. Proposal extends the longest fork
     /// chain the node is holding.
-    pub fn propose(
+    pub async fn propose(
         &mut self,
         slot: u64,
         fork_index: i64,
@@ -299,7 +332,12 @@ impl ValidatorState {
         }
 
         // Generate proposal
-        let unproposed_txs = self.unproposed_txs(fork_index)?;
+        let mut unproposed_txs = self.unproposed_txs(fork_index)?;
+        // Verify transactions and filter erroneous ones
+        let erroneous_txs = self.verify_transactions(&unproposed_txs[..], false).await?;
+        if !erroneous_txs.is_empty() {
+            unproposed_txs.retain(|x| !erroneous_txs.contains(&x));
+        }
         let mut tree = BridgeTree::<MerkleNode, MERKLE_DEPTH>::new(100);
         // The following is pretty weird, so something better should be done.
         for tx in &unproposed_txs {
@@ -579,10 +617,18 @@ impl ValidatorState {
         // Validate state transition against canonical state
         // TODO: This should be validated against fork state
         info!(target: "consensus::validator", "receive_proposal(): Starting state transition validation");
-        if let Err(e) = self.verify_transactions(&proposal.block.txs, false).await {
-            error!(target: "consensus::validator", "receive_proposal(): Transaction verifications failed: {}", e);
-            return Err(e)
-        };
+        match self.verify_transactions(&proposal.block.txs, false).await {
+            Ok(erroneous_txs) => {
+                if !erroneous_txs.is_empty() {
+                    error!(target: "consensus::validator", "Proposal contains erroneous transactions");
+                    return Err(Error::ErroneousTxsDetected)
+                }
+            }
+            Err(e) => {
+                error!(target: "consensus::validator", "receive_proposal(): Transaction verifications failed: {}", e);
+                return Err(e)
+            }
+        }
 
         // TODO: [PLACEHOLDER] Add rewards validation
 
@@ -705,9 +751,17 @@ impl ValidatorState {
             // TODO: FIXME: The state transitions have already been written, they have to be in memory
             //              until this point.
             info!(target: "consensus::validator", "Applying state transition for finalized block");
-            if let Err(e) = self.verify_transactions(&proposal.txs, true).await {
-                error!(target: "consensus::validator", "Finalized block transaction verifications failed: {}", e);
-                return Err(e)
+            match self.verify_transactions(&proposal.txs, true).await {
+                Ok(erroneous_txs) => {
+                    if !erroneous_txs.is_empty() {
+                        error!(target: "consensus::validator", "Finalized block contains erroneous transactions");
+                        return Err(Error::ErroneousTxsDetected)
+                    }
+                }
+                Err(e) => {
+                    error!(target: "consensus::validator", "Finalized block transaction verifications failed: {}", e);
+                    return Err(e)
+                }
             }
 
             // Remove proposal transactions from pending txs store
@@ -756,6 +810,11 @@ impl ValidatorState {
         self.consensus.forks = vec![];
         self.consensus.slot_checkpoints = vec![];
 
+        // Purge pending erroneous txs since canonical state has been changed
+        if let Err(e) = self.purge_pending_txs().await {
+            error!(target: "consensus::validator", "consensus: Purging pending transactions failed: {}", e);
+        }
+
         Ok((finalized, finalized_slot_checkpoints))
     }
 
@@ -777,10 +836,18 @@ impl ValidatorState {
         info!(target: "consensus::validator", "receive_blocks(): Starting state transition validations");
 
         for block in blocks {
-            if let Err(e) = self.verify_transactions(&block.txs, true).await {
-                error!(target: "consensus::validator", "receive_blocks(): Transaction verifications failed: {}", e);
-                return Err(e)
-            };
+            match self.verify_transactions(&block.txs, true).await {
+                Ok(erroneous_txs) => {
+                    if !erroneous_txs.is_empty() {
+                        error!(target: "consensus::validator", "receive_blocks(): Block contains erroneous transactions");
+                        return Err(Error::ErroneousTxsDetected)
+                    }
+                }
+                Err(e) => {
+                    error!(target: "consensus::validator", "receive_blocks(): Transaction verifications failed: {}", e);
+                    return Err(e)
+                }
+            }
         }
 
         info!(target: "consensus::validator", "receive_blocks(): All state transitions passed. Appending blocks to ledger.");
@@ -822,6 +889,11 @@ impl ValidatorState {
         info!(target: "consensus::validator", "receive_finalized_block(): Removing block transactions from pending txs store");
         self.blockchain.pending_txs.remove(&block.txs)?;
 
+        // Purge pending erroneous txs since canonical state has been changed
+        if let Err(e) = self.purge_pending_txs().await {
+            error!(target: "consensus::validator", "receive_finalized_block(): Purging pending transactions failed: {}", e);
+        }
+
         Ok(true)
     }
 
@@ -869,214 +941,182 @@ impl ValidatorState {
         Ok(())
     }
 
-    /// Validate signatures, wasm execution, and zk proofs for given transactions.
-    /// If all of those succeed, try to execute a state update for the contract calls.
-    /// Currently the verifications are sequential, and the function will skip a
-    /// transaction if any of the verifications fail.
-    /// The function takes a boolean called `write` which tells it to actually write
-    /// the state transitions to the database.
-    // TODO: This should be paralellized as if even one tx in the batch fails to verify,
-    //       we can skip it. When things are parallel, make sure to write in a deterministic
-    //       order.
-    // TODO: This function should be refactored to be more readable and efficient.
-    //       1. Get metadata
-    //       2. Verify signatures
-    //       3. Verify execution
-    //       4. Verify ZK proofs
-    //       5. (optionally) write
-    pub async fn verify_transactions(&self, txs: &[Transaction], write: bool) -> Result<()> {
-        info!(target: "consensus::validator", "Verifying {} transaction(s)", txs.len());
-        let blockchain_overlay = BlockchainOverlay::new(&self.blockchain)?;
+    /// Validate signatures, wasm execution, and zk proofs for given transaction in
+    /// provided runtimes. If all of those succeed, try to execute a state update
+    /// for the contract calls.
+    async fn verify_transaction(
+        &self,
+        blockchain_overlay: BlockchainOverlayPtr,
+        tx: &Transaction,
+    ) -> Result<()> {
+        let mut runtimes = HashMap::new();
+        let tx_hash = blake3::hash(&serialize(tx));
+        info!(target: "consensus::validator", "Verifying transaction {}", tx_hash);
+
+        // Table of public inputs used for ZK proof verification
+        let mut zkp_table = vec![];
+        // Table of public keys used for signature verification
+        let mut sig_table = vec![];
+        // State updates produced by contract execution
+        let mut updates = vec![];
+        // Map of zk proof verifying keys for the current transaction
+        let mut verifying_keys: HashMap<[u8; 32], HashMap<String, VerifyingKey>> = HashMap::new();
+
+        // Initialize the map
+        for call in tx.calls.iter() {
+            verifying_keys.insert(call.contract_id.to_bytes(), HashMap::new());
+        }
 
-        for tx in txs {
-            let tx_hash = blake3::hash(&serialize(tx));
-            info!(target: "consensus::validator", "Verifying transaction {}", tx_hash);
-
-            // Table of public inputs used for ZK proof verification
-            let mut zkp_table = vec![];
-            // Table of public keys used for signature verification
-            let mut sig_table = vec![];
-            // State updates produced by contract execution
-            let mut updates = vec![];
-            // Map of zk proof verifying keys for the current transaction
-            //let mut verifying_keys: HashMap<[u8; 32], Vec<(String, VerifyingKey)>> = HashMap::new();
-            let mut verifying_keys: HashMap<[u8; 32], HashMap<String, VerifyingKey>> =
-                HashMap::new();
-
-            // Initialize the map
-            for call in tx.calls.iter() {
-                verifying_keys.insert(call.contract_id.to_bytes(), HashMap::new());
-            }
+        // Iterate over all calls to get the metadata
+        for (idx, call) in tx.calls.iter().enumerate() {
+            info!(target: "consensus::validator", "Executing contract call {}", idx);
+
+            // Write the actual payload data
+            let mut payload = vec![];
+            payload.write_u32(idx as u32)?; // Call index
+            tx.calls.encode(&mut payload)?; // Actual call data
 
-            // Iterate over all calls to get the metadata
-            for (idx, call) in tx.calls.iter().enumerate() {
-                info!(target: "consensus::validator", "Executing contract call {}", idx);
+            // Instantiate the wasm runtime
+            let runtime_key = call.contract_id.to_string();
+            if !runtimes.contains_key(&runtime_key) {
                 let wasm = self.blockchain.wasm_bincode.get(call.contract_id)?;
+                let r = Runtime::new(&wasm, blockchain_overlay.clone(), call.contract_id)?;
+                runtimes.insert(runtime_key.clone(), r);
+            }
+            let runtime = runtimes.get_mut(&runtime_key).unwrap();
 
-                // Write the actual payload data
-                let mut payload = vec![];
-                payload.write_u32(idx as u32)?; // Call index
-                tx.calls.encode(&mut payload)?; // Actual call data
+            info!(target: "consensus::validator", "Executing \"metadata\" call");
+            let metadata = runtime.metadata(&payload)?;
 
-                // Instantiate the wasm runtime
-                let mut runtime =
-                    Runtime::new(&wasm, blockchain_overlay.clone(), call.contract_id)?;
+            // Decode the metadata retrieved from the execution
+            let mut decoder = Cursor::new(&metadata);
 
-                info!(target: "consensus::validator", "Executing \"metadata\" call");
-                let metadata = runtime.metadata(&payload)?;
+            // The tuple is (zkas_ns, public_inputs)
+            let zkp_pub: Vec<(String, Vec<pallas::Base>)> = Decodable::decode(&mut decoder)?;
 
-                // Decode the metadata retrieved from the execution
-                let mut decoder = Cursor::new(&metadata);
+            let sig_pub: Vec<PublicKey> = Decodable::decode(&mut decoder)?;
+            // TODO: Make sure we've read all the bytes above.
+            info!(target: "consensus::validator", "Successfully executed \"metadata\" call");
 
-                // The tuple is (zkas_ns, public_inputs)
-                let zkp_pub: Vec<(String, Vec<pallas::Base>)> = Decodable::decode(&mut decoder)?;
+            // Here we'll look up verifying keys and insert them into the per-contract map.
+            info!(target: "consensus::validator", "Performing VerifyingKey lookups from the sled db");
+            for (zkas_ns, _) in &zkp_pub {
+                let inner_vk_map = verifying_keys.get_mut(&call.contract_id.to_bytes()).unwrap();
 
-                let sig_pub: Vec<PublicKey> = Decodable::decode(&mut decoder)?;
-                // TODO: Make sure we've read all the bytes above.
-                info!(target: "consensus::validator", "Successfully executed \"metadata\" call");
+                if inner_vk_map.contains_key(zkas_ns.as_str()) {
+                    continue
+                }
 
-                // Here we'll look up verifying keys and insert them into the per-contract map.
-                info!(target: "consensus::validator", "Performing VerifyingKey lookups from the sled db");
-                for (zkas_ns, _) in &zkp_pub {
-                    let inner_vk_map =
-                        verifying_keys.get_mut(&call.contract_id.to_bytes()).unwrap();
+                let (_, vk) = self.blockchain.contracts.get_zkas(
+                    &self.blockchain.sled_db,
+                    &call.contract_id,
+                    zkas_ns,
+                )?;
 
-                    if inner_vk_map.contains_key(zkas_ns.as_str()) {
-                        continue
-                    }
+                inner_vk_map.insert(zkas_ns.to_string(), vk);
+            }
 
-                    let (_, vk) = self.blockchain.contracts.get_zkas(
-                        &self.blockchain.sled_db,
-                        &call.contract_id,
-                        zkas_ns,
-                    )?;
+            zkp_table.push(zkp_pub);
+            sig_table.push(sig_pub);
 
-                    inner_vk_map.insert(zkas_ns.to_string(), vk);
-                }
+            // After getting the metadata, we run the "exec" function with the same
+            // runtime and the same payload.
+            info!(target: "consensus::validator", "Executing \"exec\" call");
+            let state_update = runtime.exec(&payload)?;
 
-                zkp_table.push(zkp_pub);
-                sig_table.push(sig_pub);
+            info!(target: "consensus::validator", "Successfully executed \"exec\" call");
+            updates.push(state_update);
 
-                // After getting the metadata, we run the "exec" function with the same
-                // runtime and the same payload.
-                info!(target: "consensus::validator", "Executing \"exec\" call");
-                let state_update = runtime.exec(&payload)?;
+            // At this point we're done with the call and move on to the next one.
+        }
 
-                info!(target: "consensus::validator", "Successfully executed \"exec\" call");
-                updates.push(state_update);
+        // When we're done looping and executing over the tx's contract calls, we
+        // move on with verification. First we verify the signatures as that's
+        // cheaper, and then finally we verify the ZK proofs.
+        info!(target: "consensus::validator", "Verifying signatures for transaction {}", tx_hash);
+        if sig_table.len() != tx.signatures.len() {
+            error!(target: "consensus::validator", "Incorrect number of signatures in tx {}", tx_hash);
+            return Err(Error::InvalidSignature)
+        }
 
-                // At this point we're done with the call and move on to the next one.
+        match tx.verify_sigs(sig_table) {
+            Ok(()) => {
+                info!(target: "consensus::validator", "Signatures verification for tx {} successful", tx_hash)
             }
-
-            // When we're done looping and executing over the tx's contract calls, we
-            // move on with verification. First we verify the signatures as that's
-            // cheaper, and then finally we verify the ZK proofs.
-            info!(target: "consensus::validator", "Verifying signatures for transaction {}", tx_hash);
-            if sig_table.len() != tx.signatures.len() {
-                error!(target: "consensus::validator", "Incorrect number of signatures in tx {}", tx_hash);
-                return Err(Error::InvalidSignature)
+            Err(e) => {
+                error!(target: "consensus::validator", "Signature verification for tx {} failed: {}", tx_hash, e);
+                return Err(e)
             }
+        };
 
-            match tx.verify_sigs(sig_table) {
-                Ok(()) => {
-                    info!(target: "consensus::validator", "Signatures verification for tx {} successful", tx_hash)
-                }
-                Err(e) => {
-                    error!(target: "consensus::validator", "Signature verification for tx {} failed: {}", tx_hash, e);
-                    return Err(e)
-                }
-            };
-
-            info!(target: "consensus::validator", "Verifying ZK proofs for transaction {}", tx_hash);
-            match tx.verify_zkps(verifying_keys.clone(), zkp_table).await {
-                Ok(()) => {
-                    info!(target: "consensus::validator", "ZK proof verification for tx {} successful", tx_hash)
-                }
-                Err(e) => {
-                    error!(target: "consensus::validator", "ZK proof verification for tx {} failed: {}", tx_hash, e);
-                    return Err(e)
-                }
-            };
-
-            // After the verifications stage passes, if we're told to write, we
-            // apply the state updates.
-            assert!(tx.calls.len() == updates.len());
-
-            if write {
-                info!(target: "consensus::validator", "Performing state updates");
-                for (call, update) in tx.calls.iter().zip(updates.iter()) {
-                    // For this we instantiate the runtimes again.
-                    // TODO: Optimize this
-                    // TODO: Sum up the gas costs of previous calls during execution
-                    //       and verification and these.
-                    let wasm = self.blockchain.wasm_bincode.get(call.contract_id)?;
-
-                    let mut runtime =
-                        Runtime::new(&wasm, blockchain_overlay.clone(), call.contract_id)?;
-
-                    info!(target: "consensus::validator", "Executing \"apply\" call");
-                    // TODO: FIXME: This should be done in an atomic tx/batch
-                    runtime.apply(update)?;
-                    info!(target: "consensus::validator", "State update applied successfully")
-                }
-            } else {
-                info!(target: "consensus::validator", "Skipping apply of state updates because write=false");
+        info!(target: "consensus::validator", "Verifying ZK proofs for transaction {}", tx_hash);
+        match tx.verify_zkps(verifying_keys.clone(), zkp_table).await {
+            Ok(()) => {
+                info!(target: "consensus::validator", "ZK proof verification for tx {} successful", tx_hash)
             }
+            Err(e) => {
+                error!(target: "consensus::validator", "ZK proof verification for tx {} failed: {}", tx_hash, e);
+                return Err(e)
+            }
+        };
 
-            info!(target: "consensus::validator", "Transaction {} verified successfully", tx_hash);
+        // After the verifications stage passes we can apply the state updates.
+        assert!(tx.calls.len() == updates.len());
+
+        info!(target: "consensus::validator", "Performing state updates");
+        for (call, update) in tx.calls.iter().zip(updates.iter()) {
+            // Retrieve already initiated runtime and apply update
+            // TODO: Sum up the gas costs of previous calls during execution
+            //       and verification and these.
+            let runtime = runtimes.get_mut(&call.contract_id.to_string()).unwrap();
+            info!(target: "consensus::validator", "Executing \"apply\" call");
+            runtime.apply(update)?;
+            info!(target: "consensus::validator", "State update applied successfully")
         }
 
-        if write {
-            // The beauty of using Arc<Mutex<Struct{Arc<Mutex<>>}>>
-            blockchain_overlay.lock().unwrap().overlay.lock().unwrap().apply()?;
-        }
+        info!(target: "consensus::validator", "Transaction {} verified successfully", tx_hash);
 
         Ok(())
     }
 
-    pub async fn verify_transactions2(&self, _txs: &[Transaction], _write: bool) -> Result<()> {
-        /*
-        let mut sled_overlay = SledDbOverlay::new();
-
-        for tx in txs {
-            for call in tx {
-                let wasm = sled_overlay.wasm_bincode.get(call.contract_id)?;
-
-                let mut runtime = Runtime::new(&wasm, &mut sled_overlay, call.contract_id)?;
-                let metadata = runtime.metadata(&payload)?;
-            }
+    /// Validate a set of [`Transaction`] in sequence and apply them if all are valid.
+    /// Erroneous transactions are filtered out of the set and returned to caller.
+    /// The function takes a boolean called `write` which tells it to actually write
+    /// the state transitions to the database.
+    pub async fn verify_transactions(
+        &self,
+        txs: &[Transaction],
+        write: bool,
+    ) -> Result<Vec<Transaction>> {
+        info!(target: "consensus::validator", "Verifying {} transaction(s)", txs.len());
 
-            // verify sigs
-            // verify zk proofs
-        }
+        let mut erroneous_txs = vec![];
+        let blockchain_overlay = BlockchainOverlay::new(&self.blockchain)?;
 
         for tx in txs {
-            let mut updates = vec![];
-
-            for call in tx {
-                let wasm = sled_overlay.wasm_bincode.get(call.contract_id)?;
-
-                let mut runtime = Runtime::new(&wasm, &mut sled_overlay, call.contract_id)?;
-                let state_update = runtime.exec(&payload)?;
-                updates.push((call.contract_id, state_update));
+            if let Err(e) = self.verify_transaction(blockchain_overlay.clone(), tx).await {
+                warn!(target: "consensus::validator", "Transaction verification failed: {}", e);
+                erroneous_txs.push(tx.clone());
             }
+        }
 
-            for (cid, update) in updates {
-                let wasm = sled_overlay.wasm_bincode.get(cid)?;
-
-                let mut runtime = Runtime::new(&wasm, &mut sled_overlay, cid)?;
-                runtime.apply(update)?;
-            }
+        let lock = blockchain_overlay.lock().unwrap();
+        let overlay = lock.overlay.lock().unwrap();
+        if !erroneous_txs.is_empty() {
+            warn!(target: "consensus::validator", "Erroneous transactions found in set");
+            overlay.purge_new_trees()?;
+            return Ok(erroneous_txs)
         }
 
-        if write && failed.is_empty() {
-            sled_overlay.apply();
-            db.flush_async().await;
+        if !write {
+            info!(target: "consensus::validator", "Skipping apply of state updates because write=false");
+            overlay.purge_new_trees()?;
+            return Ok(erroneous_txs)
         }
 
-        */
+        overlay.apply()?;
 
-        Ok(())
+        Ok(erroneous_txs)
     }
 
     /// Append to canonical state received finalized slot checkpoints from block sync task.

src/contract/money/tests/txs_verification.rs

@@ -116,7 +116,7 @@ async fn txs_verification() -> Result<()> {
 
     // Now Alice can send a little bit of funds to Bob.
     // We can duplicate this transaction to simulate double spending.
-    let duplicates = 1; // Change this number to 2 to double spend
+    let duplicates = 3; // Change this number to >1 to double spend
     let mut transactions = vec![];
     let mut txs_params = vec![];
     for i in 0..duplicates {
@@ -190,18 +190,26 @@ async fn txs_verification() -> Result<()> {
     assert_eq!(txs_params.len(), duplicates);
 
     // Now we can try to execute the transactions sequentialy.
-    // The first transaction will get applied, while the second one(duplicate) will fail.
+    // Each node will detect the duplicate txs and filter them out,
+    // then only apply the first txs from the set.
+    let valid_txs = vec![transactions[0].clone()];
     info!(target: "money", "[Faucet] ==============================");
     info!(target: "money", "[Faucet] Executing Alice2Bob payment tx");
     info!(target: "money", "[Faucet] ==============================");
-    th.faucet.state.read().await.verify_transactions(&transactions, true).await?;
+    let erroneous_txs =
+        th.faucet.state.read().await.verify_transactions(&transactions, true).await?;
+    assert_eq!(erroneous_txs.len(), duplicates - 1);
+    th.faucet.state.read().await.verify_transactions(&valid_txs, true).await?;
     th.faucet.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[0].coin.inner()));
     th.faucet.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[1].coin.inner()));
 
     info!(target: "money", "[Alice] ==============================");
     info!(target: "money", "[Alice] Executing Alice2Bob payment tx");
     info!(target: "money", "[Alice] ==============================");
-    th.alice.state.read().await.verify_transactions(&transactions, true).await?;
+    let erroneous_txs =
+        th.alice.state.read().await.verify_transactions(&transactions, true).await?;
+    assert_eq!(erroneous_txs.len(), duplicates - 1);
+    th.alice.state.read().await.verify_transactions(&valid_txs, true).await?;
     th.alice.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[0].coin.inner()));
     let alice_leaf_pos = th.alice.merkle_tree.witness().unwrap();
     th.alice.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[1].coin.inner()));
@@ -209,7 +217,9 @@ async fn txs_verification() -> Result<()> {
     info!(target: "money", "[Bob] ==============================");
     info!(target: "money", "[Bob] Executing Alice2Bob payment tx");
     info!(target: "money", "[Bob] ==============================");
-    th.bob.state.read().await.verify_transactions(&transactions, true).await?;
+    let erroneous_txs = th.bob.state.read().await.verify_transactions(&transactions, true).await?;
+    assert_eq!(erroneous_txs.len(), duplicates - 1);
+    th.bob.state.read().await.verify_transactions(&valid_txs, true).await?;
     th.bob.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[0].coin.inner()));
     th.bob.merkle_tree.append(&MerkleNode::from(txs_params[0].outputs[1].coin.inner()));
     let bob_leaf_pos = th.bob.merkle_tree.witness().unwrap();

src/error.rs

@@ -272,6 +272,9 @@ pub enum Error {
     #[error("Proposer is not eligible to produce proposals")]
     ProposalProposerNotEligible,
 
+    #[error("Erroneous transactions detected")]
+    ErroneousTxsDetected,
+
     // ===============
     // Database errors
     // ===============