perf(consensus): batch block UTXOs under a shared read budget

crates/zakura-consensus/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zakura-consensus"
-version = "7.0.2-rc0"
+version = "8.0.0-rc0"
 authors.workspace = true
 description = "Implementation of Zcash consensus checks for the Zakura node. Internal crate, published to support cargo install zakura"
 license.workspace = true
@@ -67,11 +67,11 @@ zcash_primitives = { workspace = true }
 tower-fallback = { package = "zakura-tower-fallback", path = "../tower-fallback/", version = "1.2.0" }
 tower-batch-control = { package = "zakura-tower-batch-control", path = "../tower-batch-control/", version = "1.3.0" }
 
-zakura-script = { path = "../zakura-script", version = "3.2.2-rc0" }
-zakura-state = { path = "../zakura-state", version = "7.2.0-rc0" }
-zakura-node-services = { path = "../zakura-node-services", version = "3.2.3-rc0" }
+zakura-script = { path = "../zakura-script", version = "3.2.3-rc0" }
+zakura-state = { path = "../zakura-state", version = "8.0.0-rc0" }
+zakura-node-services = { path = "../zakura-node-services", version = "3.2.4-rc0" }
 zakura-chain = { path = "../zakura-chain", version = "7.0.0-rc0" }
-zakura-header-chain = { path = "../zakura-header-chain", version = "2.0.1-rc0" }
+zakura-header-chain = { path = "../zakura-header-chain", version = "2.0.2-rc0" }
 
 zcash_protocol.workspace = true
 
@@ -93,7 +93,7 @@ toml = { workspace = true }
 
 tokio = { workspace = true, features = ["full", "tracing", "test-util"] }
 
-zakura-state = { path = "../zakura-state", version = "7.2.0-rc0", features = ["proptest-impl"] }
+zakura-state = { path = "../zakura-state", version = "8.0.0-rc0", features = ["proptest-impl"] }
 zakura-chain = { path = "../zakura-chain", version = "7.0.0-rc0", features = ["proptest-impl"] }
 zakura-test = { path = "../zakura-test/", version = "2.1.0" }
 

crates/zakura-consensus/benches/worst_case_tx_verification.rs

@@ -466,6 +466,7 @@ fn build_workload(case: &BenchmarkCase, candidates: &[CandidateTx]) -> Option<Wo
             stats.verifier_checks += candidate.counts.verifier_check_counts();
 
             Request::Block {
+                utxo_resolver: None,
                 transaction_hash: candidate.transaction.hash(),
                 transaction: candidate.transaction.clone(),
                 known_outpoint_hashes: known_outpoint_hashes.clone(),

crates/zakura-consensus/src/block.rs

@@ -348,6 +348,22 @@ where
 
             let known_outpoint_hashes: Arc<HashSet<transaction::Hash>> =
                 Arc::new(known_utxos.keys().map(|outpoint| outpoint.hash).collect());
+            // Reject cross-transaction double spends before cloning historical outputs.
+            let mut spent_outpoints = HashSet::new();
+            for outpoint in block
+                .transactions
+                .iter()
+                .flat_map(|tx| tx.spent_outpoints())
+            {
+                if !spent_outpoints.insert(outpoint) {
+                    return Err(crate::error::TransactionError::DuplicateTransparentSpend(
+                        outpoint,
+                    )
+                    .into());
+                }
+            }
+            let utxo_resolver =
+                tx::BlockUtxos::for_block(&block, &known_utxos, state_service.clone());
             // Keep this guard after `known_outpoint_hashes` so its `Drop` removes the
             // pointer-keyed registration before the `Arc` address can be reused.
             let _block_batch_flush = primitives::register_block_verifier_batch_flush(
@@ -367,6 +383,7 @@ where
                         transaction: transaction.clone(),
                         known_outpoint_hashes: known_outpoint_hashes.clone(),
                         known_utxos: known_utxos.clone(),
+                        utxo_resolver: utxo_resolver.clone(),
                         height,
                         time: block.header.time,
                     });

crates/zakura-consensus/src/block/tests.rs

@@ -739,6 +739,122 @@ fn librustzcash_conversion_test_network(network_upgrade: NetworkUpgrade) -> Netw
         .expect("failed to build configured network")
 }
 
