sui_core/

consensus_handler.rs

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

use std::{
    collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque},
    hash::Hash,
    num::NonZeroUsize,
    sync::{Arc, Mutex},
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use consensus_config::Committee as ConsensusCommittee;
use consensus_core::{CommitIndex, CommitRef, ConsumerMonitor};
use consensus_types::block::BlockRef;
use consensus_types::block::TransactionIndex;
use fastcrypto_zkp::bn254::zk_login::{JWK, JwkId};
use lru::LruCache;
use mysten_common::{
    assert_reachable, assert_sometimes, debug_fatal, random_util::randomize_cache_capacity_in_tests,
};
use mysten_metrics::{
    monitored_future,
    monitored_mpsc::{self, UnboundedReceiver},
    monitored_scope, spawn_monitored_task,
};
use parking_lot::RwLockWriteGuard;
use serde::{Deserialize, Serialize};
use sui_config::node::CongestionLogConfig;
use sui_macros::{fail_point, fail_point_arg, fail_point_if};
use sui_protocol_config::{
    Chain, PerObjectCongestionControlMode, ProtocolConfig, assert_reachable_gated,
};
use sui_types::{
    authenticator_state::ActiveJwk,
    base_types::{
        AuthorityName, ConciseableName, ConsensusObjectSequenceKey, ObjectID, ObjectRef,
        SequenceNumber, TransactionDigest,
    },
    crypto::RandomnessRound,
    digests::{AdditionalConsensusStateDigest, ConsensusCommitDigest, Digest},
    executable_transaction::{
        TrustedExecutableTransaction, VerifiedExecutableTransaction,
        VerifiedExecutableTransactionWithAliases,
    },
    messages_checkpoint::{
        CheckpointSequenceNumber, CheckpointSignatureMessage, CheckpointTimestamp,
    },
    messages_consensus::{
        AuthorityCapabilitiesV2, AuthorityIndex, ConsensusDeterminedVersionAssignments,
        ConsensusPosition, ConsensusTransaction, ConsensusTransactionKey, ConsensusTransactionKind,
        ExecutionTimeObservation, SharedTransactionDenyConfig,
    },
    sui_system_state::epoch_start_sui_system_state::EpochStartSystemStateTrait,
    transaction::{
        InputObjectKind, PlainTransactionWithClaims, SenderSignedData, TransactionDataAPI,
        TransactionKey, VerifiedTransaction, WithAliases, paid_proposer_count,
    },
};
use tokio::task::JoinSet;
use tracing::{debug, error, info, instrument, trace, warn};

use crate::{
    authority::{
        AuthorityMetrics, AuthorityState, ExecutionEnv,
        authority_per_epoch_store::{
            AuthorityPerEpochStore, CancelConsensusCertificateReason, ConsensusStats,
            ConsensusStatsAPI, ExecutionIndices, ExecutionIndicesWithStatsV2,
            consensus_quarantine::ConsensusCommitOutput,
        },
        backpressure::{BackpressureManager, BackpressureSubscriber},
        congestion_log::CongestionCommitLogger,
        consensus_tx_status_cache::ConsensusTxStatus,
        execution_time_estimator::ExecutionTimeEstimator,
        shared_object_congestion_tracker::SharedObjectCongestionTracker,
        shared_object_version_manager::{AssignedTxAndVersions, AssignedVersions, Schedulable},
        transaction_deferral::{DeferralKey, DeferralReason, transaction_deferral_within_limit},
    },
    checkpoints::{
        CheckpointHeight, CheckpointRoots, CheckpointService, CheckpointServiceNotify,
        PendingCheckpoint, PendingCheckpointInfo,
    },
    consensus_throughput_calculator::ConsensusThroughputCalculator,
    consensus_types::consensus_output_api::{ConsensusCommitAPI, ParsedTransaction},
    epoch::{
        randomness::{AdvanceDkgError, DkgStatus, RandomnessManager},
        reconfiguration::ReconfigState,
    },
    execution_cache::ObjectCacheRead,
    execution_scheduler::{SettlementBatchInfo, SettlementScheduler},
    gasless_rate_limiter::ConsensusGaslessCounter,
    post_consensus_tx_reorder::PostConsensusTxReorder,
    traffic_controller::{TrafficController, policies::TrafficTally},
    transaction_deny_config_manager::TransactionDenyConfigManager,
};

/// Tracks the nature of intra-commit owned object lock conflicts for a winning transaction.
#[derive(Default)]
struct ConflictInfo {
    /// Number of conflicts on gas payment objects.
    gas_object_conflicts: u64,
    /// Number of conflicts on non-gas owned objects.
    non_gas_object_conflicts: u64,
    /// Index of the block authority that sequenced the winning (lock holder) transaction.
    winner_author: usize,
}

/// Output from filtering consensus transactions.
/// Contains the filtered transactions and any owned object locks acquired post-consensus.
struct FilteredConsensusOutput {
    transactions: Vec<(SequencedConsensusTransactionKind, u32)>,
    owned_object_locks: HashMap<ObjectRef, TransactionDigest>,
    dropped_transaction_keys: Vec<ConsensusTransactionKey>,
    // When multiple transactions in the same commit try to lock the same owned object, the transaction
    // that managed to lock it first is tracked here with the conflict info.
    contested_transaction_digests: HashMap<TransactionDigest, ConflictInfo>,
}

pub struct ConsensusHandlerInitializer {
    state: Arc<AuthorityState>,
    checkpoint_service: Arc<CheckpointService>,
    epoch_store: Arc<AuthorityPerEpochStore>,
    throughput_calculator: Arc<ConsensusThroughputCalculator>,
    backpressure_manager: Arc<BackpressureManager>,
    congestion_logger: Option<Arc<Mutex<CongestionCommitLogger>>>,
    consensus_gasless_counter: Arc<ConsensusGaslessCounter>,
}

impl ConsensusHandlerInitializer {
    pub fn new(
        state: Arc<AuthorityState>,
        checkpoint_service: Arc<CheckpointService>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        throughput_calculator: Arc<ConsensusThroughputCalculator>,
        backpressure_manager: Arc<BackpressureManager>,
        congestion_log_config: Option<CongestionLogConfig>,
    ) -> Self {
        let congestion_logger =
            congestion_log_config.and_then(|config| match CongestionCommitLogger::new(&config) {
                Ok(logger) => Some(Arc::new(Mutex::new(logger))),
                Err(e) => {
                    debug_fatal!("Failed to create congestion logger: {e}");
                    None
                }
            });
        let consensus_gasless_counter = state.consensus_gasless_counter.clone();
        Self {
            state,
            checkpoint_service,
            epoch_store,
            throughput_calculator,
            backpressure_manager,
            congestion_logger,
            consensus_gasless_counter,
        }
    }

    #[cfg(test)]
    pub(crate) fn new_for_testing(
        state: Arc<AuthorityState>,
        checkpoint_service: Arc<CheckpointService>,
    ) -> Self {
        let backpressure_manager = BackpressureManager::new_for_tests();
        let consensus_gasless_counter = state.consensus_gasless_counter.clone();
        Self {
            state: state.clone(),
            checkpoint_service,
            epoch_store: state.epoch_store_for_testing().clone(),
            throughput_calculator: Arc::new(ConsensusThroughputCalculator::new(
                None,
                state.metrics.clone(),
            )),
            backpressure_manager,
            congestion_logger: None,
            consensus_gasless_counter,
        }
    }

    pub(crate) fn new_consensus_handler(&self) -> ConsensusHandler<CheckpointService> {
        let new_epoch_start_state = self.epoch_store.epoch_start_state();
        let consensus_committee = new_epoch_start_state.get_consensus_committee();

        let settlement_scheduler = SettlementScheduler::new(
            self.state.execution_scheduler().as_ref().clone(),
            self.state.get_transaction_cache_reader().clone(),
            self.state.metrics.clone(),
        );
        ConsensusHandler::new(
            self.epoch_store.clone(),
            self.checkpoint_service.clone(),
            settlement_scheduler,
            self.state.get_object_cache_reader().clone(),
            consensus_committee,
            self.state.metrics.clone(),
            self.throughput_calculator.clone(),
            self.backpressure_manager.subscribe(),
            self.state.traffic_controller.clone(),
            self.congestion_logger.clone(),
            self.consensus_gasless_counter.clone(),
            self.state.transaction_deny_config_manager().clone(),
        )
    }
}

mod additional_consensus_state {
    use std::marker::PhantomData;

    use consensus_core::CommitRef;
    use fastcrypto::hash::HashFunction as _;
    use sui_types::{crypto::DefaultHash, digests::Digest};

    use super::*;
    /// AdditionalConsensusState tracks any in-memory state that is retained by ConsensusHandler
    /// between consensus commits. Because of crash recovery, using such data is inherently risky.
    /// In order to do this safely, we must store data from a fixed number of previous commits.
    /// Then, at start-up, that same fixed number of already processed commits is replayed to
    /// reconstruct the state.
    ///
    /// To make sure that bugs in this process appear immediately, we record the digest of this
    /// state in ConsensusCommitPrologue, so that any deviation causes an immediate fork.
    #[derive(Serialize, Deserialize)]
    pub(super) struct AdditionalConsensusState {
        commit_interval_observer: CommitIntervalObserver,
    }

    impl AdditionalConsensusState {
        pub fn new(additional_consensus_state_window_size: u32) -> Self {
            Self {
                commit_interval_observer: CommitIntervalObserver::new(
                    additional_consensus_state_window_size,
                ),
            }
        }

        /// Update all internal state based on the new commit
        pub(crate) fn observe_commit(
            &mut self,
            protocol_config: &ProtocolConfig,
            epoch_start_time: u64,
            consensus_commit: &impl ConsensusCommitAPI,
        ) -> ConsensusCommitInfo {
            self.commit_interval_observer
                .observe_commit_time(consensus_commit);

            let estimated_commit_period = self
                .commit_interval_observer
                .commit_interval_estimate()
                .unwrap_or(Duration::from_millis(
                    protocol_config.min_checkpoint_interval_ms(),
                ));

            info!("estimated commit rate: {:?}", estimated_commit_period);

            self.commit_info_impl(
                epoch_start_time,
                consensus_commit,
                Some(estimated_commit_period),
            )
        }

        fn commit_info_impl(
            &self,
            epoch_start_time: u64,
            consensus_commit: &impl ConsensusCommitAPI,
            estimated_commit_period: Option<Duration>,
        ) -> ConsensusCommitInfo {
            let leader_author = consensus_commit.leader_author_index();
            let timestamp = consensus_commit.commit_timestamp_ms();

            let timestamp = if timestamp < epoch_start_time {
                error!(
                    "Unexpected commit timestamp {timestamp} less then epoch start time {epoch_start_time}, author {leader_author:?}"
                );
                epoch_start_time
            } else {
                timestamp
            };

            ConsensusCommitInfo {
                _phantom: PhantomData,
                round: consensus_commit.leader_round(),
                timestamp,
                leader_author,
                consensus_commit_ref: consensus_commit.commit_ref(),
                rejected_transactions_digest: consensus_commit.rejected_transactions_digest(),
                additional_state_digest: Some(self.digest()),
                estimated_commit_period,
                skip_consensus_commit_prologue_in_test: false,
            }
        }

        /// Get the digest of the current state.
        fn digest(&self) -> AdditionalConsensusStateDigest {
            let mut hash = DefaultHash::new();
            bcs::serialize_into(&mut hash, self).unwrap();
            AdditionalConsensusStateDigest::new(hash.finalize().into())
        }
    }

    pub struct ConsensusCommitInfo {
        // prevent public construction
        _phantom: PhantomData<()>,

        pub round: u64,
        pub timestamp: u64,
        pub leader_author: AuthorityIndex,
        pub consensus_commit_ref: CommitRef,
        pub rejected_transactions_digest: Digest,

        additional_state_digest: Option<AdditionalConsensusStateDigest>,
        estimated_commit_period: Option<Duration>,

        pub skip_consensus_commit_prologue_in_test: bool,
    }

    impl ConsensusCommitInfo {
        pub fn new_for_test(
            commit_round: u64,
            commit_timestamp: u64,
            estimated_commit_period: Option<Duration>,
            skip_consensus_commit_prologue_in_test: bool,
        ) -> Self {
            Self {
                _phantom: PhantomData,
                round: commit_round,
                timestamp: commit_timestamp,
                leader_author: 0,
                consensus_commit_ref: CommitRef::default(),
                rejected_transactions_digest: Digest::default(),
                additional_state_digest: Some(AdditionalConsensusStateDigest::ZERO),
                estimated_commit_period,
                skip_consensus_commit_prologue_in_test,
            }
        }

        pub fn new_for_congestion_test(
            commit_round: u64,
            commit_timestamp: u64,
            estimated_commit_period: Duration,
        ) -> Self {
            Self::new_for_test(
                commit_round,
                commit_timestamp,
                Some(estimated_commit_period),
                true,
            )
        }

        pub fn additional_state_digest(&self) -> AdditionalConsensusStateDigest {
            // this method cannot be called if stateless_commit_info is used
            self.additional_state_digest
                .expect("additional_state_digest is not available")
        }

        pub fn estimated_commit_period(&self) -> Duration {
            // this method cannot be called if stateless_commit_info is used
            self.estimated_commit_period
                .expect("estimated commit period is not available")
        }

        fn consensus_commit_digest(&self) -> ConsensusCommitDigest {
            ConsensusCommitDigest::new(self.consensus_commit_ref.digest.into_inner())
        }

        fn consensus_commit_prologue_v4_transaction(
            &self,
            epoch: u64,
            consensus_determined_version_assignments: ConsensusDeterminedVersionAssignments,
            additional_state_digest: AdditionalConsensusStateDigest,
        ) -> VerifiedExecutableTransaction {
            let transaction = VerifiedTransaction::new_consensus_commit_prologue_v4(
                epoch,
                self.round,
                self.timestamp,
                self.consensus_commit_digest(),
                consensus_determined_version_assignments,
                additional_state_digest,
            );
            VerifiedExecutableTransaction::new_system(transaction, epoch)
        }

        pub fn create_consensus_commit_prologue_transaction(
            &self,
            epoch: u64,
            cancelled_txn_version_assignment: Vec<(
                TransactionDigest,
                Vec<(ConsensusObjectSequenceKey, SequenceNumber)>,
            )>,
            indirect_state_observer: IndirectStateObserver,
        ) -> VerifiedExecutableTransaction {
            let version_assignments =
                ConsensusDeterminedVersionAssignments::CancelledTransactionsV2(
                    cancelled_txn_version_assignment,
                );
            let additional_state_digest =
                indirect_state_observer.fold_with(self.additional_state_digest());

            self.consensus_commit_prologue_v4_transaction(
                epoch,
                version_assignments,
                additional_state_digest,
            )
        }
    }

    #[derive(Default)]
    pub struct IndirectStateObserver {
        hash: DefaultHash,
    }

    impl IndirectStateObserver {
        pub fn new() -> Self {
            Self::default()
        }

        pub fn observe_indirect_state<T: Serialize>(&mut self, state: &T) {
            bcs::serialize_into(&mut self.hash, state).unwrap();
        }

        pub fn fold_with(
            self,
            d1: AdditionalConsensusStateDigest,
        ) -> AdditionalConsensusStateDigest {
            let hash = self.hash.finalize();
            let d2 = AdditionalConsensusStateDigest::new(hash.into());

            let mut hasher = DefaultHash::new();
            bcs::serialize_into(&mut hasher, &d1).unwrap();
            bcs::serialize_into(&mut hasher, &d2).unwrap();
            AdditionalConsensusStateDigest::new(hasher.finalize().into())
        }
    }

    #[test]
    fn test_additional_consensus_state() {
        use crate::consensus_test_utils::TestConsensusCommit;

        fn observe(state: &mut AdditionalConsensusState, round: u64, timestamp: u64) {
            let protocol_config = ProtocolConfig::get_for_max_version_UNSAFE();
            state.observe_commit(
                &protocol_config,
                100,
                &TestConsensusCommit::empty(round, timestamp, 0),
            );
        }

        let mut s1 = AdditionalConsensusState::new(3);
        observe(&mut s1, 1, 1000);
        observe(&mut s1, 2, 2000);
        observe(&mut s1, 3, 3000);
        observe(&mut s1, 4, 4000);

        let mut s2 = AdditionalConsensusState::new(3);
        // Because state uses a ring buffer, we should get the same digest
        // even though we only added the 3 latest observations.
        observe(&mut s2, 2, 2000);
        observe(&mut s2, 3, 3000);
        observe(&mut s2, 4, 4000);

        assert_eq!(s1.digest(), s2.digest());

        observe(&mut s1, 5, 5000);
        observe(&mut s2, 5, 5000);

        assert_eq!(s1.digest(), s2.digest());
    }
}
use additional_consensus_state::AdditionalConsensusState;
pub(crate) use additional_consensus_state::{ConsensusCommitInfo, IndirectStateObserver};

struct QueuedCheckpointRoots {
    roots: CheckpointRoots,
    timestamp: CheckpointTimestamp,
    consensus_commit_ref: CommitRef,
    rejected_transactions_digest: Digest,
}

struct Chunk<
    T: crate::authority::shared_object_version_manager::AsTx = VerifiedExecutableTransaction,
> {
    schedulables: Vec<Schedulable<T>>,
    settlement: Option<Schedulable<T>>,
    height: CheckpointHeight,
}

impl<T: crate::authority::shared_object_version_manager::AsTx + Clone> Chunk<T> {
    fn all_schedulables(&self) -> impl Iterator<Item = &Schedulable<T>> + Clone {
        self.schedulables.iter().chain(self.settlement.iter())
    }

    fn all_schedulables_from(chunks: &[Self]) -> impl Iterator<Item = &Schedulable<T>> + Clone {
        chunks.iter().flat_map(|c| c.all_schedulables())
    }

    fn to_checkpoint_roots(&self) -> CheckpointRoots {
        let tx_roots: Vec<_> = self.schedulables.iter().map(|s| s.key()).collect();
        let settlement_root = self.settlement.as_ref().map(|s| s.key());
        CheckpointRoots {
            tx_roots,
            settlement_root,
            height: self.height,
        }
    }
}

impl From<Chunk<VerifiedExecutableTransactionWithAliases>> for Chunk {
    fn from(chunk: Chunk<VerifiedExecutableTransactionWithAliases>) -> Self {
        Chunk {
            schedulables: chunk.schedulables.into_iter().map(|s| s.into()).collect(),
            settlement: chunk.settlement.map(|s| s.into()),
            height: chunk.height,
        }
    }
}

// Accumulates checkpoint roots from consensus and flushes them into PendingCheckpoints.
// Roots are buffered in `pending_roots` until a flush is triggered (by max_tx overflow or
// time interval). A flush always drains all buffered roots into a single PendingCheckpoint,
// so the queue only ever holds roots for one pending checkpoint at a time.
//
// Owns an ExecutionSchedulerSender so push_chunk can send schedulables and settlement info
// for execution in one shot.
pub(crate) struct CheckpointQueue {
    last_built_timestamp: CheckpointTimestamp,
    pending_roots: VecDeque<QueuedCheckpointRoots>,
    height: u64,
    pending_tx_count: usize,
    current_checkpoint_seq: CheckpointSequenceNumber,
    max_tx: usize,
    min_checkpoint_interval_ms: u64,
    execution_scheduler_sender: ExecutionSchedulerSender,
}

impl CheckpointQueue {
    pub(crate) fn new(
        last_built_timestamp: CheckpointTimestamp,
        checkpoint_height: u64,
        next_checkpoint_seq: CheckpointSequenceNumber,
        max_tx: usize,
        min_checkpoint_interval_ms: u64,
        execution_scheduler_sender: ExecutionSchedulerSender,
    ) -> Self {
        Self {
            last_built_timestamp,
            pending_roots: VecDeque::new(),
            height: checkpoint_height,
            pending_tx_count: 0,
            current_checkpoint_seq: next_checkpoint_seq,
            max_tx,
            min_checkpoint_interval_ms,
            execution_scheduler_sender,
        }
    }

    #[cfg(test)]
    fn new_for_testing(
        last_built_timestamp: CheckpointTimestamp,
        checkpoint_height: u64,
        next_checkpoint_seq: CheckpointSequenceNumber,
        max_tx: usize,
        min_checkpoint_interval_ms: u64,
    ) -> Self {
        let (sender, _receiver) = monitored_mpsc::unbounded_channel("test_checkpoint_queue_sender");
        Self {
            last_built_timestamp,
            pending_roots: VecDeque::new(),
            height: checkpoint_height,
            pending_tx_count: 0,
            current_checkpoint_seq: next_checkpoint_seq,
            max_tx,
            min_checkpoint_interval_ms,
            execution_scheduler_sender: ExecutionSchedulerSender::new_for_testing(sender),
        }
    }

    #[cfg(test)]
    fn new_for_testing_with_sender(
        last_built_timestamp: CheckpointTimestamp,
        checkpoint_height: u64,
        next_checkpoint_seq: CheckpointSequenceNumber,
        max_tx: usize,
        min_checkpoint_interval_ms: u64,
        sender: monitored_mpsc::UnboundedSender<SchedulerMessage>,
    ) -> Self {
        Self {
            last_built_timestamp,
            pending_roots: VecDeque::new(),
            height: checkpoint_height,
            pending_tx_count: 0,
            current_checkpoint_seq: next_checkpoint_seq,
            max_tx,
            min_checkpoint_interval_ms,
            execution_scheduler_sender: ExecutionSchedulerSender::new_for_testing(sender),
        }
    }

    pub(crate) fn last_built_timestamp(&self) -> CheckpointTimestamp {
        self.last_built_timestamp
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.pending_roots.is_empty()
    }

    fn next_height(&mut self) -> u64 {
        self.height += 1;
        self.height
    }

    fn push_chunk(
        &mut self,
        chunk: Chunk,
        assigned_versions: &HashMap<TransactionKey, AssignedVersions>,
        timestamp: CheckpointTimestamp,
        consensus_commit_ref: CommitRef,
        rejected_transactions_digest: Digest,
    ) -> Vec<PendingCheckpoint> {
        let max_tx = self.max_tx;
        let user_tx_count = chunk.schedulables.len();

        let roots = chunk.to_checkpoint_roots();

        let schedulables: Vec<_> = chunk
            .schedulables
            .into_iter()
            .map(|s| {
                let versions = assigned_versions
                    .get(&s.key())
                    .cloned()
                    .unwrap_or_else(AssignedVersions::empty);
                (s, versions)
            })
            .collect();

        let mut flushed_checkpoints = Vec::new();

        if self.pending_tx_count > 0
            && self.pending_tx_count + user_tx_count > max_tx
            && let Some(checkpoint) = self.flush_forced()
        {
            flushed_checkpoints.push(checkpoint);
        }

        let settlement_info = chunk.settlement.as_ref().map(|s| {
            let settlement_key = s.key();
            let tx_keys: Vec<_> = schedulables.iter().map(|(s, _)| s.key()).collect();
            SettlementBatchInfo {
                settlement_key,
                tx_keys,
                checkpoint_height: chunk.height,
                checkpoint_seq: self.current_checkpoint_seq,
                assigned_versions: assigned_versions
                    .get(&settlement_key)
                    .cloned()
                    .unwrap_or_else(AssignedVersions::empty),
            }
        });

        self.execution_scheduler_sender
            .send(schedulables, settlement_info);

        self.pending_tx_count += user_tx_count;
        self.pending_roots.push_back(QueuedCheckpointRoots {
            roots,
            timestamp,
            consensus_commit_ref,
            rejected_transactions_digest,
        });

        flushed_checkpoints
    }

    pub(crate) fn flush(
        &mut self,
        current_timestamp: CheckpointTimestamp,
        force: bool,
    ) -> Option<PendingCheckpoint> {
        if !force && current_timestamp < self.last_built_timestamp + self.min_checkpoint_interval_ms
        {
            return None;
        }
        self.flush_forced()
    }

