iota_core/

consensus_adapter.rs

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

use std::{
    collections::HashMap,
    future::Future,
    ops::Deref,
    sync::{
        Arc,
        atomic::{AtomicU64, Ordering},
    },
    time::Instant,
};

use arc_swap::ArcSwap;
use dashmap::{DashMap, try_result::TryResult};
use futures::{
    FutureExt, StreamExt,
    future::{self, Either, select},
    pin_mut,
    stream::FuturesUnordered,
};
use iota_metrics::{GaugeGuard, InflightGuardFutureExt, LATENCY_SEC_BUCKETS, spawn_monitored_task};
use iota_sdk_types::TransactionDigest;
use iota_simulator::anemo::PeerId;
use iota_types::{
    base_types::AuthorityName,
    committee::Committee,
    error::{IotaError, IotaResult},
    fp_ensure,
    messages_consensus::{ConsensusTransaction, ConsensusTransactionKey, ConsensusTransactionKind},
};
use itertools::Itertools;
use parking_lot::RwLockReadGuard;
use prometheus_filtered::{
    Histogram, HistogramVec, IntCounterVec, IntGauge, IntGaugeVec, MetricLevel, Registry,
    register_histogram_vec_with_registry, register_histogram_with_registry,
    register_int_counter_vec_with_registry, register_int_gauge_vec_with_registry,
    register_int_gauge_with_registry,
};
use tokio::{
    sync::{Semaphore, SemaphorePermit, oneshot},
    task::JoinHandle,
    time::{
        Duration, {self},
    },
};
use tracing::{debug, info, trace, warn};

use crate::{
    authority::authority_per_epoch_store::AuthorityPerEpochStore,
    checkpoints::CheckpointStore,
    connection_monitor::ConnectionStatus,
    consensus_handler::{SequencedConsensusTransactionKey, classify},
    epoch::reconfiguration::{ReconfigState, ReconfigurationInitiator},
    metrics::LatencyObserver,
};

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

const SEQUENCING_CERTIFICATE_POSITION_BUCKETS: &[f64] = &[
    0., 1., 2., 3., 5., 10., 15., 20., 25., 30., 50., 100., 150., 200.,
];

pub struct ConsensusAdapterMetrics {
    // Certificate sequencing metrics
    pub sequencing_certificate_attempt: IntCounterVec,
    pub sequencing_certificate_success: IntCounterVec,
    pub sequencing_certificate_failures: IntCounterVec,
    pub sequencing_certificate_status: IntCounterVec,
    pub sequencing_certificate_inflight: IntGaugeVec,
    pub sequencing_acknowledge_latency: HistogramVec,
    pub sequencing_certificate_latency: HistogramVec,
    pub sequencing_certificate_authority_position: Histogram,
    pub sequencing_certificate_positions_moved: Histogram,
    pub sequencing_certificate_preceding_disconnected: Histogram,
    pub sequencing_certificate_processed: IntCounterVec,
    pub sequencing_in_flight_semaphore_wait: IntGauge,
    pub sequencing_in_flight_submissions: IntGauge,
    pub sequencing_estimated_latency: IntGauge,
    pub sequencing_resubmission_interval_ms: IntGauge,
}