+#[tokio::test]
+async fn block_verification_uses_batched_external_outputs_without_committing_them() {
+    let _init_guard = zakura_test::init();
+    let network = librustzcash_conversion_test_network(NetworkUpgrade::Nu5);
+    let outpoint = transparent::OutPoint {
+        hash: [42; 32].into(),
+        index: 0,
+    };
+    for (script_succeeds, duplicate_spend) in [(true, false), (false, false), (true, true)] {
+        let mut block: Block = zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES
+            .zcash_deserialize_into()
+            .unwrap();
+        block.transactions = vec![
+            Arc::new(v5_coinbase_transaction(
+                NetworkUpgrade::Nu5,
+                Height(1),
+                &network,
+            )),
+            Arc::new(Transaction::V5 {
+                network_upgrade: NetworkUpgrade::Nu5,
+                inputs: vec![transparent::Input::PrevOut {
+                    outpoint,
+                    unlock_script: transparent::Script::new(&[]),
+                    sequence: u32::MAX,
+                }],
+                outputs: vec![transparent::Output {
+                    value: Amount::try_from(1).unwrap(),
+                    lock_script: transparent::Script::new(&[0x51]),
+                }],
+                lock_time: LockTime::unlocked(),
+                expiry_height: Height(2),
+                sapling_shielded_data: None,
+                orchard_shielded_data: None,
+            }),
+        ];
+        if duplicate_spend {
+            let mut duplicate = (*block.transactions[1]).clone();
+            let Transaction::V5 { expiry_height, .. } = &mut duplicate else {
+                unreachable!()
+            };
+            *expiry_height = Height(3);
+            block.transactions.push(Arc::new(duplicate));
+        }
+        Arc::make_mut(&mut block.header).merkle_root = block.transactions.iter().collect();
+        let expected_hash = block.hash();
+        let state =
+            tower::service_fn(move |request| async move {
+                Ok::<_, BoxError>(match request {
+                    zs::Request::KnownBlock(_) => zs::Response::KnownBlock(None),
+                    zs::Request::AwaitUtxos(outpoints) => {
+                        assert!(
+                            !duplicate_spend,
+                            "reject duplicate spends before reading outputs"
+                        );
+                        assert_eq!(outpoints, vec![outpoint]);
+                        zs::Response::Utxos(
+                            [(
+                                outpoint,
+                                transparent::Utxo::new(
+                                    transparent::Output {
+                                        value: Amount::try_from(1).unwrap(),
+                                        lock_script: transparent::Script::new(&[
+                                            if script_succeeds { 0x51 } else { 0x00 },
+                                        ]),
+                                    },
+                                    Height(0),
+                                    false,
+                                ),
+                            )]
+                            .into_iter()
+                            .collect(),
+                        )
+                    }
+                    zs::Request::CheckBlockProposalValidity(prepared) => {
+                        assert!(!prepared.new_outputs.contains_key(&outpoint));
+                        assert_eq!(
+                            prepared.new_outputs,
+                            transparent::new_ordered_outputs(
+                                &prepared.block,
+                                &prepared.transaction_hashes
+                            )
+                        );
+                        zs::Response::ValidBlockProposal
+                    }
+                    _ => panic!("semantic block verification must use the block resolver"),
+                })
+            });
+        let transaction = transaction::Verifier::new_for_tests(&network, state);
+        let transaction = Buffer::new(BoxService::new(transaction), 1);
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(30),
+            SemanticBlockVerifier::new(&network, state, transaction)
+                .oneshot(Request::CheckProposal(Arc::new(block))),
+        )
+        .await
+        .unwrap();
+        if duplicate_spend {
+            assert!(
+                matches!(result, Err(VerifyBlockError::Transaction(
+                TransactionError::DuplicateTransparentSpend(found))) if found == outpoint),
+                "{result:?}"
+            );
+        } else if script_succeeds {
+            assert_eq!(result.unwrap(), expected_hash);
+        } else {
+            assert!(
+                matches!(
+                    result,
+                    Err(VerifyBlockError::Transaction(TransactionError::Script(_)))
+                ),
+                "{result:?}"
+            );
+        }
+    }
+}
+
 fn block_with_librustzcash_conversion_failure(
     case: LibrustzcashConversionFailure,
     network: &Network,

crates/zakura-consensus/src/transaction.rs

@@ -44,6 +44,8 @@ use zakura_state as zs;
 use crate::{error::TransactionError, primitives, script, BoxError};
 
 pub mod check;
+mod utxo_resolver;
+pub use utxo_resolver::BlockUtxos;
 #[cfg(test)]
 mod tests;
 
@@ -176,6 +178,9 @@ pub enum Request {
         known_outpoint_hashes: Arc<HashSet<transaction::Hash>>,
         /// Additional UTXOs which are known at the time of verification.
         known_utxos: Arc<HashMap<transparent::OutPoint, transparent::OrderedUtxo>>,
+        /// Shared external-output resolver supplied by the semantic block verifier.
+        /// Standalone transaction verification can use `None` for legacy state lookups.
+        utxo_resolver: Option<BlockUtxos>,
         /// The height of the block containing this transaction.
         height: block::Height,
         /// The time that the block was mined.
@@ -792,6 +797,7 @@ where
         let mut spent_outputs: Vec<Option<transparent::Output>> = vec![None; inputs.len()];
         // Stores (input_idx, outpoint) for UTXOs not found in the best chain (fetched from mempool later).
         let mut spent_mempool_outpoints: Vec<(usize, transparent::OutPoint)> = Vec::new();
+        let mut resolved_utxos = None;
 
         for (input_idx, input) in inputs.iter().enumerate() {
             if let transparent::Input::PrevOut { outpoint, .. } = input {
@@ -817,6 +823,19 @@ where
                     };
 
                     utxo
+                } else if let Request::Block {
+                    utxo_resolver: Some(resolver),
+                    ..
+                } = &req
+                {
+                    if resolved_utxos.is_none() {
+                        resolved_utxos = Some(resolver.resolve().await?);
+                    }
+                    resolved_utxos
+                        .as_ref()
+                        .expect("the resolver completed above")
+                        .get(*outpoint, state.clone())
+                        .await?
                 } else {
                     let response = state
                         .clone()