    fn flush_forced(&mut self) -> Option<PendingCheckpoint> {
        if self.pending_roots.is_empty() {
            return None;
        }

        let to_flush: Vec<_> = self.pending_roots.drain(..).collect();
        let last_root = to_flush.last().unwrap();

        let checkpoint = PendingCheckpoint {
            roots: to_flush.iter().map(|q| q.roots.clone()).collect(),
            details: PendingCheckpointInfo {
                timestamp_ms: last_root.timestamp,
                last_of_epoch: false,
                checkpoint_height: last_root.roots.height,
                consensus_commit_ref: last_root.consensus_commit_ref,
                rejected_transactions_digest: last_root.rejected_transactions_digest,
                checkpoint_seq: self.current_checkpoint_seq,
            },
        };

        self.last_built_timestamp = last_root.timestamp;
        self.pending_tx_count = 0;
        self.current_checkpoint_seq += 1;

        Some(checkpoint)
    }

    pub(crate) fn checkpoint_seq(&self) -> CheckpointSequenceNumber {
        self.current_checkpoint_seq
            .checked_sub(1)
            .expect("checkpoint_seq called before any checkpoint was assigned")
    }
}

/// Runtime shutdown cancelled blocking work while a consensus commit was being processed.
#[derive(Debug)]
pub(crate) struct RuntimeShutdown;

pub struct ConsensusHandler<C> {
    /// A store created for each epoch. ConsensusHandler is recreated each epoch, with the
    /// corresponding store. This store is also used to get the current epoch ID.
    epoch_store: Arc<AuthorityPerEpochStore>,
    /// Holds the indices, hash and stats after the last consensus commit
    /// It is used for avoiding replaying already processed transactions,
    /// checking chain consistency, and accumulating per-epoch consensus output stats.
    last_consensus_stats: ExecutionIndicesWithStatsV2,
    checkpoint_service: Arc<C>,
    /// cache reader is needed when determining the next version to assign for shared objects.
    cache_reader: Arc<dyn ObjectCacheRead>,
    /// The consensus committee used to do stake computations for deciding set of low scoring authorities
    committee: ConsensusCommittee,
    // TODO: ConsensusHandler doesn't really share metrics with AuthorityState. We could define
    // a new metrics type here if we want to.
    metrics: Arc<AuthorityMetrics>,
    /// LRU cache of processed messages. Deduplication only checks whether a key is
    /// present. The value counts how many copies of the message this validator has
    /// seen, which feeds the local staggering signal. A count is lost on eviction or
    /// restart. The next copy then starts a fresh count, so it is not charged as excess.
    processed_cache: LruCache<SequencedConsensusTransactionKey, u64>,
    /// Using the throughput calculator to record the current consensus throughput
    throughput_calculator: Arc<ConsensusThroughputCalculator>,

    additional_consensus_state: AdditionalConsensusState,

    backpressure_subscriber: BackpressureSubscriber,

    traffic_controller: Option<Arc<TrafficController>>,

    congestion_logger: Option<Arc<Mutex<CongestionCommitLogger>>>,

    consensus_gasless_counter: Arc<ConsensusGaslessCounter>,

    transaction_deny_config_manager: Arc<TransactionDenyConfigManager>,

    checkpoint_queue: Mutex<CheckpointQueue>,
}

const PROCESSED_CACHE_CAP: usize = 1024 * 1024;

fn assert_supported_protocol_config(protocol_config: &ProtocolConfig) {
    assert!(
        matches!(
            protocol_config.per_object_congestion_control_mode(),
            PerObjectCongestionControlMode::ExecutionTimeEstimate(_)
        ),
        "support for congestion control modes other than PerObjectCongestionControlMode::ExecutionTimeEstimate has been removed"
    );
    assert!(
        protocol_config.split_checkpoints_in_consensus_handler(),
        "support for splitting checkpoints outside of consensus handler has been removed"
    );
    assert!(protocol_config.ignore_execution_time_observations_after_certs_closed());
    assert!(protocol_config.record_time_estimate_processed());
    assert!(protocol_config.prepend_prologue_tx_in_consensus_commit_in_checkpoints());
    assert!(protocol_config.consensus_checkpoint_signature_key_includes_digest());
    assert!(protocol_config.authority_capabilities_v2());
    assert!(protocol_config.cancel_for_failed_dkg_early());
    assert!(protocol_config.record_consensus_determined_version_assignments_in_prologue_v2());
    assert!(protocol_config.record_additional_state_digest_in_prologue());
    assert!(protocol_config.additional_consensus_digest_indirect_state());
    assert!(protocol_config.include_cancelled_randomness_txns_in_prologue());
    assert!(protocol_config.fix_checkpoint_signature_mapping());
    assert!(protocol_config.merge_randomness_into_checkpoint());
    assert!(
        protocol_config.timestamp_based_epoch_close(),
        "support for non-timestamp-based epoch close has been removed"
    );
}

impl<C> ConsensusHandler<C> {
    pub(crate) fn new(
        epoch_store: Arc<AuthorityPerEpochStore>,
        checkpoint_service: Arc<C>,
        settlement_scheduler: SettlementScheduler,
        cache_reader: Arc<dyn ObjectCacheRead>,
        committee: ConsensusCommittee,
        metrics: Arc<AuthorityMetrics>,
        throughput_calculator: Arc<ConsensusThroughputCalculator>,
        backpressure_subscriber: BackpressureSubscriber,
        traffic_controller: Option<Arc<TrafficController>>,
        congestion_logger: Option<Arc<Mutex<CongestionCommitLogger>>>,
        consensus_gasless_counter: Arc<ConsensusGaslessCounter>,
        transaction_deny_config_manager: Arc<TransactionDenyConfigManager>,
    ) -> Self {
        assert_supported_protocol_config(epoch_store.protocol_config());

        // Recover last_consensus_stats so it is consistent across validators.
        let mut last_consensus_stats = epoch_store
            .get_last_consensus_stats()
            .expect("Should be able to read last consensus index");
        // stats is empty at the beginning of epoch.
        if !last_consensus_stats.stats.is_initialized() {
            last_consensus_stats.stats = ConsensusStats::new(committee.size());
            last_consensus_stats.checkpoint_seq = epoch_store.previous_epoch_last_checkpoint();
        }
        let max_tx = epoch_store
            .protocol_config()
            .max_transactions_per_checkpoint() as usize;
        let min_checkpoint_interval_ms = epoch_store
            .protocol_config()
            .min_checkpoint_interval_ms_as_option()
            .unwrap_or_default();
        let execution_scheduler_sender =
            ExecutionSchedulerSender::start(settlement_scheduler, epoch_store.clone());
        let commit_rate_estimate_window_size = epoch_store
            .protocol_config()
            .get_consensus_commit_rate_estimation_window_size();
        let last_built_timestamp = last_consensus_stats.last_checkpoint_flush_timestamp;
        let checkpoint_height = last_consensus_stats.height;
        let next_checkpoint_seq = last_consensus_stats.checkpoint_seq + 1;
        Self {
            epoch_store,
            last_consensus_stats,
            checkpoint_service,
            cache_reader,
            committee,
            metrics,
            processed_cache: LruCache::new(
                NonZeroUsize::new(randomize_cache_capacity_in_tests(PROCESSED_CACHE_CAP)).unwrap(),
            ),
            throughput_calculator,
            additional_consensus_state: AdditionalConsensusState::new(
                commit_rate_estimate_window_size,
            ),
            backpressure_subscriber,
            traffic_controller,
            congestion_logger,
            consensus_gasless_counter,
            transaction_deny_config_manager,
            checkpoint_queue: Mutex::new(CheckpointQueue::new(
                last_built_timestamp,
                checkpoint_height,
                next_checkpoint_seq,
                max_tx,
                min_checkpoint_interval_ms,
                execution_scheduler_sender,
            )),
        }
    }

    /// Returns the last subdag index processed by the handler.
    pub(crate) fn last_processed_subdag_index(&self) -> u64 {
        self.last_consensus_stats.index.sub_dag_index
    }

    pub(crate) fn new_for_testing(
        epoch_store: Arc<AuthorityPerEpochStore>,
        checkpoint_service: Arc<C>,
        execution_scheduler_sender: ExecutionSchedulerSender,
        cache_reader: Arc<dyn ObjectCacheRead>,
        committee: ConsensusCommittee,
        metrics: Arc<AuthorityMetrics>,
        throughput_calculator: Arc<ConsensusThroughputCalculator>,
        backpressure_subscriber: BackpressureSubscriber,
        traffic_controller: Option<Arc<TrafficController>>,
        transaction_deny_config_manager: Arc<TransactionDenyConfigManager>,
        last_consensus_stats: ExecutionIndicesWithStatsV2,
    ) -> Self {
        assert_supported_protocol_config(epoch_store.protocol_config());

        let commit_rate_estimate_window_size = epoch_store
            .protocol_config()
            .get_consensus_commit_rate_estimation_window_size();
        let max_tx = epoch_store
            .protocol_config()
            .max_transactions_per_checkpoint() as usize;
        let min_checkpoint_interval_ms = epoch_store
            .protocol_config()
            .min_checkpoint_interval_ms_as_option()
            .unwrap_or_default();
        let last_built_timestamp = last_consensus_stats.last_checkpoint_flush_timestamp;
        let checkpoint_height = last_consensus_stats.height;
        Self {
            epoch_store,
            last_consensus_stats,
            checkpoint_service,
            cache_reader,
            committee,
            metrics,
            processed_cache: LruCache::new(
                NonZeroUsize::new(randomize_cache_capacity_in_tests(PROCESSED_CACHE_CAP)).unwrap(),
            ),
            throughput_calculator,
            additional_consensus_state: AdditionalConsensusState::new(
                commit_rate_estimate_window_size,
            ),
            backpressure_subscriber,
            traffic_controller,
            congestion_logger: None,
            consensus_gasless_counter: Arc::new(ConsensusGaslessCounter::default()),
            transaction_deny_config_manager,
            checkpoint_queue: Mutex::new(CheckpointQueue::new(
                last_built_timestamp,
                checkpoint_height,
                0,
                max_tx,
                min_checkpoint_interval_ms,
                execution_scheduler_sender,
            )),
        }
    }
}

#[derive(Default)]
struct CommitHandlerInput {
    user_transactions: Vec<VerifiedExecutableTransactionWithAliases>,
    capability_notifications: Vec<AuthorityCapabilitiesV2>,
    execution_time_observations: Vec<ExecutionTimeObservation>,
    checkpoint_signature_messages: Vec<CheckpointSignatureMessage>,
    randomness_dkg_messages: Vec<(AuthorityName, Vec<u8>)>,
    randomness_dkg_confirmations: Vec<(AuthorityName, Vec<u8>)>,
    end_of_publish_transactions: Vec<AuthorityName>,
    new_jwks: Vec<(AuthorityName, JwkId, JWK)>,
    transaction_deny_config_updates: Vec<(AuthorityName, SharedTransactionDenyConfig)>,
}

/// Copy counts from one commit for the staggered-submission duplication signal. They
/// are counted before deduplication, so duplicates that get discarded still count.
#[derive(Default)]
struct StaggeringCommitCounts {
    /// Copies beyond the number of proposers the transaction paid for. Only
    /// transactions that do not restrict their proposers are counted. This is the
    /// signal's numerator.
    excess_copies: u64,
    /// The number of distinct user transactions. This is the signal's denominator.
    unique_user_txns: u64,
    /// Duplicates the transaction paid for. The signal ignores them.
    paid_duplicate_copies: u64,
    /// Duplicates of transactions that restrict their proposers. The signal ignores them.
    restricted_duplicate_copies: u64,
}

impl StaggeringCommitCounts {
    /// Returns the counts keyed by their `staggered_submission_commit_copies` label.
    fn by_kind(&self) -> [(&'static str, u64); 4] {
        [
            ("unique", self.unique_user_txns),
            ("excess", self.excess_copies),
            ("paid_duplicate", self.paid_duplicate_copies),
            ("restricted_duplicate", self.restricted_duplicate_copies),
        ]
    }
}

struct CommitHandlerState {
    dkg_failed: bool,
    randomness_round: Option<RandomnessRound>,
    output: ConsensusCommitOutput,
    indirect_state_observer: Option<IndirectStateObserver>,
    initial_reconfig_state: ReconfigState,
    // Occurrence counts within this commit. Amplification deferral uses them, so they
    // must be identical on every validator.
    occurrence_counts: HashMap<TransactionDigest, u32>,
    staggering_counts: StaggeringCommitCounts,
    // Transactions involved in same commit owned object lock contention (double-spend),
    // mapped to conflict info (gas vs non-gas breakdown).
    contested_transaction_digests: HashMap<TransactionDigest, ConflictInfo>,
}

impl CommitHandlerState {
    fn new(epoch_store: &AuthorityPerEpochStore, consensus_round: u64) -> Self {
        Self {
            output: ConsensusCommitOutput::new(consensus_round),
            dkg_failed: false,
            randomness_round: None,
            indirect_state_observer: Some(IndirectStateObserver::new()),
            initial_reconfig_state: epoch_store.get_reconfig_state_read_lock_guard().clone(),
            occurrence_counts: HashMap::new(),
            staggering_counts: StaggeringCommitCounts::default(),
            contested_transaction_digests: HashMap::new(),
        }
    }

    fn get_notifications(&self) -> Vec<SequencedConsensusTransactionKey> {
        self.output
            .get_consensus_messages_processed()
            .cloned()
            .collect()
    }

    fn init_randomness<'a, 'epoch>(
        &'a mut self,
        epoch_store: &'epoch AuthorityPerEpochStore,
        commit_info: &'a ConsensusCommitInfo,
    ) -> Option<tokio::sync::MutexGuard<'epoch, RandomnessManager>> {
        let mut randomness_manager = epoch_store.randomness_manager.get().map(|rm| {
            rm.try_lock()
                .expect("should only ever be called from the commit handler thread")
        });

        let mut dkg_failed = false;
        let randomness_round = if epoch_store.randomness_state_enabled() {
            let randomness_manager = randomness_manager
                .as_mut()
                .expect("randomness manager should exist if randomness is enabled");
            match randomness_manager.dkg_status_for_commit_round(commit_info.round) {
                DkgStatus::Pending => None,
                DkgStatus::TimedOut => {
                    dkg_failed = true;
                    None
                }
                DkgStatus::Successful => {
                    // Generate randomness for this commit if DKG is successful and we are still
                    // accepting certs.
                    if self.initial_reconfig_state.should_accept_tx() {
                        randomness_manager
                            // TODO: make infallible
                            .reserve_next_randomness(commit_info.timestamp, &mut self.output)
                            .expect("epoch ended")
                    } else {
                        None
                    }
                }
            }
        } else {
            None
        };

        if randomness_round.is_some() {
            assert!(!dkg_failed); // invariant check
        }

        self.randomness_round = randomness_round;
        self.dkg_failed = dkg_failed;

        randomness_manager
    }
}

/// Deferred transactions abandoned because the epoch close deadline forced the epoch
/// closed while they were still unscheduled.
struct AbandonedDeferredTxns {
    count: usize,
    // At most 10 (key, digest) pairs for logging.
    sample: Vec<(DeferralKey, TransactionDigest)>,
}

impl<C: CheckpointServiceNotify + Send + Sync> ConsensusHandler<C> {
    /// Called during startup to allow us to observe commits we previously processed, for crash recovery.
    /// Any state computed here must be a pure function of the commits observed, it cannot depend on any
    /// state recorded in the epoch db.
    fn handle_prior_consensus_commit(&mut self, consensus_commit: impl ConsensusCommitAPI) {
        assert!(
            self.epoch_store
                .protocol_config()
                .record_additional_state_digest_in_prologue()
        );
        let protocol_config = self.epoch_store.protocol_config();
        let epoch_start_time = self
            .epoch_store
            .epoch_start_config()
            .epoch_start_timestamp_ms();

        self.additional_consensus_state.observe_commit(
            protocol_config,
            epoch_start_time,
            &consensus_commit,
        );
    }

    #[cfg(test)]
    pub(crate) async fn handle_consensus_commit_for_test(
        &mut self,
        consensus_commit: impl ConsensusCommitAPI,
    ) {
        let transactions = consensus_commit.transactions();
        self.handle_consensus_commit(consensus_commit, transactions)
            .await
            .expect("runtime shutdown during test commit");
    }

    /// Returns `Err(RuntimeShutdown)` if runtime shutdown cancelled work that the commit depends
    /// on. The commit's output is then not persisted, and it is replayed after restart.
    #[instrument(level = "debug", skip_all, fields(epoch = self.epoch_store.epoch(), round = consensus_commit.leader_round()))]
    pub(crate) async fn handle_consensus_commit(
        &mut self,
        consensus_commit: impl ConsensusCommitAPI,
        transactions: ParsedConsensusTransactions,
    ) -> Result<(), RuntimeShutdown> {
        // This may block until one of two conditions happens:
        // - Number of uncommitted transactions in the writeback cache goes below the
        //   backpressure threshold.
        // - The highest executed checkpoint catches up to the highest certified checkpoint.
        self.backpressure_subscriber.await_no_backpressure().await;

        let epoch = self.epoch_store.epoch();

        let _scope = monitored_scope("ConsensusCommitHandler::handle_consensus_commit");

        let last_committed_round = self.last_consensus_stats.index.last_committed_round;

        self.epoch_store
            .consensus_tx_status_cache
            .update_last_committed_leader_round(last_committed_round as u32);
        self.epoch_store
            .tx_reject_reason_cache
            .set_last_committed_leader_round(last_committed_round as u32);

        let commit_info = self.additional_consensus_state.observe_commit(
            self.epoch_store.protocol_config(),
            self.epoch_store
                .epoch_start_config()
                .epoch_start_timestamp_ms(),
            &consensus_commit,
        );
        assert!(commit_info.round > last_committed_round);

        let (timestamp, leader_author, commit_sub_dag_index) =
            self.gather_commit_metadata(&consensus_commit);

        info!(
            %consensus_commit,
            "Received consensus output {}. Rejected transactions {}",
            consensus_commit.commit_ref(),
            consensus_commit.rejected_transactions_debug_string(),
        );

        self.last_consensus_stats.index = ExecutionIndices {
            last_committed_round: commit_info.round,
            sub_dag_index: commit_sub_dag_index,
            transaction_index: 0_u64,
        };

        self.metrics
            .consensus_committed_subdags
            .with_label_values(&[&leader_author.to_string()])
            .inc();

        let mut state = CommitHandlerState::new(&self.epoch_store, commit_info.round);

        let FilteredConsensusOutput {
            transactions,
            owned_object_locks,
            dropped_transaction_keys,
            contested_transaction_digests,
        } = self.filter_consensus_txns(
            state.initial_reconfig_state.clone(),
            &commit_info,
            transactions,
        );
        state.contested_transaction_digests = contested_transaction_digests;
        // Buffer owned object locks for batch write.
        if !owned_object_locks.is_empty() {
            state.output.set_owned_object_locks(owned_object_locks);
        }

        // Still record the dropped transactions as consensus message processed.
        for key in dropped_transaction_keys {
            state.output.record_consensus_message_processed(
                SequencedConsensusTransactionKey::External(key),
            );
        }
        let transactions = self.deduplicate_consensus_txns(&mut state, &commit_info, transactions);

        let mut randomness_manager = state.init_randomness(&self.epoch_store, &commit_info);

        let CommitHandlerInput {
            user_transactions,
            capability_notifications,
            execution_time_observations,
            checkpoint_signature_messages,
            randomness_dkg_messages,
            randomness_dkg_confirmations,
            end_of_publish_transactions,
            new_jwks,
            transaction_deny_config_updates,
        } = self.build_commit_handler_input(transactions);

        self.process_gasless_transactions(&commit_info, &user_transactions);
        self.process_jwks(&mut state, &commit_info, new_jwks);
        self.process_capability_notifications(capability_notifications);
        self.process_transaction_deny_config_updates(transaction_deny_config_updates);
        self.process_execution_time_observations(&mut state, execution_time_observations);
        self.process_checkpoint_signature_messages(checkpoint_signature_messages);

        self.process_dkg_updates(
            &mut state,
            &commit_info,
            randomness_manager.as_deref_mut(),
            randomness_dkg_messages,
            randomness_dkg_confirmations,
        )
        .await?;

        let mut execution_time_estimator = self
            .epoch_store
            .execution_time_estimator
            .try_lock()
            .expect("should only ever be called from the commit handler thread");

        let authenticator_state_update_transaction =
            self.create_authenticator_state_update(last_committed_round, &commit_info);

        let (
            transactions_to_schedule,
            randomness_transactions_to_schedule,
            cancelled_txns,
            randomness_state_update_transaction,
        ) = self.collect_transactions_to_schedule(
            &mut state,
            &mut execution_time_estimator,
            &commit_info,
            user_transactions,
        );

        let (should_accept_tx, lock, final_round, abandoned_deferred_txns) =
            self.handle_close_epoch(&mut state, &commit_info, end_of_publish_transactions);

        let make_checkpoint = should_accept_tx || final_round;
        if !make_checkpoint {
            // No need for any further processing
            return Ok(());
        }

        // If this is the final round, record execution time observations for storage in the
        // end-of-epoch tx.
        if final_round {
            self.record_end_of_epoch_execution_time_observations(&mut execution_time_estimator);
        }

        let consensus_commit_prologue = (!commit_info.skip_consensus_commit_prologue_in_test)
            .then_some(Schedulable::ConsensusCommitPrologue(
                epoch,
                commit_info.round,
                commit_info.consensus_commit_ref.index,
            ));

        let schedulables: Vec<_> = itertools::chain!(
            consensus_commit_prologue.into_iter(),
            authenticator_state_update_transaction
                .into_iter()
                .map(Schedulable::Transaction),
            transactions_to_schedule
                .into_iter()
                .map(Schedulable::Transaction),
        )
        .collect();

        let randomness_schedulables: Vec<_> = randomness_state_update_transaction
            .into_iter()
            .chain(
                randomness_transactions_to_schedule
                    .into_iter()
                    .map(Schedulable::Transaction),
            )
            .collect();

        let num_schedulables = schedulables.len();
        let checkpoint_height = self.create_pending_checkpoints(
            &mut state,
            &commit_info,
            schedulables,
            randomness_schedulables,
            &cancelled_txns,
            final_round,
        );

        let notifications = state.get_notifications();

        let mut stats_to_record = self.last_consensus_stats.clone();
        stats_to_record.height = checkpoint_height;
        {
            let queue = self.checkpoint_queue.lock().unwrap();
            stats_to_record.last_checkpoint_flush_timestamp = queue.last_built_timestamp();
            stats_to_record.checkpoint_seq = queue.checkpoint_seq();
        }

        state.output.record_consensus_commit_stats(stats_to_record);

        self.record_deferral_deletion(&mut state);

        self.epoch_store
            .consensus_quarantine
            .write()
            .push_consensus_output(state.output, &self.epoch_store)
            .expect("push_consensus_output should not fail");

        debug!(
            ?commit_info.round,
            "Notifying checkpoint service about new pending checkpoint(s)",
        );
        self.checkpoint_service
            .notify_checkpoint()
            .expect("failed to notify checkpoint service");

        if let Some(randomness_round) = state.randomness_round {
            randomness_manager
                .as_ref()
                .expect("randomness manager should exist if randomness round is provided")
                .generate_randomness(epoch, randomness_round);
        }

        self.epoch_store.process_notifications(notifications.iter());

        // pass lock by value to ensure that it is held until this point
        self.log_final_round(lock, final_round);

        // Report abandoned transactions only after the commit output has been pushed
        // and all notifications delivered: in configurations where debug_fatal panics, firing
        // it mid-commit would abort processing of the very commit that closes the epoch.
        if let Some(AbandonedDeferredTxns { count, sample }) = abandoned_deferred_txns {
            self.metrics
                .consensus_handler_dropped_transactions
                .with_label_values(&["epoch_close_deadline"])
                .inc_by(count as u64);
            debug_fatal!(
                "Epoch close deadline reached with unscheduled deferred transactions: count={}, sample={:?}",
                count,
                sample
            );
        }

        // update the calculated throughput
        self.throughput_calculator
            .add_transactions(timestamp, num_schedulables as u64);

        fail_point_if!("correlated-crash-after-consensus-commit-boundary", || {
            let key = [commit_sub_dag_index, epoch];
            if sui_simulator::random::deterministic_probability_once(&key, 0.01) {
                sui_simulator::task::kill_current_node(None);
            }
        });

        fail_point!("crash");
        Ok(())
    }