impl ConsensusAdapterMetrics {
    pub fn new(registry: &Registry) -> Self {
        Self {
            sequencing_certificate_attempt: register_int_counter_vec_with_registry!(
                "sequencing_certificate_attempt",
                "Counts the number of certificates the validator attempts to sequence.",
                &["tx_type"],
                registry;
                MetricLevel::Warn,
            )
            .unwrap(),
            sequencing_certificate_success: register_int_counter_vec_with_registry!(
                "sequencing_certificate_success",
                "Counts the number of successfully sequenced certificates.",
                &["tx_type"],
                registry,
            )
            .unwrap(),
            sequencing_certificate_failures: register_int_counter_vec_with_registry!(
                "sequencing_certificate_failures",
                "Counts the number of sequenced certificates that failed other than by timeout.",
                &["tx_type"],
                registry,
            )
            .unwrap(),
            sequencing_certificate_status: register_int_counter_vec_with_registry!(
                "sequencing_certificate_status",
                "The status of the certificate sequencing as reported by consensus. The status can be either sequenced or garbage collected.",
                &["tx_type", "status"],
                registry;
                MetricLevel::Warn,
            )
            .unwrap(),
            sequencing_certificate_inflight: register_int_gauge_vec_with_registry!(
                "sequencing_certificate_inflight",
                "The inflight requests to sequence certificates.",
                &["tx_type"],
                registry;
                MetricLevel::Warn,
            )
            .unwrap(),
            sequencing_acknowledge_latency: register_histogram_vec_with_registry!(
                "sequencing_acknowledge_latency",
                "The latency for acknowledgement from sequencing engine. The overall sequencing latency is measured by the sequencing_certificate_latency metric",
                &["retry", "tx_type"],
                LATENCY_SEC_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            sequencing_certificate_latency: register_histogram_vec_with_registry!(
                "sequencing_certificate_latency",
                "The latency for sequencing a certificate.",
                &["position", "tx_type", "processed_method"],
                LATENCY_SEC_BUCKETS.to_vec(),
                registry;
                MetricLevel::Warn,
            )
            .unwrap(),
            sequencing_certificate_authority_position: register_histogram_with_registry!(
                "sequencing_certificate_authority_position",
                "The position of the authority when submitted a certificate to consensus.",
                SEQUENCING_CERTIFICATE_POSITION_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            sequencing_certificate_positions_moved: register_histogram_with_registry!(
                "sequencing_certificate_positions_moved",
                "The number of authorities ahead of ourselves that were filtered out when submitting a certificate to consensus.",
                SEQUENCING_CERTIFICATE_POSITION_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            sequencing_certificate_preceding_disconnected: register_histogram_with_registry!(
                "sequencing_certificate_preceding_disconnected",
                "The number of authorities that were hashed to an earlier position that were filtered out due to being disconnected when submitting to consensus.",
                SEQUENCING_CERTIFICATE_POSITION_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            sequencing_certificate_processed: register_int_counter_vec_with_registry!(
                "sequencing_certificate_processed",
                "The number of certificates that have been processed either by consensus or checkpoint.",
                &["source"],
                registry;
                MetricLevel::Warn,
            )
            .unwrap(),
            sequencing_in_flight_semaphore_wait: register_int_gauge_with_registry!(
                "sequencing_in_flight_semaphore_wait",
                "How many requests are blocked on submit_permit.",
                registry,
            )
            .unwrap(),
            sequencing_in_flight_submissions: register_int_gauge_with_registry!(
                "sequencing_in_flight_submissions",
                "Number of transactions submitted to local consensus instance and not yet sequenced",
                registry,
            )
            .unwrap(),
            sequencing_estimated_latency: register_int_gauge_with_registry!(
                "sequencing_estimated_latency",
                "Consensus latency estimated by consensus adapter in milliseconds",
                registry,
            )
            .unwrap(),
            sequencing_resubmission_interval_ms: register_int_gauge_with_registry!(
                "sequencing_resubmission_interval_ms",
                "Resubmission interval used by consensus adapter in milliseconds",
                registry,
            )
            .unwrap(),
        }
    }

    pub fn new_test() -> Self {
        Self::new(&Registry::default())
    }
}

/// Block status for internal use in the consensus adapter.
pub enum BlockStatusInternal {
    Sequenced,
    GarbageCollected,
}

impl From<starfish_core::BlockStatus> for BlockStatusInternal {
    fn from(status: starfish_core::BlockStatus) -> Self {
        match status {
            starfish_core::BlockStatus::Sequenced(_) => BlockStatusInternal::Sequenced,
            starfish_core::BlockStatus::GarbageCollected(_) => {
                BlockStatusInternal::GarbageCollected
            }
        }
    }
}

pub type BlockStatusReceiver = oneshot::Receiver<BlockStatusInternal>;

#[mockall::automock]
pub trait SubmitToConsensus: Sync + Send + 'static {
    fn submit_to_consensus(
        &self,
        transactions: &[ConsensusTransaction],
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> IotaResult;
}

#[mockall::automock]
#[async_trait::async_trait]
pub trait ConsensusClient: Sync + Send + 'static {
    async fn submit(
        &self,
        transactions: &[ConsensusTransaction],
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> IotaResult<BlockStatusReceiver>;
}

/// Submit IOTA certificates to the consensus.
pub struct ConsensusAdapter {
    /// The network client connecting to the consensus node of this authority.
    consensus_client: Arc<dyn ConsensusClient>,

    /// The checkpoint store for the validator
    checkpoint_store: Arc<CheckpointStore>,

    /// Authority pubkey.
    authority: AuthorityName,

    /// The hard limit on the number of inflight transactions at this node.
    /// Used as the upper bound for graduated pre-consensus load shedding
    /// (`graduated_load_shedding_soft_limit_pct`) in the certificate-less
    /// (P-COOL) mode, and as the threshold for the binary
    /// cutoff in `ConsensusAdapter::check_consensus_overload()` in both
    /// certificate-less and certificate-based flows.
    max_pending_transactions: usize,

    /// Number of submitted transactions still inflight at this node.
    num_inflight_transactions: AtomicU64,

    /// Dictates the maximum position  from which will submit to consensus. Even
    /// if the is elected to submit from a higher position than this, it
    /// will "reset" to the max_submit_position.
    max_submit_position: Option<usize>,

    /// When provided it will override the current back off logic and will use
    /// this value instead as delay step.
    submit_delay_step_override: Option<Duration>,

    /// A structure to check the connection statuses populated by the Connection
    /// Monitor Listener
    connection_monitor_status: Arc<dyn CheckConnection>,

    /// A structure to check the reputation scores populated by Consensus
    low_scoring_authorities: ArcSwap<Arc<ArcSwap<HashMap<AuthorityName, u64>>>>,

    /// A structure to register metrics
    metrics: ConsensusAdapterMetrics,

    /// Semaphore limiting parallel submissions to consensus
    submit_semaphore: Semaphore,

    /// Tracks consensus submission latency for adaptive submit-delay backoff.
    latency_observer: LatencyObserver,

    /// Percentage of `max_pending_transactions` (hard limit) defining the soft
    /// limit at which graduated pre-consensus load shedding begins. When
    /// in-flight transactions are at or below the soft limit, no shedding
    /// occurs; above it, the shedding rate scales linearly from 0% to 100% at
    /// `max_pending_transactions`. Used in the certificate-less (P-COOL) mode.
    graduated_load_shedding_soft_limit_pct: u32,
}

pub trait CheckConnection: Send + Sync {
    fn check_connection(
        &self,
        ourself: &AuthorityName,
        authority: &AuthorityName,
    ) -> Option<ConnectionStatus>;
    fn update_mapping_for_epoch(&self, authority_names_to_peer_ids: HashMap<AuthorityName, PeerId>);
}

pub struct ConnectionMonitorStatus {
    /// Current connection statuses forwarded from the connection monitor
    pub connection_statuses: Arc<DashMap<PeerId, ConnectionStatus>>,
    /// A map from authority name to peer id
    pub authority_names_to_peer_ids: ArcSwap<HashMap<AuthorityName, PeerId>>,
}

pub struct ConnectionMonitorStatusForTests {}

impl ConsensusAdapter {
    /// Make a new Consensus adapter instance.
    pub fn new(
        consensus_client: Arc<dyn ConsensusClient>,
        checkpoint_store: Arc<CheckpointStore>,
        authority: AuthorityName,
        connection_monitor_status: Arc<dyn CheckConnection>,
        max_pending_transactions: usize,
        max_pending_local_submissions: usize,
        max_submit_position: Option<usize>,
        submit_delay_step_override: Option<Duration>,
        metrics: ConsensusAdapterMetrics,
        graduated_load_shedding_soft_limit_pct: u32,
    ) -> Self {
        let num_inflight_transactions = Default::default();
        let low_scoring_authorities =
            ArcSwap::from_pointee(Arc::new(ArcSwap::from_pointee(HashMap::new())));
        Self {
            consensus_client,
            checkpoint_store,
            authority,
            max_pending_transactions,
            max_submit_position,
            submit_delay_step_override,
            num_inflight_transactions,
            connection_monitor_status,
            low_scoring_authorities,
            metrics,
            submit_semaphore: Semaphore::new(max_pending_local_submissions),
            latency_observer: LatencyObserver::new(),
            graduated_load_shedding_soft_limit_pct,
        }
    }

    /// Test-only: creates a consensus adapter with a `MockConsensusClient`,
    /// default values for all parameters, and the given authority name.
    #[cfg(test)]
    pub fn new_for_testing_with_authority_name(authority_name: AuthorityName) -> Self {
        Self::new(
            Arc::new(MockConsensusClient::new()),
            CheckpointStore::new_for_tests(),
            authority_name,
            Arc::new(ConnectionMonitorStatusForTests {}),
            100_000,
            100_000,
            None,
            None,
            ConsensusAdapterMetrics::new_test(),
            50,
        )
    }

    pub fn swap_low_scoring_authorities(
        &self,
        new_low_scoring: Arc<ArcSwap<HashMap<AuthorityName, u64>>>,
    ) {
        self.low_scoring_authorities.swap(Arc::new(new_low_scoring));
    }

    pub fn submit_recovered(self: &Arc<Self>, epoch_store: &Arc<AuthorityPerEpochStore>) {
        // Currently consensus worker might lose transactions on restart, so we need to
        // resend them.
        // TODO: get_all_pending_consensus_transactions is called
        // twice when initializing AuthorityPerEpochStore and here, should not
        // be a big deal but can be optimized
        let mut recovered = epoch_store.get_all_pending_consensus_transactions();

        let is_pending_consensus_certificates_empty =
            if epoch_store.protocol_config().enable_pcool_flow() {
                // In the P-COOL flow, the list of pending consensus
                // certificates is always empty.
                true
            } else {
                epoch_store.pending_consensus_certificates_empty()
            };

        if epoch_store
            .get_reconfig_state_read_lock_guard()
            .is_reject_user_certs()
            && is_pending_consensus_certificates_empty
        {
            // If `recovered` does not contain `EndOfPublish` yet, we need to insert it.
            if !recovered
                .iter()
                .any(ConsensusTransaction::is_end_of_publish)
            {
                // There are two cases when this is needed:
                // (1) We send `EndOfPublish` message after removing pending certificates in
                // `submit_and_wait_inner`. It is possible that node will crash
                // between those two steps, in which case we might need to
                // re-introduce `EndOfPublish` message on restart.
                // (2) If node crashed inside `ConsensusAdapter::close_epoch`,
                // after reconfig lock state was written to DB and before we persisted
                // `EndOfPublish` message.
                recovered.push(ConsensusTransaction::new_end_of_publish(self.authority));
            }
        }
        debug!(
            "Submitting {:?} recovered pending consensus transactions to consensus",
            recovered.len()
        );
        for transaction in recovered {
            if transaction.is_end_of_publish() {
                info!(epoch=?epoch_store.epoch(), "Submitting EndOfPublish message to consensus");
            }
            self.submit_unchecked(&[transaction], epoch_store);
        }
    }

    fn await_submit_delay(
        &self,
        committee: &Committee,
        transactions: &[ConsensusTransaction],
    ) -> (impl Future<Output = ()>, usize, usize, usize) {
        // Use the minimum digest to compute submit delay.
        let min_digest = transactions
            .iter()
            .filter_map(|tx| match &tx.kind {
                ConsensusTransactionKind::CertifiedTransaction(certificate) => {
                    Some(certificate.digest())
                }
                ConsensusTransactionKind::UserTransactionV1(_) => {
                    // P-COOL: no submit delay needed (number of submitting validators
                    // controlled through another mechanism)
                    None
                }
                _ => None,
            })
            .min();

        let (duration, position, positions_moved, preceding_disconnected) = match min_digest {
            Some(digest) => self.await_submit_delay_user_transaction(committee, digest),
            _ => (Duration::ZERO, 0, 0, 0),
        };
        (
            tokio::time::sleep(duration),
            position,
            positions_moved,
            preceding_disconnected,
        )
    }

    fn await_submit_delay_user_transaction(
        &self,
        committee: &Committee,
        tx_digest: &TransactionDigest,
    ) -> (Duration, usize, usize, usize) {
        let (position, positions_moved, preceding_disconnected) =
            self.submission_position(committee, tx_digest);

        const DEFAULT_LATENCY: Duration = Duration::from_secs(1); // > p50 consensus latency with global deployment
        const MIN_LATENCY: Duration = Duration::from_millis(150);
        const MAX_LATENCY: Duration = Duration::from_secs(3);

        let latency = self.latency_observer.latency().unwrap_or(DEFAULT_LATENCY);
        self.metrics
            .sequencing_estimated_latency
            .set(latency.as_millis() as i64);

        let latency = std::cmp::max(latency, MIN_LATENCY);
        let latency = std::cmp::min(latency, MAX_LATENCY);
        let latency = latency * 2;
        let (delay_step, position) =
            self.override_by_max_submit_position_settings(latency, position);

        self.metrics
            .sequencing_resubmission_interval_ms
            .set(delay_step.as_millis() as i64);

        (
            delay_step * position as u32,
            position,
            positions_moved,
            preceding_disconnected,
        )
    }

    /// Overrides the latency and the position if there are defined settings for
    /// `max_submit_position` and `submit_delay_step_override`. If the
    /// `max_submit_position` has defined, then that will always be used
    /// irrespective of any so far decision. Same for the
    /// `submit_delay_step_override`.
    fn override_by_max_submit_position_settings(
        &self,
        latency: Duration,
        mut position: usize,
    ) -> (Duration, usize) {
        // Respect any manual override for position and latency from the settings
        if let Some(max_submit_position) = self.max_submit_position {
            position = std::cmp::min(position, max_submit_position);
        }

        let delay_step = self.submit_delay_step_override.unwrap_or(latency);
        (delay_step, position)
    }

    /// Check when this authority should submit the certificate to consensus.
    /// This sorts all authorities based on pseudo-random distribution derived
    /// from transaction hash.
    ///
    /// The function targets having 1 consensus transaction submitted per user
    /// transaction when system operates normally.
    ///
    /// The function returns the position of this authority when it is their
    /// turn to submit the transaction to consensus.
    fn submission_position(
        &self,
        committee: &Committee,
        tx_digest: &TransactionDigest,
    ) -> (usize, usize, usize) {
        let positions = committee.shuffle_by_stake_from_tx_digest(tx_digest);

        self.check_submission_wrt_connectivity_and_scores(positions)
    }

    /// This function runs the following algorithm to decide whether or not to
    /// submit a transaction to consensus.
    ///
    /// It takes in a deterministic list that represents positions of all the
    /// authorities. The authority in the first position will be responsible
    /// for submitting to consensus, and so we check if we are this
    /// validator, and if so, return true.
    ///
    /// If we are not in that position, we check our connectivity to the
    /// authority in that position. If we are connected to them, we can
    /// assume that they are operational and will submit the transaction. If
    /// we are not connected to them, we assume that they are not operational
    /// and we will not rely on that authority to submit the transaction. So
    /// we shift them out of the first position, and run this algorithm
    /// again on the new set of positions.
    ///
    /// This can possibly result in a transaction being submitted twice if an
    /// authority sees a false negative in connectivity to another, such as
    /// in the case of a network partition.
    ///
    /// Recursively, if the authority further ahead of us in the positions is a
    /// low performing authority, we will move our positions up one, and
    /// submit the transaction. This allows maintaining performance overall.
    /// We will only do this part for authorities that are not low performers
    /// themselves to prevent extra amplification in the case that the
    /// positions look like [low_scoring_a1, low_scoring_a2, a3]
    fn check_submission_wrt_connectivity_and_scores(
        &self,
        positions: Vec<AuthorityName>,
    ) -> (usize, usize, usize) {
        let low_scoring_authorities = self.low_scoring_authorities.load().load_full();
        if low_scoring_authorities.get(&self.authority).is_some() {
            return (positions.len(), 0, 0);
        }
        let initial_position = get_position_in_list(self.authority, positions.clone());
        let mut preceding_disconnected = 0;
        let mut before_our_position = true;

        let filtered_positions: Vec<_> = positions
            .into_iter()
            .filter(|authority| {
                let keep = self.authority == *authority; // don't filter ourself out
                if keep {
                    before_our_position = false;
                }

                // filter out any nodes that appear disconnected
                let connected = self
                    .connection_monitor_status
                    .check_connection(&self.authority, authority)
                    .unwrap_or(ConnectionStatus::Disconnected)
                    == ConnectionStatus::Connected;
                if !connected && before_our_position {
                    preceding_disconnected += 1; // used for metrics
                }

                // Filter out low scoring nodes
                let high_scoring = low_scoring_authorities.get(authority).is_none();

                keep || (connected && high_scoring)
            })
            .collect();

        let position = get_position_in_list(self.authority, filtered_positions);

        (
            position,
            initial_position - position,
            preceding_disconnected,
        )
    }

    /// This method blocks until transaction is persisted in local database
    /// It then returns handle to async task, user can join this handle to await
    /// while transaction is processed by consensus
    ///
    /// This method guarantees that once submit(but not returned async handle)
    /// returns, transaction is persisted and will eventually be sent to
    /// consensus even after restart
    ///
    /// When submitting a certificate caller **must** provide a ReconfigState
    /// lock guard
    pub fn submit(
        self: &Arc<Self>,
        transaction: ConsensusTransaction,
        lock: Option<&RwLockReadGuard<ReconfigState>>,
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> IotaResult<JoinHandle<()>> {
        self.submit_batch(&[transaction], lock, epoch_store)
    }

    pub fn submit_batch(
        self: &Arc<Self>,
        transactions: &[ConsensusTransaction],
        lock: Option<&RwLockReadGuard<ReconfigState>>,
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> IotaResult<JoinHandle<()>> {
        if transactions.len() > 1 {
            // Soft-bundle batches must be homogeneous: either all
            // CertifiedTransaction (certificate flow) or all
            // UserTransactionV1 (P-COOL flow). submit_and_wait_inner
            // assumes a single transaction kind across the batch.
            for transaction in transactions {
                // Routed through the shared user-transaction helpers rather than an
                // inline match, so a new user-transaction kind is classified in one
                // place (`ConsensusTransactionKind::is_user_transaction`) and becomes
                // soft-bundle eligible automatically.
                fp_ensure!(
                    transaction.is_user_certificate() || transaction.kind.is_user_transaction(),
                    IotaError::InvalidTxKindInSoftBundle
                );
            }
        }

        epoch_store.insert_pending_consensus_transactions(transactions, lock)?;

        Ok(self.submit_unchecked(transactions, epoch_store))
    }

    /// Returns the limit on the number of inflight transactions at this node.
    pub(super) fn max_pending_transactions(&self) -> usize {
        self.max_pending_transactions
    }

    /// Returns the percentage of `max_pending_transactions` (hard limit)
    /// defining the soft limit at which graduated pre-consensus load
    /// shedding begins. Defaults to 50%. Used in the certificate-less
    /// (P-COOL) mode.
    pub(super) fn graduated_load_shedding_soft_limit_pct(&self) -> u32 {
        self.graduated_load_shedding_soft_limit_pct
    }

    /// Returns the number of transactions currently in-flight in consensus.
    pub(super) fn num_inflight_transactions(&self) -> u64 {
        self.num_inflight_transactions.load(Ordering::Relaxed)
    }

    /// Test-only: sets the number of in-flight transactions.
    #[cfg(test)]
    pub(super) fn set_num_inflight_transactions_for_testing(&self, value: u64) {
        self.num_inflight_transactions
            .store(value, Ordering::Relaxed);
    }

    /// Returns `true` if both hard limits allow another transaction:
    /// (i) in-flight count is at or below `max_pending_transactions`, and
    /// (ii) `submit_semaphore` has at least one available permit.
    ///
    /// Uses relaxed atomic reads: the two limits are not observed atomically.
    fn check_consensus_hard_limits(&self) -> bool {
        // First check total in-flight transactions (waiting and in submission).
        // TODO: this check is redundant in the P-COOL flow - graduated
        // shedding already rejects at 100% once `num_inflight_transactions`
        // reaches `max_pending_transactions`. Remove when the certificate-based
        // flow is removed from the codebase. The semaphore check below stays in
        // either case, since it is a separate concurrency limit not covered
        // by graduated shedding.
        if self.num_inflight_transactions.load(Ordering::Relaxed) as usize
            > self.max_pending_transactions
        {
            return false;
        }

        // Then check if `submit_semaphore` has permits
        self.submit_semaphore.available_permits() > 0
    }

    /// `IotaResult` wrapper for `check_consensus_hard_limits`. Returns
    /// `TooManyTransactionsPendingConsensus` when the hard limits are exceeded.
    pub(crate) fn check_consensus_overload(&self) -> IotaResult {
        fp_ensure!(
            self.check_consensus_hard_limits(),
            IotaError::TooManyTransactionsPendingConsensus
        );

        Ok(())
    }

    fn submit_unchecked(
        self: &Arc<Self>,
        transactions: &[ConsensusTransaction],
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> JoinHandle<()> {
        // Reconfiguration lock is dropped when pending_consensus_transactions is
        // persisted, before it is handled by consensus
        let async_stage = self
            .clone()
            .submit_and_wait(transactions.to_vec(), epoch_store.clone());
        // Number of these tasks is weakly limited based on `num_inflight_transactions`.
        // (Limit is not applied atomically, and only to user transactions.)
        let join_handle = spawn_monitored_task!(async_stage);
        join_handle
    }

    async fn submit_and_wait(
        self: Arc<Self>,
        transactions: Vec<ConsensusTransaction>,
        epoch_store: Arc<AuthorityPerEpochStore>,
    ) {
        // When epoch_terminated signal is received all pending submit_and_wait_inner
        // are dropped.
        //
        // This is needed because submit_and_wait_inner waits on read_notify for
        // consensus message to be processed, which may never happen on epoch
        // boundary.
        //
        // In addition to that, within_alive_epoch ensures that all pending consensus
        // adapter tasks are stopped before reconfiguration can proceed.
        //
        // This is essential because consensus workers reuse same ports when consensus
        // restarts, this means we might be sending transactions from previous
        // epochs to consensus of new epoch if we have not had this barrier.
        epoch_store
            .within_alive_epoch(self.submit_and_wait_inner(transactions, &epoch_store))
            .await
            .ok(); // result here indicates if epoch ended earlier, we don't
        // care about it
    }

    async fn submit_and_wait_inner(
        self: Arc<Self>,
        transactions: Vec<ConsensusTransaction>,
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) {
        if transactions.is_empty() {
            return;
        }

        // submit_batch enforces that multi-tx batches (soft bundles) are
        // homogeneous: either all CertifiedTransaction or all UserTransactionV1.
        // Single-tx submits can be any kind.
        let is_soft_bundle = transactions.len() > 1;

        let mut transaction_keys = Vec::new();

        for transaction in &transactions {
            if matches!(transaction.kind, ConsensusTransactionKind::EndOfPublish(..)) {
                info!(epoch=?epoch_store.epoch(), "Submitting EndOfPublish message to consensus");
                epoch_store.record_epoch_pending_certs_process_time_metric();
            }

            let transaction_key = SequencedConsensusTransactionKey::External(transaction.key());
            transaction_keys.push(transaction_key);
        }
        let tx_type = if !is_soft_bundle {
            classify(&transactions[0])
        } else {
            "soft_bundle"
        };

        let mut guard = InflightDropGuard::acquire(&self, tx_type);

        // Create the waiter until the node's turn comes to submit to consensus
        let (await_submit, position, positions_moved, preceding_disconnected) =
            self.await_submit_delay(epoch_store.committee(), &transactions[..]);

        // Create the waiter until the transaction is processed by consensus or via
        // checkpoint
        let processed_via_consensus_or_checkpoint =
            self.await_consensus_or_checkpoint(transaction_keys.clone(), epoch_store);
        pin_mut!(processed_via_consensus_or_checkpoint);

        let processed_waiter = tokio::select! {
            // We need to wait for some delay until we submit transaction to the consensus
            _ = await_submit => Some(processed_via_consensus_or_checkpoint),

            // If epoch ends, don't wait for submit delay
            _ = epoch_store.user_certs_closed_notify() => {
                warn!(epoch = ?epoch_store.epoch(), "Epoch ended, skipping submission delay");
                Some(processed_via_consensus_or_checkpoint)
            }

            // If transaction is received by consensus or checkpoint while we wait, we are done.
            // Capture the resolved `ProcessedMethod` so the latency metric and
            // `latency_observer` accurately reflect whether the early-fire was a
            // consensus processing, a checkpoint sync, or a `dropped_tx_status_cache`
            // hit. Without this, the guard would stay at its default `Consensus`,
            // mislabeling cache-hit retries.
            method = &mut processed_via_consensus_or_checkpoint => {
                guard.processed_method = method;
                None
            }
        };

        // Log warnings for administrative transactions that fail to get sequenced
        let _monitor = if !is_soft_bundle
            && matches!(
                transactions[0].kind,
                ConsensusTransactionKind::EndOfPublish(_)
                    | ConsensusTransactionKind::CapabilityNotificationV1(_)
                    | ConsensusTransactionKind::RandomnessDkgMessage(_, _)
                    | ConsensusTransactionKind::RandomnessDkgConfirmation(_, _)
                    | ConsensusTransactionKind::OverloadNotificationV1(_, _, _)
                    | ConsensusTransactionKind::TransactionDenyRuleProposal(_)
            ) {
            let transaction_keys = transaction_keys.clone();
            Some(CancelOnDrop(spawn_monitored_task!(async {
                let mut i = 0u64;
                loop {
                    i += 1;
                    const WARN_DELAY_S: u64 = 30;
                    tokio::time::sleep(Duration::from_secs(WARN_DELAY_S)).await;
                    let total_wait = i * WARN_DELAY_S;
                    warn!(
                        "Still waiting {} seconds for transactions {:?} to commit in consensus",
                        total_wait, transaction_keys
                    );
                }
            })))
        } else {
            None
        };

        if let Some(processed_waiter) = processed_waiter {
            debug!("Submitting {:?} to consensus", transaction_keys);

            // populate the position only when this authority submits the transaction
            // to consensus
            guard.position = Some(position);
            guard.positions_moved = Some(positions_moved);
            guard.preceding_disconnected = Some(preceding_disconnected);

            let _permit: SemaphorePermit = self
                .submit_semaphore
                .acquire()
                .count_in_flight(self.metrics.sequencing_in_flight_semaphore_wait.clone())
                .await
                .expect("Consensus adapter does not close semaphore");
            let _in_flight_submission_guard =
                GaugeGuard::acquire(&self.metrics.sequencing_in_flight_submissions);

            // We enter this branch when in select above await_submit completed and
            // processed_waiter is pending This means it is time for us to
            // submit transaction to consensus
            let submit_inner = async {
                const RETRY_DELAY_STEP: Duration = Duration::from_secs(1);

                loop {
                    // Submit the transaction to consensus and return the submit result with a
                    // status waiter
                    let status_waiter = self
                        .submit_inner(
                            &transactions,
                            epoch_store,
                            &transaction_keys,
                            tx_type,
                            is_soft_bundle,
                        )
                        .await;

                    match status_waiter.await {
                        Ok(BlockStatusInternal::Sequenced) => {
                            self.metrics
                                .sequencing_certificate_status
                                .with_label_values(&[tx_type, "sequenced"])
                                .inc();
                            // Block has been sequenced. Nothing more to do, we do have guarantees
                            // that the transaction will appear in consensus output.
                            trace!(
                                "Transaction {transaction_keys:?} has been sequenced by consensus."
                            );
                            break;
                        }
                        Ok(BlockStatusInternal::GarbageCollected) => {
                            self.metrics
                                .sequencing_certificate_status
                                .with_label_values(&[tx_type, "garbage_collected"])
                                .inc();
                            // Block has been garbage collected and we have no guarantees that the
                            // transaction will appear in consensus output. We'll
                            // resubmit the transaction to consensus. If the transaction has been
                            // already "processed", then probably someone else has submitted
                            // the transaction and managed to get sequenced. Then this future will
                            // have been cancelled anyways so no need to check here on the processed
                            // output.
                            debug!(
                                "Transaction {transaction_keys:?} was garbage collected before being sequenced. Will be retried."
                            );
                            time::sleep(RETRY_DELAY_STEP).await;
                            continue;
                        }
                        Err(err) => {
                            warn!(
                                "Error while waiting for status from consensus for transactions {transaction_keys:?}, with error {:?}. Will be retried.",
                                err
                            );
                            time::sleep(RETRY_DELAY_STEP).await;
                            continue;
                        }
                    }
                }
            };

            guard.processed_method = match select(processed_waiter, submit_inner.boxed()).await {
                Either::Left((observed_via_consensus, _submit_inner)) => observed_via_consensus,
                Either::Right(((), processed_waiter)) => {
                    debug!("Submitted {transaction_keys:?} to consensus");
                    processed_waiter.await
                }
            };
        }
        debug!("{transaction_keys:?} processed by consensus");

        let consensus_keys: Vec<_> = transactions.iter().map(|t| t.key()).collect();
        epoch_store
            .remove_pending_consensus_transactions(&consensus_keys)
            .expect("Storage error when removing consensus transaction");

        let is_user_tx = is_soft_bundle
            || if epoch_store.protocol_config().enable_pcool_flow() {
                // In the P-COOL flow, `UserTransactionV1` kind corresponds
                // to user transactions.  Routed through the shared predicate
                // so a new user-transaction kind is classified in one place.
                transactions[0].kind.is_user_transaction()
            } else {
                // In the certificate mode, `CertifiedTransaction` kind corresponds
                // to user transactions.
                transactions[0].is_user_certificate()
            };
        let send_end_of_publish = if is_user_tx {
            if epoch_store.protocol_config().enable_pcool_flow() {
                // In the P-COOL flow, `EndOfPublish` is sent solely from
                // `close_epoch`. There is no pending certificate drain to
                // monitor here, and sending from this per-transaction callback
                // would produce N duplicate EndOfPublish messages (one per
                // in-flight user transaction completing after `RejectUserCerts`
                // is set).
                false
            } else {
                // In certificate mode, `EndOfPublish` is sent once the list of
                // pending consensus certificates is drained. Multiple tasks can
                // enter here concurrently with `pending_count == 0`, producing
                // duplicate messages — this is rare and does not affect
                // correctness.
                //
                // Note: there could be a race condition here where we enter
                // this check in `RejectAllCerts` state. In that case we don't
                // need to send `EndOfPublish` because the condition to enter
                // `RejectAllCerts` is when 2f+1 other validators already
                // sequenced their `EndOfPublish` message.
                //
                // TODO: This entire certificate-mode drain logic can be removed
                // once the certificate flow is fully cleaned up.
                if epoch_store
                    .get_reconfig_state_read_lock_guard()
                    .is_reject_user_certs()
                {
                    let pending_count = epoch_store.pending_consensus_certificates_count();
                    debug!(epoch=?epoch_store.epoch(), ?pending_count, "Deciding whether to send EndOfPublish");

                    pending_count == 0 // send end of epoch if no pending certificates
                } else {
                    false
                }
            }
        } else {
            false
        };
        if send_end_of_publish {
            // Spawn a separate task for EndOfPublish so that
            // submit_and_wait_inner returns promptly after the original
            // transaction is processed. Awaiting the retry loop inline
            // would hold the InflightDropGuard and inflate in-flight
            // metrics for the duration of retries.
            let adapter = self.clone();
            let epoch_store = epoch_store.clone();
            spawn_monitored_task!(async move {
                if epoch_store
                    .within_alive_epoch(adapter.submit_end_of_publish_with_retry(&epoch_store))
                    .await
                    .is_err()
                {
                    warn!(
                        epoch = ?epoch_store.epoch(),
                        "EndOfPublish submission cancelled: epoch has ended",
                    );
                }
            });
        }
        self.metrics
            .sequencing_certificate_success
            .with_label_values(&[tx_type])
            .inc();
    }

    async fn submit_inner(
        self: &Arc<Self>,
        transactions: &[ConsensusTransaction],
        epoch_store: &Arc<AuthorityPerEpochStore>,
        transaction_keys: &[SequencedConsensusTransactionKey],
        tx_type: &str,
        is_soft_bundle: bool,
    ) -> BlockStatusReceiver {
        let ack_start = Instant::now();
        let mut retries: u32 = 0;

        let status_waiter = loop {
            match self
                .consensus_client
                .submit(transactions, epoch_store)
                .await
            {
                Err(err) => {
                    // This can happen during reconfig, or when consensus has full internal buffers
                    // and needs to back pressure, so retry a few times before logging warnings.
                    if retries > 30
                        || (retries > 3 && (is_soft_bundle || !transactions[0].kind.is_dkg()))
                    {
                        warn!(
                            "Failed to submit transactions {transaction_keys:?} to consensus: {err:?}. Retry #{retries}"
                        );
                    }
                    self.metrics
                        .sequencing_certificate_failures
                        .with_label_values(&[tx_type])
                        .inc();
                    retries += 1;

                    if !is_soft_bundle && transactions[0].kind.is_dkg() {
                        // Shorter delay for DKG messages, which are time-sensitive and happen at
                        // start-of-epoch when submit errors due to active reconfig are likely.
                        time::sleep(Duration::from_millis(100)).await;
                    } else {
                        time::sleep(Duration::from_secs(10)).await;
                    };
                }
                Ok(status_waiter) => {
                    break status_waiter;
                }
            }
        };

        // we want to record the num of retries when reporting latency but to avoid
        // label cardinality we do some simple bucketing to give us a good
        // enough idea of how many retries happened associated with the latency.
        let bucket = match retries {
            0..=10 => retries.to_string(), // just report the retry count as is
            11..=20 => "between_10_and_20".to_string(),
            21..=50 => "between_20_and_50".to_string(),
            51..=100 => "between_50_and_100".to_string(),
            _ => "over_100".to_string(),
        };

        self.metrics
            .sequencing_acknowledge_latency
            .with_label_values(&[bucket.as_str(), tx_type])
            .observe(ack_start.elapsed().as_secs_f64());

        status_waiter
    }

    /// Waits for transactions to appear either to consensus output or been
    /// executed via a checkpoint (state sync).
    /// Returns the processed method, whether the transactions have been
    /// processed via consensus, or have been synced via checkpoint.
    async fn await_consensus_or_checkpoint(
        self: &Arc<Self>,
        transaction_keys: Vec<SequencedConsensusTransactionKey>,
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> ProcessedMethod {
        let notifications = FuturesUnordered::new();
        for transaction_key in transaction_keys {
            let transaction_digests = match transaction_key {
                SequencedConsensusTransactionKey::External(
                    ConsensusTransactionKey::Certificate(digest),
                )
                | SequencedConsensusTransactionKey::External(
                    ConsensusTransactionKey::UserTransaction(digest),
                ) => vec![digest],
                _ => vec![],
            };

            let checkpoint_synced_future = if let SequencedConsensusTransactionKey::External(
                ConsensusTransactionKey::CheckpointSignature(_, checkpoint_sequence_number),
            ) = transaction_key
            {
                // If the transaction is a checkpoint signature, we can also wait to get
                // notified when a checkpoint with equal or higher sequence
                // number has been already synced. This way we don't try to unnecessarily
                // sequence the signature for an already verified checkpoint.
                Either::Left(
                    self.checkpoint_store
                        .notify_read_synced_checkpoint(checkpoint_sequence_number),
                )
            } else {
                Either::Right(future::pending())
            };

            // We wait for each transaction individually to be processed by consensus,
            // executed in a checkpoint or dropped. We could equally just get
            // notified in aggregate when all transactions are processed, but
            // with this approach can get notified in a more fine-grained way as
            // transactions can be marked as processed in different ways. This
            // is mostly a concern for the soft-bundle transactions.
            let dropped_digest = transaction_digests.first().copied();
            notifications.push(async move {
                tokio::select! {
                    processed = epoch_store.consensus_messages_processed_notify(vec![transaction_key]) => {
                        processed.expect("Storage error when waiting for consensus message processed");
                        self.metrics.sequencing_certificate_processed.with_label_values(&["consensus"]).inc();
                        return ProcessedMethod::Consensus;
                    },
                    processed = epoch_store.transactions_executed_in_checkpoint_notify(transaction_digests), if !transaction_digests.is_empty() => {
                        processed.expect("Storage error when waiting for transaction executed in checkpoint");
                        self.metrics.sequencing_certificate_processed.with_label_values(&["checkpoint"]).inc();
                    }
                    _ = checkpoint_synced_future => {
                        self.metrics.sequencing_certificate_processed.with_label_values(&["synced_checkpoint"]).inc();
                    }
                    _ = async move {
                        if let Some(d) = dropped_digest {
                            let _err = epoch_store.notify_read_any_drop(d).await;
                        } else {
                            future::pending::<()>().await;
                        }
                    } => {
                        self.metrics.sequencing_certificate_processed.with_label_values(&["dropped"]).inc();
                        return ProcessedMethod::Dropped;
                    }
                }
                ProcessedMethod::Checkpoint
            });
        }

        let processed_methods = notifications.collect::<Vec<ProcessedMethod>>().await;
        if processed_methods.contains(&ProcessedMethod::Dropped) {
            ProcessedMethod::Dropped
        } else if processed_methods.contains(&ProcessedMethod::Checkpoint) {
            ProcessedMethod::Checkpoint
        } else {
            ProcessedMethod::Consensus
        }
    }

    /// Submits an `EndOfPublish` message to consensus with exponential
    /// backoff (capped at `MAX_BACKOFF`). Retries indefinitely on any
    /// error — both transient failures (e.g. DB write errors in
    /// `insert_pending_consensus_transactions`) and permanent ones (e.g.
    /// `EpochEnded` from `tables()`). A missing `EndOfPublish` would
    /// stall the epoch, so the loop never gives up on its own.
    ///
    /// Callers **must** wrap this with `epoch_store.within_alive_epoch()`
    /// to cancel retries when the epoch terminates — this is the
    /// mechanism that stops the loop on permanent `EpochEnded` errors.
    async fn submit_end_of_publish_with_retry(
        self: &Arc<Self>,
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) {
        const INITIAL_BACKOFF: Duration = Duration::from_millis(100);
        const MAX_BACKOFF: Duration = Duration::from_secs(10);

        info!(
            epoch = ?epoch_store.epoch(),
            authority = ?self.authority,
            "Sending EndOfPublish message to consensus",
        );

        let mut attempt: u32 = 0;
        loop {
            match self.submit(
                ConsensusTransaction::new_end_of_publish(self.authority),
                None,
                epoch_store,
            ) {
                Ok(_) => return,
                Err(IotaError::EpochEnded(_)) => {
                    warn!(
                        epoch = ?epoch_store.epoch(),
                        authority = ?self.authority,
                        "EndOfPublish submission stopped: epoch has ended",
                    );
                    return;
                }
                Err(err) => {
                    let backoff = (INITIAL_BACKOFF * 2u32.pow(attempt.min(10))).min(MAX_BACKOFF);
                    warn!(
                        epoch = ?epoch_store.epoch(),
                        authority = ?self.authority,
                        attempt,
                        "Failed to submit EndOfPublish, retrying in {:?}: {:?}",
                        backoff,
                        err,
                    );
                    tokio::time::sleep(backoff).await;
                    attempt = attempt.saturating_add(1);
                }
            }
        }
    }
}

impl CheckConnection for ConnectionMonitorStatus {
    fn check_connection(
        &self,
        ourself: &AuthorityName,
        authority: &AuthorityName,
    ) -> Option<ConnectionStatus> {
        if ourself == authority {
            return Some(ConnectionStatus::Connected);
        }

        let mapping = self.authority_names_to_peer_ids.load_full();
        let peer_id = match mapping.get(authority) {
            Some(p) => p,
            None => {
                warn!(
                    "failed to find peer {:?} in connection monitor listener",
                    authority
                );
                return None;
            }
        };

        let res = match self.connection_statuses.try_get(peer_id) {
            TryResult::Present(c) => Some(c.value().clone()),
            TryResult::Absent => None,
            TryResult::Locked => {
                // update is in progress, assume the status is still or becoming disconnected
                Some(ConnectionStatus::Disconnected)
            }
        };
        res
    }
    fn update_mapping_for_epoch(
        &self,
        authority_names_to_peer_ids: HashMap<AuthorityName, PeerId>,
    ) {
        self.authority_names_to_peer_ids
            .swap(Arc::new(authority_names_to_peer_ids));
    }
}

impl CheckConnection for ConnectionMonitorStatusForTests {
    fn check_connection(
        &self,
        _ourself: &AuthorityName,
        _authority: &AuthorityName,
    ) -> Option<ConnectionStatus> {
        Some(ConnectionStatus::Connected)
    }
    fn update_mapping_for_epoch(
        &self,
        _authority_names_to_peer_ids: HashMap<AuthorityName, PeerId>,
    ) {
    }
}

pub fn get_position_in_list(
    search_authority: AuthorityName,
    positions: Vec<AuthorityName>,
) -> usize {
    positions
        .into_iter()
        .find_position(|authority| *authority == search_authority)
        .expect("Couldn't find ourselves in shuffled committee")
        .0
}

impl ReconfigurationInitiator for Arc<ConsensusAdapter> {
    /// This method is called externally to begin reconfiguration.
    /// It transitions the reconfig state to reject new user transactions.
    /// `ConsensusAdapter` will send `EndOfPublish` once all pending
    /// transactions are drained (in the certificate mode) or right away
    /// (in the P-COOL flow). Submission is asynchronous —
    /// a background task handles retries so this method returns promptly.
    fn close_epoch(&self, epoch_store: &Arc<AuthorityPerEpochStore>) {
        let send_end_of_publish = {
            let reconfig_guard = epoch_store.get_reconfig_state_write_lock_guard();
            if !reconfig_guard.should_accept_user_certs() {
                // Allow caller to call this method multiple times
                return;
            }

            let send_end_of_publish = if epoch_store.protocol_config().enable_pcool_flow() {
                // In the P-COOL flow, there are no pending consensus
                // certificates, so `EndOfPublish` is always sent immediately.
                debug!(epoch=?epoch_store.epoch(), "Closing epoch in P-COOL mode");

                true
            } else {
                // In certificate mode, `EndOfPublish` is sent only once the list
                // of pending consensus certificates is drained.
                let pending_count = epoch_store.pending_consensus_certificates_count();
                debug!(epoch=?epoch_store.epoch(), ?pending_count, "Trying to close epoch");

                pending_count == 0 // send end of epoch if no pending certificates
            };

            epoch_store.close_user_certs(reconfig_guard);

            send_end_of_publish
            // reconfig_guard lock is dropped here.
        };

        if send_end_of_publish {
            // Spawned because ReconfigurationInitiator::close_epoch is
            // sync — it cannot await. This is safe: by this point
            // close_user_certs() has already set the reconfig state to
            // reject new transactions, so no further user work depends
            // on this method returning. The background task retries
            // until the message is delivered or the epoch terminates.
            let adapter = self.clone();
            let epoch_store = epoch_store.clone();
            spawn_monitored_task!(async move {
                if epoch_store
                    .within_alive_epoch(adapter.submit_end_of_publish_with_retry(&epoch_store))
                    .await
                    .is_err()
                {
                    warn!(
                        epoch = ?epoch_store.epoch(),
                        "EndOfPublish submission cancelled: epoch has ended",
                    );
                }
            });
        }
    }
}

struct CancelOnDrop<T>(JoinHandle<T>);

impl<T> Deref for CancelOnDrop<T> {
    type Target = JoinHandle<T>;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<T> Drop for CancelOnDrop<T> {
    fn drop(&mut self) {
        self.0.abort();
    }
}

/// Tracks number of inflight consensus requests and relevant metrics
struct InflightDropGuard<'a> {
    adapter: &'a ConsensusAdapter,
    start: Instant,
    position: Option<usize>,
    positions_moved: Option<usize>,
    preceding_disconnected: Option<usize>,
    tx_type: &'static str,
    processed_method: ProcessedMethod,
}

#[derive(PartialEq, Eq, Copy, Clone)]
enum ProcessedMethod {
    Consensus,
    Checkpoint,
    Dropped,
}

impl<'a> InflightDropGuard<'a> {
    pub fn acquire(adapter: &'a ConsensusAdapter, tx_type: &'static str) -> Self {
        adapter
            .num_inflight_transactions
            .fetch_add(1, Ordering::SeqCst);
        adapter
            .metrics
            .sequencing_certificate_inflight
            .with_label_values(&[tx_type])
            .inc();
        adapter
            .metrics
            .sequencing_certificate_attempt
            .with_label_values(&[tx_type])
            .inc();
        Self {
            adapter,
            start: Instant::now(),
            position: None,
            positions_moved: None,
            preceding_disconnected: None,
            tx_type,
            processed_method: ProcessedMethod::Consensus,
        }
    }
}

impl Drop for InflightDropGuard<'_> {
    fn drop(&mut self) {
        self.adapter
            .num_inflight_transactions
            .fetch_sub(1, Ordering::SeqCst);
        self.adapter
            .metrics
            .sequencing_certificate_inflight
            .with_label_values(&[self.tx_type])
            .dec();

        let position = if let Some(position) = self.position {
            self.adapter
                .metrics
                .sequencing_certificate_authority_position
                .observe(position as f64);
            position.to_string()
        } else {
            "not_submitted".to_string()
        };

        if let Some(positions_moved) = self.positions_moved {
            self.adapter
                .metrics
                .sequencing_certificate_positions_moved
                .observe(positions_moved as f64);
        };

        if let Some(preceding_disconnected) = self.preceding_disconnected {
            self.adapter
                .metrics
                .sequencing_certificate_preceding_disconnected
                .observe(preceding_disconnected as f64);
        };

        let latency = self.start.elapsed();
        let processed_method = match self.processed_method {
            ProcessedMethod::Consensus => "processed_via_consensus",
            ProcessedMethod::Checkpoint => "processed_via_checkpoint",
            ProcessedMethod::Dropped => "dropped",
        };
        self.adapter
            .metrics
            .sequencing_certificate_latency
            .with_label_values(&[position.as_str(), self.tx_type, processed_method])
            .observe(latency.as_secs_f64());

        // Only sample latency after consensus quorum is up. Otherwise, the wait for
        // consensus quorum at the beginning of an epoch can distort the sampled
        // latencies. Technically there are more system transaction types that
        // can be included in samples after the first consensus commit, but this
        // set of types should be enough.
        if self.position == Some(0) {
            // Transaction types below require quorum existed in the current epoch.
            // TODO: refactor tx_type to enum.
            let sampled = matches!(
                self.tx_type,
                "shared_certificate" | "owned_certificate" | "checkpoint_signature" | "soft_bundle"
            );
            // Exclude checkpoint-synced and dropped txs from the latency observer:
            // their latency reflects state-sync timing or rejection, not consensus
            // throughput, and would skew the observed consensus latency.
            if sampled && self.processed_method == ProcessedMethod::Consensus {
                self.adapter.latency_observer.report(latency);
            }
        }
    }
}

impl SubmitToConsensus for Arc<ConsensusAdapter> {
    fn submit_to_consensus(
        &self,
        transactions: &[ConsensusTransaction],
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> IotaResult {
        self.submit_batch(transactions, None, epoch_store)
            .map(|_| ())
    }
}

pub fn position_submit_certificate(
    committee: &Committee,
    ourselves: &AuthorityName,
    tx_digest: &TransactionDigest,
) -> usize {
    let validators = committee.shuffle_by_stake_from_tx_digest(tx_digest);
    get_position_in_list(*ourselves, validators)
}

#[cfg(test)]
mod adapter_tests {
    use std::{sync::Arc, time::Duration};

    use fastcrypto::{
        bls12381::min_sig::BLS12381PrivateKey,
        traits::{KeyPair, ToFromBytes},
    };
    use iota_sdk_types::TransactionDigest;
    use iota_types::{
        committee::Committee,
        crypto::{AuthorityKeyPair, AuthorityPublicKeyBytes},
    };

    use super::position_submit_certificate;
    use crate::{
        checkpoints::CheckpointStore,
        consensus_adapter::{
            ConnectionMonitorStatusForTests, ConsensusAdapter, ConsensusAdapterMetrics,
        },
        starfish_adapter::LazyStarfishClient,
    };

    /// A committee built from fixed keys and stakes, so the submission order
    /// these tests check depends only on the code under test, not on `rand`
    /// or on how fastcrypto generates keys.
    fn test_committee() -> Committee {
        const STAKES: [u64; 10] = [3, 7, 1, 9, 4, 4, 2, 8, 5, 6];
        let authorities = STAKES
            .iter()
            .enumerate()
            .map(|(i, stake)| {
                let key_pair = AuthorityKeyPair::from(
                    BLS12381PrivateKey::from_bytes(&[i as u8 + 1; 32]).unwrap(),
                );
                (AuthorityPublicKeyBytes::from(key_pair.public()), *stake)
            })
            .collect();
        Committee::new_for_testing_with_normalized_voting_power(0, authorities)
    }

    #[tokio::test]
    async fn test_await_submit_delay_user_transaction() {
        let committee = test_committee();

        // When we define max submit position and delay step
        let consensus_adapter = ConsensusAdapter::new(
            Arc::new(LazyStarfishClient::new()),
            CheckpointStore::new_for_tests(),
            *committee.authority_by_index(0).unwrap(),
            Arc::new(ConnectionMonitorStatusForTests {}),
            100_000,
            100_000,
            Some(1),
            Some(Duration::from_secs(2)),
            ConsensusAdapterMetrics::new_test(),
            50,
        );

        // transaction to submit
        let tx_digest = TransactionDigest::new([7; 32]);

        // Ensure that the original position is higher
        let (position, positions_moved, _) =
            consensus_adapter.submission_position(&committee, &tx_digest);
        assert_eq!(position, 8);
        assert_eq!(positions_moved, 0);

        // Make sure that position is set to max value 1
        let (delay_step, position, positions_moved, _) =
            consensus_adapter.await_submit_delay_user_transaction(&committee, &tx_digest);

        assert_eq!(position, 1);
        assert_eq!(delay_step, Duration::from_secs(2));
        assert_eq!(positions_moved, 0);

        // Without submit position and delay step
        let consensus_adapter = ConsensusAdapter::new(
            Arc::new(LazyStarfishClient::new()),
            CheckpointStore::new_for_tests(),
            *committee.authority_by_index(0).unwrap(),
            Arc::new(ConnectionMonitorStatusForTests {}),
            100_000,
            100_000,
            None,
            None,
            ConsensusAdapterMetrics::new_test(),
            50,
        );

        let (delay_step, position, positions_moved, _) =
            consensus_adapter.await_submit_delay_user_transaction(&committee, &tx_digest);

        assert_eq!(position, 8);

        // delay_step * position * 2 = 1 * 8 * 2 = 16
        assert_eq!(delay_step, Duration::from_secs(16));
        assert_eq!(positions_moved, 0);
    }

    #[test]
    fn test_position_submit_certificate() {
        let committee = test_committee();

        // many transaction digests, and account for validator selection
        const NUM_TEST_TRANSACTIONS: u32 = 1000;

        for tx_idx in 0..NUM_TEST_TRANSACTIONS {
            let mut digest_bytes = [0u8; 32];
            digest_bytes[..4].copy_from_slice(&tx_idx.to_le_bytes());
            let tx_digest = TransactionDigest::new(digest_bytes);

            let mut zero_found = false;
            for (name, _) in committee.members() {
                let f = position_submit_certificate(&committee, name, &tx_digest);
                assert!(f < committee.num_members());
                if f == 0 {
                    // One and only one validator gets position 0
                    assert!(!zero_found);
                    zero_found = true;
                }
            }
            assert!(zero_found);
        }
    }
}