Bound parallel verification of transactions

ledger/src/check_transaction_basic.rs

@@ -24,4 +24,13 @@ impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
     ) -> Result<()> {
         self.vm().check_transaction(transaction, rejected_id, rng)
     }
+
+    /// Checks that the given list of transactions are well-formed and unique.
+    pub fn check_transactions_basic<R: CryptoRng + Rng>(
+        &self,
+        transactions: &[(&Transaction<N>, Option<Field<N>>)],
+        rng: &mut R,
+    ) -> Result<()> {
+        self.vm().check_transactions(transactions, rng)
+    }
 }

synthesizer/Cargo.toml

@@ -120,6 +120,11 @@ path = "./snark"
 version = "=0.16.19"
 optional = true
 
+[dependencies.utilities]
+package = "snarkvm-utilities"
+path = "../utilities"
+version = "=0.16.19"
+
 [dependencies.aleo-std]
 version = "0.1.24"
 default-features = false

synthesizer/src/vm/finalize.rs

@@ -16,8 +16,6 @@ use super::*;
 
 use ledger_committee::{MAX_DELEGATORS, MIN_DELEGATOR_STAKE, MIN_VALIDATOR_STAKE};
 
-use rand::{rngs::StdRng, SeedableRng};
-
 impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
     /// Speculates on the given list of transactions in the VM.
     /// This function aborts all transactions that are not are well-formed or unique.
@@ -47,32 +45,16 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
         let candidate_transactions: Vec<_> = candidate_transactions.collect::<Vec<_>>();
         let candidate_transaction_ids: Vec<_> = candidate_transactions.iter().map(|tx| tx.id()).collect();
 
+        // Determine if the vm is currently processing the genesis block.
+        let is_genesis =
+            self.block_store().find_block_height_from_state_root(self.block_store().current_state_root())?.is_none();
         // If the transactions are not part of the genesis block, ensure each transaction is well-formed and unique. Abort any transactions that are not.
-        let (verified_transactions, verification_aborted_transactions) =
-            match self.block_store().find_block_height_from_state_root(self.block_store().current_state_root())? {
-                // If the current state root does not exist in the block store, then the genesis block has not been introduced yet.
-                None => (candidate_transactions, vec![]),
-                // Verify transactions for all non-genesis cases.
-                _ => {
-                    let rngs =
-                        (0..candidate_transactions.len()).map(|_| StdRng::from_seed(rng.gen())).collect::<Vec<_>>();
-                    // Verify the transactions and collect the error message if there is one.
-                    cfg_into_iter!(candidate_transactions).zip(rngs).partition_map(|(transaction, mut rng)| {
-                        // Abort the transaction if it is a fee transaction.
-                        if transaction.is_fee() {
-                            return Either::Right((
-                                transaction,
-                                "Fee transactions are not allowed in speculate".to_string(),
-                            ));
-                        }
-                        // Verify the transaction.
-                        match self.check_transaction(transaction, None, &mut rng) {
-                            Ok(_) => Either::Left(transaction),
-                            Err(e) => Either::Right((transaction, e.to_string())),
-                        }
-                    })
-                }
-            };
+        let (verified_transactions, verification_aborted_transactions) = match is_genesis {
+            // If the current state root does not exist in the block store, then the genesis block has not been introduced yet.
+            true => (candidate_transactions, vec![]),
+            // Verify transactions for all non-genesis cases.
+            false => self.prepare_transactions_for_speculate(&candidate_transactions, rng)?,
+        };
 
         // Performs a **dry-run** over the list of ratifications, solutions, and transactions.
         let (ratifications, confirmed_transactions, speculation_aborted_transactions, ratified_finalize_operations) =