    fn handle_close_epoch(
        &self,
        state: &mut CommitHandlerState,
        commit_info: &ConsensusCommitInfo,
        end_of_publish_transactions: Vec<AuthorityName>,
    ) -> (
        bool,
        Option<RwLockWriteGuard<'_, ReconfigState>>,
        bool,
        Option<AbandonedDeferredTxns>,
    ) {
        let timestamp_triggered =
            commit_info.timestamp >= self.epoch_store.next_reconfiguration_timestamp_ms();
        let deadline_reached = self
            .epoch_store
            .protocol_config()
            .epoch_close_deadline_ms_as_option()
            .is_some_and(|deadline_ms| {
                commit_info.timestamp
                    >= self
                        .epoch_store
                        .next_reconfiguration_timestamp_ms()
                        .saturating_add(deadline_ms)
            });
        let collected_eop_quorum =
            self.process_end_of_publish_transactions(state, end_of_publish_transactions);
        if timestamp_triggered || collected_eop_quorum {
            let (lock, final_round, abandoned_deferred_txns) =
                self.advance_end_of_epoch_state_machine(state, deadline_reached);
            (
                lock.should_accept_tx(),
                Some(lock),
                final_round,
                abandoned_deferred_txns,
            )
        } else {
            (true, None, false, None)
        }
    }

    fn record_end_of_epoch_execution_time_observations(
        &self,
        estimator: &mut ExecutionTimeEstimator,
    ) {
        self.epoch_store
            .end_of_epoch_execution_time_observations
            .set(estimator.take_observations())
            .expect("`stored_execution_time_observations` should only be set once at end of epoch");
    }

    fn record_deferral_deletion(&self, state: &mut CommitHandlerState) {
        let mut deferred_transactions = self
            .epoch_store
            .consensus_output_cache
            .deferred_transactions
            .lock();
        for deleted_deferred_key in state.output.get_deleted_deferred_txn_keys() {
            deferred_transactions.remove(&deleted_deferred_key);
        }
    }

    fn log_final_round(&self, lock: Option<RwLockWriteGuard<ReconfigState>>, final_round: bool) {
        if final_round {
            let epoch = self.epoch_store.epoch();
            info!(
                ?epoch,
                lock=?lock.as_ref(),
                final_round=?final_round,
                "Notified last checkpoint"
            );
            self.epoch_store.record_end_of_message_quorum_time_metric();
        }
    }

    #[allow(clippy::type_complexity)]
    fn collect_transactions_to_schedule(
        &self,
        state: &mut CommitHandlerState,
        execution_time_estimator: &mut ExecutionTimeEstimator,
        commit_info: &ConsensusCommitInfo,
        user_transactions: Vec<VerifiedExecutableTransactionWithAliases>,
    ) -> (
        Vec<VerifiedExecutableTransactionWithAliases>,
        Vec<VerifiedExecutableTransactionWithAliases>,
        BTreeMap<TransactionDigest, CancelConsensusCertificateReason>,
        Option<Schedulable<VerifiedExecutableTransactionWithAliases>>,
    ) {
        let _scope = monitored_scope("ConsensusCommitHandler::collect_transactions_to_schedule");
        let protocol_config = self.epoch_store.protocol_config();
        let epoch = self.epoch_store.epoch();

        let (ordered_txns, ordered_randomness_txns, previously_deferred_tx_digests) =
            self.merge_and_reorder_transactions(state, commit_info, user_transactions);

        let mut shared_object_congestion_tracker =
            self.init_congestion_tracker(commit_info, false, &ordered_txns);
        let mut shared_object_using_randomness_congestion_tracker =
            self.init_congestion_tracker(commit_info, true, &ordered_randomness_txns);

        let randomness_state_update_transaction = state
            .randomness_round
            .map(|round| Schedulable::RandomnessStateUpdate(epoch, round));
        debug!(
            "Randomness state update transaction: {:?}",
            randomness_state_update_transaction
                .as_ref()
                .map(|t| t.key())
        );

        let mut transactions_to_schedule = Vec::with_capacity(ordered_txns.len());
        let mut randomness_transactions_to_schedule =
            Vec::with_capacity(ordered_randomness_txns.len());
        let mut deferred_txns = BTreeMap::new();
        let mut cancelled_txns = BTreeMap::new();

        self.record_duplication_signal(state);

        for transaction in ordered_txns {
            self.handle_deferral_and_cancellation(
                state,
                &mut cancelled_txns,
                &mut deferred_txns,
                &mut transactions_to_schedule,
                protocol_config,
                commit_info,
                transaction,
                &mut shared_object_congestion_tracker,
                &previously_deferred_tx_digests,
                execution_time_estimator,
            );
        }

        for transaction in ordered_randomness_txns {
            if state.dkg_failed {
                debug!(
                    "Canceling randomness-using transaction {:?} because DKG failed",
                    transaction.tx().digest(),
                );
                cancelled_txns.insert(
                    *transaction.tx().digest(),
                    CancelConsensusCertificateReason::DkgFailed,
                );
                randomness_transactions_to_schedule.push(transaction);
                continue;
            }
            self.handle_deferral_and_cancellation(
                state,
                &mut cancelled_txns,
                &mut deferred_txns,
                &mut randomness_transactions_to_schedule,
                protocol_config,
                commit_info,
                transaction,
                &mut shared_object_using_randomness_congestion_tracker,
                &previously_deferred_tx_digests,
                execution_time_estimator,
            );
        }

        let mut total_deferred_txns = 0;
        {
            let mut deferred_transactions = self
                .epoch_store
                .consensus_output_cache
                .deferred_transactions
                .lock();
            for (key, txns) in deferred_txns.into_iter() {
                total_deferred_txns += txns.len();
                // Keys reloaded by this commit are still present in the map here
                // (record_deferral_deletion runs after this function), but no key
                // inserted here can equal one of them: reloads cover
                // future_round <= round while re-deferrals use round + 1, and a
                // randomness reload and a randomness re-deferral are mutually exclusive
                // per commit. A re-deferral does keep its original deferred_from_round,
                // though: a randomness-using transaction deferred at round F by a check
                // that precedes the randomness check (unpaid amplification, owned-object
                // double spend) gets ConsensusRound{F + 1, F}, and when reloaded at
                // F + 1 without randomness it lands on Randomness{F} - the key still
                // holding round F's fresh randomness deferrals. Overwriting that entry
                // (in memory and in the write batch below) strands the displaced
                // transactions: finalized but never reloaded or executed this epoch,
                // with their owned inputs locked until epoch end.
                debug_assert!(
                    !state
                        .output
                        .get_deleted_deferred_txn_keys()
                        .any(|deleted| deleted == key),
                    "deferral key {key:?} was reloaded by this commit and must not be re-inserted"
                );
                let txns = if protocol_config.merge_colliding_deferrals() {
                    match deferred_transactions.remove(&key) {
                        Some(mut merged) => {
                            assert_reachable_gated!(
                                "Merged colliding deferral entries instead of displacing finalized transactions.",
                                |pc| pc.merge_colliding_deferrals()
                                    && (pc.defer_owned_object_double_spend()
                                        || pc.defer_unpaid_amplification())
                            );
                            debug_assert!(
                                {
                                    let prev_digests: HashSet<_> =
                                        merged.iter().map(|t| *t.tx().digest()).collect();
                                    txns.iter().all(|t| !prev_digests.contains(t.tx().digest()))
                                },
                                "colliding deferral entries must not share transactions"
                            );
                            // Deterministic merge: previously deferred transactions keep
                            // their position ahead of this commit's.
                            merged.extend(txns);
                            merged
                        }
                        None => txns,
                    }
                } else {
                    if let Some(prev) = deferred_transactions.get(&key) {
                        // Last-writer-wins semantics must be preserved bit for bit for
                        // protocol versions without the fix; flag the stranding.
                        let new_digests: HashSet<_> =
                            txns.iter().map(|t| *t.tx().digest()).collect();
                        let displaced: Vec<_> = prev
                            .iter()
                            .map(|t| *t.tx().digest())
                            .filter(|d| !new_digests.contains(d))
                            .collect();
                        if !displaced.is_empty() {
                            debug_fatal!(
                                "Deferral key collision displaced finalized transactions: key {:?}, displaced {:?}",
                                key,
                                displaced
                            );
                        }
                    }
                    txns
                };
                deferred_transactions.insert(key, txns.clone());
                state.output.defer_transactions(key, txns);
            }
        }

        self.metrics
            .consensus_handler_deferred_transactions
            .inc_by(total_deferred_txns as u64);
        self.metrics
            .consensus_handler_cancelled_transactions
            .inc_by(cancelled_txns.len() as u64);
        self.metrics
            .consensus_handler_max_object_costs
            .with_label_values(&["regular_commit"])
            .set(shared_object_congestion_tracker.max_cost() as i64);
        self.metrics
            .consensus_handler_max_object_costs
            .with_label_values(&["randomness_commit"])
            .set(shared_object_using_randomness_congestion_tracker.max_cost() as i64);

        let congestion_commit_data = shared_object_congestion_tracker.finish_commit(commit_info);
        let randomness_congestion_commit_data =
            shared_object_using_randomness_congestion_tracker.finish_commit(commit_info);

        if let Some(logger) = &self.congestion_logger {
            let epoch = self.epoch_store.epoch();
            let mut logger = logger.lock().unwrap();
            logger.write_commit_log(epoch, commit_info, false, &congestion_commit_data);
            logger.write_commit_log(epoch, commit_info, true, &randomness_congestion_commit_data);
        }

        if let Some(tx_object_debts) = self.epoch_store.tx_object_debts.get()
            && let Err(e) = tx_object_debts.try_send(
                congestion_commit_data
                    .accumulated_debts
                    .iter()
                    .chain(randomness_congestion_commit_data.accumulated_debts.iter())
                    .map(|(id, _)| *id)
                    .collect(),
            )
        {
            info!("failed to send updated object debts to ExecutionTimeObserver: {e:?}");
        }

        state
            .output
            .set_congestion_control_object_debts(congestion_commit_data.accumulated_debts);
        state.output.set_congestion_control_randomness_object_debts(
            randomness_congestion_commit_data.accumulated_debts,
        );

        (
            transactions_to_schedule,
            randomness_transactions_to_schedule,
            cancelled_txns,
            randomness_state_update_transaction,
        )
    }

    #[allow(clippy::type_complexity)]
    fn create_pending_checkpoints(
        &self,
        state: &mut CommitHandlerState,
        commit_info: &ConsensusCommitInfo,
        schedulables: Vec<Schedulable<VerifiedExecutableTransactionWithAliases>>,
        randomness_schedulables: Vec<Schedulable<VerifiedExecutableTransactionWithAliases>>,
        cancelled_txns: &BTreeMap<TransactionDigest, CancelConsensusCertificateReason>,
        final_round: bool,
    ) -> CheckpointHeight {
        let _scope = monitored_scope("ConsensusCommitHandler::create_pending_checkpoints");
        let protocol_config = self.epoch_store.protocol_config();
        let epoch = self.epoch_store.epoch();
        let accumulators_enabled = self.epoch_store.accumulators_enabled();
        let max_transactions_per_checkpoint =
            protocol_config.max_transactions_per_checkpoint() as usize;

        let should_write_random_checkpoint = state.randomness_round.is_some()
            || (state.dkg_failed && !randomness_schedulables.is_empty());

        let mut checkpoint_queue = self.checkpoint_queue.lock().unwrap();

        let build_chunks =
            |schedulables: Vec<Schedulable<VerifiedExecutableTransactionWithAliases>>,
             queue: &mut CheckpointQueue|
             -> Vec<Chunk<VerifiedExecutableTransactionWithAliases>> {
                schedulables
                    .chunks(max_transactions_per_checkpoint)
                    .map(|chunk| {
                        let height = queue.next_height();
                        let schedulables: Vec<_> = chunk.to_vec();
                        let settlement = if accumulators_enabled {
                            Some(Schedulable::AccumulatorSettlement(epoch, height))
                        } else {
                            None
                        };
                        Chunk {
                            schedulables,
                            settlement,
                            height,
                        }
                    })
                    .collect()
            };

        let num_schedulables = schedulables.len();
        let chunked_schedulables = build_chunks(schedulables, &mut checkpoint_queue);
        if chunked_schedulables.len() > 1 {
            info!(
                "Splitting transactions into {} checkpoint chunks (num_schedulables={}, max_tx={})",
                chunked_schedulables.len(),
                num_schedulables,
                max_transactions_per_checkpoint
            );
            assert_reachable!("checkpoint split due to transaction limit");
        }
        let chunked_randomness_schedulables = if should_write_random_checkpoint {
            build_chunks(randomness_schedulables, &mut checkpoint_queue)
        } else {
            vec![]
        };

        let schedulables_for_version_assignment =
            Chunk::all_schedulables_from(&chunked_schedulables);
        let randomness_schedulables_for_version_assignment =
            Chunk::all_schedulables_from(&chunked_randomness_schedulables);

        let assigned_versions = self
            .epoch_store
            .process_consensus_transaction_shared_object_versions(
                self.cache_reader.as_ref(),
                schedulables_for_version_assignment,
                randomness_schedulables_for_version_assignment,
                cancelled_txns,
                &mut state.output,
            )
            .expect("failed to assign shared object versions");

        let consensus_commit_prologue =
            self.add_consensus_commit_prologue_transaction(state, commit_info, &assigned_versions);

        let mut chunked_schedulables = chunked_schedulables;
        let mut assigned_versions = assigned_versions;
        if let Some(consensus_commit_prologue) = consensus_commit_prologue {
            assert!(matches!(
                chunked_schedulables[0].schedulables[0],
                Schedulable::ConsensusCommitPrologue(..)
            ));
            assert!(matches!(
                assigned_versions.0[0].0,
                TransactionKey::ConsensusCommitPrologue(..)
            ));
            assigned_versions.0[0].0 =
                TransactionKey::Digest(*consensus_commit_prologue.tx().digest());
            chunked_schedulables[0].schedulables[0] =
                Schedulable::Transaction(consensus_commit_prologue);
        }

        let assigned_versions = assigned_versions.into_map();

        self.epoch_store.process_user_signatures(
            chunked_schedulables
                .iter()
                .flat_map(|c| c.all_schedulables())
                .chain(
                    chunked_randomness_schedulables
                        .iter()
                        .flat_map(|c| c.all_schedulables()),
                ),
        );

        let commit_height = chunked_randomness_schedulables
            .last()
            .or(chunked_schedulables.last())
            .map(|c| c.height)
            .expect("at least one checkpoint root must be created per commit");

        let mut pending_checkpoints = Vec::new();
        for chunk in chunked_schedulables {
            pending_checkpoints.extend(checkpoint_queue.push_chunk(
                chunk.into(),
                &assigned_versions,
                commit_info.timestamp,
                commit_info.consensus_commit_ref,
                commit_info.rejected_transactions_digest,
            ));
        }

        // We don't want to block checkpoint formation for non-randomness schedulables
        // on randomness state update. Therefore, we include randomness chunks in the
        // subsequent checkpoint. First we flush the queue, then enqueue randomness
        // for merging into the subsequent commit.
        pending_checkpoints.extend(checkpoint_queue.flush(commit_info.timestamp, final_round));

        if should_write_random_checkpoint {
            for chunk in chunked_randomness_schedulables {
                pending_checkpoints.extend(checkpoint_queue.push_chunk(
                    chunk.into(),
                    &assigned_versions,
                    commit_info.timestamp,
                    commit_info.consensus_commit_ref,
                    commit_info.rejected_transactions_digest,
                ));
            }
            if final_round {
                pending_checkpoints.extend(checkpoint_queue.flush(commit_info.timestamp, true));
            }
        }

        if final_round && let Some(last) = pending_checkpoints.last_mut() {
            last.details.last_of_epoch = true;
        }

        let queue_drained = checkpoint_queue.is_empty();
        drop(checkpoint_queue);

        for pending_checkpoint in pending_checkpoints {
            debug!(
                checkpoint_height = pending_checkpoint.details.checkpoint_height,
                roots_count = pending_checkpoint.num_roots(),
                "Writing pending checkpoint",
            );
            self.epoch_store
                .write_pending_checkpoint(&mut state.output, &pending_checkpoint);
        }

        state.output.set_checkpoint_queue_drained(queue_drained);

        commit_height
    }

