Reject conflicting mempool transactions

zebra-chain/src/sapling/shielded_data.rs

@@ -4,6 +4,8 @@
 //! The `value_balance` change is handled using the default zero value.
 //! The anchor change is handled using the `AnchorVariant` type trait.
 
+#[cfg(any(test, feature = "proptest-impl"))]
+use proptest_derive::Arbitrary;
 use serde::{de::DeserializeOwned, Serialize};
 
 use crate::{
@@ -35,6 +37,7 @@ pub struct SharedAnchor {}
 
 /// This field is not present in this transaction version.
 #[derive(Copy, Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[cfg_attr(any(test, feature = "proptest-impl"), derive(Arbitrary))]
 pub struct FieldNotPresent;
 
 impl AnchorVariant for PerSpendAnchor {

zebra-chain/src/serialization/constraint.rs

@@ -185,6 +185,18 @@ impl<T> AtLeastOne<T> {
         &self.inner[0]
     }
 
+    /// Returns a mutable reference to the first element.
+    ///
+    /// Unlike `Vec` or slice, `AtLeastOne` always has a first element.
+    pub fn first_mut(&mut self) -> &mut T {
+        &mut self.inner[0]
+    }
+
+    /// Appends an element to the back of the collection.
+    pub fn push(&mut self, element: T) {
+        self.inner.push(element);
+    }
+
     /// Returns the first and all the rest of the elements of the vector.
     ///
     /// Unlike `Vec` or slice, `AtLeastOne` always has a first element.

zebra-chain/src/transaction.rs

@@ -349,6 +349,13 @@ impl Transaction {
         }
     }
 
+    /// Access the [`transparent::OutPoint`]s spent by this transaction's [`transparent::Input`]s.
+    pub fn spent_outpoints(&self) -> impl Iterator<Item = transparent::OutPoint> + '_ {
+        self.inputs()
+            .iter()
+            .filter_map(transparent::Input::outpoint)
+    }
+
     /// Access the transparent outputs of this transaction, regardless of version.
     pub fn outputs(&self) -> &[transparent::Output] {
         match self {

zebrad/Cargo.toml

@@ -60,6 +60,7 @@ tokio = { version = "0.3.6", features = ["full", "test-util"] }
 proptest = "0.10"
 proptest-derive = "0.3"
 
+zebra-chain = { path = "../zebra-chain", features = ["proptest-impl"] }
 zebra-test = { path = "../zebra-test" }
 
 [features]

zebrad/src/components/mempool/error.rs

@@ -43,4 +43,11 @@ pub enum MempoolError {
     /// The queue's capacity is [`super::downloads::MAX_INBOUND_CONCURRENCY`].
     #[error("transaction dropped because the queue is full")]
     FullQueue,
+
+    /// The transaction has a spend conflict with another transaction already in the mempool.
+    #[error(
+        "transaction rejected because another transaction in the mempool has already spent some of \
+        its inputs"
+    )]
+    SpendConflict,
 }

zebrad/src/components/mempool/storage.rs

@@ -1,8 +1,11 @@
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::{
+    collections::{HashMap, HashSet, VecDeque},
+    hash::Hash,
+};
 
 use zebra_chain::{
     block,
-    transaction::{UnminedTx, UnminedTxId},
+    transaction::{Transaction, UnminedTx, UnminedTxId},
 };
 use zebra_consensus::error::TransactionError;
 
@@ -21,6 +24,8 @@ pub enum State {
     /// An otherwise valid mempool transaction was mined into a block, therefore
     /// no longer belongs in the mempool.
     Confirmed(block::Hash),
+    /// Rejected because it has a spend conflict with another transaction already in the mempool.
+    SpendConflict,
     /// Stayed in mempool for too long without being mined.
     // TODO(2021-09-09): Implement ZIP-203: Validate Transaction Expiry Height.
     // TODO(2021-09-09): https://github.com/ZcashFoundation/zebra/issues/2387
@@ -58,6 +63,7 @@ impl Storage {
                 State::Confirmed(block_hash) => MempoolError::InBlock(*block_hash),
                 State::Excess => MempoolError::Excess,
                 State::LowFee => MempoolError::LowFee,
+                State::SpendConflict => MempoolError::SpendConflict,
             });
         }
 
@@ -69,6 +75,16 @@ impl Storage {
             return Err(MempoolError::InMempool);
         }
 
+        // If `tx` spends an UTXO already spent by another transaction in the mempool or reveals a
+        // nullifier already revealed by another transaction in the mempool, reject that
+        // transaction.
+        //
+        // TODO: Consider replacing the transaction in the mempool if the fee is higher (#2781).
+        if self.check_spend_conflicts(&tx) {
+            self.rejected.insert(tx.id, State::SpendConflict);
+            return Err(MempoolError::Rejected);
+        }
+
         // Then, we insert into the pool.
         self.verified.push_front(tx);
 
@@ -146,4 +162,38 @@ impl Storage {
         self.verified.clear();
         self.rejected.clear();
     }
