consensus: integrate block, transaction Verifiers

zebra-consensus/src/block.rs

@@ -16,6 +16,7 @@ use std::{
 };
 
 use chrono::Utc;
+use futures::stream::FuturesUnordered;
 use futures_util::FutureExt;
 use thiserror::Error;
 use tower::{Service, ServiceExt};
@@ -23,12 +24,16 @@ use tower::{Service, ServiceExt};
 use zebra_chain::{
     block::{self, Block},
     parameters::Network,
+    parameters::NetworkUpgrade,
     work::equihash,
 };
 use zebra_state as zs;
 
-use crate::error::*;
-use crate::BoxError;
+use crate::{
+    error::*,
+    transaction::{self, VerifyTransactionError},
+};
+use crate::{script, BoxError};
 
 mod check;
 mod subsidy;
@@ -37,16 +42,15 @@ mod tests;
 
 /// A service that verifies blocks.
 #[derive(Debug)]
-pub struct BlockVerifier<S>
-where
-    S: Service<zs::Request, Response = zs::Response, Error = BoxError> + Send + Clone + 'static,
-    S::Future: Send + 'static,
-{
+pub struct BlockVerifier<S> {
     /// The network to be verified.
     network: Network,
 
     /// The underlying state service, possibly wrapped in other services.
     state_service: S,
+
+    /// The transaction verification service
+    transaction_verifier: transaction::Verifier<S>,
 }
 
 #[non_exhaustive]
@@ -68,6 +72,8 @@ pub enum VerifyBlockError {
     Time(zebra_chain::block::BlockTimeError),
     #[error("unable to commit block after semantic verification")]
     Commit(#[source] BoxError),
+    #[error("invalid transaction")]
+    Transaction(#[source] VerifyTransactionError),
 }
 
 impl<S> BlockVerifier<S>
@@ -76,9 +82,14 @@ where
     S::Future: Send + 'static,
 {
     pub fn new(network: Network, state_service: S) -> Self {
+        let branch = NetworkUpgrade::Sapling.branch_id().unwrap();
+        let script_verifier = script::Verifier::new(state_service.clone(), branch);
+        let transaction_verifier = transaction::Verifier::new(script_verifier);
+
         Self {
             network,
             state_service,
+            transaction_verifier,
         }
     }
 }
@@ -102,6 +113,7 @@ where
 
     fn call(&mut self, block: Arc<Block>) -> Self::Future {
         let mut state_service = self.state_service.clone();
+        let mut transaction_verifier = self.transaction_verifier.clone();
         let network = self.network;
 
         // TODO(jlusby): Error = Report, handle errors from state_service.
@@ -159,6 +171,23 @@ where
             metrics::gauge!("block.verified.block.height", height.0 as _);
             metrics::counter!("block.verified.block.count", 1);
 
+            let mut async_checks = FuturesUnordered::new();
+
+            for transaction in &block.transactions {
+                let req = transaction::Request::Block(transaction.clone());
+                let rsp = transaction_verifier
+                    .ready_and()
+                    .await
+                    .expect("transaction verifier is always ready")
+                    .call(req);
+                async_checks.push(rsp);
+            }
+
+            use futures::StreamExt;
+            while let Some(result) = async_checks.next().await {
+                result.map_err(VerifyBlockError::Transaction)?;
+            }
+
             // Finally, submit the block for contextual verification.
             match state_service
                 .ready_and()