sui_core/unit_tests/

consensus_test_utils.rs

// Copyright (c) Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

//! Common test utilities for consensus handler testing

use std::collections::HashSet;
use std::sync::Arc;

use consensus_core::BlockStatus;
use consensus_types::block::BlockRef;
use itertools::Itertools;
use parking_lot::Mutex;
use prometheus::Registry;
use sui_types::digests::{Digest, TransactionDigest};
use sui_types::error::SuiResult;
use sui_types::executable_transaction::VerifiedExecutableTransaction;
use sui_types::messages_consensus::{
    AuthorityIndex, ConsensusPosition, ConsensusTransaction, ConsensusTransactionKind,
};
use sui_types::sui_system_state::epoch_start_sui_system_state::EpochStartSystemStateTrait;
use sui_types::transaction::VerifiedTransaction;

use crate::authority::authority_per_epoch_store::{
    AuthorityPerEpochStore, ExecutionIndicesWithStatsV2,
};
use crate::authority::backpressure::BackpressureManager;
use crate::authority::shared_object_version_manager::Schedulable;
use crate::authority::{AuthorityMetrics, AuthorityState, ExecutionEnv};
use crate::consensus_adapter::{
    BlockStatusReceiver, ConsensusAdapter, ConsensusAdapterMetrics, ConsensusClient,
};
use crate::consensus_handler::{
    ConsensusHandler, ExecutionSchedulerSender, SequencedConsensusTransaction,
    SequencedConsensusTransactionKind,
};
use crate::consensus_throughput_calculator::ConsensusThroughputCalculator;
use crate::consensus_types::consensus_output_api::{ConsensusCommitAPI, ParsedTransaction};
use crate::mock_consensus::with_block_status;

pub(crate) type CapturedTransactions = Arc<Mutex<Vec<crate::consensus_handler::SchedulerMessage>>>;

pub struct TestConsensusCommit {
    pub transactions: Vec<ConsensusTransaction>,
    pub round: u64,
    pub timestamp_ms: u64,
    pub sub_dag_index: u64,
    /// Indices into `transactions` reported as rejected by consensus voting.
    rejected_indices: HashSet<usize>,
    transaction_authors: Option<Vec<consensus_config::AuthorityIndex>>,
}

impl TestConsensusCommit {
    pub fn new(
        transactions: Vec<ConsensusTransaction>,
        round: u64,
        timestamp_ms: u64,
        sub_dag_index: u64,
    ) -> Self {
        Self {
            transactions,
            round,
            timestamp_ms,
            sub_dag_index,
            rejected_indices: HashSet::new(),
            transaction_authors: None,
        }
    }

    pub fn empty(round: u64, timestamp_ms: u64, sub_dag_index: u64) -> Self {
        Self::new(vec![], round, timestamp_ms, sub_dag_index)
    }

    pub fn with_rejected_indices(mut self, indices: impl IntoIterator<Item = usize>) -> Self {
        self.rejected_indices = indices.into_iter().collect();
        self
    }

    pub fn with_transaction_authors(mut self, authors: impl IntoIterator<Item = u32>) -> Self {
        self.transaction_authors = Some(
            authors
                .into_iter()
                .map(consensus_config::AuthorityIndex::new_for_test)
                .collect(),
        );
        self
    }
}

impl std::fmt::Display for TestConsensusCommit {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "TestConsensusCommitAPI(round={}, timestamp_ms={}, sub_dag_index={})",
            self.round, self.timestamp_ms, self.sub_dag_index
        )
    }
}

impl ConsensusCommitAPI for TestConsensusCommit {
    fn commit_ref(&self) -> consensus_core::CommitRef {
        consensus_core::CommitRef::default()
    }

    fn leader_round(&self) -> u64 {
        self.round
    }

    fn leader_author_index(&self) -> AuthorityIndex {
        0
    }

    fn commit_timestamp_ms(&self) -> u64 {
        self.timestamp_ms
    }

    fn commit_sub_dag_index(&self) -> u64 {
        self.sub_dag_index
    }

    fn transactions(&self) -> Vec<(BlockRef, Vec<ParsedTransaction>)> {
        if let Some(authors) = &self.transaction_authors {
            return self
                .transactions
                .iter()
                .zip_eq(authors)
                .enumerate()
                .map(|(i, (tx, author))| {
                    let block_ref = BlockRef {
                        author: *author,
                        round: self.round as u32,
                        digest: Default::default(),
                    };
                    let parsed_tx = ParsedTransaction {
                        transaction: tx.clone(),
                        rejected: self.rejected_indices.contains(&i),
                        serialized_len: 0,
                    };
                    (block_ref, vec![parsed_tx])
                })
                .collect();
        }

        let block_ref = BlockRef {
            author: consensus_config::AuthorityIndex::ZERO,
            round: self.round as u32,
            digest: Default::default(),
        };

        let parsed_txs: Vec<ParsedTransaction> = self
            .transactions
            .iter()
            .enumerate()
            .map(|(i, tx)| ParsedTransaction {
                transaction: tx.clone(),
                rejected: self.rejected_indices.contains(&i),
                serialized_len: 0,
            })
            .collect();

        vec![(block_ref, parsed_txs)]
    }