+
+    /// Checks if the `tx` transaction has spend conflicts with another transaction in the mempool.
+    ///
+    /// Two transactions have a spend conflict if they spent the same UTXO or if they reveal the
+    /// same nullifier.
+    fn check_spend_conflicts(&self, tx: &UnminedTx) -> bool {
+        self.has_spend_conflicts(tx, Transaction::spent_outpoints)
+            || self.has_spend_conflicts(tx, Transaction::sprout_nullifiers)
+            || self.has_spend_conflicts(tx, Transaction::sapling_nullifiers)
+            || self.has_spend_conflicts(tx, Transaction::orchard_nullifiers)
+    }
+
+    /// Checks if the `tx` transaction has any spend conflicts with the transactions in the mempool
+    /// for the provided output type obtained through the `extractor`.
+    fn has_spend_conflicts<'slf, 'tx, Extractor, Outputs>(
+        &'slf self,
+        tx: &'tx UnminedTx,
+        extractor: Extractor,
+    ) -> bool
+    where
+        'slf: 'tx,
+        Extractor: Fn(&'tx Transaction) -> Outputs,
+        Outputs: IntoIterator,
+        Outputs::Item: Eq + Hash + 'tx,
+    {
+        // TODO: This algorithm should be improved to avoid a performance impact when the mempool
+        // size is increased (#2784).
+        let new_outputs: HashSet<_> = extractor(&tx.transaction).into_iter().collect();
+
+        self.verified
+            .iter()
+            .flat_map(|tx| extractor(&tx.transaction))
+            .any(|output| new_outputs.contains(&output))
+    }
 }

zebrad/src/components/mempool/storage/tests.rs

@@ -1,103 +1,11 @@
 use std::ops::RangeBounds;
 
-use super::*;
-
 use zebra_chain::{
     block::Block, parameters::Network, serialization::ZcashDeserializeInto, transaction::UnminedTx,
 };
 
-use color_eyre::eyre::Result;
-
-#[test]
-fn mempool_storage_crud_mainnet() {
-    zebra_test::init();
-
-    let network = Network::Mainnet;
-
-    // Create an empty storage instance
-    let mut storage: Storage = Default::default();
-
-    // Get one (1) unmined transaction
-    let unmined_tx = unmined_transactions_in_blocks(.., network)
-        .next()
-        .expect("at least one unmined transaction");
-
-    // Insert unmined tx into the mempool.
-    let _ = storage.insert(unmined_tx.clone());
-
-    // Check that it is in the mempool, and not rejected.
-    assert!(storage.contains(&unmined_tx.id));
-
-    // Remove tx
-    let _ = storage.remove(&unmined_tx.id);
-
-    // Check that it is /not/ in the mempool.
-    assert!(!storage.contains(&unmined_tx.id));
-}
-
-#[test]
-fn mempool_storage_basic() -> Result<()> {
-    zebra_test::init();
-
-    mempool_storage_basic_for_network(Network::Mainnet)?;
-    mempool_storage_basic_for_network(Network::Testnet)?;
-
-    Ok(())
-}
-
-fn mempool_storage_basic_for_network(network: Network) -> Result<()> {
-    // Create an empty storage
-    let mut storage: Storage = Default::default();
-
-    // Get transactions from the first 10 blocks of the Zcash blockchain
-    let unmined_transactions: Vec<_> = unmined_transactions_in_blocks(..=10, network).collect();
-    let total_transactions = unmined_transactions.len();
-
-    // Insert them all to the storage
-    for unmined_transaction in unmined_transactions.clone() {
-        storage.insert(unmined_transaction)?;
-    }
-
-    // Separate transactions into the ones expected to be in the mempool and those expected to be
-    // rejected.
-    let rejected_transaction_count = total_transactions - MEMPOOL_SIZE;
-    let expected_to_be_rejected = &unmined_transactions[..rejected_transaction_count];
-    let expected_in_mempool = &unmined_transactions[rejected_transaction_count..];
-
-    // Only MEMPOOL_SIZE should land in verified
-    assert_eq!(storage.verified.len(), MEMPOOL_SIZE);
-
-    // The rest of the transactions will be in rejected
-    assert_eq!(storage.rejected.len(), rejected_transaction_count);
-
-    // Make sure the last MEMPOOL_SIZE transactions we sent are in the verified
-    for tx in expected_in_mempool {
-        assert!(storage.contains(&tx.id));
-    }
-
-    // Anything greater should not be in the verified
-    for tx in expected_to_be_rejected {
-        assert!(!storage.contains(&tx.id));
-    }
-
-    // Query all the ids we have for rejected, get back `total - MEMPOOL_SIZE`
-    let all_ids: HashSet<UnminedTxId> = unmined_transactions.iter().map(|tx| tx.id).collect();
-
-    // Convert response to a `HashSet` as we need a fixed order to compare.
-    let rejected_response: HashSet<UnminedTxId> =
-        storage.rejected_transactions(all_ids).into_iter().collect();
-
-    let rejected_ids = expected_to_be_rejected.iter().map(|tx| tx.id).collect();
-
-    assert_eq!(rejected_response, rejected_ids);
-
-    // Use `contains_rejected` to make sure the first id stored is now rejected
-    assert!(storage.contains_rejected(&expected_to_be_rejected[0].id));
-    // Use `contains_rejected` to make sure the last id stored is not rejected
-    assert!(!storage.contains_rejected(&expected_in_mempool[0].id));
-
-    Ok(())
-}
+mod prop;
+mod vectors;
 
 pub fn unmined_transactions_in_blocks(
     block_height_range: impl RangeBounds<u32>,