Merge pull request #2975 from ljedrz/feat/replace_vm_locks_with_channel

[Feat] Ensure sequential processing of potential Ledger writes

Author: vicsn

Cargo.lock

@@ -19,6 +19,9 @@ use anyhow::Context;
 
 impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
     /// Returns a candidate for the next block in the ledger, using a committed subdag and its transmissions.
+    ///    
+    /// # Panics
+    /// This function panics if called from an async context.
     pub fn prepare_advance_to_next_quorum_block<R: Rng + CryptoRng>(
         &self,
         subdag: Subdag<N>,
@@ -51,6 +54,9 @@ impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
     }
 
     /// Returns a candidate for the next block in the ledger.
+    ///
+    /// # Panics
+    /// This function panics if called from an async context.
     pub fn prepare_advance_to_next_beacon_block<R: Rng + CryptoRng>(
         &self,
         private_key: &PrivateKey<N>,
@@ -92,6 +98,9 @@ impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
     }
 
     /// Adds the given block as the next block in the ledger.
+    ///
+    /// # Panics
+    /// This function panics if called from an async context.
     pub fn advance_to_next_block(&self, block: &Block<N>) -> Result<()> {
         // Acquire the write lock on the current block.
         let mut current_block = self.current_block.write();
@@ -212,6 +221,9 @@ where
 
 impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
     /// Constructs a block template for the next block in the ledger.
+    ///
+    /// # Panics
+    /// This function panics if called from an async context.
     #[allow(clippy::type_complexity)]
     fn construct_block_template<R: Rng + CryptoRng>(
         &self,

ledger/src/advance.rs

@@ -21,6 +21,9 @@ use anyhow::{Context, bail};
 
 impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
     /// Checks the given block is valid next block.
+    ///
+    /// # Panics
+    /// This function panics if called from an async context.
     pub fn check_next_block<R: CryptoRng + Rng>(&self, block: &Block<N>, rng: &mut R) -> Result<()> {
         let height = block.height();
         let latest_block = self.latest_block();

ledger/src/check_next_block.rs

@@ -194,6 +194,9 @@ impl<N: Network> TestChainBuilder<N> {
     }
 
     /// Create multiple blocks, with additional parameters.
+    ///
+    /// # Panics
+    /// This function panics if called from an async context.
     pub fn generate_blocks_with_opts(
         &mut self,
         num_blocks: usize,

ledger/src/test_helpers/chain_builder.rs

@@ -148,6 +148,10 @@ optional = true
 workspace = true
 features = [ "preserve_order" ]
 
+[dependencies.tokio]
+version = "1"
+features = [ "sync" ]
+
 [dependencies.tracing]
 workspace = true
 

synthesizer/Cargo.toml

@@ -26,11 +26,15 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
     /// Returns the confirmed transactions, aborted transaction IDs,
     /// and finalize operations from pre-ratify and post-ratify.
     ///
-    /// Note: This method is used to create a new block (including the genesis block).
+    /// # Note
+    /// This method is used to create a new block (including the genesis block).
     ///   - If `coinbase_reward = None`, then the `ratifications` will not be modified.
     ///   - If `coinbase_reward = Some(coinbase_reward)`, then the method will append a
     ///     `Ratify::BlockReward(block_reward)` and `Ratify::PuzzleReward(puzzle_reward)`
     ///     to the front of the `ratifications` list.
+    ///
+    /// # Panics
+    /// This function panics if called from an async context.
     #[inline]
     #[allow(clippy::too_many_arguments)]
     pub fn speculate<'a, R: Rng + CryptoRng>(
@@ -67,8 +71,8 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
                 time_since_last_block,
                 coinbase_reward,
                 candidate_ratifications,
-                candidate_solutions,
-                verified_transactions.into_iter(),
+                candidate_solutions.clone(),
+                verified_transactions.into_iter().cloned().collect(),
             )?;
 
         // Get the aborted transaction ids.
@@ -103,6 +107,9 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
     /// This function also ensure that the given transactions are well-formed and unique.
     ///
     /// Returns the finalize operations from pre-ratify and post-ratify.
+    ///
+    /// # Panics
+    /// This function panics if called from an async context.
     #[inline]
     pub fn check_speculate<R: Rng + CryptoRng>(
         &self,
@@ -138,8 +145,8 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
                 time_since_last_block,
                 None,
                 candidate_ratifications,
-                solutions,
-                candidate_transactions.iter(),
+                solutions.clone(),
+                candidate_transactions,
             )?;
 
         // Ensure the ratifications after speculation match.
@@ -196,29 +203,66 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
     /// Returns the ratifications, confirmed transactions, aborted transactions,
     /// and finalize operations from pre-ratify and post-ratify.
     ///
-    /// Note: This method is used by `VM::speculate` and `VM::check_speculate`.
+    /// # Note
+    /// This method is used by `VM::speculate` and `VM::check_speculate`.
     ///   - If `coinbase_reward = None`, then the `ratifications` will not be modified.
     ///   - If `coinbase_reward = Some(coinbase_reward)`, then the method will append a
     ///     `Ratify::BlockReward(block_reward)` and `Ratify::PuzzleReward(puzzle_reward)`
     ///     to the front of the `ratifications` list.
-    fn atomic_speculate<'a>(
+    ///
+    /// # Panics
+    /// This function panics if called from an async context.
+    fn atomic_speculate(
         &self,
         state: FinalizeGlobalState,
         time_since_last_block: i64,
         coinbase_reward: Option<u64>,
         ratifications: Vec<Ratify<N>>,
-        solutions: &Solutions<N>,
-        transactions: impl ExactSizeIterator<Item = &'a Transaction<N>>,
+        solutions: Solutions<N>,
+        transactions: Vec<Transaction<N>>,
+    ) -> Result<(
+        Ratifications<N>,
+        Vec<ConfirmedTransaction<N>>,
+        Vec<(Transaction<N>, String)>,
+        Vec<FinalizeOperation<N>>,
+    )> {
+        let sequential_op = SequentialOperation::AtomicSpeculate(
+            state,
+            time_since_last_block,
+            coinbase_reward,
+            ratifications,
+            solutions,
+            transactions,
+        );
+        let Some(SequentialOperationResult::AtomicSpeculate(ret)) = self.run_sequential_operation(sequential_op) else {
+            bail!("Already shutting down");
+        };
+
+        ret
+    }
+
+    /// Internal function called when invoking [`Self::atomic_speculate`].
+    ///
+    /// # Note
+    /// This function must only be called from the sequential operation thread.
+    ///
+    /// # Panics
+    /// This function panics if not called from the sequential operation thread.
+    pub(crate) fn atomic_speculate_inner(
+        &self,
+        state: FinalizeGlobalState,
+        time_since_last_block: i64,
+        coinbase_reward: Option<u64>,
+        ratifications: Vec<Ratify<N>>,
+        solutions: Solutions<N>,
+        transactions: Vec<Transaction<N>>,
     ) -> Result<(
         Ratifications<N>,
         Vec<ConfirmedTransaction<N>>,
         Vec<(Transaction<N>, String)>,
         Vec<FinalizeOperation<N>>,
     )> {
-        // Acquire the atomic lock, which is needed to ensure this function is not called concurrently
-        // with other `atomic_finalize!` macro calls, which will cause a `bail!` to be triggered erroneously.
-        // Note: This lock must be held for the entire scope of the call to `atomic_finalize!`.
-        let _atomic_lock = self.atomic_lock.lock();
+        self.ensure_sequential_processing();
 
         let timer = timer!("VM::atomic_speculate");
 
@@ -319,7 +363,7 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
 
                 // Determine if the transaction should be aborted.
                 if let Some(reason) = self.should_abort_transaction(
-                    transaction,
+                    &transaction,
                     &transition_ids,
                     &input_ids,
                     &output_ids,
@@ -336,7 +380,7 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
                 // Process the transaction in an isolated atomic batch.
                 // - If the transaction succeeds, the finalize operations are stored.
                 // - If the transaction fails, the atomic batch is aborted and no finalize operations are stored.
-                let outcome = match transaction {
+                let outcome = match &transaction {
                     // The finalize operation here involves appending the 'stack',
                     // and adding the program to the finalize tree.
                     Transaction::Deploy(_, _, program_owner, deployment, fee) => {
@@ -535,7 +579,7 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
             let post_ratifications = reward_ratifications.iter().chain(post_ratifications);
 
             // Process the post-ratifications.
-            match Self::atomic_post_ratify::<false>(&self.puzzle, store, state, post_ratifications, solutions) {
+            match Self::atomic_post_ratify::<false>(&self.puzzle, store, state, post_ratifications, &solutions) {
                 // Store the finalize operations from the post-ratify.
                 Ok(operations) => ratified_finalize_operations.extend(operations),
                 // Note: This will abort the entire atomic batch.
@@ -570,10 +614,9 @@ impl<N: Network, C: ConsensusStorage<N>> VM<N, C> {
         solutions: &Solutions<N>,
         transactions: &Transactions<N>,
     ) -> Result<Vec<FinalizeOperation<N>>> {
-        // Acquire the atomic lock, which is needed to ensure this function is not called concurrently
-        // with other `atomic_finalize!` macro calls, which will cause a `bail!` to be triggered erroneously.
-        // Note: This lock must be held for the entire scope of the call to `atomic_finalize!`.
-        let _atomic_lock = self.atomic_lock.lock();
+        // The tests may run this method ad-hoc, outside of the context of add_next_block.
+        #[cfg(not(test))]
+        self.ensure_sequential_processing();
 
         let timer = timer!("VM::atomic_finalize");
 
@@ -1848,8 +1891,8 @@ finalize transfer_public:
                 CurrentNetwork::BLOCK_TIME as i64,
                 None,
                 vec![],
-                &None.into(),
-                [deployment_transaction].iter(),
+                None.into(),
+                vec![deployment_transaction],
             )
             .unwrap();
         assert_eq!(candidate_transactions.len(), 1);
@@ -1937,15 +1980,15 @@ finalize transfer_public:
         vm.check_transaction(&bond_validator_transaction, None, rng).unwrap();
 
         // Speculate on the transactions.
-        let transactions = [bond_validator_transaction.clone()];
+        let transactions = vec![bond_validator_transaction.clone()];
         let (_, confirmed_transactions, _, _) = vm
             .atomic_speculate(
                 sample_finalize_state(1),
                 CurrentNetwork::BLOCK_TIME as i64,
                 None,
                 vec![],
-                &None.into(),
-                transactions.iter(),
+                None.into(),
+                transactions,
             )
             .unwrap();
 
@@ -2063,15 +2106,15 @@ finalize transfer_public:
         vm.check_transaction(&bond_validator_transaction, None, rng).unwrap();
 
         // Speculate on the transactions.
-        let transactions = [bond_validator_transaction.clone()];
+        let transactions = vec![bond_validator_transaction.clone()];
         let (_, confirmed_transactions, _, _) = vm
             .atomic_speculate(
                 sample_finalize_state(1),
                 CurrentNetwork::BLOCK_TIME as i64,
                 None,
                 vec![],
-                &None.into(),
-                transactions.iter(),
+                None.into(),
+                transactions,
             )
             .unwrap();
 
@@ -2083,15 +2126,15 @@ finalize transfer_public:
         );
 
         // Speculate on the transactions.
-        let transactions = [bond_validator_transaction.clone()];
+        let transactions = vec![bond_validator_transaction.clone()];
         let (_, confirmed_transactions, aborted_transaction_ids, _) = vm
             .atomic_speculate(
                 sample_finalize_state(CurrentNetwork::CONSENSUS_HEIGHT(ConsensusVersion::V3).unwrap()),
                 CurrentNetwork::BLOCK_TIME as i64,
                 None,
                 vec![],
-                &None.into(),
-                transactions.iter(),
+                None.into(),
+                transactions,
             )
             .unwrap();
 
@@ -2194,15 +2237,15 @@ finalize transfer_public:
         // Transfer_10 -> Balance = 30 - 10 = 20
         // Transfer_20 -> Balance = 20 - 20 = 0
         {
-            let transactions = [mint_10.clone(), transfer_10.clone(), transfer_20.clone()];
+            let transactions = vec![mint_10.clone(), transfer_10.clone(), transfer_20.clone()];
             let (_, confirmed_transactions, aborted_transaction_ids, _) = vm
                 .atomic_speculate(
                     sample_finalize_state(1),
                     CurrentNetwork::BLOCK_TIME as i64,
                     None,
                     vec![],
-                    &None.into(),
-                    transactions.iter(),
+                    None.into(),
+                    transactions,
                 )
                 .unwrap();
 
@@ -2222,15 +2265,15 @@ finalize transfer_public:
         // Mint_20 -> Balance = 10 + 20 = 30
         // Transfer_30 -> Balance = 30 - 30 = 0
         {
-            let transactions = [transfer_20.clone(), mint_10.clone(), mint_20.clone(), transfer_30.clone()];
+            let transactions = vec![transfer_20.clone(), mint_10.clone(), mint_20.clone(), transfer_30.clone()];
             let (_, confirmed_transactions, aborted_transaction_ids, _) = vm
                 .atomic_speculate(
                     sample_finalize_state(1),
                     CurrentNetwork::BLOCK_TIME as i64,
                     None,
                     vec![],
-                    &None.into(),
-                    transactions.iter(),
+                    None.into(),
+                    transactions,
                 )
                 .unwrap();
 
@@ -2250,15 +2293,15 @@ finalize transfer_public:
         // Transfer_20 -> Balance = 20 - 20 = 0
         // Transfer_10 -> Balance = 0 - 10 = -10 (should be rejected)
         {
-            let transactions = [transfer_20.clone(), transfer_10.clone()];
+            let transactions = vec![transfer_20.clone(), transfer_10.clone()];
             let (_, confirmed_transactions, aborted_transaction_ids, _) = vm
                 .atomic_speculate(
                     sample_finalize_state(1),
                     CurrentNetwork::BLOCK_TIME as i64,
                     None,
                     vec![],
-                    &None.into(),
-                    transactions.iter(),
+                    None.into(),
+                    transactions,
                 )
                 .unwrap();
 
@@ -2282,15 +2325,15 @@ finalize transfer_public:
         // Transfer_20 -> Balance = 10 - 20 = -10 (should be rejected)
         // Transfer_10 -> Balance = 10 - 10 = 0
         {
-            let transactions = [mint_20.clone(), transfer_30.clone(), transfer_20.clone(), transfer_10.clone()];
+            let transactions = vec![mint_20.clone(), transfer_30.clone(), transfer_20.clone(), transfer_10.clone()];
             let (_, confirmed_transactions, aborted_transaction_ids, _) = vm
                 .atomic_speculate(
                     sample_finalize_state(1),
                     CurrentNetwork::BLOCK_TIME as i64,
                     None,
                     vec![],
-                    &None.into(),
-                    transactions.iter(),
+                    None.into(),
+                    transactions,
                 )
                 .unwrap();
 

synthesizer/src/vm/finalize.rs

@@ -25,3 +25,6 @@ mod macros;
 
 mod rewards;
 pub use rewards::*;
+
+mod sequential_op;
+pub use sequential_op::*;