    fn rejected_transactions_digest(&self) -> Digest {
        Digest::default()
    }

    fn rejected_transactions_debug_string(&self) -> String {
        "no rejected transactions from TestConsensusCommit".to_string()
    }
}

pub struct TestConsensusHandlerSetup<C> {
    pub consensus_handler: ConsensusHandler<C>,
    pub captured_transactions: CapturedTransactions,
    pub metrics: Arc<AuthorityMetrics>,
}

/// Makes a consensus adapter with the standard test wiring (limits, metrics), backed by the
/// given consensus client.
pub fn make_consensus_adapter_with_client_for_test(
    state: &Arc<AuthorityState>,
    client: Arc<dyn ConsensusClient>,
    max_pending_local_submissions: usize,
) -> Arc<ConsensusAdapter> {
    Arc::new(ConsensusAdapter::new(
        client,
        state.checkpoint_store.clone(),
        state.name,
        100_000,
        max_pending_local_submissions,
        ConsensusAdapterMetrics::new_test(),
        Arc::new(tokio::sync::Notify::new()),
    ))
}

pub fn make_consensus_adapter_for_test(
    state: Arc<AuthorityState>,
    process_via_checkpoint: HashSet<TransactionDigest>,
    execute: bool,
    mock_block_status_receivers: Vec<BlockStatusReceiver>,
) -> Arc<ConsensusAdapter> {
    make_consensus_adapter_for_test_with_submit_limit(
        state,
        process_via_checkpoint,
        execute,
        mock_block_status_receivers,
        100_000,
    )
}

pub fn make_consensus_adapter_for_test_with_submit_limit(
    state: Arc<AuthorityState>,
    process_via_checkpoint: HashSet<TransactionDigest>,
    execute: bool,
    mock_block_status_receivers: Vec<BlockStatusReceiver>,
    max_pending_local_submissions: usize,
) -> Arc<ConsensusAdapter> {
    #[derive(Clone)]
    struct SubmitDirectly {
        state: Arc<AuthorityState>,
        process_via_checkpoint: HashSet<TransactionDigest>,
        execute: bool,
        mock_block_status_receivers: Arc<Mutex<Vec<BlockStatusReceiver>>>,
    }

    #[async_trait::async_trait]
    impl ConsensusClient for SubmitDirectly {
        async fn submit(
            &self,
            transactions: &[ConsensusTransaction],
            epoch_store: &Arc<AuthorityPerEpochStore>,
        ) -> SuiResult<(Vec<ConsensusPosition>, BlockStatusReceiver)> {
            // If transactions are empty, then we are performing a ping check and will attempt to ping consensus and simulate a transaction submission to consensus.
            if transactions.is_empty() {
                return Ok((
                    vec![ConsensusPosition::ping(epoch_store.epoch(), BlockRef::MIN)],
                    with_block_status(BlockStatus::Sequenced(BlockRef::MIN)),
                ));
            }

            let num_transactions = transactions.len();
            let mut executed_via_checkpoint = 0;

            // Simple processing - just mark transactions for checkpoint execution if needed
            for txn in transactions {
                if let ConsensusTransactionKind::UserTransactionV2(tx) = &txn.kind {
                    let transaction_digest = tx.tx().digest();
                    if self.process_via_checkpoint.contains(transaction_digest) {
                        epoch_store
                            .insert_finalized_transactions(vec![*transaction_digest].as_slice(), 10)
                            .expect("Should not fail");
                        executed_via_checkpoint += 1;
                    }
                }
            }

            let sequenced_transactions: Vec<SequencedConsensusTransaction> = transactions
                .iter()
                .map(|txn| SequencedConsensusTransaction::new_test(txn.clone()))
                .collect();

            let keys = sequenced_transactions
                .iter()
                .map(|tx| tx.key())
                .collect::<Vec<_>>();

            // Only execute transactions if explicitly requested and not via checkpoint
            if self.execute {
                for tx in sequenced_transactions {
                    if let Some(transaction_digest) = tx.transaction.executable_transaction_digest()
                    {
                        // Skip if already executed via checkpoint
                        if self.process_via_checkpoint.contains(&transaction_digest) {
                            continue;
                        }

                        // Extract executable transaction from consensus transaction
                        let executable_tx = match &tx.transaction {
                            SequencedConsensusTransactionKind::External(ext) => match &ext.kind {
                                ConsensusTransactionKind::UserTransactionV2(tx) => {
                                    Some(VerifiedExecutableTransaction::new_from_consensus(
                                        VerifiedTransaction::new_unchecked(tx.tx().clone()),
                                        0,
                                    ))
                                }
                                _ => None,
                            },
                            SequencedConsensusTransactionKind::System(sys_tx) => {
                                Some(sys_tx.clone())
                            }
                        };

                        if let Some(exec_tx) = executable_tx {
                            let versions = epoch_store.assign_shared_object_versions_for_tests(
                                self.state.get_object_cache_reader().as_ref(),
                                std::slice::from_ref(&exec_tx),
                            )?;

                            let assigned_version = versions
                                .into_map()
                                .into_iter()
                                .next()
                                .map(|(_, v)| v)
                                .unwrap_or_default();

                            self.state.execution_scheduler().enqueue(
                                vec![(
                                    Schedulable::Transaction(exec_tx),
                                    ExecutionEnv::new().with_assigned_versions(assigned_version),
                                )],
                                epoch_store,
                            );
                        }
                    }
                }
            }

            epoch_store.process_notifications(keys.iter());

            assert_eq!(
                executed_via_checkpoint,
                self.process_via_checkpoint.len(),
                "Some transactions were not executed via checkpoint"
            );

            assert!(
                !self.mock_block_status_receivers.lock().is_empty(),
                "No mock submit responses left"
            );

            let mut consensus_positions = Vec::new();
            for index in 0..num_transactions {
                consensus_positions.push(ConsensusPosition {
                    epoch: epoch_store.epoch(),
                    index: index as u16,
                    block: BlockRef::MIN,
                });
            }

            Ok((
                consensus_positions,
                self.mock_block_status_receivers.lock().remove(0),
            ))
        }
    }
    // Make a new consensus adapter instance.
    let client = Arc::new(SubmitDirectly {
        state: state.clone(),
        process_via_checkpoint,
        execute,
        mock_block_status_receivers: Arc::new(Mutex::new(mock_block_status_receivers)),
    });
    make_consensus_adapter_with_client_for_test(&state, client, max_pending_local_submissions)
}

