Reject conflicting mempool transactions

Reject including a transaction in the mempool if it spends outputs
already spent by, or reveals nullifiers already revealed by another
transaction in the mempool.

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 conflicted with another transaction already in the mempool.
+    Conflict,
     /// Stayed in mempool for too long without being mined.
     // TODO(2021-08-20): set expiration at 2 weeks? This is what Bitcoin does.
     Expired,
@@ -57,6 +62,7 @@ impl Storage {
                 State::Confirmed(block_hash) => MempoolError::InBlock(*block_hash),
                 State::Excess => MempoolError::Excess,
                 State::LowFee => MempoolError::LowFee,
+                State::Conflict => MempoolError::Conflict,
             });
         }
 
@@ -68,6 +74,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 (#xxxx).
+        if self.check_spend_conflicts(&tx) {
+            self.rejected.insert(tx.id, State::Conflict);
+            return Err(MempoolError::Rejected);
+        }
+
         // Then, we insert into the pool.
         self.verified.push_front(tx);
 
@@ -139,4 +155,35 @@ impl Storage {
             .filter(|tx| self.rejected.contains_key(tx))
             .collect()
     }
+
+    /// Checks if the `tx` transaction conflicts with another transaction in the mempool.
+    ///
+    /// Two transactions conflict if they spent the same UTXO or if they reveal the same nullifier.
+    fn check_spend_conflicts(&self, tx: &UnminedTx) -> bool {
+        self.has_conflicts(tx, Transaction::inputs)
+            || self.has_conflicts(tx, Transaction::sprout_nullifiers)
+            || self.has_conflicts(tx, Transaction::sapling_nullifiers)
+            || self.has_conflicts(tx, Transaction::orchard_nullifiers)
+    }
+
+    /// Checks if the `tx` transaction has any conflicts with the transactions in the mempool for
+    /// the provided output type obtrained through the `extractor`.
+    fn has_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,
+    {
+        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))
+    }
 }