    // Adds the consensus commit prologue transaction to the beginning of input `transactions` to update
    // the system clock used in all transactions in the current consensus commit.
    // Returns the root of the consensus commit prologue transaction if it was added to the input.
    fn add_consensus_commit_prologue_transaction<'a>(
        &'a self,
        state: &'a mut CommitHandlerState,
        commit_info: &'a ConsensusCommitInfo,
        assigned_versions: &AssignedTxAndVersions,
    ) -> Option<VerifiedExecutableTransactionWithAliases> {
        {
            if commit_info.skip_consensus_commit_prologue_in_test {
                return None;
            }
        }

        let mut cancelled_txn_version_assignment = Vec::new();

        for (txn_key, assigned_versions) in assigned_versions.0.iter() {
            let Some(d) = txn_key.as_digest() else {
                continue;
            };

            if assigned_versions
                .shared_object_versions
                .iter()
                .any(|(_, version)| version.is_cancelled())
            {
                assert_reachable!("cancelled transactions");
                cancelled_txn_version_assignment
                    .push((*d, assigned_versions.shared_object_versions.clone()));
            }
        }

        fail_point_arg!(
            "additional_cancelled_txns_for_tests",
            |additional_cancelled_txns: Vec<(
                TransactionDigest,
                Vec<(ConsensusObjectSequenceKey, SequenceNumber)>
            )>| {
                cancelled_txn_version_assignment.extend(additional_cancelled_txns);
            }
        );

        let transaction = commit_info.create_consensus_commit_prologue_transaction(
            self.epoch_store.epoch(),
            cancelled_txn_version_assignment,
            state.indirect_state_observer.take().unwrap(),
        );
        Some(VerifiedExecutableTransactionWithAliases::no_aliases(
            transaction,
        ))
    }

    /// Passes this commit's copy counts to the staggered-submission signal and updates
    /// the signal metrics.
    ///
    /// The signal measures copies, not transactions, because copies are what waste
    /// bandwidth. A copy counts as excess when the transaction has already reached
    /// the number of proposers it paid for, or the free minimum if that is larger.
    /// This count carries across commits while the transaction stays in the processed
    /// cache. Copies are counted on arrival, so a deferred transaction counts as unique
    /// only once.
    ///
    /// The signal always runs. It changes staggering only when both the protocol flag
    /// and the node config enable it. Otherwise the metrics still show what it would do.
    fn record_duplication_signal(&self, state: &CommitHandlerState) {
        self.metrics
            .staggered_submission_excess_copies
            .observe(state.staggering_counts.excess_copies as f64);
        // These counters let dashboards recompute the ratio over any time window.
        for (kind, copies) in state.staggering_counts.by_kind() {
            self.metrics
                .staggered_submission_commit_copies
                .with_label_values(&[kind])
                .inc_by(copies);
        }
        let staggered = self.epoch_store.staggered_submission();
        let outcome = staggered.record_commit(
            state.staggering_counts.excess_copies,
            state.staggering_counts.unique_user_txns,
        );
        let transition = outcome.transition;
        self.metrics
            .staggered_submission_duplication_ratio
            .set(outcome.duplication_ratio);
        // The gauges are set on every commit so they reset when a new epoch starts.
        self.metrics
            .staggered_submission_signal_band
            .set(staggered.signal_band() as i64);
        self.metrics
            .staggered_submission_active
            .set(staggered.is_active() as i64);
        if let Some(band) = transition {
            self.metrics
                .staggered_submission_signal_transitions
                .with_label_values(&[crate::staggered_submission::signal_band_label(band)])
                .inc();
            self.metrics
                .staggered_submission_band_duration_commits
                .with_label_values(&[crate::staggered_submission::signal_band_label(
                    outcome.previous_band,
                )])
                .observe(outcome.commits_in_previous_band as f64);
        }
    }

    fn handle_deferral_and_cancellation(
        &self,
        state: &mut CommitHandlerState,
        cancelled_txns: &mut BTreeMap<TransactionDigest, CancelConsensusCertificateReason>,
        deferred_txns: &mut BTreeMap<DeferralKey, Vec<VerifiedExecutableTransactionWithAliases>>,
        scheduled_txns: &mut Vec<VerifiedExecutableTransactionWithAliases>,
        protocol_config: &ProtocolConfig,
        commit_info: &ConsensusCommitInfo,
        transaction: VerifiedExecutableTransactionWithAliases,
        shared_object_congestion_tracker: &mut SharedObjectCongestionTracker,
        previously_deferred_tx_digests: &HashMap<TransactionDigest, DeferralKey>,
        execution_time_estimator: &ExecutionTimeEstimator,
    ) {
        let tx_digest = *transaction.tx().digest();

        // Check for unpaid amplification before other deferral checks.
        // SIP-45: the gas price pays for `paid_proposer_count` submissions.
        // Transactions with more duplicates than paid for are deferred.
        //
        // A transaction that names its proposers has already had its amplification bounded by
        // validity_check, which sizes the proposer set against the gas price, so there is nothing
        // left to charge for here. A set recorded for another epoch is ignored and so bounds
        // nothing, leaving the transaction subject to deferral like any other.
        let restricts_proposers = transaction
            .tx()
            .transaction_data()
            .expiration()
            .restricts_proposers(self.epoch_store.epoch());
        if restricts_proposers {
            assert_reachable!("committed a proposer-restricted transaction");
        }
        if protocol_config.defer_unpaid_amplification() && !restricts_proposers {
            let occurrence_count = state
                .occurrence_counts
                .get(&tx_digest)
                .copied()
                .unwrap_or(0);

            let allowed_count = paid_proposer_count(
                transaction.tx().transaction_data().gas_price(),
                self.epoch_store.reference_gas_price(),
            );

            if occurrence_count as u64 > allowed_count {
                self.metrics
                    .consensus_handler_unpaid_amplification_deferrals
                    .inc();

                let deferred_from_round = previously_deferred_tx_digests
                    .get(&tx_digest)
                    .map(|k| k.deferred_from_round())
                    .unwrap_or(commit_info.round);

                let deferral_key = DeferralKey::new_for_consensus_round(
                    commit_info.round + 1,
                    deferred_from_round,
                );

                if transaction_deferral_within_limit(
                    &deferral_key,
                    protocol_config.max_deferral_rounds_for_congestion_control(),
                ) {
                    debug!(
                        "Deferring transaction {:?} due to unpaid amplification (count={}, allowed={})",
                        tx_digest, occurrence_count, allowed_count
                    );
                    deferred_txns
                        .entry(deferral_key)
                        .or_default()
                        .push(transaction);
                    return;
                }
            }
        }

        // Check for owned object double-spend: if this transaction won an owned object
        // lock while another transaction in the same commit tried to lock the same object,
        // defer it as a penalty.
        if let Some(conflict_info) = state.contested_transaction_digests.get(&tx_digest) {
            self.metrics.consensus_handler_double_spend_deferrals.inc();
            self.metrics
                .consensus_handler_double_spend_conflict_count
                .with_label_values(&["gas_object"])
                .observe(conflict_info.gas_object_conflicts as f64);
            self.metrics
                .consensus_handler_double_spend_conflict_count
                .with_label_values(&["non_gas_object"])
                .observe(conflict_info.non_gas_object_conflicts as f64);
            // Attribute the winning side of the conflict to the authority that sequenced the
            // contested holder transaction.
            self.metrics
                .consensus_handler_double_spend_conflicting_authority
                .with_label_values(&[
                    self.authority_hostname(conflict_info.winner_author),
                    "winner",
                ])
                .inc();

            if protocol_config.defer_owned_object_double_spend() {
                let deferred_from_round = previously_deferred_tx_digests
                    .get(&tx_digest)
                    .map(|k| k.deferred_from_round())
                    .unwrap_or(commit_info.round);

                let deferral_key = DeferralKey::new_for_consensus_round(
                    commit_info.round + 1,
                    deferred_from_round,
                );

                if transaction_deferral_within_limit(
                    &deferral_key,
                    protocol_config.max_deferral_rounds_for_congestion_control(),
                ) {
                    debug!(
                        "Deferring transaction {:?} due to owned object double-spend contention \
                        (gas_conflicts={}, non_gas_conflicts={})",
                        tx_digest,
                        conflict_info.gas_object_conflicts,
                        conflict_info.non_gas_object_conflicts,
                    );
                    assert_reachable_gated!(
                        "Successfully deferred transaction attempting to double spend owned object.",
                        |pc| pc.defer_owned_object_double_spend()
                    );
                    if transaction.tx().transaction_data().uses_randomness()
                        && state.randomness_round.is_none()
                    {
                        // Precondition for the deferral-key collision flagged in
                        // collect_transactions_to_schedule: at the next commit without
                        // randomness this transaction re-defers to Randomness{original
                        // round}, the key that round's fresh randomness deferrals are
                        // stored under.
                        assert_reachable_gated!(
                            "Double-spend deferred a randomness-using transaction at a round without randomness.",
                            |pc| pc.defer_owned_object_double_spend()
                        );
                    }
                    deferred_txns
                        .entry(deferral_key)
                        .or_default()
                        .push(transaction);
                    return;
                }
            }
        }

        let tx_cost = shared_object_congestion_tracker.get_tx_cost(
            execution_time_estimator,
            transaction.tx(),
            state.indirect_state_observer.as_mut().unwrap(),
        );

        let deferral_info = self.epoch_store.should_defer(
            transaction.tx(),
            commit_info,
            state.dkg_failed,
            state.randomness_round.is_some(),
            previously_deferred_tx_digests,
            shared_object_congestion_tracker,
        );

        if let Some((deferral_key, deferral_reason)) = deferral_info {
            debug!(
                "Deferring consensus certificate for transaction {:?} until {:?}",
                tx_digest, deferral_key
            );

            match deferral_reason {
                DeferralReason::RandomnessNotReady => {
                    deferred_txns
                        .entry(deferral_key)
                        .or_default()
                        .push(transaction);
                }
                DeferralReason::SharedObjectCongestion(congested_objects) => {
                    self.metrics.consensus_handler_congested_transactions.inc();
                    if transaction_deferral_within_limit(
                        &deferral_key,
                        protocol_config.max_deferral_rounds_for_congestion_control(),
                    ) {
                        deferred_txns
                            .entry(deferral_key)
                            .or_default()
                            .push(transaction);
                    } else {
                        assert_sometimes!(
                            transaction.tx().data().transaction_data().uses_randomness(),
                            "cancelled randomness-using transaction"
                        );
                        assert_sometimes!(
                            !transaction.tx().data().transaction_data().uses_randomness(),
                            "cancelled non-randomness-using transaction"
                        );

                        // Cancel the transaction that has been deferred for too long.
                        debug!(
                            "Cancelling consensus transaction {:?} with deferral key {:?} due to congestion on objects {:?}",
                            tx_digest, deferral_key, congested_objects
                        );
                        cancelled_txns.insert(
                            tx_digest,
                            CancelConsensusCertificateReason::CongestionOnObjects(
                                congested_objects,
                            ),
                        );
                        scheduled_txns.push(transaction);
                    }
                }
            }
        } else {
            // Update object execution cost for all scheduled transactions
            shared_object_congestion_tracker.bump_object_execution_cost(tx_cost, transaction.tx());
            scheduled_txns.push(transaction);
        }
    }

    fn merge_and_reorder_transactions(
        &self,
        state: &mut CommitHandlerState,
        commit_info: &ConsensusCommitInfo,
        user_transactions: Vec<VerifiedExecutableTransactionWithAliases>,
    ) -> (
        Vec<VerifiedExecutableTransactionWithAliases>,
        Vec<VerifiedExecutableTransactionWithAliases>,
        HashMap<TransactionDigest, DeferralKey>,
    ) {
        let protocol_config = self.epoch_store.protocol_config();

        let (mut txns, mut randomness_txns, previously_deferred_tx_digests) =
            self.load_deferred_transactions(state, commit_info);

        txns.reserve(user_transactions.len());
        randomness_txns.reserve(user_transactions.len());

        // There may be randomness transactions in `txns`, which were deferred due to congestion.
        // They must be placed back into `randomness_txns`.
        let mut txns: Vec<_> = txns
            .into_iter()
            .filter_map(|tx| {
                if tx.tx().transaction_data().uses_randomness() {
                    randomness_txns.push(tx);
                    None
                } else {
                    Some(tx)
                }
            })
            .collect();

        for txn in user_transactions {
            if txn.tx().transaction_data().uses_randomness() {
                randomness_txns.push(txn);
            } else {
                txns.push(txn);
            }
        }

        PostConsensusTxReorder::reorder(
            &mut txns,
            protocol_config.consensus_transaction_ordering(),
        );
        PostConsensusTxReorder::reorder(
            &mut randomness_txns,
            protocol_config.consensus_transaction_ordering(),
        );

        (txns, randomness_txns, previously_deferred_tx_digests)
    }

    fn load_deferred_transactions(
        &self,
        state: &mut CommitHandlerState,
        commit_info: &ConsensusCommitInfo,
    ) -> (
        Vec<VerifiedExecutableTransactionWithAliases>,
        Vec<VerifiedExecutableTransactionWithAliases>,
        HashMap<TransactionDigest, DeferralKey>,
    ) {
        let mut previously_deferred_tx_digests = HashMap::new();

        let deferred_txs: Vec<_> = self
            .epoch_store
            .load_deferred_transactions_for_up_to_consensus_round_v2(
                &mut state.output,
                commit_info.round,
            )
            .into_iter()
            .flat_map(|(key, txns)| txns.into_iter().map(move |tx| (key, tx)))
            .map(|(key, tx)| {
                previously_deferred_tx_digests.insert(*tx.tx().digest(), key);
                tx
            })
            .collect();
        trace!(
            "loading deferred transactions: {:?}",
            deferred_txs.iter().map(|tx| tx.tx().digest())
        );

        let deferred_randomness_txs = if state.dkg_failed || state.randomness_round.is_some() {
            let txns: Vec<_> = self
                .epoch_store
                .load_deferred_transactions_for_randomness_v2(&mut state.output)
                .into_iter()
                .flat_map(|(key, txns)| txns.into_iter().map(move |tx| (key, tx)))
                .map(|(key, tx)| {
                    previously_deferred_tx_digests.insert(*tx.tx().digest(), key);
                    tx
                })
                .collect();
            trace!(
                "loading deferred randomness transactions: {:?}",
                txns.iter().map(|tx| tx.tx().digest())
            );
            txns
        } else {
            vec![]
        };

        (
            deferred_txs,
            deferred_randomness_txs,
            previously_deferred_tx_digests,
        )
    }

    fn init_congestion_tracker(
        &self,
        commit_info: &ConsensusCommitInfo,
        for_randomness: bool,
        txns: &[VerifiedExecutableTransactionWithAliases],
    ) -> SharedObjectCongestionTracker {
        #[allow(unused_mut)]
        let mut ret = SharedObjectCongestionTracker::from_protocol_config(
            self.epoch_store
                .consensus_quarantine
                .read()
                .load_initial_object_debts(
                    &self.epoch_store,
                    commit_info.round,
                    for_randomness,
                    txns,
                ),
            self.epoch_store.protocol_config(),
            for_randomness,
            self.congestion_logger.is_some(),
        );

        fail_point_arg!(
            "initial_congestion_tracker",
            |tracker: SharedObjectCongestionTracker| {
                info!(
                    "Initialize shared_object_congestion_tracker to  {:?}",
                    tracker
                );
                ret = tracker;
            }
        );

        ret
    }

    fn process_gasless_transactions(
        &self,
        commit_info: &ConsensusCommitInfo,
        user_transactions: &[VerifiedExecutableTransactionWithAliases],
    ) {
        let gasless_count = user_transactions
            .iter()
            .filter(|txn| txn.tx().transaction_data().is_gasless_transaction())
            .count() as u64;
        self.consensus_gasless_counter
            .record_commit(commit_info.timestamp, gasless_count);
    }

    fn process_jwks(
        &self,
        state: &mut CommitHandlerState,
        commit_info: &ConsensusCommitInfo,
        new_jwks: Vec<(AuthorityName, JwkId, JWK)>,
    ) {
        for (authority_name, jwk_id, jwk) in new_jwks {
            self.epoch_store.record_jwk_vote(
                &mut state.output,
                commit_info.round,
                authority_name,
                &jwk_id,
                &jwk,
            );
        }
    }

    fn process_capability_notifications(
        &self,
        capability_notifications: Vec<AuthorityCapabilitiesV2>,
    ) {
        for capabilities in capability_notifications {
            self.epoch_store
                .record_capabilities_v2(&capabilities)
                .expect("db error");
        }
    }

    /// Applies deny-config updates at commit time. The live path already applies them at
    /// block verification in `SuiTxValidator` (re-application here is dropped as a stale
    /// generation), but a validator that is catching up can process commits without
    /// verifying every block in them, so committed updates must also be applied here.
    fn process_transaction_deny_config_updates(
        &self,
        updates: Vec<(AuthorityName, SharedTransactionDenyConfig)>,
    ) {
        for (author, update) in updates {
            self.transaction_deny_config_manager
                .apply_updates(author, vec![update]);
        }
    }

    fn process_execution_time_observations(
        &self,
        state: &mut CommitHandlerState,
        execution_time_observations: Vec<ExecutionTimeObservation>,
    ) {
        let _scope = monitored_scope("ConsensusCommitHandler::process_execution_time_observations");
        let mut execution_time_estimator = self
            .epoch_store
            .execution_time_estimator
            .try_lock()
            .expect("should only ever be called from the commit handler thread");

        for ExecutionTimeObservation {
            authority,
            generation,
            estimates,
        } in execution_time_observations
        {
            let authority_index = self
                .epoch_store
                .committee()
                .authority_index(&authority)
                .unwrap();
            execution_time_estimator.process_observations_from_consensus(
                authority_index,
                Some(generation),
                &estimates,
            );
            state
                .output
                .insert_execution_time_observation(authority_index, generation, estimates);
        }
    }

    fn process_checkpoint_signature_messages(
        &self,
        checkpoint_signature_messages: Vec<CheckpointSignatureMessage>,
    ) {
        for checkpoint_signature_message in checkpoint_signature_messages {
            self.checkpoint_service
                .notify_checkpoint_signature(&checkpoint_signature_message)
                .expect("db error");
        }
    }

    async fn process_dkg_updates(
        &self,
        state: &mut CommitHandlerState,
        commit_info: &ConsensusCommitInfo,
        randomness_manager: Option<&mut RandomnessManager>,
        randomness_dkg_messages: Vec<(AuthorityName, Vec<u8>)>,
        randomness_dkg_confirmations: Vec<(AuthorityName, Vec<u8>)>,
    ) -> Result<(), RuntimeShutdown> {
        if !self.epoch_store.randomness_state_enabled() {
            let num_dkg_messages = randomness_dkg_messages.len();
            let num_dkg_confirmations = randomness_dkg_confirmations.len();
            if num_dkg_messages + num_dkg_confirmations > 0 {
                debug_fatal!(
                    "received {} RandomnessDkgMessage and {} RandomnessDkgConfirmation messages when randomness is not enabled",
                    num_dkg_messages,
                    num_dkg_confirmations
                );
            }
            return Ok(());
        }

        let randomness_manager =
            randomness_manager.expect("randomness manager should exist if randomness is enabled");

        let randomness_dkg_updates =
            self.process_randomness_dkg_messages(randomness_manager, randomness_dkg_messages);

        let randomness_dkg_confirmation_updates = self.process_randomness_dkg_confirmations(
            state,
            randomness_manager,
            randomness_dkg_confirmations,
        );

        let protocol_config = self.epoch_store.protocol_config();
        // Once timeout failure is derived per commit, merge and completion can only make progress
        // when new consensus material arrives.
        let always_advance_dkg_to_resolution = !protocol_config
            .allow_dkg_completion_after_timeout()
            && (protocol_config.always_advance_dkg_to_resolution()
                || (self.epoch_store.get_chain() == Chain::Mainnet
                    && self.epoch_store.epoch() >= 1143))
            && randomness_manager.dkg_status() == DkgStatus::Pending;

        if randomness_dkg_updates
            || randomness_dkg_confirmation_updates
            || always_advance_dkg_to_resolution
        {
            match randomness_manager
                .advance_dkg(&mut state.output, commit_info.round)
                .await
            {
                Ok(()) => {}
                Err(AdvanceDkgError::RuntimeShutdown) => return Err(RuntimeShutdown),
                Err(AdvanceDkgError::Sui(e)) => panic!("failed to advance randomness DKG: {e:?}"),
            }
        }
        Ok(())
    }

    fn process_randomness_dkg_messages(
        &self,
        randomness_manager: &mut RandomnessManager,
        randomness_dkg_messages: Vec<(AuthorityName, Vec<u8>)>,
    ) -> bool /* randomness state updated */ {
        if randomness_dkg_messages.is_empty() {
            return false;
        }

        let mut randomness_state_updated = false;
        for (authority, bytes) in randomness_dkg_messages {
            match bcs::from_bytes(&bytes) {
                Ok(message) => {
                    randomness_manager
                        .add_message(&authority, message)
                        // TODO: make infallible
                        .expect("epoch ended");
                    randomness_state_updated = true;
                }

                Err(e) => {
                    warn!(
                        "Failed to deserialize RandomnessDkgMessage from {:?}: {e:?}",
                        authority.concise(),
                    );
                }
            }
        }

        randomness_state_updated
    }

    fn process_randomness_dkg_confirmations(
        &self,
        state: &mut CommitHandlerState,
        randomness_manager: &mut RandomnessManager,
        randomness_dkg_confirmations: Vec<(AuthorityName, Vec<u8>)>,
    ) -> bool /* randomness state updated */ {
        if randomness_dkg_confirmations.is_empty() {
            return false;
        }

        let mut randomness_state_updated = false;
        for (authority, bytes) in randomness_dkg_confirmations {
            match bcs::from_bytes(&bytes) {
                Ok(message) => {
                    randomness_manager
                        .add_confirmation(&mut state.output, &authority, message)
                        // TODO: make infallible
                        .expect("epoch ended");
                    randomness_state_updated = true;
                }
                Err(e) => {
                    warn!(
                        "Failed to deserialize RandomnessDkgConfirmation from {:?}: {e:?}",
                        authority.concise(),
                    );
                }
            }
        }

        randomness_state_updated
    }

    /// Returns true if we have collected a quorum of end of publish messages (either in this round or a previous round).
    fn process_end_of_publish_transactions(
        &self,
        state: &mut CommitHandlerState,
        end_of_publish_transactions: Vec<AuthorityName>,
    ) -> bool {
        let mut eop_aggregator = self.epoch_store.end_of_publish.try_lock().expect(
            "No contention on end_of_publish as it is only accessed from consensus handler",
        );

        if eop_aggregator.has_quorum() {
            return true;
        }

        if end_of_publish_transactions.is_empty() {
            return false;
        }

        for authority in end_of_publish_transactions {
            info!("Received EndOfPublish from {:?}", authority.concise());

            // It is ok to just release lock here as this function is the only place that transition into RejectAllCerts state
            // And this function itself is always executed from consensus task
            state.output.insert_end_of_publish(authority);
            if eop_aggregator
                .insert_generic(authority, ())
                .is_quorum_reached()
            {
                debug!(
                    "Collected enough end_of_publish messages with last message from validator {:?}",
                    authority.concise(),
                );
                return true;
            }
        }

        false
    }

    /// Once the timestamp deadline is reached or 2f+1 EndOfPublish messages are collected, we call
    /// this function every round until the epoch ends.
    fn advance_end_of_epoch_state_machine(
        &self,
        state: &mut CommitHandlerState,
        deadline_reached: bool,
    ) -> (
        RwLockWriteGuard<'_, ReconfigState>,
        bool, // true if final round
        Option<AbandonedDeferredTxns>,
    ) {
        let mut reconfig_state = self.epoch_store.get_reconfig_state_write_lock_guard();
        let start_state_is_reject_all_tx = reconfig_state.is_reject_all_tx();

        // Record the close time only on the first entry into the state machine.
        // A manual epoch close records it when closing user certs.
        if reconfig_state.should_accept_user_certs() {
            self.epoch_store.record_epoch_close_time_once();
        }

        reconfig_state.close_all_certs();

        let commit_has_deferred_txns = state.output.has_deferred_transactions();
        let previous_commits_have_deferred_txns = !self.epoch_store.deferred_transactions_empty();
        let has_deferred_txns = commit_has_deferred_txns || previous_commits_have_deferred_txns;

        // The epoch closes normally only once all deferred transactions have been scheduled;
        // past the deadline it closes regardless, abandoning whatever remains.
        let should_close = !has_deferred_txns || deadline_reached;
        let final_round = should_close && !start_state_is_reject_all_tx;

        let mut abandoned_deferred_txns = None;
        if final_round {
            info!("Transitioning to RejectAllTx");
            if has_deferred_txns {
                // Closing with deferred transactions remaining implies the deadline forced it.
                debug_assert!(deadline_reached);
                abandoned_deferred_txns = self.abandon_deferred_transactions(state);
            }
            reconfig_state.close_all_tx();
        } else if !should_close {
            debug!(
                "Blocking end of epoch on deferred transactions, from previous commits?={}, from this commit?={}",
                previous_commits_have_deferred_txns, commit_has_deferred_txns,
            );
        }

        state.output.store_reconfig_state(reconfig_state.clone());

        (reconfig_state, final_round, abandoned_deferred_txns)
    }

    /// Abandons every deferred transaction still unscheduled when the epoch close deadline
    /// forces the epoch closed: drops this commit's newly staged deferrals and stages all
    /// remaining deferral keys for deletion.
    ///
    /// (Without this, post-close commits would reload and re-defer them forever, growing the
    /// cache without bound.)
    fn abandon_deferred_transactions(
        &self,
        state: &mut CommitHandlerState,
    ) -> Option<AbandonedDeferredTxns> {
        state.output.clear_deferred_transactions();

        let already_deleted: BTreeSet<_> = state.output.get_deleted_deferred_txn_keys().collect();
        let mut count = 0;
        let mut sample = Vec::new();
        let mut abandoned_keys = Vec::new();
        {
            let deferred_transactions = self
                .epoch_store
                .consensus_output_cache
                .deferred_transactions
                .lock();
            for (key, txns) in deferred_transactions.iter() {
                if already_deleted.contains(key) {
                    // Loaded and scheduled by this commit; not abandoned.
                    continue;
                }
                abandoned_keys.push(*key);
                count += txns.len();
                for tx in txns {
                    if sample.len() < 10 {
                        sample.push((*key, *tx.tx().digest()));
                    }
                }
            }
        }
        if abandoned_keys.is_empty() {
            return None;
        }
        state
            .output
            .delete_loaded_deferred_transactions(&abandoned_keys);
        (count > 0).then_some(AbandonedDeferredTxns { count, sample })
    }

    fn gather_commit_metadata(
        &self,
        consensus_commit: &impl ConsensusCommitAPI,
    ) -> (u64, AuthorityIndex, u64) {
        let timestamp = consensus_commit.commit_timestamp_ms();
        let leader_author = consensus_commit.leader_author_index();
        let commit_sub_dag_index = consensus_commit.commit_sub_dag_index();

        let system_time_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_millis() as i64;

        let consensus_timestamp_bias_ms = system_time_ms - (timestamp as i64);
        let consensus_timestamp_bias_seconds = consensus_timestamp_bias_ms as f64 / 1000.0;
        self.metrics
            .consensus_timestamp_bias
            .observe(consensus_timestamp_bias_seconds);

        let epoch_start = self
            .epoch_store
            .epoch_start_config()
            .epoch_start_timestamp_ms();
        let timestamp = if timestamp < epoch_start {
            error!(
                "Unexpected commit timestamp {timestamp} less then epoch start time {epoch_start}, author {leader_author}"
            );
            epoch_start
        } else {
            timestamp
        };

        (timestamp, leader_author, commit_sub_dag_index)
    }

    fn create_authenticator_state_update(
        &self,
        last_committed_round: u64,
        commit_info: &ConsensusCommitInfo,
    ) -> Option<VerifiedExecutableTransactionWithAliases> {
        // Load all jwks that became active in the previous round, and commit them in this round.
        // We want to delay one round because none of the transactions in the previous round could
        // have been authenticated with the jwks that became active in that round.
        //
        // Because of this delay, jwks that become active in the last round of the epoch will
        // never be committed. That is ok, because in the new epoch, the validators should
        // immediately re-submit these jwks, and they can become active then.
        let new_jwks = self
            .epoch_store
            .get_new_jwks(last_committed_round)
            .expect("Unrecoverable error in consensus handler");

        if !new_jwks.is_empty() {
            let authenticator_state_update_transaction = authenticator_state_update_transaction(
                &self.epoch_store,
                commit_info.round,
                new_jwks,
            );
            debug!(
                "adding AuthenticatorStateUpdate({:?}) tx: {:?}",
                authenticator_state_update_transaction.digest(),
                authenticator_state_update_transaction,
            );

            Some(VerifiedExecutableTransactionWithAliases::no_aliases(
                authenticator_state_update_transaction,
            ))
        } else {
            None
        }
    }

    // Returns the hostname of the block author at the given committee index, used as a metric
    // label. Falls back to "unknown" if the index is out of bounds.
    fn authority_hostname(&self, author: usize) -> &str {
        self.committee
            .to_authority_index(author)
            .map(|index| self.committee.authority(index).hostname.as_str())
            .unwrap_or("unknown")
    }

    // Filters out rejected or deprecated transactions.
    // Returns FilteredConsensusOutput containing transactions and owned_object_locks.
    #[instrument(level = "trace", skip_all)]
    fn filter_consensus_txns(
        &mut self,
        initial_reconfig_state: ReconfigState,
        commit_info: &ConsensusCommitInfo,
        block_transactions: ParsedConsensusTransactions,
    ) -> FilteredConsensusOutput {
        let _scope = monitored_scope("ConsensusCommitHandler::filter_consensus_txns");
        let mut transactions = Vec::new();
        let mut owned_object_locks = HashMap::new();
        let mut dropped_transaction_keys = Vec::new();
        // Consensus transaction status updates are collected here and flushed in a
        // single batched write (one lock acquisition, notifications outside the lock)
        // at the end of the commit, rather than one write+notify per transaction.
        let mut status_updates: Vec<(ConsensusPosition, ConsensusTxStatus)> = Vec::new();
        let mut contested_transaction_digests: HashMap<TransactionDigest, ConflictInfo> =
            HashMap::new();
        // Block author for each transaction that successfully acquired owned object locks, so we
        // can attribute the winning side of a double-spend conflict to the authority that
        // sequenced it.
        let mut lock_holder_authors: HashMap<TransactionDigest, usize> = HashMap::new();
        let epoch = self.epoch_store.epoch();
        let mut num_finalized_user_transactions = vec![0; self.committee.size()];
        let mut num_rejected_user_transactions = vec![0; self.committee.size()];
        let mut num_dropped_user_transactions = vec![0; self.committee.size()];

        // Prefetch the cross-commit owned-object lock state for the whole commit in one
        // batched read. These locks are constant for the duration of a commit (new locks
        // are only written at commit end), so this replaces the per-transaction
        // quarantine+DB lookup that try_acquire_owned_object_locks_post_consensus used to
        // do. Over-reading refs of transactions that are later filtered out is harmless.
        let existing_locks = {
            let mut prefetch_refs: Vec<ObjectRef> = Vec::new();
            for (_block, parsed_transactions) in &block_transactions {
                for parsed in parsed_transactions {
                    if let ConsensusTransactionKind::UserTransactionV2(tx_with_claims) =
                        &parsed.transaction.kind
                        && let Some(refs) = owned_object_refs_to_lock(tx_with_claims)
                    {
                        prefetch_refs.extend(refs);
                    }
                }
            }
            prefetch_refs.sort();
            prefetch_refs.dedup();
            self.epoch_store.get_owned_object_locks_map(&prefetch_refs)
        };

        for (block, parsed_transactions) in block_transactions {
            let author = block.author.value();
            let author_hostname = self.committee.authority(block.author).hostname.as_str();
            // TODO: consider only messages within 1~3 rounds of the leader?
            self.last_consensus_stats.stats.inc_num_messages(author);

            // Set the "ping" transaction status for this block. This is necessary as there might be some ping requests waiting for the ping transaction to be certified.
            status_updates.push((
                ConsensusPosition::ping(epoch, block),
                ConsensusTxStatus::Finalized,
            ));

            for (tx_index, parsed) in parsed_transactions.into_iter().enumerate() {
                let position = ConsensusPosition {
                    epoch,
                    block,
                    index: tx_index as TransactionIndex,
                };

                // Transaction has appeared in consensus output, we can increment the submission count
                // for this tx for DoS protection.
                if let Some(tx) = parsed.transaction.kind.as_user_transaction() {
                    let digest = tx.digest();
                    if let Some((spam_weight, submitter_client_addrs)) = self
                        .epoch_store
                        .submitted_transaction_cache
                        .increment_submission_count(digest)
                    {
                        if let Some(ref traffic_controller) = self.traffic_controller {
                            debug!(
                                "Transaction {digest} exceeded submission limits, spam_weight: {spam_weight:?} applied to {} client addresses",
                                submitter_client_addrs.len()
                            );

                            // Apply spam weight to all client addresses that submitted this transaction
                            for addr in submitter_client_addrs {
                                traffic_controller.tally(
                                    TrafficTally::new(Some(addr), None, None, spam_weight.clone())
                                        .with_method(
                                            "consensus_submission_limit_exceeded".to_string(),
                                        ),
                                );
                            }
                        } else {
                            warn!(
                                "Transaction {digest} exceeded submission limits, spam_weight: {spam_weight:?} for {} client addresses (traffic controller not configured)",
                                submitter_client_addrs.len()
                            );
                        }
                    }
                }

                // Record metrics for every committed transaction, regardless of whether it was
                // accepted or rejected by consensus, so we measure the full committed output.
                let kind = classify(&parsed.transaction);
                let outcome = if parsed.rejected {
                    "rejected"
                } else {
                    "accepted"
                };
                self.metrics
                    .consensus_handler_processed
                    .with_label_values(&[kind, outcome])
                    .inc();
                self.metrics.observe_consensus_handler_transaction_size(
                    kind,
                    outcome,
                    parsed.serialized_len,
                );
                // Per-author breakdown is only tracked for user transactions, since that is where
                // rejections and author-attributable spam are meaningful. Keeping the block author
                // label scoped to user transactions also bounds metric cardinality.
                if parsed.transaction.is_user_transaction() {
                    self.metrics
                        .consensus_handler_processed_user_transactions
                        .with_label_values(&[outcome, author_hostname])
                        .inc();
                }

                if parsed.rejected {
                    if parsed.transaction.is_user_transaction() {
                        status_updates.push((position, ConsensusTxStatus::Rejected));
                        num_rejected_user_transactions[author] += 1;
                    }
                    // Skip processing rejected transactions.
                    continue;
                }

                if parsed.transaction.is_user_transaction() {
                    self.last_consensus_stats
                        .stats
                        .inc_num_user_transactions(author);
                }

                if !initial_reconfig_state.should_accept_consensus_certs() {
                    // (Note: we no longer need to worry about the previously deferred condition, since we are only
                    // processing newly-received transactions at this time).
                    match &parsed.transaction.kind {
                        ConsensusTransactionKind::UserTransactionV2(_)
                        // deprecated and ignore later, but added for exhaustive match
                        | ConsensusTransactionKind::UserTransaction(_)
                        | ConsensusTransactionKind::CertifiedTransaction(_)
                        | ConsensusTransactionKind::CapabilityNotification(_)
                        | ConsensusTransactionKind::CapabilityNotificationV2(_)
                        | ConsensusTransactionKind::EndOfPublish(_)
                        // Note: we no longer have to check protocol_config.ignore_execution_time_observations_after_certs_closed()
                        | ConsensusTransactionKind::ExecutionTimeObservation(_)
                        | ConsensusTransactionKind::NewJWKFetched(_, _, _)
                        | ConsensusTransactionKind::UpdateTransactionDenyConfig(_) => {
                            // Deterministic drop: the reconfig state is derived from prior
                            // commits, so all validators ignore this transaction. Record a
                            // terminal status for user transactions so status waiters are
                            // not leaked — no consensus-processed signal will ever come for
                            // them, and epoch termination (the only other backstop) can
                            // itself stall when reconfiguration is stuck.
                            if parsed.transaction.is_user_transaction() {
                                status_updates.push((position, ConsensusTxStatus::Dropped));
                                num_dropped_user_transactions[author] += 1;
                                self.metrics
                                    .consensus_handler_dropped_transactions
                                    .with_label_values(&["end_of_epoch"])
                                    .inc();
                            }
                            debug!(
                                "Ignoring consensus transaction {:?} because of end of epoch",
                                parsed.transaction.key()
                            );
                            continue;
                        }

                        // These are the message types that are still processed even if !should_accept_consensus_certs()
                        ConsensusTransactionKind::CheckpointSignature(_)
                        | ConsensusTransactionKind::CheckpointSignatureV2(_)
                        | ConsensusTransactionKind::RandomnessStateUpdate(_, _)
                        | ConsensusTransactionKind::RandomnessDkgMessage(_, _)
                        | ConsensusTransactionKind::RandomnessDkgConfirmation(_, _) => ()
                    }
                }

                if !initial_reconfig_state.should_accept_tx() {
                    match &parsed.transaction.kind {
                        ConsensusTransactionKind::RandomnessDkgConfirmation(_, _)
                        | ConsensusTransactionKind::RandomnessDkgMessage(_, _) => continue,
                        _ => {}
                    }
                }

                // Handle deprecated messages
                match &parsed.transaction.kind {
                    ConsensusTransactionKind::CapabilityNotification(_)
                    | ConsensusTransactionKind::RandomnessStateUpdate(_, _)
                    | ConsensusTransactionKind::CheckpointSignature(_) => {
                        debug_fatal!(
                            "BUG: saw deprecated tx {:?}for commit round {}",
                            parsed.transaction.key(),
                            commit_info.round
                        );
                        continue;
                    }
                    _ => {}
                }

                if parsed.transaction.is_user_transaction() {
                    let author_name = self
                        .epoch_store
                        .committee()
                        .authority_by_index(author as u32)
                        .unwrap();
                    if self
                        .epoch_store
                        .has_received_end_of_publish_from(author_name)
                    {
                        // In some edge cases, consensus might resend previously seen certificate after EndOfPublish
                        // An honest validator should not send a new transaction after EndOfPublish. Whether the
                        // transaction is duplicate or not, we filter it out here.
                        // Deterministic drop (the EndOfPublish set is accumulated from
                        // prior commits): record a terminal status so waiters are not
                        // leaked, as with the certs-closed drop above.
                        status_updates.push((position, ConsensusTxStatus::Dropped));
                        num_dropped_user_transactions[author] += 1;
                        self.metrics
                            .consensus_handler_dropped_transactions
                            .with_label_values(&["end_of_publish"])
                            .inc();
                        warn!(
                            "Ignoring consensus transaction {:?} from authority {:?}, which already sent EndOfPublish message to consensus",
                            author_name.concise(),
                            parsed.transaction.key(),
                        );
                        continue;
                    }
                }

                // Perform post-consensus owned object conflict detection. If lock acquisition
                // fails, the transaction has invalid/conflicting owned inputs and should be dropped.
                // This must happen AFTER all filtering checks above to avoid acquiring locks
                // for transactions that will be dropped (e.g., during epoch change).
                // Only applies to UserTransactionV2 - other transaction types don't need lock acquisition.
                if let ConsensusTransactionKind::UserTransactionV2(tx_with_claims) =
                    &parsed.transaction.kind
                {
                    let tx = tx_with_claims.tx();
                    let Some(owned_object_refs) = owned_object_refs_to_lock(tx_with_claims) else {
                        // Invalid input object error is deterministic across all validators.
                        self.metrics
                            .consensus_handler_dropped_transactions
                            .with_label_values(&["invalid_input"])
                            .inc();
                        status_updates.push((position, ConsensusTxStatus::Dropped));
                        num_dropped_user_transactions[author] += 1;
                        // Record the concrete input error as the reject reason so effects
                        // waiters get a terminal, non-retriable error instead of a bare
                        // Dropped with no reason (which clients treat as retriable).
                        if let Err(e) = tx.transaction_data().input_objects() {
                            self.epoch_store
                                .set_rejection_vote_reason(position, &e.into());
                        }
                        dropped_transaction_keys.push(parsed.transaction.key());
                        debug_fatal!("Invalid input objects for transaction {}", tx.digest());
                        continue;
                    };

                    match self
                        .epoch_store
                        .try_acquire_owned_object_locks_post_consensus(
                            &owned_object_refs,
                            *tx.digest(),
                            &owned_object_locks,
                            &existing_locks,
                        ) {
                        Ok(new_locks) => {
                            owned_object_locks.extend(new_locks.into_iter());
                            lock_holder_authors.entry(*tx.digest()).or_insert(author);
                            // Lock acquisition succeeded - now set Finalized status
                            status_updates.push((position, ConsensusTxStatus::Finalized));
                            num_finalized_user_transactions[author] += 1;
                        }
                        Err(e) => {
                            // Flag intra-commit double-spend: if the conflict is with
                            // a transaction in this commit (in owned_object_locks), the
                            // holder should be deferred as penalty.
                            let gas_object_ids: HashSet<ObjectID> = tx
                                .transaction_data()
                                .gas()
                                .iter()
                                .map(|obj_ref| obj_ref.0)
                                .collect();
                            let mut is_intra_commit_conflict = false;
                            for obj_ref in &owned_object_refs {
                                if let Some(holder_digest) = owned_object_locks.get(obj_ref) {
                                    is_intra_commit_conflict = true;
                                    let info = contested_transaction_digests
                                        .entry(*holder_digest)
                                        .or_default();
                                    info.winner_author = lock_holder_authors
                                        .get(holder_digest)
                                        .copied()
                                        .unwrap_or(author);
                                    if gas_object_ids.contains(&obj_ref.0) {
                                        info.gas_object_conflicts += 1;
                                    } else {
                                        info.non_gas_object_conflicts += 1;
                                    }
                                }
                            }
                            // Attribute the losing side of the conflict to the authority that
                            // sequenced this dropped transaction. Counted once per loser, even
                            // if it conflicted on multiple objects.
                            if is_intra_commit_conflict {
                                self.metrics
                                    .consensus_handler_double_spend_conflicting_authority
                                    .with_label_values(&[self.authority_hostname(author), "loser"])
                                    .inc();
                            }
                            debug!("Dropping transaction {}: {}", tx.digest(), e);
                            self.metrics
                                .consensus_handler_dropped_transactions
                                .with_label_values(&["lock_conflict"])
                                .inc();
                            status_updates.push((position, ConsensusTxStatus::Dropped));
                            num_dropped_user_transactions[author] += 1;
                            self.epoch_store.set_rejection_vote_reason(position, &e);
                            dropped_transaction_keys.push(parsed.transaction.key());
                            continue;
                        }
                    }
                }

                let transaction = SequencedConsensusTransactionKind::External(parsed.transaction);
                transactions.push((transaction, author as u32));
            }
        }

        // Flush all collected status updates in one batched write. None of the logic
        // above reads transaction status back, so deferring visibility to here (still
        // before dedup/processing) is equivalent to setting each status inline.
        self.epoch_store.set_consensus_tx_statuses(status_updates);

        for (i, authority) in self.committee.authorities() {
            let hostname = &authority.hostname;
            self.metrics
                .consensus_committed_messages
                .with_label_values(&[hostname])
                .set(self.last_consensus_stats.stats.get_num_messages(i.value()) as i64);
            self.metrics
                .consensus_committed_user_transactions
                .with_label_values(&[hostname])
                .set(
                    self.last_consensus_stats
                        .stats
                        .get_num_user_transactions(i.value()) as i64,
                );
            self.metrics
                .consensus_finalized_user_transactions
                .with_label_values(&[hostname])
                .add(num_finalized_user_transactions[i.value()] as i64);
            self.metrics
                .consensus_rejected_user_transactions
                .with_label_values(&[hostname])
                .add(num_rejected_user_transactions[i.value()] as i64);
            self.metrics
                .consensus_dropped_user_transactions
                .with_label_values(&[hostname])
                .add(num_dropped_user_transactions[i.value()] as i64);
        }

        FilteredConsensusOutput {
            transactions,
            owned_object_locks,
            dropped_transaction_keys,
            contested_transaction_digests,
        }
    }

    fn deduplicate_consensus_txns(
        &mut self,
        state: &mut CommitHandlerState,
        commit_info: &ConsensusCommitInfo,
        transactions: Vec<(SequencedConsensusTransactionKind, u32)>,
    ) -> Vec<VerifiedSequencedConsensusTransaction> {
        let _scope = monitored_scope("ConsensusCommitHandler::deduplicate_consensus_txns");
        let epoch = self.epoch_store.epoch();
        let rgp = self.epoch_store.reference_gas_price().max(1);
        let mut all_transactions = Vec::new();

        // Track occurrence counts for each transaction key within this commit.
        // Also serves as the deduplication set (count > 1 means duplicate within commit).
        let mut occurrence_counts: HashMap<SequencedConsensusTransactionKey, u32> = HashMap::new();
        // Keys being seen for the first time (not duplicates from previous commits).
        let mut first_commit_keys: HashSet<SequencedConsensusTransactionKey> = HashSet::new();

        for (seq, (transaction, cert_origin)) in transactions.into_iter().enumerate() {
            // In process_consensus_transactions_and_commit_boundary(), we will add a system consensus commit
            // prologue transaction, which will be the first transaction in this consensus commit batch.
            // Therefore, the transaction sequence number starts from 1 here.
            let current_tx_index = ExecutionIndices {
                last_committed_round: commit_info.round,
                sub_dag_index: commit_info.consensus_commit_ref.index.into(),
                transaction_index: (seq + 1) as u64,
            };

            self.last_consensus_stats.index = current_tx_index;

            let certificate_author = *self
                .epoch_store
                .committee()
                .authority_by_index(cert_origin)
                .unwrap();

            let sequenced_transaction = SequencedConsensusTransaction {
                certificate_author_index: cert_origin,
                certificate_author,
                consensus_index: current_tx_index,
                transaction,
            };

            let Some(verified_transaction) = self
                .epoch_store
                .verify_consensus_transaction(sequenced_transaction)
            else {
                continue;
            };

            let key = verified_transaction.0.key();

            if let Some(tx_digest) = key.user_transaction_digest() {
                self.epoch_store
                    .cache_recently_finalized_transaction(tx_digest);
            }

            // Increment count and check if this is a duplicate within this commit.
            // This replaces the separate processed_set HashSet.
            let count = occurrence_counts.entry(key.clone()).or_insert(0);
            *count += 1;
            let in_commit = *count > 1;

            let (total_occurrences, in_cache) = match self.processed_cache.get_mut(&key) {
                Some(total) => {
                    *total = total.saturating_add(1);
                    (*total, true)
                }
                None => {
                    // If the key was evicted during this commit, the commit's own count
                    // is still correct. Only counts from earlier commits are lost.
                    let total = u64::from(*count);
                    self.processed_cache.put(key.clone(), total);
                    (total, false)
                }
            };
            let user_transaction = match &verified_transaction.0.transaction {
                SequencedConsensusTransactionKind::External(transaction) => {
                    transaction.kind.as_user_transaction()
                }
                SequencedConsensusTransactionKind::System(_) => None,
            };
            if let Some(transaction) = user_transaction {
                let tx_data = transaction.data().transaction_data();
                if !tx_data.expiration().restricts_proposers(epoch) {
                    let allowance = paid_proposer_count(tx_data.gas_price(), rgp);
                    if total_occurrences > allowance {
                        state.staggering_counts.excess_copies += 1;
                    } else if total_occurrences > 1 {
                        state.staggering_counts.paid_duplicate_copies += 1;
                    }
                } else if total_occurrences > 1 {
                    state.staggering_counts.restricted_duplicate_copies += 1;
                }
            }
            if in_commit || in_cache {
                self.metrics.skipped_consensus_txns_cache_hit.inc();
                continue;
            }
            if self
                .epoch_store
                .is_consensus_message_processed(&key)
                .expect("db error")
            {
                self.metrics.skipped_consensus_txns.inc();
                continue;
            }

            state.staggering_counts.unique_user_txns += u64::from(user_transaction.is_some());
            first_commit_keys.insert(key.clone());

            state.output.record_consensus_message_processed(key);

            all_transactions.push(verified_transaction);
        }

        for key in first_commit_keys {
            if let Some(&count) = occurrence_counts.get(&key)
                && count > 1
            {
                self.metrics
                    .consensus_handler_duplicate_tx_count
                    .observe(count as f64);
            }
        }

        // Copy user transaction occurrence counts to state for unpaid amplification detection.
        assert!(
            state.occurrence_counts.is_empty(),
            "occurrence_counts should be empty before populating"
        );
        state.occurrence_counts.reserve(occurrence_counts.len());
        state.occurrence_counts.extend(
            occurrence_counts
                .into_iter()
                .filter_map(|(key, count)| key.user_transaction_digest().map(|d| (d, count))),
        );

        all_transactions
    }

    fn build_commit_handler_input(
        &self,
        transactions: Vec<VerifiedSequencedConsensusTransaction>,
    ) -> CommitHandlerInput {
        let _scope = monitored_scope("ConsensusCommitHandler::build_commit_handler_input");
        let epoch = self.epoch_store.epoch();
        let mut commit_handler_input = CommitHandlerInput::default();

        for VerifiedSequencedConsensusTransaction(transaction) in transactions.into_iter() {
            match transaction.transaction {
                SequencedConsensusTransactionKind::External(consensus_transaction) => {
                    match consensus_transaction.kind {
                        // === User transactions ===
                        ConsensusTransactionKind::UserTransactionV2(tx) => {
                            // Extract the aliases claim (required) from the claims
                            let used_alias_versions = tx.aliases();
                            let inner_tx = tx.into_tx();
                            // Safe because transactions are certified by consensus.
                            let tx = VerifiedTransaction::new_unchecked(inner_tx);
                            // TODO(fastpath): accept position in consensus, after plumbing consensus round, authority index, and transaction index here.
                            let transaction =
                                VerifiedExecutableTransaction::new_from_consensus(tx, epoch);
                            if let Some(used_alias_versions) = used_alias_versions {
                                commit_handler_input
                                    .user_transactions
                                    .push(WithAliases::new(transaction, used_alias_versions));
                            } else {
                                commit_handler_input.user_transactions.push(
                                    VerifiedExecutableTransactionWithAliases::no_aliases(
                                        transaction,
                                    ),
                                );
                            }
                        }

                        // === State machines ===
                        ConsensusTransactionKind::EndOfPublish(authority_public_key_bytes) => {
                            commit_handler_input
                                .end_of_publish_transactions
                                .push(authority_public_key_bytes);
                        }
                        ConsensusTransactionKind::NewJWKFetched(
                            authority_public_key_bytes,
                            jwk_id,
                            jwk,
                        ) => {
                            commit_handler_input.new_jwks.push((
                                authority_public_key_bytes,
                                jwk_id,
                                jwk,
                            ));
                        }
                        ConsensusTransactionKind::RandomnessDkgMessage(
                            authority_public_key_bytes,
                            items,
                        ) => {
                            commit_handler_input
                                .randomness_dkg_messages
                                .push((authority_public_key_bytes, items));
                        }
                        ConsensusTransactionKind::RandomnessDkgConfirmation(
                            authority_public_key_bytes,
                            items,
                        ) => {
                            commit_handler_input
                                .randomness_dkg_confirmations
                                .push((authority_public_key_bytes, items));
                        }
                        ConsensusTransactionKind::CapabilityNotificationV2(
                            authority_capabilities_v2,
                        ) => {
                            commit_handler_input
                                .capability_notifications
                                .push(authority_capabilities_v2);
                        }
                        ConsensusTransactionKind::ExecutionTimeObservation(
                            execution_time_observation,
                        ) => {
                            commit_handler_input
                                .execution_time_observations
                                .push(execution_time_observation);
                        }
                        ConsensusTransactionKind::CheckpointSignatureV2(
                            checkpoint_signature_message,
                        ) => {
                            commit_handler_input
                                .checkpoint_signature_messages
                                .push(*checkpoint_signature_message);
                        }
                        ConsensusTransactionKind::UpdateTransactionDenyConfig(msg) => {
                            commit_handler_input
                                .transaction_deny_config_updates
                                .push((transaction.certificate_author, *msg));
                        }

                        // Deprecated messages, filtered earlier by filter_consensus_txns()
                        // or rejected by SuiTxValidator. Kept for exhaustiveness.
                        ConsensusTransactionKind::CheckpointSignature(_)
                        | ConsensusTransactionKind::RandomnessStateUpdate(_, _)
                        | ConsensusTransactionKind::CapabilityNotification(_)
                        | ConsensusTransactionKind::CertifiedTransaction(_)
                        | ConsensusTransactionKind::UserTransaction(_) => {
                            unreachable!("filtered earlier")
                        }
                    }
                }
                // TODO: I think we can delete this, it was only used to inject randomness state update into the tx stream.
                SequencedConsensusTransactionKind::System(_verified_envelope) => unreachable!(),
            }
        }

        commit_handler_input
    }
}