/// Creates a ConsensusHandler for testing with a mock ExecutionSchedulerSender that captures transactions
pub async fn setup_consensus_handler_for_testing_with_checkpoint_service<C>(
    authority: &Arc<AuthorityState>,
    checkpoint_service: Arc<C>,
) -> TestConsensusHandlerSetup<C>
where
    C: Send + Sync + 'static,
{
    let epoch_store = authority.epoch_store_for_testing();
    let consensus_committee = epoch_store.epoch_start_state().get_consensus_committee();
    let metrics = Arc::new(AuthorityMetrics::new(&Registry::new()));
    let throughput_calculator = ConsensusThroughputCalculator::new(None, metrics.clone());
    let backpressure_manager = BackpressureManager::new_for_tests();
    let last_consensus_stats = ExecutionIndicesWithStatsV2 {
        stats: crate::authority::authority_per_epoch_store::ConsensusStats::new(
            consensus_committee.size(),
        ),
        ..Default::default()
    };

    let captured_transactions: CapturedTransactions = Arc::new(Mutex::new(Vec::new()));
    let captured_tx_clone = captured_transactions.clone();

    let (tx_sender, mut receiver) =
        mysten_metrics::monitored_mpsc::unbounded_channel("test_execution_scheduler");

    tokio::spawn(async move {
        while let Some(item) = receiver.recv().await {
            captured_tx_clone.lock().push(item);
        }
    });

    let execution_scheduler_sender = ExecutionSchedulerSender::new_for_testing(tx_sender);

    let consensus_handler = ConsensusHandler::new_for_testing(
        epoch_store.clone(),
        checkpoint_service,
        execution_scheduler_sender,
        authority.get_object_cache_reader().clone(),
        consensus_committee,
        metrics.clone(),
        Arc::new(throughput_calculator),
        backpressure_manager.subscribe(),
        authority.traffic_controller.clone(),
        authority.transaction_deny_config_manager().clone(),
        last_consensus_stats,
    );

    TestConsensusHandlerSetup {
        consensus_handler,
        captured_transactions,
        metrics,
    }
}

/// Creates a ConsensusHandler for testing with CheckpointServiceNoop
#[cfg(test)]
pub async fn setup_consensus_handler_for_testing(
    authority: &Arc<AuthorityState>,
) -> TestConsensusHandlerSetup<crate::checkpoints::CheckpointServiceNoop> {
    setup_consensus_handler_for_testing_with_checkpoint_service(
        authority,
        Arc::new(crate::checkpoints::CheckpointServiceNoop {}),
    )
    .await
}