Delete old consensus handler (#24107)

- Rewrite tests (with extensive help from Claude) that depend on old
consensus handler
- delete old consensus handler

Author: mystenmark

crates/sui-core/src/authority/authority_per_epoch_store.rs

@@ -2,22 +2,21 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::checkpoints::CheckpointServiceNoop;
-use crate::consensus_handler::SequencedConsensusTransaction;
-
 use core::default::Default;
 use fastcrypto::hash::MultisetHash;
 use fastcrypto::traits::KeyPair;
 use sui_protocol_config::Chain;
 use sui_types::base_types::FullObjectRef;
 use sui_types::crypto::{AccountKeyPair, AuthorityKeyPair};
-use sui_types::messages_consensus::ConsensusTransaction;
 use sui_types::utils::to_sender_signed_transaction;
 
-use super::shared_object_version_manager::{AssignedTxAndVersions, AssignedVersions};
+use super::shared_object_version_manager::AssignedVersions;
 use super::test_authority_builder::TestAuthorityBuilder;
 use super::*;
 
+#[cfg(test)]
+use super::shared_object_version_manager::{AssignedTxAndVersions, Schedulable};
+
 pub async fn send_and_confirm_transaction(
     authority: &AuthorityState,
     transaction: Transaction,
@@ -427,27 +426,20 @@ pub async fn send_consensus(
     authority: &AuthorityState,
     cert: &VerifiedCertificate,
 ) -> AssignedVersions {
-    let transaction = SequencedConsensusTransaction::new_test(
-        ConsensusTransaction::new_certificate_message(&authority.name, cert.clone().into_inner()),
-    );
-
-    let (_, assigned_versions) = authority
+    let assigned_versions = authority
         .epoch_store_for_testing()
-        .process_consensus_transactions_for_tests(
-            vec![transaction],
-            &Arc::new(CheckpointServiceNoop {}),
+        .assign_shared_object_versions_for_tests(
             authority.get_object_cache_reader().as_ref(),
-            &authority.metrics,
-            true,
+            &vec![VerifiedExecutableTransaction::new_from_certificate(
+                cert.clone(),
+            )],
         )
-        .await
         .unwrap();
 
     let assigned_versions = assigned_versions
-        .0
-        .into_iter()
-        .next()
-        .map(|(_, v)| v)
+        .into_map()
+        .get(&cert.key())
+        .cloned()
         .unwrap_or_else(|| AssignedVersions::new(vec![], None));
 
     let certs = vec![(
@@ -466,53 +458,70 @@ pub async fn send_consensus_no_execution(
     authority: &AuthorityState,
     cert: &VerifiedCertificate,
 ) -> AssignedVersions {
-    let transaction = SequencedConsensusTransaction::new_test(
-        ConsensusTransaction::new_certificate_message(&authority.name, cert.clone().into_inner()),
-    );
-
-    // Call process_consensus_transaction() instead of handle_consensus_transaction(), to avoid actually executing cert.
+    // Use the simpler assign_shared_object_versions_for_tests API to avoid actually executing cert.
     // This allows testing cert execution independently.
-    let (_, assigned_versions) = authority
+    let assigned_versions = authority
         .epoch_store_for_testing()
-        .process_consensus_transactions_for_tests(
-            vec![transaction],
-            &Arc::new(CheckpointServiceNoop {}),
+        .assign_shared_object_versions_for_tests(
             authority.get_object_cache_reader().as_ref(),
-            &authority.metrics,
-            true,
+            &vec![VerifiedExecutableTransaction::new_from_certificate(
+                cert.clone(),
+            )],
         )
-        .await
         .unwrap();
-    assert_eq!(assigned_versions.0.len(), 1);
-    assigned_versions.0.into_iter().next().unwrap().1
+
+    assigned_versions
+        .into_map()
+        .get(&cert.key())
+        .cloned()
+        .unwrap_or_else(|| AssignedVersions::new(vec![], None))
 }
 
-pub async fn send_batch_consensus_no_execution(
+#[cfg(test)]
+pub async fn send_batch_consensus_no_execution<C>(
     authority: &AuthorityState,
     certificates: &[VerifiedCertificate],
-    skip_consensus_commit_prologue_in_test: bool,
-) -> (Vec<Schedulable>, AssignedTxAndVersions) {
-    let transactions = certificates
+    consensus_handler: &mut crate::consensus_handler::ConsensusHandler<C>,
+    captured_transactions: &crate::consensus_test_utils::CapturedTransactions,
+) -> (Vec<Schedulable>, AssignedTxAndVersions)
+where
+    C: crate::checkpoints::CheckpointServiceNotify + Send + Sync + 'static,
+{
+    use crate::consensus_test_utils::TestConsensusCommit;
+    use sui_types::messages_consensus::ConsensusTransaction;
+
+    let consensus_transactions: Vec<ConsensusTransaction> = certificates
         .iter()
         .map(|cert| {
-            SequencedConsensusTransaction::new_test(ConsensusTransaction::new_certificate_message(
-                &authority.name,
-                cert.clone().into_inner(),
-            ))
+            ConsensusTransaction::new_certificate_message(&authority.name, cert.clone().into())
         })
         .collect();
 
-    // Call process_consensus_transaction() instead of handle_consensus_transaction(), to avoid actually executing cert.
-    // This allows testing cert execution independently.
-    authority
-        .epoch_store_for_testing()
-        .process_consensus_transactions_for_tests(
-            transactions,
-            &Arc::new(CheckpointServiceNoop {}),
-            authority.get_object_cache_reader().as_ref(),
-            &authority.metrics,
-            skip_consensus_commit_prologue_in_test,
-        )
-        .await
-        .unwrap()
+    // Determine appropriate round and timestamp
+    let epoch_store = authority.epoch_store_for_testing();
+    let round = epoch_store.get_highest_pending_checkpoint_height() + 1;
+    let timestamp_ms = epoch_store.epoch_start_state().epoch_start_timestamp_ms();
+    let sub_dag_index = 0;
+
+    let commit =
+        TestConsensusCommit::new(consensus_transactions, round, timestamp_ms, sub_dag_index);
+
+    consensus_handler
+        .handle_consensus_commit_for_test(commit)
+        .await;
+
+    // Wait for captured transactions to be available
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+    let (scheduled_txns, assigned_tx_and_versions) = {
+        let mut captured = captured_transactions.lock();
+        assert!(
+            !captured.is_empty(),
+            "Expected transactions to be scheduled"
+        );
+        let (scheduled_txns, assigned_tx_and_versions, _) = captured.remove(0);
+        (scheduled_txns, assigned_tx_and_versions)
+    };
+
+    (scheduled_txns, assigned_tx_and_versions)
 }

crates/sui-core/src/authority/authority_test_utils.rs

@@ -6,7 +6,6 @@ use crate::authority::authority_per_epoch_store::{
 };
 use crate::authority::transaction_deferral::DeferralKey;
 use crate::checkpoints::BuilderCheckpointSummary;
-use crate::consensus_handler::SequencedConsensusTransactionKind;
 use crate::epoch::randomness::SINGLETON_KEY;
 use dashmap::DashMap;
 use fastcrypto_tbls::{dkg_v1, nodes::PartyId};
@@ -23,9 +22,7 @@ use sui_types::crypto::RandomnessRound;
 use sui_types::error::SuiResult;
 use sui_types::execution::ExecutionTimeObservationKey;
 use sui_types::messages_checkpoint::{CheckpointContents, CheckpointSequenceNumber};
-use sui_types::messages_consensus::{
-    AuthorityIndex, ConsensusTransaction, ConsensusTransactionKind,
-};
+use sui_types::messages_consensus::AuthorityIndex;
 use sui_types::{
     base_types::{ConsensusObjectSequenceKey, ObjectID},
     digests::TransactionDigest,
@@ -193,14 +190,6 @@ impl ConsensusCommitOutput {
     }
 
     pub fn defer_transactions(
-        &mut self,
-        key: DeferralKey,
-        transactions: Vec<VerifiedSequencedConsensusTransaction>,
-    ) {
-        self.deferred_txns.push((key, transactions));
-    }
-
-    pub fn defer_transactions_v2(
         &mut self,
         key: DeferralKey,
         transactions: Vec<VerifiedExecutableTransaction>,
@@ -301,14 +290,11 @@ impl ConsensusCommitOutput {
             batch.insert_batch(&tables.next_shared_object_versions_v2, next_versions)?;
         }
 
-        // TODO(consensus-handler-rewrite): delete the old structures once commit handler rewrite is complete
-        batch.delete_batch(&tables.deferred_transactions, &self.deleted_deferred_txns)?;
         batch.delete_batch(
             &tables.deferred_transactions_v2,
             &self.deleted_deferred_txns,
         )?;
 
-        batch.insert_batch(&tables.deferred_transactions, self.deferred_txns)?;
         batch.insert_batch(
             &tables.deferred_transactions_v2,
             self.deferred_txns_v2.into_iter().map(|(key, txs)| {
@@ -402,10 +388,6 @@ impl ConsensusCommitOutput {
 pub(crate) struct ConsensusOutputCache {
     // deferred transactions is only used by consensus handler so there should never be lock contention
     // - hence no need for a DashMap.
-    // TODO(consensus-handler-rewrite): remove this once we no longer need to support the old consensus handler
-    pub(crate) deferred_transactions:
-        Mutex<BTreeMap<DeferralKey, Vec<VerifiedSequencedConsensusTransaction>>>,
-
     pub(crate) deferred_transactions_v2:
         Mutex<BTreeMap<DeferralKey, Vec<VerifiedExecutableTransaction>>>,
 
@@ -423,10 +405,6 @@ impl ConsensusOutputCache {
         epoch_start_configuration: &EpochStartConfiguration,
         tables: &AuthorityEpochTables,
     ) -> Self {
-        let deferred_transactions = tables
-            .get_all_deferred_transactions()
-            .expect("load deferred transactions cannot fail");
-
         let deferred_transactions_v2 = tables
             .get_all_deferred_transactions_v2()
             .expect("load deferred transactions cannot fail");
@@ -439,7 +417,6 @@ impl ConsensusOutputCache {
         let executed_in_epoch_cache_capacity = 50_000;
 
         Self {
-            deferred_transactions: Mutex::new(deferred_transactions),
             deferred_transactions_v2: Mutex::new(deferred_transactions_v2),
             user_signatures_for_checkpoints: Default::default(),
             executed_in_epoch: RwLock::new(DashMap::with_shard_amount(2048)),
@@ -915,94 +892,7 @@ impl ConsensusOutputQuarantine {
             .next()
     }
 
-    pub(super) fn load_initial_object_debts(
-        &self,
-        epoch_store: &AuthorityPerEpochStore,
-        current_round: Round,
-        for_randomness: bool,
-        transactions: &[VerifiedSequencedConsensusTransaction],
-    ) -> SuiResult<impl IntoIterator<Item = (ObjectID, u64)>> {
-        let protocol_config = epoch_store.protocol_config();
-        let tables = epoch_store.tables()?;
-        let default_per_commit_budget = protocol_config
-            .max_accumulated_txn_cost_per_object_in_mysticeti_commit_as_option()
-            .unwrap_or(0);
-        let (hash_table, db_table, per_commit_budget) = if for_randomness {
-            (
-                &self.congestion_control_randomness_object_debts,
-                &tables.congestion_control_randomness_object_debts,
-                protocol_config
-                    .max_accumulated_randomness_txn_cost_per_object_in_mysticeti_commit_as_option()
-                    .unwrap_or(default_per_commit_budget),
-            )
-        } else {
-            (
-                &self.congestion_control_object_debts,
-                &tables.congestion_control_object_debts,
-                default_per_commit_budget,
-            )
-        };
-        let mut shared_input_object_ids: Vec<_> = transactions
-            .iter()
-            .filter_map(|tx| {
-                match &tx.0.transaction {
-                    SequencedConsensusTransactionKind::External(ConsensusTransaction {
-                        kind: ConsensusTransactionKind::CertifiedTransaction(tx),
-                        ..
-                    }) => Some(itertools::Either::Left(
-                        tx.shared_input_objects().map(|obj| obj.id),
-                    )),
-                    SequencedConsensusTransactionKind::External(ConsensusTransaction {
-                        kind: ConsensusTransactionKind::UserTransaction(tx),
-                        ..
-                    // Bug fix that required a protocol flag.
-                    }) if protocol_config.use_mfp_txns_in_load_initial_object_debts() => Some(
-                        itertools::Either::Right(tx.shared_input_objects().map(|obj| obj.id)),
-                    ),
-                    _ => None,
-                }
-            })
-            .flatten()
-            .collect();
-        shared_input_object_ids.sort();
-        shared_input_object_ids.dedup();
-
-        let results = do_fallback_lookup(
-            &shared_input_object_ids,
-            |object_id| {
-                if let Some(debt) = hash_table.get(object_id) {
-                    CacheResult::Hit(Some(debt.into_v1()))
-                } else {
-                    CacheResult::Miss
-                }
-            },
-            |object_ids| {
-                db_table
-                    .multi_get(object_ids)
-                    .expect("db error")
-                    .into_iter()
-                    .map(|debt| debt.map(|debt| debt.into_v1()))
-                    .collect()
-            },
-        );
-
-        Ok(results
-            .into_iter()
-            .zip(shared_input_object_ids)
-            .filter_map(|(debt, object_id)| debt.map(|debt| (debt, object_id)))
-            .map(move |((round, debt), object_id)| {
-                // Stored debts already account for the budget of the round in which
-                // they were accumulated. Application of budget from future rounds to
-                // the debt is handled here.
-                assert!(current_round > round);
-                let num_rounds = current_round - round - 1;
-                let debt = debt.saturating_sub(per_commit_budget * num_rounds);
-                (object_id, debt)
-            }))
-    }
-
-    // TODO: Remove the above version and rename this without _v2
-    pub(crate) fn load_initial_object_debts_v2(
+    pub(crate) fn load_initial_object_debts(
         &self,
         epoch_store: &AuthorityPerEpochStore,
         current_round: Round,