@@ -127,15 +109,16 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
     ) -> Result<Vec<FinalizeOperation<N>>> {
         let timer = timer!("VM::check_speculate");
 
+        // Retrieve the transactions and their rejected IDs.
+        let transactions_and_rejected_ids = transactions
+            .iter()
+            .map(|transaction| transaction.to_rejected_id().map(|rejected_id| (transaction.deref(), rejected_id)))
+            .collect::<Result<Vec<_>>>()?;
         // Ensure each transaction is well-formed and unique.
         // NOTE: We perform the transaction checks here prior to `atomic_speculate` because we must
         // ensure that the `Fee` transactions are valid. We can't unify the transaction checks in `atomic_speculate`
         // because we run speculation on the unconfirmed variant of the transactions.
-        let rngs = (0..transactions.len()).map(|_| StdRng::from_seed(rng.gen())).collect::<Vec<_>>();
-        cfg_iter!(transactions).zip(rngs).try_for_each(|(transaction, mut rng)| {
-            self.check_transaction(transaction, transaction.to_rejected_id()?, &mut rng)
-                .map_err(|e| anyhow!("Invalid transaction found in the transactions list: {e}"))
-        })?;
+        self.check_transactions(&transactions_and_rejected_ids, rng)?;
 
         // Reconstruct the candidate ratifications to verify the speculation.
         let candidate_ratifications = ratifications.iter().cloned().collect::<Vec<_>>();

synthesizer/src/vm/mod.rs

@@ -57,12 +57,14 @@ use ledger_store::{
 };
 use synthesizer_process::{deployment_cost, execution_cost, Authorization, Process, Trace};
 use synthesizer_program::{FinalizeGlobalState, FinalizeOperation, FinalizeStoreTrait, Program};
+use utilities::handle_halting;
 
 use aleo_std::prelude::{finish, lap, timer};
 use indexmap::{IndexMap, IndexSet};
 use itertools::Either;
 use lru::LruCache;
 use parking_lot::{Mutex, RwLock};
+use rand::{rngs::StdRng, SeedableRng};
 use std::{num::NonZeroUsize, sync::Arc};
 
 #[cfg(not(feature = "serial"))]
@@ -701,7 +703,6 @@ function compute:
         // Construct the new block header.
         let (ratifications, transactions, aborted_transaction_ids, ratified_finalize_operations) =
             vm.speculate(sample_finalize_state(1), None, vec![], &None.into(), transactions.iter(), rng)?;
-        assert!(aborted_transaction_ids.is_empty());
 
         // Construct the metadata associated with the block.
         let metadata = Metadata::new(
@@ -1268,12 +1269,12 @@ function do:
     }
 
     #[test]
-    #[should_panic]
     fn test_deployment_synthesis_underreport() {
         let rng = &mut TestRng::default();
 
         // Initialize a private key.
         let private_key = sample_genesis_private_key(rng);
+        let address = Address::try_from(&private_key).unwrap();
 
         // Initialize the genesis block.
         let genesis = sample_genesis_block(rng);
@@ -1318,7 +1319,33 @@ function do:
         let adjusted_transaction = Transaction::Deploy(txid, program_owner, Box::new(adjusted_deployment), fee);
 
         // Verify the deployment transaction. It should panic when enforcing the first constraint over the vk limit.
-        let _ = vm.check_transaction(&adjusted_transaction, None, rng);
+        let result = vm.check_transaction(&adjusted_transaction, None, rng);
+        assert!(result.is_err());
+        // Check that the error message mentions the panic.
+        if let Err(err) = result {
+            assert!(err.to_string().contains("panic"));
+        }
+
+        // Create a standard transaction
+        // Prepare the inputs.
+        let inputs = [
+            Value::<CurrentNetwork>::from_str(&address.to_string()).unwrap(),
+            Value::<CurrentNetwork>::from_str("1u64").unwrap(),
+        ]
+        .into_iter();
+
+        // Execute.
+        let transaction =
+            vm.execute(&private_key, ("credits.aleo", "transfer_public"), inputs, None, 0, None, rng).unwrap();
+
+        // Check that the deployment transaction will be aborted if injected into a block.
+        let block = sample_next_block(&vm, &private_key, &[transaction, adjusted_transaction.clone()], rng).unwrap();
+
+        // Check that the block aborts the deployment transaction.
+        assert_eq!(block.aborted_transaction_ids(), &vec![adjusted_transaction.id()]);
+
+        // Update the VM.
+        vm.add_next_block(&block).unwrap();
     }
 
     #[test]

synthesizer/src/vm/verify.rs

@@ -31,6 +31,87 @@ macro_rules! ensure_is_unique {
     };
 }
 
+impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
+    /// The maximum number of deployments to verify in parallel.
+    const MAX_PARALLEL_DEPLOY_VERIFICATIONS: usize = 5;
+    /// The maximum number of executions to verify in parallel.
+    const MAX_PARALLEL_EXECUTE_VERIFICATIONS: usize = 1000;
+
+    /// Verifies the list of transactions in the VM. On failure, returns an error.
+    pub fn check_transactions<R: CryptoRng + Rng>(
+        &self,
+        transactions: &[(&Transaction<N>, Option<Field<N>>)],
+        rng: &mut R,
+    ) -> Result<()> {
+        // Separate the transactions into deploys and executions.
+        let (deployments, executions): (Vec<_>, Vec<_>) = transactions.iter().partition(|(tx, _)| tx.is_deploy());
+        // Chunk the deploys and executions into groups for parallel verification.
+        let deployments_for_verification = deployments.chunks(Self::MAX_PARALLEL_DEPLOY_VERIFICATIONS);
+        let executions_for_verification = executions.chunks(Self::MAX_PARALLEL_EXECUTE_VERIFICATIONS);
+
+        // Verify the transactions in batches.
+        for transactions in deployments_for_verification.chain(executions_for_verification) {
+            // Ensure each transaction is well-formed and unique.
+            let rngs = (0..transactions.len()).map(|_| StdRng::from_seed(rng.gen())).collect::<Vec<_>>();
+            cfg_iter!(transactions).zip(rngs).try_for_each(|((transaction, rejected_id), mut rng)| {
+                self.check_transaction(transaction, *rejected_id, &mut rng)
+                    .map_err(|e| anyhow!("Invalid transaction found in the transactions list: {e}"))
+            })?;
+        }
+
+        Ok(())
+    }
+
+    /// Prepares the transactions for speculation. This function returns a list of valid transactions
+    /// and a list of transactions to be aborted.
+    ///
+    /// Transactions will be aborted if they are a fee transaction, or if they are invalid.
+    pub(crate) fn prepare_transactions_for_speculate<'a, R: CryptoRng + Rng>(
+        &self,
+        transactions: &[&'a Transaction<N>],
+        rng: &mut R,
+    ) -> Result<(Vec<&'a Transaction<N>>, Vec<(&'a Transaction<N>, String)>)> {
+        // Construct the list of valid and invalid transactions.
+        let mut valid_transactions = Vec::with_capacity(transactions.len());
+        let mut aborted_transactions = Vec::with_capacity(transactions.len());
+
+        // Separate the transactions into deploys and executions.
+        let (deployments, executions): (Vec<&Transaction<N>>, Vec<&Transaction<N>>) =
+            transactions.iter().partition(|tx| tx.is_deploy());
+        // Chunk the deploys and executions into groups for parallel verification.
+        let deployments_for_verification = deployments.chunks(Self::MAX_PARALLEL_DEPLOY_VERIFICATIONS);
+        let executions_for_verification = executions.chunks(Self::MAX_PARALLEL_EXECUTE_VERIFICATIONS);
+
+        // Verify the transactions in batches and separate the valid and invalid transactions.
+        for transactions in deployments_for_verification.chain(executions_for_verification) {
+            let rngs = (0..transactions.len()).map(|_| StdRng::from_seed(rng.gen())).collect::<Vec<_>>();
+            // Verify the transactions and collect the error message if there is one.
+            let (valid, invalid): (Vec<_>, Vec<_>) =
+                cfg_into_iter!(transactions).zip(rngs).partition_map(|(transaction, mut rng)| {
+                    // Abort the transaction if it is a fee transaction.
+                    if transaction.is_fee() {
+                        return Either::Right((
+                            *transaction,
+                            "Fee transactions are not allowed in speculate".to_string(),
+                        ));
+                    }
+                    // Verify the transaction.
+                    match self.check_transaction(transaction, None, &mut rng) {
+                        Ok(_) => Either::Left(*transaction),
+                        Err(e) => Either::Right((*transaction, e.to_string())),
+                    }
+                });
+
+            // Collect the valid and aborted transactions.
+            valid_transactions.extend(valid);
+            aborted_transactions.extend(invalid);
+        }
+
+        // Return the valid and invalid transactions.
+        Ok((valid_transactions, aborted_transactions))
+    }
+}
+
 impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
     /// Verifies the transaction in the VM. On failure, returns an error.
     #[inline]
@@ -121,7 +202,13 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
                 }
                 // Verify the deployment if it has not been verified before.
                 if !is_partially_verified {
-                    self.check_deployment_internal(deployment, rng)?;
+                    // Catch any panics during verification.
+                    match handle_halting!(panic::AssertUnwindSafe(|| {
+                        self.check_deployment_internal(deployment, rng)
+                    })) {
+                        Ok(result) => result?,
+                        Err(_) => bail!("Transaction '{id}' panicked during verification"),
+                    }
                 }
             }
             Transaction::Execute(id, execution, _) => {
@@ -133,8 +220,13 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
                 if self.block_store().contains_rejected_deployment_or_execution_id(&execution_id)? {
                     bail!("Transaction '{id}' contains a previously rejected execution")
                 }
-                // Verify the execution.
-                self.check_execution_internal(execution, is_partially_verified)?;
+                // Verify the execution and catch any panics.
+                match handle_halting!(panic::AssertUnwindSafe(|| {
+                    self.check_execution_internal(execution, is_partially_verified)
+                })) {
+                    Ok(result) => result?,
+                    Err(_) => bail!("Transaction '{id}' panicked during verification"),
+                }
             }
             Transaction::Fee(..) => { /* no-op */ }
         }