/// Sends transactions to the execution scheduler in a separate task,
/// to avoid blocking consensus handler.
pub(crate) type SchedulerMessage = (
    Vec<(Schedulable, AssignedVersions)>,
    Option<SettlementBatchInfo>,
);

#[derive(Clone)]
pub(crate) struct ExecutionSchedulerSender {
    sender: monitored_mpsc::UnboundedSender<SchedulerMessage>,
}

impl ExecutionSchedulerSender {
    fn start(
        settlement_scheduler: SettlementScheduler,
        epoch_store: Arc<AuthorityPerEpochStore>,
    ) -> Self {
        let (sender, recv) = monitored_mpsc::unbounded_channel("execution_scheduler_sender");
        spawn_monitored_task!(Self::run(recv, settlement_scheduler, epoch_store));
        Self { sender }
    }

    pub(crate) fn new_for_testing(
        sender: monitored_mpsc::UnboundedSender<SchedulerMessage>,
    ) -> Self {
        Self { sender }
    }

    fn send(
        &self,
        transactions: Vec<(Schedulable, AssignedVersions)>,
        settlement: Option<SettlementBatchInfo>,
    ) {
        let _ = self.sender.send((transactions, settlement));
    }

    async fn run(
        mut recv: monitored_mpsc::UnboundedReceiver<SchedulerMessage>,
        settlement_scheduler: SettlementScheduler,
        epoch_store: Arc<AuthorityPerEpochStore>,
    ) {
        while let Some((transactions, settlement)) = recv.recv().await {
            let _guard = monitored_scope("ConsensusHandler::enqueue");
            let txns = transactions
                .into_iter()
                .map(|(txn, versions)| (txn, ExecutionEnv::new().with_assigned_versions(versions)))
                .collect();
            if let Some(settlement) = settlement {
                settlement_scheduler.enqueue_v2(txns, settlement, &epoch_store);
            } else {
                settlement_scheduler.enqueue(txns, &epoch_store);
            }
        }
    }
}

