change(state): Check block and transaction Sprout anchors in parallel (#5742)

* parallelize anchors checks for blocks

* parallelize anchors checks for unmined_tx

* reverts par_iter in block_sapling_orchard_anchors_refer_to_final_treestates

* moves fetch_sprout_final_treestates out of rayon thread

zebra-state/src/service/check/anchors.rs

@@ -3,6 +3,8 @@
 
 use std::{collections::HashMap, sync::Arc};
 
+use rayon::prelude::*;
+
 use zebra_chain::{
     block::{Block, Height},
     sprout,
@@ -406,11 +408,7 @@ pub(crate) fn block_sprout_anchors_refer_to_treestates(
         "received sprout final treestate anchors",
     );
 
-    block
+    let check_tx_sprout_anchors = |(tx_index_in_block, transaction)| {
-        .transactions
-        .iter()
-        .enumerate()
-        .try_for_each(|(tx_index_in_block, transaction)| {
         sprout_anchors_refer_to_treestates(
             &sprout_final_treestates,
             transaction,
@@ -420,7 +418,25 @@ pub(crate) fn block_sprout_anchors_refer_to_treestates(
         )?;
 
         Ok(())
-        })
+    };
+
+    // The overhead for a parallel iterator is unwarranted if sprout_final_treestates is empty
+    // because it will either return an error for the first transaction or only check that `joinsplit_data`
+    // is `None` for each transaction.
+    if sprout_final_treestates.is_empty() {
+        // The block has no valid sprout anchors
+        block
+            .transactions
+            .iter()
+            .enumerate()
+            .try_for_each(check_tx_sprout_anchors)
+    } else {
+        block
+            .transactions
+            .par_iter()
+            .enumerate()
+            .try_for_each(check_tx_sprout_anchors)
+    }
 }
 
 /// Accepts a [`ZebraDb`], an optional [`Option<Arc<Chain>>`](Chain), and an [`UnminedTx`].
@@ -444,6 +460,8 @@ pub(crate) fn tx_anchors_refer_to_final_treestates(
         None,
     )?;
 
+    // If there are no sprout transactions in the block, avoid running a rayon scope
+    if unmined_tx.transaction.has_sprout_joinsplit_data() {
         let mut sprout_final_treestates = HashMap::new();
 
         fetch_sprout_final_treestates(
@@ -455,19 +473,31 @@ pub(crate) fn tx_anchors_refer_to_final_treestates(
             None,
         );
 
+        let mut sprout_anchors_result = None;
+        rayon::in_place_scope_fifo(|s| {
+            // This check is expensive, because it updates a note commitment tree for each sprout JoinSplit.
+            // Since we could be processing attacker-controlled mempool transactions, we need to run each one
+            // in its own thread, separately from tokio's blocking I/O threads. And if we are under heavy load,
+            // we want verification to finish in order, so that later transactions can't delay earlier ones.
+            s.spawn_fifo(|_s| {
                 tracing::trace!(
                     sprout_final_treestate_count = ?sprout_final_treestates.len(),
                     ?sprout_final_treestates,
                     "received sprout final treestate anchors",
                 );
 
-    sprout_anchors_refer_to_treestates(
+                sprout_anchors_result = Some(sprout_anchors_refer_to_treestates(
                     &sprout_final_treestates,
                     &unmined_tx.transaction,
                     unmined_tx.id.mined_id(),
                     None,
                     None,
-    )?;
+                ));
+            });
+        });
+
+        sprout_anchors_result.expect("scope has finished")?;
+    }
 
     Ok(())
 }