fix(consensus): Avoid a concurrency bug when verifying transactions in blocks that are already present in the mempool

zebra-consensus/src/transaction.rs

@@ -402,7 +402,7 @@ where
         async move {
             tracing::trace!(?tx_id, ?req, "got tx verify request");
 
-            if let Some(result) = Self::try_find_verified_unmined_tx(&req, mempool.clone()).await {
+            if let Some(result) = Self::find_verified_unmined_tx(&req, mempool.clone(), state.clone()).await {
                 let verified_tx = result?;
 
                 return Ok(Response::Block {
@@ -645,17 +645,22 @@ where
     }
 
     /// Attempts to find a transaction in the mempool by its transaction hash and checks
-    /// that all of its dependencies are available in the block.
+    /// that all of its dependencies are available in the block or in the state.  Waits
+    /// for UTXOs being spent by the given transaction to arrive in the state if they're
+    /// not found elsewhere.
     ///
     /// Returns [`Some(Ok(VerifiedUnminedTx))`](VerifiedUnminedTx) if successful,
     /// None if the transaction id was not found in the mempool,
     /// or `Some(Err(TransparentInputNotFound))` if the transaction was found, but some of its
-    /// dependencies are missing in the block.
-    async fn try_find_verified_unmined_tx(
+    /// dependencies were not found in the block or state after a timeout.
+    async fn find_verified_unmined_tx(
         req: &Request,
         mempool: Option<Timeout<Mempool>>,
+        state: Timeout<ZS>,
     ) -> Option<Result<VerifiedUnminedTx, TransactionError>> {
-        if req.is_mempool() || req.transaction().is_coinbase() {
+        let tx = req.transaction();
+
+        if req.is_mempool() || tx.is_coinbase() {
             return None;
         }
 
@@ -664,7 +669,7 @@ where
         let tx_id = req.tx_mined_id();
 
         let mempool::Response::TransactionWithDeps {
-            transaction,
+            transaction: verified_tx,
             dependencies,
         } = mempool
             .oneshot(mempool::Request::TransactionWithDepsByMinedId(tx_id))
@@ -676,17 +681,35 @@ where
 
         // Note: This does not verify that the spends are in order, the spend order
         //       should be verified during contextual validation in zebra-state.
-        let has_all_tx_deps = dependencies
+        let missing_deps: HashSet<_> = dependencies
             .into_iter()
-            .all(|dependency_id| known_outpoint_hashes.contains(&dependency_id));
+            .filter(|dependency_id| !known_outpoint_hashes.contains(dependency_id))
+            .collect();
 
-        let result = if has_all_tx_deps {
-            Ok(transaction)
-        } else {
-            Err(TransactionError::TransparentInputNotFound)
-        };
+        if missing_deps.is_empty() {
+            return Some(Ok(verified_tx));
+        }
+
+        let missing_outpoints = tx.inputs().iter().filter_map(|input| {
+            if let transparent::Input::PrevOut { outpoint, .. } = input {
+                missing_deps.contains(&outpoint.hash).then_some(outpoint)
+            } else {
+                None
+            }
+        });
+
+        for missing_outpoint in missing_outpoints {
+            let query = state
+                .clone()
+                .oneshot(zebra_state::Request::AwaitUtxo(*missing_outpoint));
+            match query.await {
+                Ok(zebra_state::Response::Utxo(_)) => {}
+                Err(_) => return Some(Err(TransactionError::TransparentInputNotFound)),
+                _ => unreachable!("AwaitUtxo always responds with Utxo"),
+            };
+        }
 
-        Some(result)
+        Some(Ok(verified_tx))
     }
 
     /// Wait for the UTXOs that are being spent by the given transaction.
@@ -732,7 +755,10 @@ where
                         .clone()
                         .oneshot(zs::Request::UnspentBestChainUtxo(*outpoint));
 
-                    let zebra_state::Response::UnspentBestChainUtxo(utxo) = query.await? else {
+                    let zebra_state::Response::UnspentBestChainUtxo(utxo) = query
+                        .await
+                        .map_err(|_| TransactionError::TransparentInputNotFound)?
+                    else {
                         unreachable!("UnspentBestChainUtxo always responds with Option<Utxo>")
                     };
 

zebra-consensus/src/transaction/tests.rs

@@ -753,22 +753,23 @@ async fn skips_verification_of_block_transactions_in_mempool() {
         transparent::Input::Coinbase { .. } => panic!("requires a non-coinbase transaction"),
     };
 
+    let mut state_clone = state.clone();
     tokio::spawn(async move {
-        state
+        state_clone
             .expect_request(zebra_state::Request::BestChainNextMedianTimePast)
             .await
             .expect("verifier should call mock state service with correct request")
             .respond(zebra_state::Response::BestChainNextMedianTimePast(
                 DateTime32::MAX,
             ));
 
-        state
+        state_clone
             .expect_request(zebra_state::Request::UnspentBestChainUtxo(input_outpoint))
             .await
             .expect("verifier should call mock state service with correct request")
             .respond(zebra_state::Response::UnspentBestChainUtxo(None));
 
-        state
+        state_clone
             .expect_request_that(|req| {
                 matches!(
                     req,
@@ -780,20 +781,20 @@ async fn skips_verification_of_block_transactions_in_mempool() {
             .respond(zebra_state::Response::ValidBestChainTipNullifiersAndAnchors);
     });
 
+    let utxo = known_utxos
+        .get(&input_outpoint)
+        .expect("input outpoint should exist in known_utxos")
+        .utxo
+        .clone();
+
     let mut mempool_clone = mempool.clone();
+    let output = utxo.output.clone();
     tokio::spawn(async move {
         mempool_clone
             .expect_request(mempool::Request::AwaitOutput(input_outpoint))
             .await
             .expect("verifier should call mock state service with correct request")
-            .respond(mempool::Response::UnspentOutput(
-                known_utxos
-                    .get(&input_outpoint)
-                    .expect("input outpoint should exist in known_utxos")
-                    .utxo
-                    .output
-                    .clone(),
-            ));
+            .respond(mempool::Response::UnspentOutput(output));
     });
 
     let verifier_response = verifier
@@ -825,11 +826,11 @@ async fn skips_verification_of_block_transactions_in_mempool() {
 
     let mut mempool_clone = mempool.clone();
     tokio::spawn(async move {
-        for _ in 0..2 {
+        for _ in 0..3 {
             mempool_clone
                 .expect_request(mempool::Request::TransactionWithDepsByMinedId(tx_hash))
                 .await
-                .expect("verifier should call mock state service with correct request")
+                .expect("verifier should call mock mempool service with correct request")
                 .respond(mempool::Response::TransactionWithDeps {
                     transaction: transaction.clone(),
                     dependencies: [input_outpoint.hash].into(),
@@ -855,6 +856,23 @@ async fn skips_verification_of_block_transactions_in_mempool() {
         panic!("unexpected response variant from transaction verifier for Block request")
     };
 
+    tokio::spawn(async move {
+        state
+            .expect_request(zebra_state::Request::AwaitUtxo(input_outpoint))
+            .await
+            .expect("verifier should call mock state service with correct request")
+            .respond(zebra_state::Response::Utxo(utxo));
+    });
+
+    let crate::transaction::Response::Block { .. } = verifier
+        .clone()
+        .oneshot(make_request.clone()(Arc::new(HashSet::new())))
+        .await
+        .expect("should succeed after calling state service")
+    else {
+        panic!("unexpected response variant from transaction verifier for Block request")
+    };
+
     let verifier_response_err = *verifier
         .clone()
         .oneshot(make_request(Arc::new(HashSet::new())))
@@ -875,7 +893,7 @@ async fn skips_verification_of_block_transactions_in_mempool() {
     // already the mempool.
     assert_eq!(
         mempool.poll_count(),
-        4,
+        5,
         "the mempool service should have been polled 4 times"
     );
 }

zebra-node-services/src/mempool/transaction_dependencies.rs

@@ -11,6 +11,13 @@ pub struct TransactionDependencies {
     /// a mempool transaction. Used during block template construction
     /// to exclude transactions from block templates unless all of the
     /// transactions they depend on have been included.
+    ///
+    /// # Note
+    ///
+    /// Dependencies that have been mined into blocks are not removed here until those blocks have
+    /// been committed to the best chain. Dependencies that have been committed onto side chains, or
+    /// which are in the verification pipeline but have not yet been committed to the best chain,
+    /// are not removed here unless and until they arrive in the best chain, and the mempool is polled.
     dependencies: HashMap<transaction::Hash, HashSet<transaction::Hash>>,
 
     /// Lists of transaction ids in the mempool that spend UTXOs created

zebra-state/src/request.rs

@@ -733,7 +733,8 @@ pub enum Request {
     ///
     /// This request is purely informational, and there are no guarantees about
     /// whether the UTXO remains unspent or is on the best chain, or any chain.
-    /// Its purpose is to allow asynchronous script verification.
+    /// Its purpose is to allow asynchronous script verification or to wait until
+    /// the UTXO arrives in the state before validating dependant transactions.
     ///
     /// # Correctness
     ///