/// Capacity of the channel from the deserialize worker to the commit handler. Small/bounded: lets
/// the worker prepare ~1 commit ahead (pipelining) while applying backpressure when the handler is
/// behind, instead of buffering parsed commits unboundedly.
const CONSENSUS_HANDLER_DESERIALIZE_CHANNEL_CAPACITY: usize = 2;

/// Transactions BCS-deserialized out of a consensus commit, grouped by block. Produced by the
/// deserialize worker and consumed by the commit handler, so parsing stays off the handler's
/// critical path.
type ParsedConsensusTransactions = Vec<(BlockRef, Vec<ParsedTransaction>)>;

/// Manages the lifetime of tasks handling the commits and transactions output by consensus.
pub(crate) struct MysticetiConsensusHandler {
    tasks: JoinSet<()>,
}

impl MysticetiConsensusHandler {
    pub(crate) fn new(
        last_processed_commit_at_startup: CommitIndex,
        mut consensus_handler: ConsensusHandler<CheckpointService>,
        mut commit_receiver: UnboundedReceiver<consensus_core::CommittedSubDag>,
        consumer_monitor: Arc<ConsumerMonitor>,
    ) -> Self {
        debug!(
            last_processed_commit_at_startup,
            "Starting consensus replay"
        );
        let mut tasks = JoinSet::new();

        // Stage 1 — deserialize worker: BCS-parses each commit's transactions off the handler's
        // critical path, so parsing overlaps the handler processing the previous commit. The
        // channel is bounded (small) so the worker prepares ~1 commit ahead but applies
        // backpressure (rather than buffering parsed commits unboundedly) when the handler is the
        // bottleneck. Single-threaded; ordering is preserved (one worker, FIFO channel, one handler).
        let (parsed_sender, mut parsed_receiver) = monitored_mpsc::channel(
            "consensus_deserialized_commits",
            CONSENSUS_HANDLER_DESERIALIZE_CHANNEL_CAPACITY,
        );
        tasks.spawn(monitored_future!(async move {
            while let Some(consensus_commit) = commit_receiver.recv().await {
                let transactions: ParsedConsensusTransactions = {
                    let _scope = monitored_scope("ConsensusCommitHandler::deserialize_worker");
                    consensus_commit.transactions()
                };
                // The send is intentionally outside the scope above: on the bounded channel it
                // blocks when the handler is the bottleneck, and that idle-wait would otherwise
                // inflate the deserialize_worker scope (making it read as CPU work, not waiting).
                if parsed_sender
                    .send((consensus_commit, transactions))
                    .await
                    .is_err()
                {
                    break;
                }
            }
        }));

        // Stage 2 — commit handler: processes pre-parsed commits in order.
        tasks.spawn(monitored_future!(async move {
            // TODO: pause when execution is overloaded, so consensus can detect the backpressure.
            while let Some((consensus_commit, transactions)) = parsed_receiver.recv().await {
                let commit_index = consensus_commit.commit_ref.index;
                if commit_index <= last_processed_commit_at_startup {
                    consensus_handler.handle_prior_consensus_commit(consensus_commit);
                } else if let Err(e) = consensus_handler
                    .handle_consensus_commit(consensus_commit, transactions)
                    .await
                {
                    info!(
                        "Consensus commit {commit_index} was not handled ({e:?}); it will be replayed after restart"
                    );
                    break;
                }
                consumer_monitor.set_highest_handled_commit(commit_index);
            }
        }));
        Self { tasks }
    }

    pub(crate) async fn abort(&mut self) {
        self.tasks.shutdown().await;
    }
}

fn authenticator_state_update_transaction(
    epoch_store: &AuthorityPerEpochStore,
    round: u64,
    mut new_active_jwks: Vec<ActiveJwk>,
) -> VerifiedExecutableTransaction {
    let epoch = epoch_store.epoch();
    new_active_jwks.sort();

    info!("creating authenticator state update transaction");
    assert!(epoch_store.authenticator_state_enabled());
    let transaction = VerifiedTransaction::new_authenticator_state_update(
        epoch,
        round,
        new_active_jwks,
        epoch_store
            .epoch_start_config()
            .authenticator_obj_initial_shared_version()
            .expect("authenticator state obj must exist"),
    );
    VerifiedExecutableTransaction::new_system(transaction, epoch)
}

/// The owned (non-immutable `ImmOrOwnedMoveObject`) object refs that a
/// `UserTransactionV2` must lock post-consensus. Immutable objects are excluded as
/// they can be used concurrently. Returns `None` if the transaction's input objects
/// are invalid. Used both to prefetch existing locks for the whole commit and, per
/// transaction, to perform conflict detection — keeping the two in sync.
fn owned_object_refs_to_lock(
    tx_with_claims: &PlainTransactionWithClaims,
) -> Option<Vec<ObjectRef>> {
    let immutable_object_ids: HashSet<ObjectID> =
        tx_with_claims.get_immutable_objects().into_iter().collect();
    let input_objects = tx_with_claims
        .tx()
        .transaction_data()
        .input_objects()
        .ok()?;
    Some(
        input_objects
            .iter()
            .filter_map(|obj| match obj {
                InputObjectKind::ImmOrOwnedMoveObject(obj_ref)
                    if !immutable_object_ids.contains(&obj_ref.0) =>
                {
                    Some(*obj_ref)
                }
                _ => None,
            })
            .collect(),
    )
}

/// Label for the `sequencing_certificate_*` metrics.
pub(crate) fn tx_type_label(transactions: &[ConsensusTransaction]) -> &'static str {
    match transactions {
        [transaction] => classify(transaction),
        _ => "soft_bundle",
    }
}

pub(crate) fn classify(transaction: &ConsensusTransaction) -> &'static str {
    match &transaction.kind {
        // Deprecated and rejected by SuiTxValidator; never classified in practice.
        ConsensusTransactionKind::CertifiedTransaction(_) => "_deprecated_certificate",
        ConsensusTransactionKind::CheckpointSignature(_) => "checkpoint_signature",
        ConsensusTransactionKind::CheckpointSignatureV2(_) => "checkpoint_signature",
        ConsensusTransactionKind::EndOfPublish(_) => "end_of_publish",
        ConsensusTransactionKind::CapabilityNotification(_) => "capability_notification",
        ConsensusTransactionKind::CapabilityNotificationV2(_) => "capability_notification_v2",
        ConsensusTransactionKind::NewJWKFetched(_, _, _) => "new_jwk_fetched",
        ConsensusTransactionKind::RandomnessStateUpdate(_, _) => "randomness_state_update",
        ConsensusTransactionKind::RandomnessDkgMessage(_, _) => "randomness_dkg_message",
        ConsensusTransactionKind::RandomnessDkgConfirmation(_, _) => "randomness_dkg_confirmation",
        ConsensusTransactionKind::UserTransaction(_) => "_deprecated_user_transaction",
        ConsensusTransactionKind::UserTransactionV2(tx) => {
            if tx.tx().is_consensus_tx() {
                "shared_user_transaction_v2"
            } else {
                "owned_user_transaction_v2"
            }
        }
        ConsensusTransactionKind::ExecutionTimeObservation(_) => "execution_time_observation",
        ConsensusTransactionKind::UpdateTransactionDenyConfig(_) => {
            "update_transaction_deny_config"
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SequencedConsensusTransaction {
    pub certificate_author_index: AuthorityIndex,
    pub certificate_author: AuthorityName,
    pub consensus_index: ExecutionIndices,
    pub transaction: SequencedConsensusTransactionKind,
}

#[derive(Debug, Clone)]
#[allow(clippy::large_enum_variant)]
pub enum SequencedConsensusTransactionKind {
    External(ConsensusTransaction),
    System(VerifiedExecutableTransaction),
}

impl Serialize for SequencedConsensusTransactionKind {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let serializable = SerializableSequencedConsensusTransactionKind::from(self);
        serializable.serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for SequencedConsensusTransactionKind {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let serializable =
            SerializableSequencedConsensusTransactionKind::deserialize(deserializer)?;
        Ok(serializable.into())
    }
}

// We can't serialize SequencedConsensusTransactionKind directly because it contains a
// VerifiedExecutableTransaction, which is not serializable (by design). This wrapper allows us to
// convert to a serializable format easily.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[allow(clippy::large_enum_variant)]
enum SerializableSequencedConsensusTransactionKind {
    External(ConsensusTransaction),
    System(TrustedExecutableTransaction),
}

impl From<&SequencedConsensusTransactionKind> for SerializableSequencedConsensusTransactionKind {
    fn from(kind: &SequencedConsensusTransactionKind) -> Self {
        match kind {
            SequencedConsensusTransactionKind::External(ext) => {
                SerializableSequencedConsensusTransactionKind::External(ext.clone())
            }
            SequencedConsensusTransactionKind::System(txn) => {
                SerializableSequencedConsensusTransactionKind::System(txn.clone().serializable())
            }
        }
    }
}

impl From<SerializableSequencedConsensusTransactionKind> for SequencedConsensusTransactionKind {
    fn from(kind: SerializableSequencedConsensusTransactionKind) -> Self {
        match kind {
            SerializableSequencedConsensusTransactionKind::External(ext) => {
                SequencedConsensusTransactionKind::External(ext)
            }
            SerializableSequencedConsensusTransactionKind::System(txn) => {
                SequencedConsensusTransactionKind::System(txn.into())
            }
        }
    }
}

#[derive(Serialize, Deserialize, Clone, Hash, PartialEq, Eq, Debug, Ord, PartialOrd)]
pub enum SequencedConsensusTransactionKey {
    External(ConsensusTransactionKey),
    System(TransactionDigest),
}

impl SequencedConsensusTransactionKey {
    pub fn user_transaction_digest(&self) -> Option<TransactionDigest> {
        match self {
            SequencedConsensusTransactionKey::External(key) => match key {
                ConsensusTransactionKey::Certificate(digest) => Some(*digest),
                _ => None,
            },
            SequencedConsensusTransactionKey::System(_) => None,
        }
    }
}

impl SequencedConsensusTransactionKind {
    pub fn key(&self) -> SequencedConsensusTransactionKey {
        match self {
            SequencedConsensusTransactionKind::External(ext) => {
                SequencedConsensusTransactionKey::External(ext.key())
            }
            SequencedConsensusTransactionKind::System(txn) => {
                SequencedConsensusTransactionKey::System(*txn.digest())
            }
        }
    }

    pub fn get_tracking_id(&self) -> u64 {
        match self {
            SequencedConsensusTransactionKind::External(ext) => ext.get_tracking_id(),
            SequencedConsensusTransactionKind::System(_txn) => 0,
        }
    }

    pub fn is_executable_transaction(&self) -> bool {
        match self {
            SequencedConsensusTransactionKind::External(ext) => ext.is_user_transaction(),
            SequencedConsensusTransactionKind::System(_) => true,
        }
    }

    pub fn executable_transaction_digest(&self) -> Option<TransactionDigest> {
        match self {
            SequencedConsensusTransactionKind::External(ext) => match &ext.kind {
                ConsensusTransactionKind::UserTransactionV2(txn) => Some(*txn.tx().digest()),
                _ => None,
            },
            SequencedConsensusTransactionKind::System(txn) => Some(*txn.digest()),
        }
    }

    pub fn is_end_of_publish(&self) -> bool {
        match self {
            SequencedConsensusTransactionKind::External(ext) => {
                matches!(ext.kind, ConsensusTransactionKind::EndOfPublish(..))
            }
            SequencedConsensusTransactionKind::System(_) => false,
        }
    }
}

impl SequencedConsensusTransaction {
    pub fn sender_authority(&self) -> AuthorityName {
        self.certificate_author
    }

    pub fn key(&self) -> SequencedConsensusTransactionKey {
        self.transaction.key()
    }

    pub fn is_end_of_publish(&self) -> bool {
        if let SequencedConsensusTransactionKind::External(ref transaction) = self.transaction {
            matches!(transaction.kind, ConsensusTransactionKind::EndOfPublish(..))
        } else {
            false
        }
    }

    pub fn try_take_execution_time_observation(&mut self) -> Option<ExecutionTimeObservation> {
        if let SequencedConsensusTransactionKind::External(ConsensusTransaction {
            kind: ConsensusTransactionKind::ExecutionTimeObservation(observation),
            ..
        }) = &mut self.transaction
        {
            Some(std::mem::take(observation))
        } else {
            None
        }
    }

    pub fn is_system(&self) -> bool {
        matches!(
            self.transaction,
            SequencedConsensusTransactionKind::System(_)
        )
    }

    pub fn is_user_tx_with_randomness(&self, randomness_state_enabled: bool) -> bool {
        if !randomness_state_enabled {
            // If randomness is disabled, these should be processed same as a tx without randomness,
            // which will eventually fail when the randomness state object is not found.
            return false;
        }
        match &self.transaction {
            SequencedConsensusTransactionKind::External(ConsensusTransaction {
                kind: ConsensusTransactionKind::UserTransactionV2(txn),
                ..
            }) => txn.tx().transaction_data().uses_randomness(),
            _ => false,
        }
    }

    pub fn as_consensus_txn(&self) -> Option<&SenderSignedData> {
        match &self.transaction {
            SequencedConsensusTransactionKind::External(ConsensusTransaction {
                kind: ConsensusTransactionKind::UserTransactionV2(txn),
                ..
            }) if txn.tx().is_consensus_tx() => Some(txn.tx().data()),
            SequencedConsensusTransactionKind::System(txn) if txn.is_consensus_tx() => {
                Some(txn.data())
            }
            _ => None,
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VerifiedSequencedConsensusTransaction(pub SequencedConsensusTransaction);

#[cfg(test)]
impl VerifiedSequencedConsensusTransaction {
    pub fn new_test(transaction: ConsensusTransaction) -> Self {
        Self(SequencedConsensusTransaction::new_test(transaction))
    }
}

impl SequencedConsensusTransaction {
    pub fn new_test(transaction: ConsensusTransaction) -> Self {
        Self {
            certificate_author_index: 0,
            certificate_author: AuthorityName::ZERO,
            consensus_index: Default::default(),
            transaction: SequencedConsensusTransactionKind::External(transaction),
        }
    }
}

#[derive(Serialize, Deserialize)]
pub(crate) struct CommitIntervalObserver {
    ring_buffer: VecDeque<u64>,
}

impl CommitIntervalObserver {
    pub fn new(window_size: u32) -> Self {
        Self {
            ring_buffer: VecDeque::with_capacity(window_size as usize),
        }
    }

    pub fn observe_commit_time(&mut self, consensus_commit: &impl ConsensusCommitAPI) {
        let commit_time = consensus_commit.commit_timestamp_ms();
        if self.ring_buffer.len() == self.ring_buffer.capacity() {
            self.ring_buffer.pop_front();
        }
        self.ring_buffer.push_back(commit_time);
    }

    pub fn commit_interval_estimate(&self) -> Option<Duration> {
        if self.ring_buffer.len() <= 1 {
            None
        } else {
            let first = self.ring_buffer.front().unwrap();
            let last = self.ring_buffer.back().unwrap();
            let duration = last.saturating_sub(*first);
            let num_commits = self.ring_buffer.len() as u64;
            Some(Duration::from_millis(duration.div_ceil(num_commits)))
        }
    }
}

#[cfg(test)]
#[path = "unit_tests/staggered_submission_signal_tests.rs"]
mod staggering_signal_tests;

#[cfg(test)]
mod tests {
    use consensus_core::{
        BlockAPI, CommitDigest, CommitRef, CommittedSubDag, TestBlock, Transaction, VerifiedBlock,
    };
    use futures::pin_mut;
    use prometheus::Registry;
    use sui_macros::sim_test;
    use sui_protocol_config::{ConsensusTransactionOrdering, ProtocolConfig};
    use sui_types::{
        base_types::ExecutionDigests,
        base_types::{AuthorityName, FullObjectRef, ObjectID, SuiAddress, random_object_ref},
        committee::Committee,
        crypto::deterministic_random_account_key,
        gas::GasCostSummary,
        message_envelope::Message,
        messages_checkpoint::{
            CheckpointContents, CheckpointSignatureMessage, CheckpointSummary,
            SignedCheckpointSummary,
        },
        messages_consensus::ConsensusTransaction,
        object::Object,
        transaction::{
            CertifiedTransaction, TransactionData, TransactionDataAPI, VerifiedCertificate,
        },
    };

    use sui_types::SUI_RANDOMNESS_STATE_OBJECT_ID;
    use sui_types::programmable_transaction_builder::ProgrammableTransactionBuilder;
    use sui_types::transaction::{ObjectArg, SharedObjectMutability};

    use super::*;
    use crate::{
        authority::{
            authority_per_epoch_store::ConsensusStatsAPI,
            consensus_tx_status_cache::NotifyReadConsensusTxStatusResult,
            test_authority_builder::TestAuthorityBuilder,
        },
        checkpoints::CheckpointServiceNoop,
        consensus_adapter::consensus_tests::test_user_transaction,
        consensus_test_utils::{TestConsensusCommit, setup_consensus_handler_for_testing},
        post_consensus_tx_reorder::PostConsensusTxReorder,
    };

    fn epoch_close_deadline_config(deadline_ms: Option<u64>) -> ProtocolConfig {
        let mut protocol_config = ProtocolConfig::get_for_max_version_UNSAFE();
        if let Some(deadline_ms) = deadline_ms {
            protocol_config.set_epoch_close_deadline_ms_for_testing(deadline_ms);
        } else {
            protocol_config.disable_epoch_close_deadline_ms_for_testing();
        }
        protocol_config
    }

    #[tokio::test]
    async fn test_consensus_handler_max_transaction_size() {
        let metrics = AuthorityMetrics::new(&Registry::new());

        metrics.observe_consensus_handler_transaction_size("class_a", "accepted", 100);
        metrics.observe_consensus_handler_transaction_size("class_a", "rejected", 80);
        metrics.observe_consensus_handler_transaction_size("class_b", "accepted", 50);

        assert_eq!(
            metrics
                .consensus_handler_max_transaction_size
                .with_label_values(&["class_a"])
                .get(),
            100
        );
        assert_eq!(
            metrics
                .consensus_handler_max_transaction_size
                .with_label_values(&["class_b"])
                .get(),
            50
        );

        metrics.observe_consensus_handler_transaction_size("class_a", "rejected", 120);

        assert_eq!(
            metrics
                .consensus_handler_max_transaction_size
                .with_label_values(&["class_a"])
                .get(),
            120
        );
    }

    #[sim_test]
    async fn test_epoch_close_deadline_preserves_pre_deadline_blocking() {
        let state = TestAuthorityBuilder::new()
            .with_protocol_config(epoch_close_deadline_config(Some(100)))
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();
        epoch_store.insert_deferred_transactions_for_test(
            DeferralKey::new_for_consensus_round(u64::MAX, 1),
            vec![user_txn(1)],
        );
        let scheduled_end = epoch_store.next_reconfiguration_timestamp_ms();
        let mut setup = setup_consensus_handler_for_testing(&state).await;

        setup
            .consensus_handler
            .handle_consensus_commit_for_test(TestConsensusCommit::empty(1, scheduled_end + 99, 1))
            .await;

        let reconfig_state = epoch_store.get_reconfig_state_read_lock_guard();
        assert!(reconfig_state.is_reject_all_certs());
        assert!(!reconfig_state.is_reject_all_tx());
        assert_eq!(
            epoch_store.get_all_deferred_transactions_for_test().len(),
            1
        );
        assert!(
            epoch_store
                .get_pending_checkpoints(None)
                .iter()
                .all(|(_, checkpoint)| !checkpoint.details.last_of_epoch)
        );
        assert_eq!(
            setup
                .consensus_handler
                .metrics
                .consensus_handler_dropped_transactions
                .with_label_values(&["epoch_close_deadline"])
                .get(),
            0
        );
    }

    #[sim_test]
    async fn test_epoch_close_deadline_none_preserves_indefinite_blocking() {
        let state = TestAuthorityBuilder::new()
            .with_protocol_config(epoch_close_deadline_config(None))
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();
        epoch_store.insert_deferred_transactions_for_test(
            DeferralKey::new_for_consensus_round(u64::MAX, 1),
            vec![user_txn(1)],
        );
        let scheduled_end = epoch_store.next_reconfiguration_timestamp_ms();
        let mut setup = setup_consensus_handler_for_testing(&state).await;

        setup
            .consensus_handler
            .handle_consensus_commit_for_test(TestConsensusCommit::empty(
                1,
                scheduled_end.saturating_add(1_000_000),
                1,
            ))
            .await;

        let reconfig_state = epoch_store.get_reconfig_state_read_lock_guard();
        assert!(reconfig_state.is_reject_all_certs());
        assert!(!reconfig_state.is_reject_all_tx());
    }

    #[sim_test]
    async fn test_epoch_close_deadline_is_inert_without_deferred_transactions() {
        let state = TestAuthorityBuilder::new()
            .with_protocol_config(epoch_close_deadline_config(Some(100)))
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();
        let scheduled_end = epoch_store.next_reconfiguration_timestamp_ms();
        let mut setup = setup_consensus_handler_for_testing(&state).await;

        // Commit timestamp is past the deadline, so deadline_reached is true — with no
        // deferred transactions this must close cleanly with nothing abandoned (no
        // debug_fatal panic, metric stays zero).
        setup
            .consensus_handler
            .handle_consensus_commit_for_test(TestConsensusCommit::empty(1, scheduled_end + 100, 1))
            .await;

        assert!(
            epoch_store
                .get_reconfig_state_read_lock_guard()
                .is_reject_all_tx()
        );
        let checkpoints = epoch_store.get_pending_checkpoints(None);
        assert!(checkpoints.last().unwrap().1.details.last_of_epoch);
        assert_eq!(
            setup
                .consensus_handler
                .metrics
                .consensus_handler_dropped_transactions
                .with_label_values(&["epoch_close_deadline"])
                .get(),
            0
        );
    }

    #[sim_test]
    async fn test_epoch_close_deadline_counts_abandoned_transactions_and_closes_first() {
        let state = TestAuthorityBuilder::new()
            .with_protocol_config(epoch_close_deadline_config(Some(100)))
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();
        let key = DeferralKey::new_for_consensus_round(u64::MAX, 1);
        epoch_store.insert_deferred_transactions_for_test(key, vec![user_txn(1), user_txn(2)]);
        let setup = setup_consensus_handler_for_testing(&state).await;
        let mut handler_state = CommitHandlerState::new(&epoch_store, 1);

        let (reconfig_state, final_round, abandoned) = setup
            .consensus_handler
            .advance_end_of_epoch_state_machine(&mut handler_state, true);

        assert!(reconfig_state.is_reject_all_tx());
        assert!(final_round);
        let abandoned = abandoned.expect("deferred transactions must be reported as abandoned");
        assert_eq!(abandoned.count, 2);
        assert_eq!(abandoned.sample.len(), 2);
        // The abandoned key must be staged for deletion so the final commit clears both the
        // db table and (via record_deferral_deletion) the in-memory cache.
        assert!(
            handler_state
                .output
                .get_deleted_deferred_txn_keys()
                .any(|deleted| deleted == key)
        );
    }

    #[sim_test]
    async fn test_epoch_close_deadline_does_not_report_transactions_drained_in_commit() {
        let state = TestAuthorityBuilder::new()
            .with_protocol_config(epoch_close_deadline_config(Some(100)))
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();
        let key = DeferralKey::new_for_consensus_round(1, 0);
        epoch_store.insert_deferred_transactions_for_test(key, vec![user_txn(1), user_txn(2)]);
        let setup = setup_consensus_handler_for_testing(&state).await;
        let mut handler_state = CommitHandlerState::new(&epoch_store, 1);
        handler_state
            .output
            .delete_loaded_deferred_transactions(&[key]);

        let (reconfig_state, final_round, abandoned) = setup
            .consensus_handler
            .advance_end_of_epoch_state_machine(&mut handler_state, true);

        assert!(reconfig_state.is_reject_all_tx());
        assert!(final_round);
        assert!(abandoned.is_none());
    }

    // debug_fatal only panics when crash_on_debug() is true (debug_assertions, msim, or
    // SUI_ENABLE_DEBUG_ASSERTIONS); in a plain release build it logs and continues, so the
    // should_panic expectation would fail there.
    #[cfg(debug_assertions)]
    #[tokio::test(flavor = "current_thread")]
    #[should_panic(
        expected = "Epoch close deadline reached with unscheduled deferred transactions"
    )]
    async fn test_epoch_close_deadline_timestamp_jump_abandons_fresh_deferral() {
        use sui_protocol_config::{ExecutionTimeEstimateParams, PerObjectCongestionControlMode};

        let execution_time_params = ExecutionTimeEstimateParams {
            target_utilization: 1,
            allowed_txn_cost_overage_burst_limit_us: 0,
            max_estimate_us: u64::MAX,
            randomness_scalar: 100,
            stored_observations_num_included_checkpoints: 10,
            stored_observations_limit: u64::MAX,
            stake_weighted_median_threshold: 0,
            default_none_duration_for_new_keys: true,
            observations_chunk_size: None,
        };
        let mut protocol_config = epoch_close_deadline_config(Some(100));
        protocol_config.set_per_object_congestion_control_mode_for_testing(
            PerObjectCongestionControlMode::ExecutionTimeEstimate(execution_time_params),
        );
        protocol_config.set_max_deferral_rounds_for_congestion_control_for_testing(1_000);

        let (sender, keypair) = deterministic_random_account_key();
        let gas_objects: Vec<_> = (0..4)
            .map(|_| Object::with_id_owner_for_testing(ObjectID::random(), sender))
            .collect();
        let shared_object = Object::shared_for_testing();
        let mut starting_objects = gas_objects.clone();
        starting_objects.push(shared_object.clone());
        let state = TestAuthorityBuilder::new()
            .with_starting_objects(&starting_objects)
            .with_protocol_config(protocol_config)
            .build()
            .await;
        let mut consensus_transactions = Vec::new();
        for gas_object in gas_objects {
            let transaction = test_user_transaction(
                &state,
                sender,
                &keypair,
                gas_object,
                vec![shared_object.clone()],
            )
            .await;
            consensus_transactions.push(ConsensusTransaction::new_user_transaction_v2_message(
                &state.name,
                transaction.into(),
            ));
        }
        let epoch_store = state.epoch_store_for_testing();
        let scheduled_end = epoch_store.next_reconfiguration_timestamp_ms();
        let mut setup = setup_consensus_handler_for_testing(&state).await;

        setup
            .consensus_handler
            .handle_consensus_commit_for_test(TestConsensusCommit::new(
                consensus_transactions,
                1,
                scheduled_end + 100,
                1,
            ))
            .await;
    }

    #[sim_test]
    async fn test_consensus_commit_handler() {
        telemetry_subscribers::init_for_testing();

        // GIVEN
        // 1 account keypair
        let (sender, keypair) = deterministic_random_account_key();
        // 12 gas objects.
        let gas_objects: Vec<Object> = (0..12)
            .map(|_| Object::with_id_owner_for_testing(ObjectID::random(), sender))
            .collect();
        // 4 owned objects.
        let owned_objects: Vec<Object> = (0..4)
            .map(|_| Object::with_id_owner_for_testing(ObjectID::random(), sender))
            .collect();
        // 6 shared objects.
        let shared_objects: Vec<Object> = (0..6)
            .map(|_| Object::shared_for_testing())
            .collect::<Vec<_>>();
        let mut all_objects = gas_objects.clone();
        all_objects.extend(owned_objects.clone());
        all_objects.extend(shared_objects.clone());

        let network_config =
            sui_swarm_config::network_config_builder::ConfigBuilder::new_with_temp_dir()
                .with_objects(all_objects.clone())
                .build();

        let state = TestAuthorityBuilder::new()
            .with_network_config(&network_config, 0)
            .build()
            .await;

        let epoch_store = state.epoch_store_for_testing().clone();
        let new_epoch_start_state = epoch_store.epoch_start_state();
        let consensus_committee = new_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 settlement_scheduler = SettlementScheduler::new(
            state.execution_scheduler().as_ref().clone(),
            state.get_transaction_cache_reader().clone(),
            state.metrics.clone(),
        );
        let mut consensus_handler = ConsensusHandler::new(
            epoch_store,
            Arc::new(CheckpointServiceNoop {}),
            settlement_scheduler,
            state.get_object_cache_reader().clone(),
            consensus_committee.clone(),
            metrics,
            Arc::new(throughput_calculator),
            backpressure_manager.subscribe(),
            state.traffic_controller.clone(),
            None,
            state.consensus_gasless_counter.clone(),
            state.transaction_deny_config_manager().clone(),
        );

        // AND create test user transactions alternating between owned and shared input.
        let mut user_transactions = vec![];
        for (i, gas_object) in gas_objects[0..8].iter().enumerate() {
            let input_object = if i % 2 == 0 {
                owned_objects.get(i / 2).unwrap().clone()
            } else {
                shared_objects.get(i / 2).unwrap().clone()
            };
            let transaction = test_user_transaction(
                &state,
                sender,
                &keypair,
                gas_object.clone(),
                vec![input_object],
            )
            .await;
            user_transactions.push(transaction);
        }

        // AND create 4 more user transactions with remaining gas objects and 2 shared objects.
        // Having more txns on the same shared object may get deferred.
        for (i, gas_object) in gas_objects[8..12].iter().enumerate() {
            let shared_object = if i < 2 {
                shared_objects[4].clone()
            } else {
                shared_objects[5].clone()
            };
            let transaction = test_user_transaction(
                &state,
                sender,
                &keypair,
                gas_object.clone(),
                vec![shared_object],
            )
            .await;
            user_transactions.push(transaction);
        }

        // AND create block for each user transaction
        let mut blocks = Vec::new();
        for (i, consensus_transaction) in user_transactions
            .iter()
            .cloned()
            .map(|t| ConsensusTransaction::new_user_transaction_v2_message(&state.name, t.into()))
            .enumerate()
        {
            let transaction_bytes = bcs::to_bytes(&consensus_transaction).unwrap();
            let block = VerifiedBlock::new_for_test(
                TestBlock::new(100 + i as u32, (i % consensus_committee.size()) as u32)
                    .set_transactions(vec![Transaction::new(transaction_bytes)])
                    .build(),
            );

            blocks.push(block);
        }

        // AND create the consensus commit
        let leader_block = blocks[0].clone();
        let committed_sub_dag = CommittedSubDag::new(
            leader_block.reference(),
            blocks.clone(),
            leader_block.timestamp_ms(),
            CommitRef::new(10, CommitDigest::MIN),
        );

        // Test that the consensus handler respects backpressure.
        backpressure_manager.set_backpressure(true);
        // Default watermarks are 0,0 which will suppress the backpressure.
        backpressure_manager.update_highest_certified_checkpoint(1);

        // AND process the consensus commit once
        {
            let waiter =
                consensus_handler.handle_consensus_commit_for_test(committed_sub_dag.clone());
            pin_mut!(waiter);

            // waiter should not complete within 5 seconds
            tokio::time::timeout(std::time::Duration::from_secs(5), &mut waiter)
                .await
                .unwrap_err();

            // lift backpressure
            backpressure_manager.set_backpressure(false);

            // waiter completes now.
            tokio::time::timeout(std::time::Duration::from_secs(100), waiter)
                .await
                .unwrap();
        }

        // THEN check the consensus stats
        let num_blocks = blocks.len();
        let num_transactions = user_transactions.len();
        let last_consensus_stats_1 = consensus_handler.last_consensus_stats.clone();
        assert_eq!(
            last_consensus_stats_1.index.transaction_index,
            num_transactions as u64
        );
        assert_eq!(last_consensus_stats_1.index.sub_dag_index, 10_u64);
        assert_eq!(last_consensus_stats_1.index.last_committed_round, 100_u64);
        assert_eq!(
            last_consensus_stats_1.stats.get_num_messages(0),
            num_blocks as u64
        );
        assert_eq!(
            last_consensus_stats_1.stats.get_num_user_transactions(0),
            num_transactions as u64
        );

        // THEN check for execution status of user transactions.
        for (i, t) in user_transactions.iter().enumerate() {
            let digest = t.tx().digest();
            if tokio::time::timeout(
                std::time::Duration::from_secs(10),
                state.notify_read_effects_for_testing("", *digest),
            )
            .await
            .is_ok()
            {
                // Effects exist as expected.
            } else {
                panic!("User transaction {} {} did not execute", i, digest);
            }
        }

        // THEN check for no inflight or suspended transactions.
        state.execution_scheduler().check_empty_for_testing().await;
    }

    #[sim_test]
    async fn test_dropped_owned_object_lock_conflict_is_marked_processed() {
        telemetry_subscribers::init_for_testing();

        let (sender, keypair) = deterministic_random_account_key();
        let gas_objects: Vec<Object> = (0..2)
            .map(|_| Object::with_id_owner_for_testing(ObjectID::random(), sender))
            .collect();
        let owned_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);
        let mut all_objects = gas_objects.clone();
        all_objects.push(owned_object.clone());

        let state = TestAuthorityBuilder::new()
            .with_starting_objects(&all_objects)
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();
        let owned_object_ref = state
            .get_object(&owned_object.id())
            .unwrap()
            .compute_object_reference();

        let winner = test_user_transaction(
            &state,
            sender,
            &keypair,
            gas_objects[0].clone(),
            vec![owned_object.clone()],
        )
        .await;
        let loser = test_user_transaction(
            &state,
            sender,
            &keypair,
            gas_objects[1].clone(),
            vec![owned_object.clone()],
        )
        .await;

        let winner_digest = *winner.tx().digest();
        let loser_digest = *loser.tx().digest();
        assert_ne!(winner_digest, loser_digest);

        let winner_consensus_tx =
            ConsensusTransaction::new_user_transaction_v2_message(&state.name, winner.into());
        let loser_consensus_tx =
            ConsensusTransaction::new_user_transaction_v2_message(&state.name, loser.into());
        let winner_key = SequencedConsensusTransactionKey::External(winner_consensus_tx.key());
        let loser_key = SequencedConsensusTransactionKey::External(loser_consensus_tx.key());

        let round = 100;
        let commit = TestConsensusCommit::new(
            vec![winner_consensus_tx, loser_consensus_tx],
            round as u64,
            1_000,
            10,
        );
        let mut setup = setup_consensus_handler_for_testing(&state).await;
        setup
            .consensus_handler
            .handle_consensus_commit_for_test(commit)
            .await;
        assert_eq!(
            setup
                .consensus_handler
                .metrics
                .consensus_handler_dropped_transactions
                .with_label_values(&["lock_conflict"])
                .get(),
            1
        );

        let block = BlockRef {
            author: consensus_config::AuthorityIndex::ZERO,
            round,
            digest: Default::default(),
        };
        assert!(matches!(
            epoch_store
                .consensus_tx_status_cache
                .notify_read_transaction_status(ConsensusPosition {
                    epoch: epoch_store.epoch(),
                    block,
                    index: 0,
                })
                .await,
            NotifyReadConsensusTxStatusResult::Status(ConsensusTxStatus::Finalized)
        ));
        assert!(matches!(
            epoch_store
                .consensus_tx_status_cache
                .notify_read_transaction_status(ConsensusPosition {
                    epoch: epoch_store.epoch(),
                    block,
                    index: 1,
                })
                .await,
            NotifyReadConsensusTxStatusResult::Status(ConsensusTxStatus::Dropped)
        ));

        let locks = epoch_store.get_owned_object_locks_map(&[owned_object_ref]);
        assert_eq!(locks.get(&owned_object_ref), Some(&winner_digest));
        assert!(
            epoch_store
                .is_consensus_message_processed(&winner_key)
                .unwrap()
        );
        assert!(
            epoch_store
                .is_consensus_message_processed(&loser_key)
                .unwrap()
        );
        // The processed notification resolves for the dropped key as well — this is
        // the signal consensus adapter waiters block on.
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            epoch_store.consensus_messages_processed_notify(vec![loser_key]),
        )
        .await
        .expect("processed notification for dropped transaction should resolve");
    }

    #[sim_test]
    async fn test_rejected_transaction_sets_status_and_is_not_marked_processed() {
        telemetry_subscribers::init_for_testing();

        let (sender, keypair) = deterministic_random_account_key();
        let gas_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);
        let owned_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);

        let state = TestAuthorityBuilder::new()
            .with_starting_objects(&[gas_object.clone(), owned_object.clone()])
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();

        let transaction =
            test_user_transaction(&state, sender, &keypair, gas_object, vec![owned_object]).await;
        let consensus_tx =
            ConsensusTransaction::new_user_transaction_v2_message(&state.name, transaction.into());
        let key = SequencedConsensusTransactionKey::External(consensus_tx.key());

        let round = 100;
        let commit = TestConsensusCommit::new(vec![consensus_tx], round as u64, 1_000, 10)
            .with_rejected_indices([0]);
        let mut setup = setup_consensus_handler_for_testing(&state).await;
        setup
            .consensus_handler
            .handle_consensus_commit_for_test(commit)
            .await;

        // The rejected position receives a terminal Rejected status.
        let block = BlockRef {
            author: consensus_config::AuthorityIndex::ZERO,
            round,
            digest: Default::default(),
        };
        assert!(matches!(
            epoch_store
                .consensus_tx_status_cache
                .notify_read_transaction_status(ConsensusPosition {
                    epoch: epoch_store.epoch(),
                    block,
                    index: 0,
                })
                .await,
            NotifyReadConsensusTxStatusResult::Status(ConsensusTxStatus::Rejected)
        ));
        // Unlike dropped transactions, a rejected transaction must NOT be marked
        // consensus-processed: rejection is per-position, and the digest must stay
        // resubmittable within the epoch so a later occurrence can be finalized.
        assert!(!epoch_store.is_consensus_message_processed(&key).unwrap());
    }

    #[sim_test]
    async fn test_user_transaction_ignored_at_epoch_close_sets_dropped_status() {
        telemetry_subscribers::init_for_testing();

        let (sender, keypair) = deterministic_random_account_key();
        let gas_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);
        let owned_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);

        let state = TestAuthorityBuilder::new()
            .with_starting_objects(&[gas_object.clone(), owned_object.clone()])
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();

        let transaction =
            test_user_transaction(&state, sender, &keypair, gas_object, vec![owned_object]).await;
        let consensus_tx =
            ConsensusTransaction::new_user_transaction_v2_message(&state.name, transaction.into());
        let key = SequencedConsensusTransactionKey::External(consensus_tx.key());

        // Close the epoch to the point where consensus certs are no longer accepted.
        {
            let mut guard = epoch_store.get_reconfig_state_write_lock_guard();
            guard.close_all_certs();
        }

        let round = 100;
        let commit = TestConsensusCommit::new(vec![consensus_tx], round as u64, 1_000, 10);
        let mut setup = setup_consensus_handler_for_testing(&state).await;
        setup
            .consensus_handler
            .handle_consensus_commit_for_test(commit)
            .await;

        // The ignored position receives a terminal Dropped status so submission and
        // effects waiters are not leaked until epoch termination.
        let block = BlockRef {
            author: consensus_config::AuthorityIndex::ZERO,
            round,
            digest: Default::default(),
        };
        let status = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            epoch_store
                .consensus_tx_status_cache
                .notify_read_transaction_status(ConsensusPosition {
                    epoch: epoch_store.epoch(),
                    block,
                    index: 0,
                }),
        )
        .await
        .expect("transaction ignored at epoch close should receive a terminal status");
        assert!(matches!(
            status,
            NotifyReadConsensusTxStatusResult::Status(ConsensusTxStatus::Dropped)
        ));
        assert!(!epoch_store.is_consensus_message_processed(&key).unwrap());
        assert_eq!(
            setup
                .consensus_handler
                .metrics
                .consensus_handler_dropped_transactions
                .with_label_values(&["end_of_epoch"])
                .get(),
            1
        );
    }

    #[sim_test]
    async fn test_user_transaction_after_end_of_publish_sets_dropped_status() {
        telemetry_subscribers::init_for_testing();

        let (sender, keypair) = deterministic_random_account_key();
        let gas_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);
        let owned_object = Object::with_id_owner_for_testing(ObjectID::random(), sender);

        let state = TestAuthorityBuilder::new()
            .with_starting_objects(&[gas_object.clone(), owned_object.clone()])
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();

        let transaction =
            test_user_transaction(&state, sender, &keypair, gas_object, vec![owned_object]).await;
        let consensus_tx =
            ConsensusTransaction::new_user_transaction_v2_message(&state.name, transaction.into());
        let key = SequencedConsensusTransactionKey::External(consensus_tx.key());

        // Record that this authority (the block author in TestConsensusCommit) already
        // sent EndOfPublish, without advancing the reconfig state, so the commit below
        // exercises the post-EndOfPublish author filter rather than the certs-closed one.
        epoch_store
            .end_of_publish
            .try_lock()
            .unwrap()
            .insert_generic(state.name, ());

        let round = 100;
        let commit = TestConsensusCommit::new(vec![consensus_tx], round as u64, 1_000, 10);
        let mut setup = setup_consensus_handler_for_testing(&state).await;
        setup
            .consensus_handler
            .handle_consensus_commit_for_test(commit)
            .await;

        let block = BlockRef {
            author: consensus_config::AuthorityIndex::ZERO,
            round,
            digest: Default::default(),
        };
        let status = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            epoch_store
                .consensus_tx_status_cache
                .notify_read_transaction_status(ConsensusPosition {
                    epoch: epoch_store.epoch(),
                    block,
                    index: 0,
                }),
        )
        .await
        .expect("transaction ignored after EndOfPublish should receive a terminal status");
        assert!(matches!(
            status,
            NotifyReadConsensusTxStatusResult::Status(ConsensusTxStatus::Dropped)
        ));
        assert!(!epoch_store.is_consensus_message_processed(&key).unwrap());
        assert_eq!(
            setup
                .consensus_handler
                .metrics
                .consensus_handler_dropped_transactions
                .with_label_values(&["end_of_publish"])
                .get(),
            1
        );
    }

    fn to_short_strings(txs: Vec<VerifiedExecutableTransactionWithAliases>) -> Vec<String> {
        txs.into_iter()
            .map(|tx| format!("transaction({})", tx.tx().transaction_data().gas_price()))
            .collect()
    }

    #[test]
    fn test_order_by_gas_price() {
        let mut v = vec![user_txn(42), user_txn(100)];
        PostConsensusTxReorder::reorder(&mut v, ConsensusTransactionOrdering::ByGasPrice);
        assert_eq!(
            to_short_strings(v),
            vec![
                "transaction(100)".to_string(),
                "transaction(42)".to_string(),
            ]
        );

        let mut v = vec![
            user_txn(1200),
            user_txn(12),
            user_txn(1000),
            user_txn(42),
            user_txn(100),
            user_txn(1000),
        ];
        PostConsensusTxReorder::reorder(&mut v, ConsensusTransactionOrdering::ByGasPrice);
        assert_eq!(
            to_short_strings(v),
            vec![
                "transaction(1200)".to_string(),
                "transaction(1000)".to_string(),
                "transaction(1000)".to_string(),
                "transaction(100)".to_string(),
                "transaction(42)".to_string(),
                "transaction(12)".to_string(),
            ]
        );
    }

    #[sim_test]
    async fn test_checkpoint_signature_dedup() {
        telemetry_subscribers::init_for_testing();

        let network_config =
            sui_swarm_config::network_config_builder::ConfigBuilder::new_with_temp_dir().build();
        let state = TestAuthorityBuilder::new()
            .with_network_config(&network_config, 0)
            .build()
            .await;

        let epoch_store = state.epoch_store_for_testing().clone();
        let consensus_committee = epoch_store.epoch_start_state().get_consensus_committee();

        let make_signed = || {
            let epoch = epoch_store.epoch();
            let contents =
                CheckpointContents::new_with_digests_only_for_tests([ExecutionDigests::random()]);
            let summary = CheckpointSummary::new(
                &ProtocolConfig::get_for_max_version_UNSAFE(),
                epoch,
                42, // sequence number
                10, // network_total_transactions
                &contents,
                None, // previous_digest
                GasCostSummary::default(),
                None,       // end_of_epoch_data
                0,          // timestamp
                Vec::new(), // randomness_rounds
                Vec::new(), // checkpoint_artifact_digests
            );
            SignedCheckpointSummary::new(epoch, summary, &*state.secret, state.name)
        };

        // Prepare V2 pair: same (authority, seq), different digests => different keys
        let v2_s1 = make_signed();
        let v2_s1_clone = v2_s1.clone();
        let v2_digest_a = v2_s1.data().digest();
        let v2_a =
            ConsensusTransaction::new_checkpoint_signature_message_v2(CheckpointSignatureMessage {
                summary: v2_s1,
            });

        let v2_s2 = make_signed();
        let v2_digest_b = v2_s2.data().digest();
        let v2_b =
            ConsensusTransaction::new_checkpoint_signature_message_v2(CheckpointSignatureMessage {
                summary: v2_s2,
            });

        assert_ne!(v2_digest_a, v2_digest_b);

        // Create an exact duplicate with same digest to exercise valid dedup
        assert_eq!(v2_s1_clone.data().digest(), v2_digest_a);
        let v2_dup =
            ConsensusTransaction::new_checkpoint_signature_message_v2(CheckpointSignatureMessage {
                summary: v2_s1_clone,
            });

        let to_tx = |ct: &ConsensusTransaction| Transaction::new(bcs::to_bytes(ct).unwrap());
        let block = VerifiedBlock::new_for_test(
            TestBlock::new(100, 0)
                .set_transactions(vec![to_tx(&v2_a), to_tx(&v2_b), to_tx(&v2_dup)])
                .build(),
        );
        let commit = CommittedSubDag::new(
            block.reference(),
            vec![block.clone()],
            block.timestamp_ms(),
            CommitRef::new(10, CommitDigest::MIN),
        );

        let metrics = Arc::new(AuthorityMetrics::new(&Registry::new()));
        let throughput = ConsensusThroughputCalculator::new(None, metrics.clone());
        let backpressure = BackpressureManager::new_for_tests();
        let settlement_scheduler = SettlementScheduler::new(
            state.execution_scheduler().as_ref().clone(),
            state.get_transaction_cache_reader().clone(),
            state.metrics.clone(),
        );
        let mut handler = ConsensusHandler::new(
            epoch_store.clone(),
            Arc::new(CheckpointServiceNoop {}),
            settlement_scheduler,
            state.get_object_cache_reader().clone(),
            consensus_committee.clone(),
            metrics,
            Arc::new(throughput),
            backpressure.subscribe(),
            state.traffic_controller.clone(),
            None,
            state.consensus_gasless_counter.clone(),
            state.transaction_deny_config_manager().clone(),
        );

        handler.handle_consensus_commit_for_test(commit).await;

        use crate::consensus_handler::SequencedConsensusTransactionKey as SK;
        use sui_types::messages_consensus::ConsensusTransactionKey as CK;

        // V2 distinct digests: both must be processed. If these were collapsed to one CheckpointSeq num, only one would process.
        let v2_key_a = SK::External(CK::CheckpointSignatureV2(state.name, 42, v2_digest_a));
        let v2_key_b = SK::External(CK::CheckpointSignatureV2(state.name, 42, v2_digest_b));
        assert!(
            epoch_store
                .is_consensus_message_processed(&v2_key_a)
                .unwrap()
        );
        assert!(
            epoch_store
                .is_consensus_message_processed(&v2_key_b)
                .unwrap()
        );
    }

    #[sim_test]
    async fn test_verify_consensus_transaction_filters_mismatched_authorities() {
        telemetry_subscribers::init_for_testing();

        let network_config =
            sui_swarm_config::network_config_builder::ConfigBuilder::new_with_temp_dir().build();
        let state = TestAuthorityBuilder::new()
            .with_network_config(&network_config, 0)
            .build()
            .await;

        let epoch_store = state.epoch_store_for_testing().clone();
        let consensus_committee = epoch_store.epoch_start_state().get_consensus_committee();

        // Create a different authority than our test authority
        use fastcrypto::traits::KeyPair;
        let (_, wrong_keypair) = sui_types::crypto::get_authority_key_pair();
        let wrong_authority: AuthorityName = wrong_keypair.public().into();

        // Create EndOfPublish transaction with mismatched authority
        let mismatched_eop = ConsensusTransaction::new_end_of_publish(wrong_authority);

        // Create valid EndOfPublish transaction with correct authority
        let valid_eop = ConsensusTransaction::new_end_of_publish(state.name);

        // Create CheckpointSignature with mismatched authority
        let epoch = epoch_store.epoch();
        let contents =
            CheckpointContents::new_with_digests_only_for_tests([ExecutionDigests::random()]);
        let summary = CheckpointSummary::new(
            &ProtocolConfig::get_for_max_version_UNSAFE(),
            epoch,
            42, // sequence number
            10, // network_total_transactions
            &contents,
            None, // previous_digest
            GasCostSummary::default(),
            None,       // end_of_epoch_data
            0,          // timestamp
            Vec::new(), // randomness_rounds
            Vec::new(), // checkpoint commitments
        );

        // Create a signed checkpoint with the wrong authority
        let mismatched_checkpoint_signed =
            SignedCheckpointSummary::new(epoch, summary.clone(), &wrong_keypair, wrong_authority);
        let mismatched_checkpoint_digest = mismatched_checkpoint_signed.data().digest();
        let mismatched_checkpoint =
            ConsensusTransaction::new_checkpoint_signature_message_v2(CheckpointSignatureMessage {
                summary: mismatched_checkpoint_signed,
            });

        // Create a valid checkpoint signature with correct authority
        let valid_checkpoint_signed =
            SignedCheckpointSummary::new(epoch, summary, &*state.secret, state.name);
        let valid_checkpoint_digest = valid_checkpoint_signed.data().digest();
        let valid_checkpoint =
            ConsensusTransaction::new_checkpoint_signature_message_v2(CheckpointSignatureMessage {
                summary: valid_checkpoint_signed,
            });

        let to_tx = |ct: &ConsensusTransaction| Transaction::new(bcs::to_bytes(ct).unwrap());

        // Create a block with both valid and invalid transactions
        let block = VerifiedBlock::new_for_test(
            TestBlock::new(100, 0)
                .set_transactions(vec![
                    to_tx(&mismatched_eop),
                    to_tx(&valid_eop),
                    to_tx(&mismatched_checkpoint),
                    to_tx(&valid_checkpoint),
                ])
                .build(),
        );
        let commit = CommittedSubDag::new(
            block.reference(),
            vec![block.clone()],
            block.timestamp_ms(),
            CommitRef::new(10, CommitDigest::MIN),
        );

        let metrics = Arc::new(AuthorityMetrics::new(&Registry::new()));
        let throughput = ConsensusThroughputCalculator::new(None, metrics.clone());
        let backpressure = BackpressureManager::new_for_tests();
        let settlement_scheduler = SettlementScheduler::new(
            state.execution_scheduler().as_ref().clone(),
            state.get_transaction_cache_reader().clone(),
            state.metrics.clone(),
        );
        let mut handler = ConsensusHandler::new(
            epoch_store.clone(),
            Arc::new(CheckpointServiceNoop {}),
            settlement_scheduler,
            state.get_object_cache_reader().clone(),
            consensus_committee.clone(),
            metrics,
            Arc::new(throughput),
            backpressure.subscribe(),
            state.traffic_controller.clone(),
            None,
            state.consensus_gasless_counter.clone(),
            state.transaction_deny_config_manager().clone(),
        );

        handler.handle_consensus_commit_for_test(commit).await;

        use crate::consensus_handler::SequencedConsensusTransactionKey as SK;
        use sui_types::messages_consensus::ConsensusTransactionKey as CK;

        // Check that valid transactions were processed
        let valid_eop_key = SK::External(CK::EndOfPublish(state.name));
        assert!(
            epoch_store
                .is_consensus_message_processed(&valid_eop_key)
                .unwrap(),
            "Valid EndOfPublish should have been processed"
        );

        let valid_checkpoint_key = SK::External(CK::CheckpointSignatureV2(
            state.name,
            42,
            valid_checkpoint_digest,
        ));
        assert!(
            epoch_store
                .is_consensus_message_processed(&valid_checkpoint_key)
                .unwrap(),
            "Valid CheckpointSignature should have been processed"
        );

        // Check that mismatched authority transactions were NOT processed (filtered out by verify_consensus_transaction)
        let mismatched_eop_key = SK::External(CK::EndOfPublish(wrong_authority));
        assert!(
            !epoch_store
                .is_consensus_message_processed(&mismatched_eop_key)
                .unwrap(),
            "Mismatched EndOfPublish should NOT have been processed (filtered by verify_consensus_transaction)"
        );

        let mismatched_checkpoint_key = SK::External(CK::CheckpointSignatureV2(
            wrong_authority,
            42,
            mismatched_checkpoint_digest,
        ));
        assert!(
            !epoch_store
                .is_consensus_message_processed(&mismatched_checkpoint_key)
                .unwrap(),
            "Mismatched CheckpointSignature should NOT have been processed (filtered by verify_consensus_transaction)"
        );
    }

    /// Committed deny-config updates are applied by the commit handler even though the
    /// live path already applies them at block verification: a validator that is
    /// catching up can process commits without verifying every block. Updates are
    /// attributed to the consensus block author, so spoofed authority claims are
    /// still dropped.
    #[sim_test]
    async fn test_deny_config_updates_applied_at_commit() {
        telemetry_subscribers::init_for_testing();

        let _guard = ProtocolConfig::apply_overrides_for_testing(|_, mut c| {
            c.set_share_transaction_deny_config_in_consensus_for_testing(true);
            c
        });

        // A single-validator committee, so block author index 0 maps to `state.name`.
        let network_config =
            sui_swarm_config::network_config_builder::ConfigBuilder::new_with_temp_dir()
                .committee_size(std::num::NonZeroUsize::new(1).unwrap())
                .build();
        let state = TestAuthorityBuilder::new()
            .with_network_config(&network_config, 0)
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing().clone();
        let consensus_committee = epoch_store.epoch_start_state().get_consensus_committee();
        let manager = state.transaction_deny_config_manager().clone();

        let now_ms = crate::authority::AuthorityState::unixtime_now_ms();
        let make_update = |authority, generation| {
            ConsensusTransaction::new_update_transaction_deny_config(
                SharedTransactionDenyConfig::V1(
                    sui_types::messages_consensus::SharedTransactionDenyConfigV1 {
                        authority,
                        generation,
                        rules: Some(sui_types::transaction_deny_rules::TransactionDenyRules {
                            package_publish_disabled: true,
                            ..Default::default()
                        }),
                    },
                ),
            )
        };

        let spoofed = make_update(AuthorityName::ZERO, now_ms + 1);
        let sane = make_update(state.name, now_ms);

        let to_tx = |ct: &ConsensusTransaction| Transaction::new(bcs::to_bytes(ct).unwrap());
        let block = VerifiedBlock::new_for_test(
            TestBlock::new(100, 0)
                .set_transactions(vec![to_tx(&spoofed), to_tx(&sane)])
                .build(),
        );
        let commit = CommittedSubDag::new(
            block.reference(),
            vec![block.clone()],
            block.timestamp_ms(),
            CommitRef::new(10, CommitDigest::MIN),
        );

        let metrics = Arc::new(AuthorityMetrics::new(&Registry::new()));
        let throughput = ConsensusThroughputCalculator::new(None, metrics.clone());
        let backpressure = BackpressureManager::new_for_tests();
        let settlement_scheduler = SettlementScheduler::new(
            state.execution_scheduler().as_ref().clone(),
            state.get_transaction_cache_reader().clone(),
            state.metrics.clone(),
        );
        let mut handler = ConsensusHandler::new(
            epoch_store.clone(),
            Arc::new(CheckpointServiceNoop {}),
            settlement_scheduler,
            state.get_object_cache_reader().clone(),
            consensus_committee.clone(),
            metrics,
            Arc::new(throughput),
            backpressure.subscribe(),
            state.traffic_controller.clone(),
            None,
            state.consensus_gasless_counter.clone(),
            state.transaction_deny_config_manager().clone(),
        );

        handler.handle_consensus_commit_for_test(commit).await;

        let snapshot = manager.peer_configs_snapshot();
        assert_eq!(
            snapshot.get(&state.name).map(|msg| msg.generation()),
            Some(now_ms),
            "committed update from the block author should be applied at commit time"
        );
        assert!(
            !snapshot.contains_key(&AuthorityName::ZERO),
            "spoofed authority claim should be dropped"
        );
    }

    /// A transaction whose shared inputs include the randomness state object, so
    /// `uses_randomness()` is true. Gas is random, so every call yields a new digest.
    fn randomness_user_txn() -> VerifiedExecutableTransactionWithAliases {
        let (committee, keypairs) = Committee::new_simple_test_committee();
        let (sender, sender_keypair) = deterministic_random_account_key();
        let mut builder = ProgrammableTransactionBuilder::new();
        builder
            .obj(ObjectArg::SharedObject {
                id: SUI_RANDOMNESS_STATE_OBJECT_ID,
                initial_shared_version: 1.into(),
                mutability: SharedObjectMutability::Immutable,
            })
            .unwrap();
        let tx = sui_types::transaction::Transaction::from_data_and_signer(
            TransactionData::new_programmable(
                sender,
                vec![random_object_ref()],
                builder.finish(),
                1_000_000,
                1_000,
            ),
            vec![&sender_keypair],
        );
        let tx = VerifiedExecutableTransaction::new_from_certificate(
            VerifiedCertificate::new_unchecked(
                CertifiedTransaction::new_from_keypairs_for_testing(
                    tx.into_data(),
                    &keypairs,
                    &committee,
                ),
            ),
        );
        VerifiedExecutableTransactionWithAliases::no_aliases(tx)
    }

    /// Seeds the deferral-key collision: a randomness-using transaction deferred at
    /// round 1 by a check that precedes the randomness check (owned-object double spend)
    /// carries ConsensusRound{2, 1}. The test authority's DKG never completes, so no
    /// commit generates randomness: reloaded at round 2, that transaction re-defers to
    /// Randomness{1} - the key still holding round 1's fresh randomness deferrals.
    /// Returns the digests of (parked randomness deferral, re-deferring transaction).
    fn seed_deferral_key_collision(
        epoch_store: &AuthorityPerEpochStore,
    ) -> (TransactionDigest, TransactionDigest) {
        let parked = randomness_user_txn();
        let redeferred = randomness_user_txn();
        let parked_digest = *parked.tx().digest();
        let redeferred_digest = *redeferred.tx().digest();
        // Round 1, no randomness generated: fresh randomness-using transactions were
        // deferred under Randomness{1}...
        epoch_store.insert_deferred_transactions_for_test(
            DeferralKey::new_for_randomness(1),
            vec![parked],
        );
        // ...while another randomness-using transaction won a contested owned-object
        // lock in that commit and was double-spend-deferred under ConsensusRound{2, 1}.
        epoch_store.insert_deferred_transactions_for_test(
            DeferralKey::new_for_consensus_round(2, 1),
            vec![redeferred],
        );
        (parked_digest, redeferred_digest)
    }

    /// With merge_colliding_deferrals enabled (current protocol version), the round 2
    /// re-deferral merges into Randomness{1} instead of displacing its transactions.
    #[sim_test]
    async fn test_deferral_key_collision_merges_entries() {
        let state = TestAuthorityBuilder::new().build().await;
        let epoch_store = state.epoch_store_for_testing();
        let (parked_digest, redeferred_digest) = seed_deferral_key_collision(&epoch_store);

        let mid_epoch = epoch_store
            .next_reconfiguration_timestamp_ms()
            .saturating_sub(10_000);
        let mut setup = setup_consensus_handler_for_testing(&state).await;
        setup
            .consensus_handler
            .handle_consensus_commit_for_test(TestConsensusCommit::empty(2, mid_epoch, 1))
            .await;

        let deferred = epoch_store.get_all_deferred_transactions_for_test();
        assert_eq!(deferred.len(), 1);
        let (key, txns) = &deferred[0];
        assert_eq!(*key, DeferralKey::new_for_randomness(1));
        let digests: Vec<_> = txns.iter().map(|t| *t.tx().digest()).collect();
        // Previously parked transactions keep their position ahead of the re-deferral.
        assert_eq!(digests, vec![parked_digest, redeferred_digest]);
    }

    /// With merge_colliding_deferrals disabled, the last-writer-wins insert of older
    /// protocol versions is preserved and the collision sensor flags the displaced
    /// transactions (debug_fatal panics under test configuration).
    #[tokio::test(flavor = "current_thread")]
    #[should_panic(expected = "Deferral key collision displaced finalized transactions")]
    async fn test_deferral_key_collision_displaces_randomness_deferrals() {
        let mut protocol_config = ProtocolConfig::get_for_max_version_UNSAFE();
        protocol_config.set_merge_colliding_deferrals_for_testing(false);
        let state = TestAuthorityBuilder::new()
            .with_protocol_config(protocol_config)
            .build()
            .await;
        let epoch_store = state.epoch_store_for_testing();
        seed_deferral_key_collision(&epoch_store);

        let mid_epoch = epoch_store
            .next_reconfiguration_timestamp_ms()
            .saturating_sub(10_000);
        let mut setup = setup_consensus_handler_for_testing(&state).await;
        setup
            .consensus_handler
            .handle_consensus_commit_for_test(TestConsensusCommit::empty(2, mid_epoch, 1))
            .await;
    }

    fn user_txn(gas_price: u64) -> VerifiedExecutableTransactionWithAliases {
        let (committee, keypairs) = Committee::new_simple_test_committee();
        let (sender, sender_keypair) = deterministic_random_account_key();
        let tx = sui_types::transaction::Transaction::from_data_and_signer(
            TransactionData::new_transfer(
                SuiAddress::default(),
                FullObjectRef::from_fastpath_ref(random_object_ref()),
                sender,
                random_object_ref(),
                1000 * gas_price,
                gas_price,
            ),
            vec![&sender_keypair],
        );
        let tx = VerifiedExecutableTransaction::new_from_certificate(
            VerifiedCertificate::new_unchecked(
                CertifiedTransaction::new_from_keypairs_for_testing(
                    tx.into_data(),
                    &keypairs,
                    &committee,
                ),
            ),
        );
        VerifiedExecutableTransactionWithAliases::no_aliases(tx)
    }

    mod checkpoint_queue_tests {
        use super::*;
        use consensus_core::CommitRef;
        use sui_types::digests::Digest;

        fn make_chunk(tx_count: usize, height: u64) -> Chunk {
            Chunk {
                schedulables: (0..tx_count)
                    .map(|_| Schedulable::Transaction(user_txn(1000).into_tx()))
                    .collect(),
                settlement: None,
                height,
            }
        }

        fn make_commit_ref(index: u32) -> CommitRef {
            CommitRef {
                index,
                digest: CommitDigest::MIN,
            }
        }

        fn default_versions() -> HashMap<TransactionKey, AssignedVersions> {
            HashMap::new()
        }

        #[test]
        fn test_flush_all_checkpoint_roots() {
            let mut queue = CheckpointQueue::new_for_testing(0, 0, 0, 1000, 0);
            let versions = default_versions();

            queue.push_chunk(
                make_chunk(5, 1),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );
            queue.push_chunk(
                make_chunk(3, 2),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );

            let pending = queue.flush(1000, true);

            assert!(pending.is_some());
            assert!(queue.pending_roots.is_empty());
        }

        #[test]
        fn test_flush_respects_min_checkpoint_interval() {
            let min_interval = 200;
            let mut queue = CheckpointQueue::new_for_testing(1000, 0, 0, 1000, min_interval);
            let versions = default_versions();

            queue.push_chunk(
                make_chunk(5, 1),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );

            let pending = queue.flush(1000 + min_interval - 1, false);
            assert!(pending.is_none());
            assert_eq!(queue.pending_roots.len(), 1);

            let pending = queue.flush(1000 + min_interval, false);
            assert!(pending.is_some());
            assert!(queue.pending_roots.is_empty());
        }

        #[test]
        fn test_push_chunk_flushes_when_exceeds_max() {
            let max_tx = 10;
            let mut queue = CheckpointQueue::new_for_testing(1000, 0, 0, max_tx, 0);
            let versions = default_versions();

            queue.push_chunk(
                make_chunk(max_tx / 2 + 1, 1),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );

            let flushed = queue.push_chunk(
                make_chunk(max_tx / 2 + 1, 2),
                &versions,
                1000,
                make_commit_ref(2),
                Digest::default(),
            );

            assert_eq!(flushed.len(), 1);
            assert_eq!(queue.pending_roots.len(), 1);
        }

        #[test]
        fn test_multiple_chunks_merged_into_one_checkpoint() {
            let mut queue = CheckpointQueue::new_for_testing(0, 0, 0, 1000, 200);
            let versions = default_versions();

            queue.push_chunk(
                make_chunk(10, 1),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );
            queue.push_chunk(
                make_chunk(10, 2),
                &versions,
                1000,
                make_commit_ref(2),
                Digest::default(),
            );
            queue.push_chunk(
                make_chunk(10, 3),
                &versions,
                1000,
                make_commit_ref(3),
                Digest::default(),
            );

            let pending = queue.flush(1000, true).unwrap();

            assert_eq!(pending.roots.len(), 3);
        }

        #[test]
        fn test_push_chunk_handles_overflow() {
            let max_tx = 10;
            let mut queue = CheckpointQueue::new_for_testing(0, 0, 0, max_tx, 0);
            let versions = default_versions();

            let flushed1 = queue.push_chunk(
                make_chunk(max_tx / 2, 1),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );
            assert!(flushed1.is_empty());

            let flushed2 = queue.push_chunk(
                make_chunk(max_tx / 2, 2),
                &versions,
                1000,
                make_commit_ref(2),
                Digest::default(),
            );
            assert!(flushed2.is_empty());

            let flushed3 = queue.push_chunk(
                make_chunk(max_tx / 2, 3),
                &versions,
                1000,
                make_commit_ref(3),
                Digest::default(),
            );
            assert_eq!(flushed3.len(), 1);

            let pending = queue.flush(1000, true);

            for p in pending.iter().chain(flushed3.iter()) {
                let tx_count: usize = p.roots.iter().map(|r| r.tx_roots.len()).sum();
                assert!(tx_count <= max_tx);
            }
        }

        #[test]
        fn test_checkpoint_uses_last_chunk_height() {
            let mut queue = CheckpointQueue::new_for_testing(0, 0, 0, 1000, 0);
            let versions = default_versions();

            queue.push_chunk(
                make_chunk(10, 100),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );
            queue.push_chunk(
                make_chunk(10, 200),
                &versions,
                1000,
                make_commit_ref(2),
                Digest::default(),
            );

            let pending = queue.flush(1000, true).unwrap();

            assert_eq!(pending.details.checkpoint_height, 200);
        }

        #[test]
        fn test_last_built_timestamp_updated_on_flush() {
            let mut queue = CheckpointQueue::new_for_testing(0, 0, 0, 1000, 0);
            let versions = default_versions();

            queue.push_chunk(
                make_chunk(10, 1),
                &versions,
                5000,
                make_commit_ref(1),
                Digest::default(),
            );

            assert_eq!(queue.last_built_timestamp, 0);

            let _ = queue.flush(5000, true);

            assert_eq!(queue.last_built_timestamp, 5000);
        }

        #[test]
        fn test_settlement_info_sent_through_channel() {
            let mut queue = CheckpointQueue::new_for_testing(0, 0, 5, 1000, 0);
            let versions = default_versions();

            let chunk1 = Chunk {
                schedulables: vec![
                    Schedulable::ConsensusCommitPrologue(0, 1, 0),
                    Schedulable::ConsensusCommitPrologue(0, 2, 0),
                    Schedulable::ConsensusCommitPrologue(0, 3, 0),
                ],
                settlement: Some(Schedulable::AccumulatorSettlement(1, 1)),
                height: 1,
            };

            let chunk2 = Chunk {
                schedulables: vec![
                    Schedulable::ConsensusCommitPrologue(0, 4, 0),
                    Schedulable::ConsensusCommitPrologue(0, 5, 0),
                ],
                settlement: Some(Schedulable::AccumulatorSettlement(1, 2)),
                height: 2,
            };

            queue.push_chunk(
                chunk1,
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );
            queue.push_chunk(
                chunk2,
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );
        }

        #[test]
        fn test_settlement_checkpoint_seq_correct_after_flush() {
            let max_tx = 10;
            let initial_seq = 5;
            let (sender, mut receiver) = monitored_mpsc::unbounded_channel("test_settlement_seq");
            let mut queue =
                CheckpointQueue::new_for_testing_with_sender(0, 0, initial_seq, max_tx, 0, sender);
            let versions = default_versions();

            // Push a chunk that partially fills the queue (no flush).
            let chunk1 = Chunk {
                schedulables: (0..max_tx / 2 + 1)
                    .map(|_| Schedulable::Transaction(user_txn(1000).into_tx()))
                    .collect(),
                settlement: Some(Schedulable::AccumulatorSettlement(1, 1)),
                height: 1,
            };
            queue.push_chunk(
                chunk1,
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );

            // Drain the first message from the channel.
            let msg1 = receiver.try_recv().unwrap();
            let settlement1 = msg1.1.unwrap();
            assert_eq!(settlement1.checkpoint_seq, initial_seq);

            // Push a second chunk that triggers a flush of chunk1's roots.
            let chunk2 = Chunk {
                schedulables: (0..max_tx / 2 + 1)
                    .map(|_| Schedulable::Transaction(user_txn(1000).into_tx()))
                    .collect(),
                settlement: Some(Schedulable::AccumulatorSettlement(1, 2)),
                height: 2,
            };
            let flushed = queue.push_chunk(
                chunk2,
                &versions,
                1000,
                make_commit_ref(2),
                Digest::default(),
            );
            assert_eq!(flushed.len(), 1);
            assert_eq!(flushed[0].details.checkpoint_seq, initial_seq);

            // The second settlement must have checkpoint_seq = initial_seq + 1,
            // because the flush incremented current_checkpoint_seq.
            let msg2 = receiver.try_recv().unwrap();
            let settlement2 = msg2.1.unwrap();
            assert_eq!(settlement2.checkpoint_seq, initial_seq + 1);

            // Flush the remaining roots and verify the PendingCheckpoint's seq
            // matches the settlement's seq.
            let pending = queue.flush_forced().unwrap();
            assert_eq!(pending.details.checkpoint_seq, settlement2.checkpoint_seq);
        }

        #[test]
        fn test_checkpoint_seq_increments_on_flush() {
            let mut queue = CheckpointQueue::new_for_testing(0, 0, 10, 1000, 0);
            let versions = default_versions();

            queue.push_chunk(
                make_chunk(5, 1),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );

            let pending = queue.flush(1000, true).unwrap();

            assert_eq!(pending.details.checkpoint_seq, 10);
            assert_eq!(queue.current_checkpoint_seq, 11);
        }

        #[test]
        fn test_multiple_chunks_with_overflow() {
            let max_tx = 10;
            let mut queue = CheckpointQueue::new_for_testing(0, 0, 0, max_tx, 0);
            let versions = default_versions();

            let flushed1 = queue.push_chunk(
                make_chunk(max_tx / 2 + 1, 1),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );
            let flushed2 = queue.push_chunk(
                make_chunk(max_tx / 2 + 1, 2),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );
            let flushed3 = queue.push_chunk(
                make_chunk(max_tx / 2 + 1, 3),
                &versions,
                1000,
                make_commit_ref(1),
                Digest::default(),
            );

            let all_flushed: Vec<_> = flushed1
                .into_iter()
                .chain(flushed2)
                .chain(flushed3)
                .collect();
            assert_eq!(all_flushed.len(), 2);
            assert_eq!(queue.pending_roots.len(), 1);

            for p in &all_flushed {
                let tx_count: usize = p.roots.iter().map(|r| r.tx_roots.len()).sum();
                assert!(tx_count <= max_tx);
            }
        }
    }
}