sui_core/

authority_server.rs

// Copyright (c) 2021, Facebook, Inc. and its affiliates
// Copyright (c) Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use anyhow::Result;
use async_trait::async_trait;
use fastcrypto::traits::KeyPair;
use futures::{TryFutureExt, future};
use itertools::Itertools as _;
use moka::sync::Cache;
use mysten_common::ZipDebugEqIteratorExt;
use mysten_common::{assert_reachable, debug_fatal};
use mysten_metrics::spawn_monitored_task;
use parking_lot::Mutex;
use prometheus::{
    Gauge, Histogram, HistogramVec, IntCounter, IntCounterVec, IntGauge, Registry,
    register_gauge_with_registry, register_histogram_vec_with_registry,
    register_histogram_with_registry, register_int_counter_vec_with_registry,
    register_int_counter_with_registry, register_int_gauge_with_registry,
};
use std::{
    collections::{HashMap, HashSet},
    io,
    net::{IpAddr, SocketAddr},
    sync::Arc,
    time::{Duration, Instant, SystemTime},
};
use sui_network::{
    api::{Validator, ValidatorServer},
    tonic,
    validator::server::SUI_TLS_SERVER_NAME,
};
use sui_types::effects::TransactionEffectsAPI;
use sui_types::message_envelope::Message;
use sui_types::messages_consensus::{ConsensusTransaction, ConsensusTransactionKey};
use sui_types::messages_grpc::{
    ObjectInfoRequest, ObjectInfoResponse, RawSubmitTxResponse, SystemStateRequest,
    TransactionInfoRequest, TransactionInfoResponse,
};
use sui_types::multiaddr::Multiaddr;
use sui_types::object::Object;
use sui_types::sui_system_state::SuiSystemState;
use sui_types::traffic_control::{ClientIdSource, Weight};
use sui_types::{
    base_types::ObjectID,
    digests::{TransactionDigest, TransactionEffectsDigest},
    error::{SuiErrorKind, UserInputError},
};
use sui_types::{
    effects::TransactionEffects,
    messages_grpc::{
        ExecutedData, RawSubmitTxRequest, RawWaitForEffectsRequest, RawWaitForEffectsResponse,
        SubmitTxResult, WaitForEffectsRequest, WaitForEffectsResponse,
    },
};
use sui_types::{effects::TransactionEvents, messages_grpc::SubmitTxType};
use sui_types::{error::*, transaction::*};
use sui_types::{
    fp_ensure,
    messages_checkpoint::{
        CheckpointRequest, CheckpointRequestV2, CheckpointResponse, CheckpointResponseV2,
    },
};
use tokio::time::timeout;
use tonic::metadata::{Ascii, MetadataValue};
use tracing::{debug, error, info, instrument};

use crate::admission_queue::{AdmissionQueueContext, AdmissionQueueManager};
use crate::gasless_rate_limiter::GaslessRateLimiter;
use crate::{
    authority::{AuthorityState, consensus_tx_status_cache::ConsensusTxStatus},
    consensus_adapter::{ConsensusAdapter, ConsensusAdapterMetrics, ConsensusOverloadChecker},
    consensus_handler::SequencedConsensusTransactionKey,
    traffic_controller::{TrafficController, parse_ip, policies::TrafficTally},
};
use crate::{
    authority::{
        authority_per_epoch_store::AuthorityPerEpochStore,
        consensus_tx_status_cache::NotifyReadConsensusTxStatusResult,
    },
    checkpoints::CheckpointStore,
    mysticeti_adapter::LazyMysticetiClient,
};
use sui_config::local_ip_utils::new_local_tcp_address_for_testing;

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

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

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

pub struct AuthorityServerHandle {
    server_handle: sui_network::validator::server::Server,
}

impl AuthorityServerHandle {
    pub async fn join(self) -> Result<(), io::Error> {
        self.server_handle.handle().wait_for_shutdown().await;
        Ok(())
    }

    pub async fn kill(self) -> Result<(), io::Error> {
        self.server_handle.handle().shutdown().await;
        Ok(())
    }

    pub fn address(&self) -> &Multiaddr {
        self.server_handle.local_addr()
    }
}

pub struct AuthorityServer {
    address: Multiaddr,
    pub state: Arc<AuthorityState>,
    consensus_adapter: Arc<ConsensusAdapter>,
    pub metrics: Arc<ValidatorServiceMetrics>,
}

impl AuthorityServer {
    pub fn new_for_test_with_consensus_adapter(
        state: Arc<AuthorityState>,
        consensus_adapter: Arc<ConsensusAdapter>,
    ) -> Self {
        let address = new_local_tcp_address_for_testing();
        let metrics = Arc::new(ValidatorServiceMetrics::new_for_tests());

        Self {
            address,
            state,
            consensus_adapter,
            metrics,
        }
    }

    pub fn new_for_test(state: Arc<AuthorityState>) -> Self {
        let slot_freed_notify = Arc::new(tokio::sync::Notify::new());
        let consensus_adapter = Arc::new(ConsensusAdapter::new(
            Arc::new(LazyMysticetiClient::new()),
            CheckpointStore::new_for_tests(),
            state.name,
            100_000,
            100_000,
            ConsensusAdapterMetrics::new_test(),
            slot_freed_notify,
        ));
        Self::new_for_test_with_consensus_adapter(state, consensus_adapter)
    }

    pub async fn spawn_for_test(self) -> Result<AuthorityServerHandle, io::Error> {
        let address = self.address.clone();
        self.spawn_with_bind_address_for_test(address).await
    }

    pub async fn spawn_with_bind_address_for_test(
        self,
        address: Multiaddr,
    ) -> Result<AuthorityServerHandle, io::Error> {
        let tls_config = sui_tls::create_rustls_server_config(
            self.state.config.network_key_pair().copy().private(),
            SUI_TLS_SERVER_NAME.to_string(),
        );
        let config = mysten_network::config::Config::new();
        let server = sui_network::validator::server::ServerBuilder::from_config(
            &config,
            mysten_network::metrics::DefaultMetricsCallbackProvider::default(),
        )
        .add_service(ValidatorServer::new(ValidatorService::new_for_tests(
            self.state,
            self.consensus_adapter,
            self.metrics,
        )))
        .bind(&address, Some(tls_config))
        .await
        .unwrap();
        let local_addr = server.local_addr().to_owned();
        info!("Listening to traffic on {local_addr}");
        let handle = AuthorityServerHandle {
            server_handle: server,
        };
        Ok(handle)
    }
}

pub struct ValidatorServiceMetrics {
    pub signature_errors: IntCounter,
    pub tx_verification_latency: Histogram,
    pub handle_transaction_latency: Histogram,
    pub handle_transaction_consensus_latency: Histogram,
    pub handle_submit_transaction_consensus_latency: HistogramVec,
    pub handle_wait_for_effects_ping_latency: HistogramVec,

    handle_submit_transaction_latency: HistogramVec,
    handle_submit_transaction_bytes: HistogramVec,
    handle_submit_transaction_batch_size: HistogramVec,

    num_rejected_tx_during_overload: IntCounterVec,
    submission_rejected_transactions: IntCounterVec,
    submission_suppressed_already_processed: IntCounterVec,
    submission_suppressed_recently_submitted: IntCounterVec,
    recently_submitted_cache_size: IntGauge,
    recently_submitted_resubmission_interval: Histogram,
    submission_suppressed_inflight: IntCounterVec,
    inflight_transactions: IntGauge,
    connection_ip_not_found: IntCounter,
    forwarded_header_parse_error: IntCounter,
    forwarded_header_invalid: IntCounter,
    forwarded_header_not_included: IntCounter,
    client_id_source_config_mismatch: IntCounter,
    x_forwarded_for_num_hops: Gauge,
    pub gasless_rate_limited_count: IntCounter,
    pub gasless_submission_outcomes: IntCounterVec,
}

impl ValidatorServiceMetrics {
    pub fn new(registry: &Registry) -> Self {
        Self {
            signature_errors: register_int_counter_with_registry!(
                "total_signature_errors",
                "Number of transaction signature errors",
                registry,
            )
            .unwrap(),
            tx_verification_latency: register_histogram_with_registry!(
                "validator_service_tx_verification_latency",
                "Latency of verifying a transaction",
                mysten_metrics::SUBSECOND_LATENCY_SEC_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            handle_transaction_latency: register_histogram_with_registry!(
                "validator_service_handle_transaction_latency",
                "Latency of handling a transaction",
                mysten_metrics::SUBSECOND_LATENCY_SEC_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            handle_transaction_consensus_latency: register_histogram_with_registry!(
                "validator_service_handle_transaction_consensus_latency",
                "Latency of handling a user transaction sent through consensus",
                mysten_metrics::COARSE_LATENCY_SEC_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            handle_submit_transaction_consensus_latency: register_histogram_vec_with_registry!(
                "validator_service_submit_transaction_consensus_latency",
                "Latency of submitting a user transaction sent through consensus",
                &["req_type"],
                mysten_metrics::COARSE_LATENCY_SEC_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            handle_submit_transaction_latency: register_histogram_vec_with_registry!(
                "validator_service_submit_transaction_latency",
                "Latency of submit transaction handler",
                &["req_type"],
                mysten_metrics::LATENCY_SEC_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            handle_wait_for_effects_ping_latency: register_histogram_vec_with_registry!(
                "validator_service_handle_wait_for_effects_ping_latency",
                "Latency of handling a ping request for wait_for_effects",
                &["req_type"],
                mysten_metrics::SUBSECOND_LATENCY_SEC_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            handle_submit_transaction_bytes: register_histogram_vec_with_registry!(
                "validator_service_submit_transaction_bytes",
                "The size of transactions in the submit transaction request",
                &["req_type"],
                mysten_metrics::BYTES_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            handle_submit_transaction_batch_size: register_histogram_vec_with_registry!(
                "validator_service_submit_transaction_batch_size",
                "The number of transactions in the submit transaction request",
                &["req_type"],
                mysten_metrics::COUNT_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            num_rejected_tx_during_overload: register_int_counter_vec_with_registry!(
                "validator_service_num_rejected_tx_during_overload",
                "Number of rejected transaction due to system overload",
                &["error_type"],
                registry,
            )
            .unwrap(),
            submission_rejected_transactions: register_int_counter_vec_with_registry!(
                "validator_service_submission_rejected_transactions",
                "Number of transactions rejected during submission",
                &["reason"],
                registry,
            )
            .unwrap(),
            submission_suppressed_already_processed: register_int_counter_vec_with_registry!(
                "validator_service_submission_suppressed_already_processed",
                "Number of submitted transactions suppressed because consensus had already \
                 processed them this epoch (re-submission of already-processed transactions)",
                &["req_type"],
                registry,
            )
            .unwrap(),
            submission_suppressed_recently_submitted: register_int_counter_vec_with_registry!(
                "validator_service_submission_suppressed_recently_submitted",
                "Number of submitted transactions suppressed because the same transaction was \
                 submitted within the recent-submission window",
                &["req_type"],
                registry,
            )
            .unwrap(),
            recently_submitted_cache_size: register_int_gauge_with_registry!(
                "validator_service_recently_submitted_cache_size",
                "Approximate number of transaction digests held in the recent-submission duplicate-suppression cache",
                registry,
            )
            .unwrap(),
            recently_submitted_resubmission_interval: register_histogram_with_registry!(
                "validator_service_recently_submitted_resubmission_interval_seconds",
                "Time between a transaction being recorded and a duplicate resubmission of it being suppressed",
                mysten_metrics::SUBSECOND_LATENCY_SEC_BUCKETS.to_vec(),
                registry,
            )
            .unwrap(),
            submission_suppressed_inflight: register_int_counter_vec_with_registry!(
                "validator_service_submission_suppressed_inflight",
                "Number of submitted transactions suppressed because the same transaction was \
                 already being handled by a concurrent in-flight submit request",
                &["req_type"],
                registry,
            )
            .unwrap(),
            inflight_transactions: register_int_gauge_with_registry!(
                "validator_service_inflight_transactions",
                "Number of transactions from inflight submit requests",
                registry,
            )
            .unwrap(),
            connection_ip_not_found: register_int_counter_with_registry!(
                "validator_service_connection_ip_not_found",
                "Number of times connection IP was not extractable from request",
                registry,
            )
            .unwrap(),
            forwarded_header_parse_error: register_int_counter_with_registry!(
                "validator_service_forwarded_header_parse_error",
                "Number of times x-forwarded-for header could not be parsed",
                registry,
            )
            .unwrap(),
            forwarded_header_invalid: register_int_counter_with_registry!(
                "validator_service_forwarded_header_invalid",
                "Number of times x-forwarded-for header was invalid",
                registry,
            )
            .unwrap(),
            forwarded_header_not_included: register_int_counter_with_registry!(
                "validator_service_forwarded_header_not_included",
                "Number of times x-forwarded-for header was (unexpectedly) not included in request",
                registry,
            )
            .unwrap(),
            client_id_source_config_mismatch: register_int_counter_with_registry!(
                "validator_service_client_id_source_config_mismatch",
                "Number of times detected that client id source config doesn't agree with x-forwarded-for header",
                registry,
            )
            .unwrap(),
            x_forwarded_for_num_hops: register_gauge_with_registry!(
                "validator_service_x_forwarded_for_num_hops",
                "Number of hops in x-forwarded-for header",
                registry,
            )
            .unwrap(),
            gasless_rate_limited_count: register_int_counter_with_registry!(
                "validator_service_gasless_rate_limited_count",
                "Number of gasless transactions rejected by rate limiter",
                registry,
            )
            .unwrap(),
            gasless_submission_outcomes: register_int_counter_vec_with_registry!(
                "validator_service_gasless_submission_outcomes",
                "Number of valid gasless transaction submissions by outcome",
                &["outcome"],
                registry,
            )
            .unwrap(),
        }
    }

    pub fn new_for_tests() -> Self {
        let registry = Registry::new();
        Self::new(&registry)
    }
}

/// Where `handle_submit_transaction` routes a request.
enum AdmissionQueueSubmitMode {
    /// Admit via the gas-price priority queue.
    Queue,
    /// Submit directly to consensus, bypassing the queue — used when the queue
    /// is turned off by config, temporarily disabled by failover, or for a ping
    /// request. Individual txs are rejected when consensus is saturated
    /// (pre-queue behavior).
    Direct,
}

#[derive(Clone)]
pub struct ValidatorService {
    state: Arc<AuthorityState>,
    consensus_adapter: Arc<ConsensusAdapter>,
    metrics: Arc<ValidatorServiceMetrics>,
    traffic_controller: Option<Arc<TrafficController>>,
    client_id_source: Option<ClientIdSource>,
    gasless_limiter: GaslessRateLimiter,
    admission_queue: Option<AdmissionQueueContext>,
    /// Digests submitted within the last `recent_submission_window` (value: when recorded), to
    /// drop duplicate resubmissions before they reach consensus.
    recently_submitted: Cache<TransactionDigest, Instant>,
    /// How long a transaction is suppressed after submission (from node config).
    recent_submission_window: Duration,
    /// Digests currently being handled by an in-flight submit handler. Acquired atomically at
    /// entry and removed (then demoted into `recently_submitted`) when the handler returns.
    inflight_transactions: Arc<Mutex<HashSet<TransactionDigest>>>,
}

/// Assumed peak distinct-submission rate, used to size the dedup cache (per window).
const RECENT_SUBMISSION_PEAK_TPS: u64 = 50_000;

impl ValidatorService {
    pub fn new(
        state: Arc<AuthorityState>,
        consensus_adapter: Arc<ConsensusAdapter>,
        validator_metrics: Arc<ValidatorServiceMetrics>,
        client_id_source: Option<ClientIdSource>,
        admission_queue: Option<AdmissionQueueContext>,
    ) -> Self {
        let traffic_controller = state.traffic_controller.clone();
        let gasless_limiter = GaslessRateLimiter::new(state.consensus_gasless_counter.clone());
        let recent_submission_window = state.config.recent_submission_dedup_window();
        Self {
            state,
            consensus_adapter,
            metrics: validator_metrics,
            traffic_controller,
            client_id_source,
            gasless_limiter,
            admission_queue,
            recently_submitted: Self::new_recently_submitted_cache(recent_submission_window),
            recent_submission_window,
            inflight_transactions: Arc::new(Mutex::new(HashSet::new())),
        }
    }

    fn new_recently_submitted_cache(window: Duration) -> Cache<TransactionDigest, Instant> {
        // Memory backstop only; the window bounds the cache, and amplified duplicates do not add
        // entries (they share a digest). Sized for roughly one window at peak throughput.
        let max_capacity = window.as_secs().max(1) * RECENT_SUBMISSION_PEAK_TPS;
        Cache::builder()
            .time_to_live(window)
            .max_capacity(max_capacity)
            .build()
    }

    pub fn new_for_tests(
        state: Arc<AuthorityState>,
        consensus_adapter: Arc<ConsensusAdapter>,
        metrics: Arc<ValidatorServiceMetrics>,
    ) -> Self {
        let gasless_limiter = GaslessRateLimiter::new(state.consensus_gasless_counter.clone());
        let epoch_store = state.epoch_store_for_testing().clone();
        let slot_freed_notify = Arc::new(tokio::sync::Notify::new());
        let manager = Arc::new(AdmissionQueueManager::new_for_tests(
            consensus_adapter.clone(),
            slot_freed_notify,
        ));
        let admission_queue = Some(AdmissionQueueContext::spawn(manager, epoch_store));
        let recent_submission_window = state.config.recent_submission_dedup_window();
        Self {
            state,
            consensus_adapter,
            metrics,
            traffic_controller: None,
            client_id_source: None,
            gasless_limiter,
            admission_queue,
            recently_submitted: Self::new_recently_submitted_cache(recent_submission_window),
            recent_submission_window,
            inflight_transactions: Arc::new(Mutex::new(HashSet::new())),
        }
    }

    pub fn validator_state(&self) -> &Arc<AuthorityState> {
        &self.state
    }

    /// Test method that performs transaction validation without going through gRPC.
    pub fn handle_transaction_for_testing(&self, transaction: Transaction) -> SuiResult<()> {
        let epoch_store = self.state.load_epoch_store_one_call_per_task();

        // Validity check (basic structural validation)
        transaction.validity_check(&epoch_store.tx_validity_check_context())?;

        // Signature verification
        let transaction = epoch_store
            .verify_transaction_require_no_aliases(transaction)?
            .into_tx();

        // Validate the transaction
        self.state
            .handle_vote_transaction(&epoch_store, transaction)?;

        Ok(())
    }

    /// Test method that performs transaction validation with overload checking.
    /// Used for testing validator overload behavior.
    pub fn handle_transaction_for_testing_with_overload_check(
        &self,
        transaction: Transaction,
    ) -> SuiResult<()> {
        let epoch_store = self.state.load_epoch_store_one_call_per_task();

        // Validity check (basic structural validation)
        transaction.validity_check(&epoch_store.tx_validity_check_context())?;

        // Check system overload
        self.state.check_system_overload(
            transaction.data(),
            self.state.check_system_overload_at_signing(),
        )?;

        // Signature verification
        let transaction = epoch_store
            .verify_transaction_require_no_aliases(transaction)?
            .into_tx();

        // Validate the transaction
        self.state
            .handle_vote_transaction(&epoch_store, transaction)?;

        Ok(())
    }

    /// Collect the IDs of input objects that are immutable.
    /// This is used to create the ImmutableInputObjects claim for consensus messages.
    async fn collect_immutable_object_ids(
        &self,
        tx: &VerifiedTransaction,
        state: &AuthorityState,
    ) -> SuiResult<Vec<ObjectID>> {
        let input_objects = tx.data().transaction_data().input_objects()?;

        // Collect object IDs from ImmOrOwnedMoveObject inputs
        let object_ids: Vec<ObjectID> = input_objects
            .iter()
            .filter_map(|obj| match obj {
                InputObjectKind::ImmOrOwnedMoveObject((id, _, _)) => Some(*id),
                _ => None,
            })
            .collect();
        if object_ids.is_empty() {
            return Ok(vec![]);
        }

        // Load objects from cache and filter to immutable ones
        let objects = state.get_object_cache_reader().get_objects(&object_ids);

        // All objects should be found, since owned input objects have been validated to exist.
        objects
            .into_iter()
            .zip_debug_eq(object_ids.iter())
            .filter_map(|(obj, id)| {
                let Some(o) = obj else {
                    return Some(Err::<ObjectID, SuiError>(
                        SuiErrorKind::UserInputError {
                            error: UserInputError::ObjectNotFound {
                                object_id: *id,
                                version: None,
                            },
                        }
                        .into(),
                    ));
                };
                if o.is_immutable() {
                    Some(Ok(*id))
                } else {
                    None
                }
            })
            .collect::<SuiResult<Vec<ObjectID>>>()
    }

    #[instrument(
        name = "ValidatorService::handle_submit_transaction",
        level = "error",
        skip_all,
        err(level = "debug")
    )]
    async fn handle_submit_transaction(
        &self,
        request: tonic::Request<RawSubmitTxRequest>,
    ) -> WrappedServiceResponse<RawSubmitTxResponse> {
        let Self {
            state,
            consensus_adapter: _,
            metrics,
            traffic_controller: _,
            client_id_source,
            gasless_limiter: _,
            admission_queue: _,
            recently_submitted: _,
            recent_submission_window: _,
            inflight_transactions: _,
        } = self.clone();

        let submitter_client_addr = if let Some(client_id_source) = &client_id_source {
            self.get_client_ip_addr(&request, client_id_source)
        } else {
            self.get_client_ip_addr(&request, &ClientIdSource::SocketAddr)
        };

        let inner = request.into_inner();
        let start_epoch = state.load_epoch_store_one_call_per_task().epoch();

        let next_epoch = start_epoch + 1;
        let mut max_retries = 1;

        let mut inflight_guard = InflightTransactionsGuard::new(self);
        loop {
            let res = self
                .handle_submit_transaction_inner(
                    &state,
                    &metrics,
                    &inner,
                    submitter_client_addr,
                    &mut inflight_guard,
                )
                .await;
            match res {
                Ok((response, weight)) => return Ok((tonic::Response::new(response), weight)),
                Err(err) => {
                    if max_retries > 0
                        && let SuiErrorKind::ValidatorHaltedAtEpochEnd = err.as_inner()
                    {
                        max_retries -= 1;

                        debug!(
                            "ValidatorHaltedAtEpochEnd. Will retry after validator reconfigures"
                        );

                        if let Ok(Ok(new_epoch)) =
                            timeout(Duration::from_secs(15), state.wait_for_epoch(next_epoch)).await
                        {
                            assert_reachable!("retry submission at epoch end");
                            if new_epoch >= next_epoch {
                                continue;
                            }
                            // wait_for_epoch guarantees >= target; < would indicate a bug there.
                            debug_fatal!(
                                "wait_for_epoch returned early: expected >= {}, got {}",
                                next_epoch,
                                new_epoch
                            );
                        }
                    }
                    return Err(err.into());
                }
            }
        }
    }

    async fn handle_submit_transaction_inner(
        &self,
        state: &AuthorityState,
        metrics: &ValidatorServiceMetrics,
        request: &RawSubmitTxRequest,
        submitter_client_addr: Option<IpAddr>,
        inflight_guard: &mut InflightTransactionsGuard,
    ) -> SuiResult<(RawSubmitTxResponse, Weight)> {
        let epoch_store = state.load_epoch_store_one_call_per_task();
        let submit_type = SubmitTxType::try_from(request.submit_type).map_err(|e| {
            SuiErrorKind::GrpcMessageDeserializeError {
                type_info: "RawSubmitTxRequest.submit_type".to_string(),
                error: e.to_string(),
            }
        })?;

        let is_ping_request = submit_type == SubmitTxType::Ping;
        if is_ping_request {
            fp_ensure!(
                request.transactions.is_empty(),
                SuiErrorKind::InvalidRequest(format!(
                    "Ping request cannot contain {} transactions",
                    request.transactions.len()
                ))
                .into()
            );
        } else {
            // Ensure default and soft bundle requests contain at least one transaction.
            fp_ensure!(
                !request.transactions.is_empty(),
                SuiErrorKind::InvalidRequest(
                    "At least one transaction needs to be submitted".to_string(),
                )
                .into()
            );
        }

        // NOTE: for soft bundle requests, the system tries to sequence the transactions in the same order
        // if they use the same gas price. But this is only done with best effort.
        // Transactions in a soft bundle can be individually rejected or deferred, without affecting
        // other transactions in the same bundle.
        let is_soft_bundle_request = submit_type == SubmitTxType::SoftBundle;

        let max_num_transactions = if is_soft_bundle_request {
            // Soft bundle cannot contain too many transactions.
            // Otherwise it is hard to include all of them in a single block.
            epoch_store.protocol_config().max_soft_bundle_size()
        } else {
            // Still enforce a limit even when transactions do not need to be in the same block.
            epoch_store
                .protocol_config()
                .max_num_transactions_in_block()
        };
        fp_ensure!(
            request.transactions.len() <= max_num_transactions as usize,
            SuiErrorKind::InvalidRequest(format!(
                "Too many transactions in request: {} vs {}",
                request.transactions.len(),
                max_num_transactions
            ))
            .into()
        );

        // Transaction digests.
        let mut tx_digests = Vec::with_capacity(request.transactions.len());
        // Transactions to submit to consensus.
        let mut consensus_transactions = Vec::with_capacity(request.transactions.len());
        // Indexes of transactions above in the request transactions.
        let mut transaction_indexes = Vec::with_capacity(request.transactions.len());
        // Results corresponding to each transaction in the request.
        let mut results: Vec<Option<SubmitTxResult>> = vec![None; request.transactions.len()];
        // Total size of all transactions in the request.
        let mut total_size_bytes = 0;
        // Whether the request contains any gasless transaction.
        let mut has_gasless = false;
        // Set when a transaction duplicates an in-flight submission at admission. Tracked
        // separately because it is detected after the per-tx results are finalized (those
        // remain Submitted), so it cannot be derived from the results alone.
        let mut duplicate_at_admission = false;
        // First gas price seen in this soft bundle.
        let mut expected_soft_bundle_gas_price = None;
        // Transaction digests seen in this request attempt, used to reject repeated transactions.
        let mut request_digests = HashSet::new();

        let req_type = if is_ping_request {
            "ping"
        } else if request.transactions.len() == 1 {
            "single_transaction"
        } else if is_soft_bundle_request {
            "soft_bundle"
        } else {
            "batch"
        };

        let _handle_tx_metrics_guard = metrics
            .handle_submit_transaction_latency
            .with_label_values(&[req_type])
            .start_timer();

        let submit_mode = self.classify_submit_mode(is_ping_request);

        for (idx, tx_bytes) in request.transactions.iter().enumerate() {
            let transaction = match bcs::from_bytes::<Transaction>(tx_bytes) {
                Ok(txn) => txn,
                Err(e) => {
                    // Ok to fail the request when any transaction is invalid.
                    return Err(SuiErrorKind::TransactionDeserializationError {
                        error: format!("Failed to deserialize transaction at index {}: {}", idx, e),
                    }
                    .into());
                }
            };

            // Ok to fail the request when any transaction is invalid.
            let tx_size = transaction.validity_check(&epoch_store.tx_validity_check_context())?;
            let tx_digest = *transaction.digest();

            // A request must not repeat a transaction.
            if !request_digests.insert(tx_digest) {
                let error: SuiError = SuiErrorKind::UserInputError {
                    error: UserInputError::RepeatedTransactions { digest: tx_digest },
                }
                .into();
                // Reject individual repeated transactions in batch.
                if is_soft_bundle_request {
                    return Err(error);
                }
                results[idx] = Some(SubmitTxResult::Rejected { error });
                continue;
            }

            // Soft bundles require all transactions to use the same gas price.
            if is_soft_bundle_request {
                let gas_price = transaction.data().transaction_data().gas_price();
                if let Some(expected) = expected_soft_bundle_gas_price {
                    fp_ensure!(
                        gas_price == expected,
                        SuiErrorKind::UserInputError {
                            error: UserInputError::GasPriceMismatchError {
                                digest: tx_digest,
                                expected,
                                actual: gas_price,
                            }
                        }
                        .into()
                    );
                } else {
                    expected_soft_bundle_gas_price = Some(gas_price);
                }
            }

            let is_gasless = transaction
                .data()
                .transaction_data()
                .is_gasless_transaction();

            if is_gasless {
                has_gasless = true;
                metrics
                    .gasless_submission_outcomes
                    .with_label_values(&["attempted"])
                    .inc();
            }

            let overload_check_res = state.check_system_overload(
                transaction.data(),
                state.check_system_overload_at_signing(),
            );
            if let Err(error) = overload_check_res {
                metrics
                    .num_rejected_tx_during_overload
                    .with_label_values(&[error.as_ref()])
                    .inc();
                if is_gasless {
                    metrics
                        .gasless_submission_outcomes
                        .with_label_values(&["rejected_overload"])
                        .inc();
                }
                results[idx] = Some(SubmitTxResult::Rejected { error });
                continue;
            }

            // Use the pre-queue per-tx consensus overload reject on the direct
            // submission path (queue off, failover, or ping).
            if matches!(submit_mode, AdmissionQueueSubmitMode::Direct)
                && let Err(error) = self.consensus_adapter.check_consensus_overload()
            {
                state.update_overload_metrics("consensus");
                metrics
                    .num_rejected_tx_during_overload
                    .with_label_values(&[error.as_ref()])
                    .inc();
                if is_gasless {
                    metrics
                        .gasless_submission_outcomes
                        .with_label_values(&["rejected_overload"])
                        .inc();
                }
                results[idx] = Some(SubmitTxResult::Rejected { error });
                continue;
            }

            if is_gasless
                && !self
                    .gasless_limiter
                    .try_acquire(epoch_store.protocol_config())
            {
                metrics.gasless_rate_limited_count.inc();
                metrics
                    .gasless_submission_outcomes
                    .with_label_values(&["rejected_rate_limited"])
                    .inc();
                results[idx] = Some(SubmitTxResult::Rejected {
                    error: SuiErrorKind::ValidatorOverloadedRetryAfter {
                        retry_after_secs: 1,
                    }
                    .into(),
                });
                continue;
            }

            // Ok to fail the request when any signature is invalid.
            let verified_transaction = {
                let _metrics_guard = metrics.tx_verification_latency.start_timer();
                if epoch_store.protocol_config().address_aliases() {
                    match epoch_store.verify_transaction_with_current_aliases(transaction) {
                        Ok(tx) => tx,
                        Err(e) => {
                            metrics.signature_errors.inc();
                            return Err(e);
                        }
                    }
                } else {
                    match epoch_store.verify_transaction_require_no_aliases(transaction) {
                        Ok(tx) => tx,
                        Err(e) => {
                            metrics.signature_errors.inc();
                            return Err(e);
                        }
                    }
                }
            };

            debug!(
                ?tx_digest,
                "handle_submit_transaction: verified transaction"
            );

            // Check if the transaction has executed, before checking input objects
            // which could have been consumed.
            if let Some(effects) = state
                .get_transaction_cache_reader()
                .get_executed_effects(&tx_digest)
            {
                let effects_digest = effects.digest();
                if let Ok(executed_data) = self.complete_executed_data(effects).await {
                    let executed_result = SubmitTxResult::Executed {
                        effects_digest,
                        details: Some(executed_data),
                    };
                    results[idx] = Some(executed_result);
                    debug!(?tx_digest, "handle_submit_transaction: already executed");
                    continue;
                }
            }

            if self
                .state
                .get_transaction_cache_reader()
                .transaction_executed_in_last_epoch(&tx_digest, epoch_store.epoch())
            {
                results[idx] = Some(SubmitTxResult::Rejected {
                    error: UserInputError::TransactionAlreadyExecuted { digest: tx_digest }.into(),
                });
                debug!(
                    ?tx_digest,
                    "handle_submit_transaction: transaction already executed in previous epoch"
                );
                continue;
            }

            // Suppress resubmission of transactions consensus already processed this epoch:
            // executed transactions whose effects details could not be reconstructed above (e.g.
            // objects pruned), sequenced-but-deferred transactions, and dropped transactions.
            let consensus_key = SequencedConsensusTransactionKey::External(
                ConsensusTransactionKey::Certificate(tx_digest),
            );
            if epoch_store.is_consensus_message_processed(&consensus_key)? {
                // Prefer a concrete, non-retriable error over the generic, retriable
                // TransactionProcessing suppression. A processed-but-unexecuted digest is
                // commonly a dropped owned-object conflict loser; surfacing the terminal
                // error lets the client stop retrying instead of polling for effects that
                // will never come.
                //
                // First check the epoch owned-object lock table with the same conflict
                // logic the consensus handler uses post-consensus. Locks are never
                // released within an epoch, so this reports the conflict even before the
                // winner executes, while the loser's input versions still validate as
                // live.
                if let Ok(input_objects) = verified_transaction
                    .tx()
                    .data()
                    .transaction_data()
                    .input_objects()
                {
                    let immutable_object_ids = self
                        .collect_immutable_object_ids(verified_transaction.tx(), state)
                        .await?;
                    let owned_object_refs: Vec<_> = 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();
                    let existing_locks =
                        epoch_store.get_owned_object_locks_map(&owned_object_refs)?;
                    if let Err(error) = epoch_store.try_acquire_owned_object_locks_post_consensus(
                        &owned_object_refs,
                        tx_digest,
                        &HashMap::new(),
                        &existing_locks,
                    ) {
                        debug!(
                            ?tx_digest,
                            "handle_submit_transaction: processed transaction rejected on lock conflict: {error}"
                        );
                        metrics
                            .submission_rejected_transactions
                            .with_label_values(&[error.to_variant_name()])
                            .inc();
                        results[idx] = Some(SubmitTxResult::Rejected { error });
                        continue;
                    }
                }
                // Then revalidate against live state, which surfaces the terminal
                // stale-version error once the conflict winner has executed.
                if let Err(error) =
                    state.handle_vote_transaction(&epoch_store, verified_transaction.tx().clone())
                {
                    // The transaction may have executed while being validated (e.g. it was
                    // deferred rather than dropped).
                    if let Some(effects) = state
                        .get_transaction_cache_reader()
                        .get_executed_effects(&tx_digest)
                    {
                        let effects_digest = effects.digest();
                        if let Ok(executed_data) = self.complete_executed_data(effects).await {
                            results[idx] = Some(SubmitTxResult::Executed {
                                effects_digest,
                                details: Some(executed_data),
                            });
                            continue;
                        }
                    }
                    debug!(
                        ?tx_digest,
                        "handle_submit_transaction: processed transaction rejected on revalidation: {error}"
                    );
                    metrics
                        .submission_rejected_transactions
                        .with_label_values(&[error.to_variant_name()])
                        .inc();
                    results[idx] = Some(SubmitTxResult::Rejected { error });
                    continue;
                }
                // Validation passed, so this processed digest may still be executable.
                // Return retriable TransactionProcessing rather than resubmitting it to consensus.
                // A later client retry can observe effects or a concrete terminal validation error.
                metrics
                    .submission_suppressed_already_processed
                    .with_label_values(&[req_type])
                    .inc();
                results[idx] = Some(SubmitTxResult::Rejected {
                    error: SuiErrorKind::TransactionProcessing {
                        digest: tx_digest,
                        status: "consensus message processed".to_string(),
                    }
                    .into(),
                });
                debug!(
                    ?tx_digest,
                    "handle_submit_transaction: consensus message already processed"
                );
                continue;
            }

            // Atomically acquire the digest for the duration of this handler. Reject concurrent
            // and recent duplicates and record the result per result index.
            match inflight_guard.try_acquire(tx_digest) {
                AcquireOutcome::Acquired | AcquireOutcome::AlreadyAcquiredByThisRequest => {
                    // Continue to process the transaction and submit to consensus.
                }
                AcquireOutcome::AlreadyAcquiredByAnotherRequest => {
                    metrics
                        .submission_suppressed_inflight
                        .with_label_values(&[req_type])
                        .inc();
                    results[idx] = Some(SubmitTxResult::Rejected {
                        error: SuiErrorKind::TransactionSubmitted { digest: tx_digest }.into(),
                    });
                    debug!(
                        ?tx_digest,
                        "handle_submit_transaction: concurrent submission in progress"
                    );
                    continue;
                }
                AcquireOutcome::RecentlyProcessed { since } => {
                    metrics
                        .submission_suppressed_recently_submitted
                        .with_label_values(&[req_type])
                        .inc();
                    metrics
                        .recently_submitted_resubmission_interval
                        .observe(since.as_secs_f64());
                    results[idx] = Some(SubmitTxResult::Rejected {
                        error: SuiErrorKind::TransactionSubmitted { digest: tx_digest }.into(),
                    });
                    debug!(?tx_digest, "handle_submit_transaction: recently processed");
                    continue;
                }
            }

            debug!(
                ?tx_digest,
                "handle_submit_transaction: waiting for fastpath dependency objects"
            );
            if !state
                .wait_for_fastpath_dependency_objects(
                    verified_transaction.tx(),
                    epoch_store.epoch(),
                )
                .await?
            {
                debug!(
                    ?tx_digest,
                    "fastpath input objects are still unavailable after waiting"
                );
            }

            match state.handle_vote_transaction(&epoch_store, verified_transaction.tx().clone()) {
                Ok(_) => { /* continue processing */ }
                Err(e) => {
                    // Check if transaction has been executed while being validated.
                    // This is an edge case so checking executed effects twice is acceptable.
                    if let Some(effects) = state
                        .get_transaction_cache_reader()
                        .get_executed_effects(&tx_digest)
                    {
                        let effects_digest = effects.digest();
                        if let Ok(executed_data) = self.complete_executed_data(effects).await {
                            let executed_result = SubmitTxResult::Executed {
                                effects_digest,
                                details: Some(executed_data),
                            };
                            results[idx] = Some(executed_result);
                            continue;
                        }
                    }

                    // When the transaction has not been executed, record the error for the transaction.
                    debug!(?tx_digest, "Transaction rejected during submission: {e}");
                    metrics
                        .submission_rejected_transactions
                        .with_label_values(&[e.to_variant_name()])
                        .inc();
                    results[idx] = Some(SubmitTxResult::Rejected { error: e });
                    continue;
                }
            }

            // Create claims with aliases and / or immutable objects.
            let mut claims = vec![];

            let immutable_object_ids = self
                .collect_immutable_object_ids(verified_transaction.tx(), state)
                .await?;
            if !immutable_object_ids.is_empty() {
                claims.push(TransactionClaim::ImmutableInputObjects(
                    immutable_object_ids,
                ));
            }

            let (tx, aliases) = verified_transaction.into_inner();
            if epoch_store.protocol_config().address_aliases() {
                if epoch_store
                    .protocol_config()
                    .fix_checkpoint_signature_mapping()
                {
                    claims.push(TransactionClaim::AddressAliasesV2(aliases));
                } else {
                    let v1_aliases: Vec<_> = tx
                        .data()
                        .intent_message()
                        .value
                        .required_signers()
                        .into_iter()
                        .zip_eq(aliases.into_iter().map(|(_, seq)| seq))
                        .collect();
                    #[allow(deprecated)]
                    claims.push(TransactionClaim::AddressAliases(
                        nonempty::NonEmpty::from_vec(v1_aliases)
                            .expect("must have at least one required_signer"),
                    ));
                }
            }

            let tx_with_claims = TransactionWithClaims::new(tx.into(), claims);

            consensus_transactions.push(ConsensusTransaction::new_user_transaction_v2_message(
                &state.name,
                tx_with_claims,
            ));
            if is_gasless {
                metrics
                    .gasless_submission_outcomes
                    .with_label_values(&["submitted"])
                    .inc();
            }

            transaction_indexes.push(idx);
            tx_digests.push(tx_digest);
            total_size_bytes += tx_size;
        }

        if consensus_transactions.is_empty() && !is_ping_request {
            let spam_weight = Self::request_spam_weight(
                &results,
                has_gasless,
                duplicate_at_admission,
                is_ping_request,
            );
            let response = Self::try_from_submit_tx_response(results)?;
            return Ok((response, spam_weight));
        }

        // Set the max bytes size of the soft bundle to be half of the consensus max transactions in block size.
        // We do this to account for serialization overheads and to ensure that the soft bundle is not too large
        // when is attempted to be posted via consensus.
        let max_transaction_bytes = if is_soft_bundle_request {
            epoch_store
                .protocol_config()
                .consensus_max_transactions_in_block_bytes()
                / 2
        } else {
            epoch_store
                .protocol_config()
                .consensus_max_transactions_in_block_bytes()
        };
        fp_ensure!(
            total_size_bytes <= max_transaction_bytes as usize,
            SuiErrorKind::UserInputError {
                error: UserInputError::TotalTransactionSizeTooLargeInBatch {
                    size: total_size_bytes,
                    limit: max_transaction_bytes,
                },
            }
            .into()
        );

        metrics
            .handle_submit_transaction_bytes
            .with_label_values(&[req_type])
            .observe(total_size_bytes as f64);
        metrics
            .handle_submit_transaction_batch_size
            .with_label_values(&[req_type])
            .observe(consensus_transactions.len() as f64);

        let _latency_metric_guard = metrics
            .handle_submit_transaction_consensus_latency
            .with_label_values(&[req_type])
            .start_timer();

        if is_soft_bundle_request {
            // We only allow the `consensus_transactions` to be empty for ping requests. This is how it should and is be treated from the downstream components.
            // For any other case, having an empty `consensus_transactions` vector is an invalid state and we should have never reached at this point.
            assert!(
                !consensus_transactions.is_empty(),
                "A valid soft bundle must have at least one transaction"
            );
        }

        // Soft bundles are inserted as a single queue entry.
        // Individual transactions are each inserted separately.
        let tx_groups: Vec<Vec<ConsensusTransaction>> = if is_soft_bundle_request || is_ping_request
        {
            vec![consensus_transactions]
        } else {
            consensus_transactions
                .into_iter()
                .map(|t| vec![t])
                .collect()
        };

        // Map each submission group back to the (result index, digest) of the transactions it
        // contains, so a per-group outcome — consensus positions, or an "already processing"
        // error — can be recorded against each individual transaction. Soft bundles submit as a
        // single group; individual transactions submit one group each.
        let group_tx_meta = if is_soft_bundle_request {
            vec![
                transaction_indexes
                    .into_iter()
                    .zip_eq(tx_digests)
                    .collect::<Vec<_>>(),
            ]
        } else {
            transaction_indexes
                .into_iter()
                .zip_eq(tx_digests)
                .map(|pair| vec![pair])
                .collect::<Vec<_>>()
        };

        // Collect one result per submission group WITHOUT short-circuiting. An
        // already-processing transaction is reported per-tx as a retriable below;
        // any other error fails the whole request, after all groups have settled.
        // Soft bundles submit as a single group; individual transactions submit one group each.
        let group_results = match submit_mode {
            AdmissionQueueSubmitMode::Direct => {
                let futures = tx_groups.into_iter().map(|txns| {
                    debug!(
                        "handle_submit_transaction: submitting consensus transactions ({}): {}",
                        req_type,
                        txns.iter().map(|t| t.local_display()).join(", ")
                    );
                    self.consensus_adapter.submit_and_get_positions(
                        txns,
                        &epoch_store,
                        submitter_client_addr,
                    )
                });
                future::join_all(futures).await
            }
            AdmissionQueueSubmitMode::Queue => {
                let aq = self
                    .admission_queue
                    .as_ref()
                    .expect("Queue mode implies admission_queue is Some")
                    .load();
                let mut receivers = Vec::with_capacity(tx_groups.len());
                for txns in tx_groups {
                    let gas_price = Self::extract_gas_price(&txns);
                    let (rx, newly_inserted) = aq
                        .try_insert(gas_price, txns, submitter_client_addr)
                        .await?;
                    if !newly_inserted {
                        // Duplicate of an in-flight submission; flag the request as spam. The
                        // per-tx result is still Submitted, so this is tracked separately.
                        duplicate_at_admission = true;
                    }
                    receivers.push(rx);
                }
                future::join_all(receivers.into_iter().map(|rx| async move {
                    match rx.await {
                        Ok(result) => result.map_err(SuiError::from),
                        Err(_) => Err(SuiError::from(
                            SuiErrorKind::TooManyTransactionsPendingConsensus,
                        )),
                    }
                }))
                .await
            }
        };

        if is_ping_request {
            // For ping requests there is a single group returning the special consensus position.
            let consensus_positions = group_results
                .into_iter()
                .next()
                .expect("Ping request must have exactly one submission group")?;
            assert_eq!(consensus_positions.len(), 1);
            results.push(Some(SubmitTxResult::Submitted {
                consensus_position: consensus_positions[0],
            }));
        } else {
            for (group_result, txns_meta) in group_results.into_iter().zip_debug_eq(group_tx_meta) {
                match group_result {
                    Ok(consensus_positions) => {
                        for ((idx, tx_digest), consensus_position) in
                            txns_meta.into_iter().zip_debug_eq(consensus_positions)
                        {
                            debug!(
                                ?tx_digest,
                                "handle_submit_transaction: submitted consensus transaction at {}",
                                consensus_position,
                            );
                            results[idx] = Some(SubmitTxResult::Submitted { consensus_position });
                        }
                    }
                    // The transaction(s) in this group are already being processed by consensus.
                    // Report per-tx as a retriable rejection rather than failing the whole request.
                    Err(err) => {
                        let SuiErrorKind::TransactionProcessing { status, .. } =
                            err.as_inner().clone()
                        else {
                            return Err(err);
                        };
                        // For TransactionProcessing error, ensure the per txn result has the correct digest.
                        for (idx, tx_digest) in txns_meta {
                            debug!(
                                ?tx_digest,
                                "handle_submit_transaction: transaction already processing: {err}"
                            );
                            // Same suppression the upfront `is_consensus_message_processed` check
                            // records, just detected during submission instead of before it. The
                            // two paths are mutually exclusive, so this does not double-count.
                            metrics
                                .submission_suppressed_already_processed
                                .with_label_values(&[req_type])
                                .inc();
                            results[idx] = Some(SubmitTxResult::Rejected {
                                error: SuiErrorKind::TransactionProcessing {
                                    digest: tx_digest,
                                    status: status.clone(),
                                }
                                .into(),
                            });
                        }
                    }
                }
            }
        }

        let spam_weight = Self::request_spam_weight(
            &results,
            has_gasless,
            duplicate_at_admission,
            is_ping_request,
        );
        let response = Self::try_from_submit_tx_response(results)?;
        Ok((response, spam_weight))
    }

    /// Traffic-control spam weight for a whole submit request. The request is spam unless it is
    /// entirely accepted gas-chargable work.
    fn request_spam_weight(
        results: &[Option<SubmitTxResult>],
        has_gasless: bool,
        duplicate_at_admission: bool,
        is_ping: bool,
    ) -> Weight {
        if is_ping || has_gasless || duplicate_at_admission {
            return Weight::one();
        }
        for result in results {
            let Some(result) = result else {
                // `results` is expected to be fully populated (every entry `Some`) for the
                // request's transactions; a missing entry is a bug and is conservatively
                // treated as spam.
                debug_fatal!("transaction outcome unset when computing spam weight");
                return Weight::one();
            };
            if Self::submission_spam_weight(result) == Weight::one() {
                return Weight::one();
            }
        }
        Weight::zero()
    }

    fn submission_spam_weight(result: &SubmitTxResult) -> Weight {
        match result {
            SubmitTxResult::Submitted { .. } => Weight::zero(),
            // Non-submitted results can't be charged.
            SubmitTxResult::Executed { .. } | SubmitTxResult::Rejected { .. } => Weight::one(),
        }
    }

    fn try_from_submit_tx_response(
        results: Vec<Option<SubmitTxResult>>,
    ) -> Result<RawSubmitTxResponse, SuiError> {
        let mut raw_results = Vec::new();
        for (i, result) in results.into_iter().enumerate() {
            let result = result.ok_or_else(|| SuiErrorKind::GenericAuthorityError {
                error: format!("Missing transaction result at {}", i),
            })?;
            let raw_result = result.try_into()?;
            raw_results.push(raw_result);
        }
        Ok(RawSubmitTxResponse {
            results: raw_results,
        })
    }

    /// Extract the gas price from a batch of consensus transactions.
    /// Returns the minimum gas price in the batch, or 0 if no user transactions.
    fn extract_gas_price(transactions: &[ConsensusTransaction]) -> u64 {
        use sui_types::messages_consensus::ConsensusTransactionKind;
        transactions
            .iter()
            .filter_map(|tx| match &tx.kind {
                ConsensusTransactionKind::CertifiedTransaction(cert) => Some(cert.gas_price()),
                ConsensusTransactionKind::UserTransaction(t) => {
                    Some(t.data().transaction_data().gas_price())
                }
                ConsensusTransactionKind::UserTransactionV2(t) => {
                    Some(t.tx().data().transaction_data().gas_price())
                }
                _ => None,
            })
            .min()
            .unwrap_or(0)
    }

    fn classify_submit_mode(&self, is_ping_request: bool) -> AdmissionQueueSubmitMode {
        let Some(aq) = &self.admission_queue else {
            return AdmissionQueueSubmitMode::Direct;
        };

        // Ping requests carry no transactions and must not wait behind queued
        // work; submit them directly to consensus.
        if is_ping_request {
            return AdmissionQueueSubmitMode::Direct;
        }

        // If the queue actor is stuck, fall back to direct submission with the
        // pre-queue saturation reject until it resumes making progress.
        if aq.load().failover_tripped() {
            return AdmissionQueueSubmitMode::Direct;
        }

        AdmissionQueueSubmitMode::Queue
    }

    async fn collect_effects_data(
        &self,
        effects: &TransactionEffects,
        include_events: bool,
        include_input_objects: bool,
        include_output_objects: bool,
    ) -> SuiResult<(Option<TransactionEvents>, Vec<Object>, Vec<Object>)> {
        let events = if include_events && effects.events_digest().is_some() {
            Some(
                self.state
                    .get_transaction_events(effects.transaction_digest())?,
            )
        } else {
            None
        };

        let input_objects = if include_input_objects {
            self.state.get_transaction_input_objects(effects)?
        } else {
            vec![]
        };

        let output_objects = if include_output_objects {
            self.state.get_transaction_output_objects(effects)?
        } else {
            vec![]
        };

        Ok((events, input_objects, output_objects))
    }
}

type WrappedServiceResponse<T> = Result<(tonic::Response<T>, Weight), tonic::Status>;

/// RAII guard tracking the transaction digests a single submit request is actively handling, so
/// concurrent duplicates can be rejected. On drop, each acquired digest is removed from
/// the in-flight set and demoted into `recently_submitted` cache, so resubmissions
/// arriving shortly after the handler returns are still suppressed.
struct InflightTransactionsGuard {
    // Handle to inflight map and recently submitted cache.
    inflight: Arc<Mutex<HashSet<TransactionDigest>>>,
    recently_submitted: Cache<TransactionDigest, Instant>,
    window: Duration,
    metrics: Arc<ValidatorServiceMetrics>,
    /// Digests this request successfully acquired.
    acquired: HashSet<TransactionDigest>,
}

enum AcquireOutcome {
    /// Transaction digest newly acquired by this request.
    Acquired,
    /// Transaction digest already acquired by this request before this internal retry attempt.
    AlreadyAcquiredByThisRequest,
    /// Transaction digest being handled by another concurrent request — reject this index.
    AlreadyAcquiredByAnotherRequest,
    /// Transaction digest recently processed — reject this index.
    RecentlyProcessed { since: Duration },
}

impl InflightTransactionsGuard {
    fn new(service: &ValidatorService) -> Self {
        Self {
            inflight: service.inflight_transactions.clone(),
            recently_submitted: service.recently_submitted.clone(),
            window: service.recent_submission_window,
            metrics: service.metrics.clone(),
            acquired: HashSet::new(),
        }
    }

    fn try_acquire(&mut self, digest: TransactionDigest) -> AcquireOutcome {
        // A retry of this own request re-acquires the same digests.
        if self.acquired.contains(&digest) {
            return AcquireOutcome::AlreadyAcquiredByThisRequest;
        }

        // Suppress resubmissions of recently processed transactions.
        if let Some(outcome) = self.recently_processed_outcome(digest) {
            return outcome;
        }

        // Atomic check-and-acquire against concurrent in-flight transactions.
        {
            let mut set = self.inflight.lock();
            // Only continue processing the transaction if it is not already inflight.
            if !set.insert(digest) {
                return AcquireOutcome::AlreadyAcquiredByAnotherRequest;
            }
            self.metrics.inflight_transactions.set(set.len() as i64);
        }

        // Without this re-check, a duplicated transaction arriving between the first check and
        // the digest acquisition could slip through.
        if let Some(outcome) = self.recently_processed_outcome(digest) {
            let mut set = self.inflight.lock();
            set.remove(&digest);
            self.metrics.inflight_transactions.set(set.len() as i64);
            return outcome;
        }

        self.acquired.insert(digest);
        AcquireOutcome::Acquired
    }

    fn recently_processed_outcome(&self, digest: TransactionDigest) -> Option<AcquireOutcome> {
        let recorded_at = self.recently_submitted.get(&digest)?;
        let since = recorded_at.elapsed();
        (since < self.window).then_some(AcquireOutcome::RecentlyProcessed { since })
    }
}

impl Drop for InflightTransactionsGuard {
    fn drop(&mut self) {
        if self.acquired.is_empty() {
            return;
        }
        // Demote inflight transactions to recently submitted cache before taking the in-flight lock,
        // to avoid cleaning up the cache with the lock.
        let now = Instant::now();
        for digest in &self.acquired {
            self.recently_submitted.insert(*digest, now);
        }
        {
            let mut set = self.inflight.lock();
            for digest in &self.acquired {
                set.remove(digest);
            }
            self.metrics.inflight_transactions.set(set.len() as i64);
        }
        self.metrics
            .recently_submitted_cache_size
            .set(self.recently_submitted.entry_count() as i64);
    }
}

impl ValidatorService {
    async fn handle_submit_transaction_impl(
        &self,
        request: tonic::Request<RawSubmitTxRequest>,
    ) -> WrappedServiceResponse<RawSubmitTxResponse> {
        self.handle_submit_transaction(request).await
    }

    async fn wait_for_effects_impl(
        &self,
        request: tonic::Request<RawWaitForEffectsRequest>,
    ) -> WrappedServiceResponse<RawWaitForEffectsResponse> {
        let request: WaitForEffectsRequest = request.into_inner().try_into()?;
        let epoch_store = self.state.load_epoch_store_one_call_per_task();
        let response = timeout(
            // TODO(fastpath): Tune this once we have a good estimate of the typical delay.
            Duration::from_secs(20),
            epoch_store
                .within_alive_epoch(self.wait_for_effects_response(request, &epoch_store))
                .map_err(|_| SuiErrorKind::EpochEnded(epoch_store.epoch())),
        )
        .await
        .map_err(|_| tonic::Status::internal("Timeout waiting for effects"))???
        .try_into()?;
        Ok((tonic::Response::new(response), Weight::zero()))
    }

    #[instrument(name= "ValidatorService::wait_for_effects_response", level = "debug", skip_all, fields(consensus_position = ?request.consensus_position))]
    async fn wait_for_effects_response(
        &self,
        request: WaitForEffectsRequest,
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> SuiResult<WaitForEffectsResponse> {
        if request.ping_type.is_some() {
            return timeout(
                Duration::from_secs(10),
                self.ping_response(request, epoch_store),
            )
            .await
            .map_err(|_| SuiErrorKind::TimeoutError)?;
        }

        let Some(tx_digest) = request.transaction_digest else {
            return Err(SuiErrorKind::InvalidRequest(
                "Transaction digest is required for wait for effects requests".to_string(),
            )
            .into());
        };
        let tx_digests = [tx_digest];

        // When consensus_position is provided, also watch the consensus status cache
        // so rejected/dropped transactions get a timely response instead of waiting
        // forever for effects that will never be produced.
        let consensus_status_future = async {
            let consensus_position = match request.consensus_position {
                Some(pos) => pos,
                None => return futures::future::pending().await,
            };
            let consensus_tx_status_cache = &epoch_store.consensus_tx_status_cache;
            consensus_tx_status_cache.check_position_too_ahead(&consensus_position)?;
            match consensus_tx_status_cache
                .notify_read_transaction_status(consensus_position)
                .await
            {
                NotifyReadConsensusTxStatusResult::Status(
                    ConsensusTxStatus::Rejected | ConsensusTxStatus::Dropped,
                ) => Ok(WaitForEffectsResponse::Rejected {
                    error: epoch_store.get_rejection_vote_reason(consensus_position),
                }),
                NotifyReadConsensusTxStatusResult::Status(ConsensusTxStatus::Finalized) => {
                    // Effects will be produced — yield to let the effects future win.
                    futures::future::pending().await
                }
                NotifyReadConsensusTxStatusResult::Expired(round) => {
                    Ok(WaitForEffectsResponse::Expired {
                        epoch: epoch_store.epoch(),
                        round: Some(round),
                    })
                }
            }
        };

        tokio::select! {
            effects_result = self.state
                .get_transaction_cache_reader()
                .notify_read_executed_effects_may_fail(
                    "AuthorityServer::wait_for_effects::notify_read_executed_effects_finalized",
                    &tx_digests,
                ) => {
                let effects = effects_result?.pop().unwrap();
                let effects_digest = effects.digest();
                let details = if request.include_details {
                    Some(self.complete_executed_data(effects).await?)
                } else {
                    None
                };
                Ok(WaitForEffectsResponse::Executed {
                    effects_digest,
                    details,
                })
            }
            status_response = consensus_status_future => {
                status_response
            }
        }
    }

    #[instrument(level = "error", skip_all, err(level = "debug"))]
    async fn ping_response(
        &self,
        request: WaitForEffectsRequest,
        epoch_store: &Arc<AuthorityPerEpochStore>,
    ) -> SuiResult<WaitForEffectsResponse> {
        let consensus_tx_status_cache = &epoch_store.consensus_tx_status_cache;

        let Some(consensus_position) = request.consensus_position else {
            return Err(SuiErrorKind::InvalidRequest(
                "Consensus position is required for Ping requests".to_string(),
            )
            .into());
        };

        // We assume that the caller has already checked for the existence of the `ping` field, but handling it gracefully here.
        let Some(ping) = request.ping_type else {
            return Err(SuiErrorKind::InvalidRequest(
                "Ping type is required for ping requests".to_string(),
            )
            .into());
        };

        let _metrics_guard = self
            .metrics
            .handle_wait_for_effects_ping_latency
            .with_label_values(&[ping.as_str()])
            .start_timer();

        consensus_tx_status_cache.check_position_too_ahead(&consensus_position)?;

        let details = if request.include_details {
            Some(Box::new(ExecutedData::default()))
        } else {
            None
        };

        let status = consensus_tx_status_cache
            .notify_read_transaction_status(consensus_position)
            .await;
        match status {
            NotifyReadConsensusTxStatusResult::Status(status) => match status {
                ConsensusTxStatus::Rejected | ConsensusTxStatus::Dropped => {
                    Ok(WaitForEffectsResponse::Rejected {
                        error: epoch_store.get_rejection_vote_reason(consensus_position),
                    })
                }
                ConsensusTxStatus::Finalized => Ok(WaitForEffectsResponse::Executed {
                    effects_digest: TransactionEffectsDigest::ZERO,
                    details,
                }),
            },
            NotifyReadConsensusTxStatusResult::Expired(round) => {
                Ok(WaitForEffectsResponse::Expired {
                    epoch: epoch_store.epoch(),
                    round: Some(round),
                })
            }
        }
    }

    async fn complete_executed_data(
        &self,
        effects: TransactionEffects,
    ) -> SuiResult<Box<ExecutedData>> {
        let (events, input_objects, output_objects) = self
            .collect_effects_data(
                &effects, /* include_events */ true, /* include_input_objects */ true,
                /* include_output_objects */ true,
            )
            .await?;
        Ok(Box::new(ExecutedData {
            effects,
            events,
            input_objects,
            output_objects,
        }))
    }

    async fn object_info_impl(
        &self,
        request: tonic::Request<ObjectInfoRequest>,
    ) -> WrappedServiceResponse<ObjectInfoResponse> {
        let request = request.into_inner();
        let response = self.state.handle_object_info_request(request).await?;
        Ok((tonic::Response::new(response), Weight::one()))
    }

    async fn transaction_info_impl(
        &self,
        request: tonic::Request<TransactionInfoRequest>,
    ) -> WrappedServiceResponse<TransactionInfoResponse> {
        let request = request.into_inner();
        let response = self.state.handle_transaction_info_request(request).await?;
        Ok((tonic::Response::new(response), Weight::one()))
    }

    async fn checkpoint_impl(
        &self,
        request: tonic::Request<CheckpointRequest>,
    ) -> WrappedServiceResponse<CheckpointResponse> {
        let request = request.into_inner();
        let response = self.state.handle_checkpoint_request(&request)?;
        Ok((tonic::Response::new(response), Weight::one()))
    }

    async fn checkpoint_v2_impl(
        &self,
        request: tonic::Request<CheckpointRequestV2>,
    ) -> WrappedServiceResponse<CheckpointResponseV2> {
        let request = request.into_inner();
        let response = self.state.handle_checkpoint_request_v2(&request)?;
        Ok((tonic::Response::new(response), Weight::one()))
    }

    async fn get_system_state_object_impl(
        &self,
        _request: tonic::Request<SystemStateRequest>,
    ) -> WrappedServiceResponse<SuiSystemState> {
        let response = self
            .state
            .get_object_cache_reader()
            .get_sui_system_state_object_unsafe()?;
        Ok((tonic::Response::new(response), Weight::one()))
    }

    async fn validator_health_impl(
        &self,
        _request: tonic::Request<sui_types::messages_grpc::RawValidatorHealthRequest>,
    ) -> WrappedServiceResponse<sui_types::messages_grpc::RawValidatorHealthResponse> {
        let state = &self.state;

        // Get epoch store once for both metrics
        let epoch_store = state.load_epoch_store_one_call_per_task();

        // Get in-flight execution transactions from execution scheduler
        let num_inflight_execution_transactions =
            state.execution_scheduler().num_pending_certificates() as u64;

        // Get in-flight consensus transactions from consensus adapter
        let num_inflight_consensus_transactions =
            self.consensus_adapter.num_inflight_transactions();

        // Get last committed leader round from epoch store
        let last_committed_leader_round = epoch_store
            .consensus_tx_status_cache
            .get_last_committed_leader_round()
            .unwrap_or(0);

        // Get last locally built checkpoint sequence
        let last_locally_built_checkpoint = epoch_store
            .last_built_checkpoint_summary()
            .ok()
            .flatten()
            .map(|(_, summary)| summary.sequence_number)
            .unwrap_or(0);

        let typed_response = sui_types::messages_grpc::ValidatorHealthResponse {
            num_inflight_consensus_transactions,
            num_inflight_execution_transactions,
            last_locally_built_checkpoint,
            last_committed_leader_round,
        };

        let raw_response = typed_response
            .try_into()
            .map_err(|e: sui_types::error::SuiError| {
                tonic::Status::internal(format!("Failed to serialize health response: {}", e))
            })?;

        Ok((tonic::Response::new(raw_response), Weight::one()))
    }

    fn get_client_ip_addr<T>(
        &self,
        request: &tonic::Request<T>,
        source: &ClientIdSource,
    ) -> Option<IpAddr> {
        let forwarded_header = request.metadata().get_all("x-forwarded-for").iter().next();

        if let Some(header) = forwarded_header {
            let num_hops = header
                .to_str()
                .map(|h| h.split(',').count().saturating_sub(1))
                .unwrap_or(0);

            self.metrics.x_forwarded_for_num_hops.set(num_hops as f64);
        }

        match source {
            ClientIdSource::SocketAddr => {
                let socket_addr: Option<SocketAddr> = request.remote_addr();

                // We will hit this case if the IO type used does not
                // implement Connected or when using a unix domain socket.
                // TODO: once we have confirmed that no legitimate traffic
                // is hitting this case, we should reject such requests that
                // hit this case.
                if let Some(socket_addr) = socket_addr {
                    Some(socket_addr.ip())
                } else {
                    if cfg!(msim) {
                        // Ignore the error from simtests.
                    } else if cfg!(test) {
                        panic!("Failed to get remote address from request");
                    } else {
                        self.metrics.connection_ip_not_found.inc();
                        error!("Failed to get remote address from request");
                    }
                    None
                }
            }
            ClientIdSource::XForwardedFor(num_hops) => {
                let do_header_parse = |op: &MetadataValue<Ascii>| {
                    match op.to_str() {
                        Ok(header_val) => {
                            let header_contents =
                                header_val.split(',').map(str::trim).collect::<Vec<_>>();
                            if *num_hops == 0 {
                                error!(
                                    "x-forwarded-for: 0 specified. x-forwarded-for contents: {:?}. Please assign nonzero value for \
                                    number of hops here, or use `socket-addr` client-id-source type if requests are not being proxied \
                                    to this node. Skipping traffic controller request handling.",
                                    header_contents,
                                );
                                return None;
                            }
                            let contents_len = header_contents.len();
                            if contents_len < *num_hops {
                                error!(
                                    "x-forwarded-for header value of {:?} contains {} values, but {} hops were specified. \
                                    Expected at least {} values. Please correctly set the `x-forwarded-for` value under \
                                    `client-id-source` in the node config.",
                                    header_contents, contents_len, num_hops, contents_len,
                                );
                                self.metrics.client_id_source_config_mismatch.inc();
                                return None;
                            }
                            let Some(client_ip) = header_contents.get(contents_len - num_hops)
                            else {
                                error!(
                                    "x-forwarded-for header value of {:?} contains {} values, but {} hops were specified. \
                                    Expected at least {} values. Skipping traffic controller request handling.",
                                    header_contents, contents_len, num_hops, contents_len,
                                );
                                return None;
                            };
                            parse_ip(client_ip).or_else(|| {
                                self.metrics.forwarded_header_parse_error.inc();
                                None
                            })
                        }
                        Err(e) => {
                            // TODO: once we have confirmed that no legitimate traffic
                            // is hitting this case, we should reject such requests that
                            // hit this case.
                            self.metrics.forwarded_header_invalid.inc();
                            error!("Invalid UTF-8 in x-forwarded-for header: {:?}", e);
                            None
                        }
                    }
                };
                if let Some(op) = request.metadata().get("x-forwarded-for") {
                    do_header_parse(op)
                } else if let Some(op) = request.metadata().get("X-Forwarded-For") {
                    do_header_parse(op)
                } else {
                    self.metrics.forwarded_header_not_included.inc();
                    error!(
                        "x-forwarded-for header not present for request despite node configuring x-forwarded-for tracking type"
                    );
                    None
                }
            }
        }
    }

    async fn handle_traffic_req(&self, client: Option<IpAddr>) -> Result<(), tonic::Status> {
        if let Some(traffic_controller) = &self.traffic_controller {
            if !traffic_controller.check(&client, &None).await {
                // Entity in blocklist
                Err(tonic::Status::from_error(
                    SuiErrorKind::TooManyRequests.into(),
                ))
            } else {
                Ok(())
            }
        } else {
            Ok(())
        }
    }

    fn handle_traffic_resp<T>(
        &self,
        client: Option<IpAddr>,
        wrapped_response: WrappedServiceResponse<T>,
        method_name: &str,
    ) -> Result<tonic::Response<T>, tonic::Status> {
        let (error, spam_weight, unwrapped_response) = match wrapped_response {
            Ok((result, spam_weight)) => (None, spam_weight.clone(), Ok(result)),
            Err(status) => (
                Some(SuiError::from(status.clone())),
                Weight::zero(),
                Err(status.clone()),
            ),
        };

        if let Some(traffic_controller) = self.traffic_controller.clone() {
            traffic_controller.tally(TrafficTally {
                direct: client,
                through_fullnode: None,
                error_info: error.map(|e| {
                    let error_type = String::from(e.clone().as_ref());
                    let error_weight = normalize(e);
                    (error_weight, error_type)
                }),
                spam_weight,
                timestamp: SystemTime::now(),
                method: Some(method_name.to_string()),
            })
        }
        unwrapped_response
    }
}

// TODO: refine error matching here
fn normalize(err: SuiError) -> Weight {
    match err.as_inner() {
        SuiErrorKind::UserInputError {
            error: UserInputError::IncorrectUserSignature { .. },
        } => Weight::one(),
        SuiErrorKind::InvalidSignature { .. }
        | SuiErrorKind::SignerSignatureAbsent { .. }
        | SuiErrorKind::SignerSignatureNumberMismatch { .. }
        | SuiErrorKind::IncorrectSigner { .. }
        | SuiErrorKind::UnknownSigner { .. }
        | SuiErrorKind::WrongEpoch { .. } => Weight::one(),
        _ => Weight::zero(),
    }
}

/// Implements generic pre- and post-processing. Since this is on the critical
/// path, any heavy lifting should be done in a separate non-blocking task
/// unless it is necessary to override the return value.
#[macro_export]
macro_rules! handle_with_decoration {
    ($self:ident, $func_name:ident, $request:ident, $method_name:expr) => {{
        if $self.client_id_source.is_none() {
            return $self.$func_name($request).await.map(|(result, _)| result);
        }

        let client = $self.get_client_ip_addr(&$request, $self.client_id_source.as_ref().unwrap());

        // check if either IP is blocked, in which case return early
        $self.handle_traffic_req(client.clone()).await?;

        // handle traffic tallying
        let wrapped_response = $self.$func_name($request).await;
        $self.handle_traffic_resp(client, wrapped_response, $method_name)
    }};
}

#[async_trait]
impl Validator for ValidatorService {
    async fn submit_transaction(
        &self,
        request: tonic::Request<RawSubmitTxRequest>,
    ) -> Result<tonic::Response<RawSubmitTxResponse>, tonic::Status> {
        let validator_service = self.clone();

        // Spawns a task which handles the transaction. The task will unconditionally continue
        // processing in the event that the client connection is dropped.
        spawn_monitored_task!(async move {
            // NB: traffic tally wrapping handled within the task rather than on task exit
            // to prevent an attacker from subverting traffic control by severing the connection
            handle_with_decoration!(
                validator_service,
                handle_submit_transaction_impl,
                request,
                "submit_transaction"
            )
        })
        .await
        .unwrap()
    }

    async fn wait_for_effects(
        &self,
        request: tonic::Request<RawWaitForEffectsRequest>,
    ) -> Result<tonic::Response<RawWaitForEffectsResponse>, tonic::Status> {
        handle_with_decoration!(self, wait_for_effects_impl, request, "wait_for_effects")
    }

    async fn object_info(
        &self,
        request: tonic::Request<ObjectInfoRequest>,
    ) -> Result<tonic::Response<ObjectInfoResponse>, tonic::Status> {
        handle_with_decoration!(self, object_info_impl, request, "object_info")
    }

    async fn transaction_info(
        &self,
        request: tonic::Request<TransactionInfoRequest>,
    ) -> Result<tonic::Response<TransactionInfoResponse>, tonic::Status> {
        handle_with_decoration!(self, transaction_info_impl, request, "transaction_info")
    }

    async fn checkpoint(
        &self,
        request: tonic::Request<CheckpointRequest>,
    ) -> Result<tonic::Response<CheckpointResponse>, tonic::Status> {
        handle_with_decoration!(self, checkpoint_impl, request, "checkpoint")
    }

    async fn checkpoint_v2(
        &self,
        request: tonic::Request<CheckpointRequestV2>,
    ) -> Result<tonic::Response<CheckpointResponseV2>, tonic::Status> {
        handle_with_decoration!(self, checkpoint_v2_impl, request, "checkpoint_v2")
    }

    async fn get_system_state_object(
        &self,
        request: tonic::Request<SystemStateRequest>,
    ) -> Result<tonic::Response<SuiSystemState>, tonic::Status> {
        handle_with_decoration!(
            self,
            get_system_state_object_impl,
            request,
            "get_system_state_object"
        )
    }

    async fn validator_health(
        &self,
        request: tonic::Request<sui_types::messages_grpc::RawValidatorHealthRequest>,
    ) -> Result<tonic::Response<sui_types::messages_grpc::RawValidatorHealthResponse>, tonic::Status>
    {
        handle_with_decoration!(self, validator_health_impl, request, "validator_health")
    }
}

#[cfg(test)]
mod inflight_guard_tests {
    use super::*;
    use prometheus::Registry;

    fn make_guard(
        inflight: Arc<Mutex<HashSet<TransactionDigest>>>,
        cache: Cache<TransactionDigest, Instant>,
        metrics: Arc<ValidatorServiceMetrics>,
    ) -> InflightTransactionsGuard {
        InflightTransactionsGuard {
            inflight,
            recently_submitted: cache,
            window: Duration::from_secs(10),
            metrics,
            acquired: HashSet::new(),
        }
    }

    #[test]
    fn concurrent_acquire_rejects_other_request_and_is_idempotent_for_owner() {
        let inflight = Arc::new(Mutex::new(HashSet::new()));
        let cache = ValidatorService::new_recently_submitted_cache(Duration::from_secs(10));
        let metrics = Arc::new(ValidatorServiceMetrics::new(&Registry::new()));
        let digest = TransactionDigest::random();

        let mut g1 = make_guard(inflight.clone(), cache.clone(), metrics.clone());
        let mut g2 = make_guard(inflight.clone(), cache.clone(), metrics.clone());

        // First handler acquires it.
        assert!(matches!(g1.try_acquire(digest), AcquireOutcome::Acquired));
        assert_eq!(metrics.inflight_transactions.get(), 1);

        // A concurrent handler sees another in-flight owner and is rejected.
        assert!(matches!(
            g2.try_acquire(digest),
            AcquireOutcome::AlreadyAcquiredByAnotherRequest
        ));

        // The owning handler re-acquiring (epoch-end retry) is NOT a duplicate.
        assert!(matches!(
            g1.try_acquire(digest),
            AcquireOutcome::AlreadyAcquiredByThisRequest
        ));
    }

    #[test]
    fn drop_demotes_into_recently_processed_outcome() {
        let inflight = Arc::new(Mutex::new(HashSet::new()));
        let cache = ValidatorService::new_recently_submitted_cache(Duration::from_secs(10));
        let metrics = Arc::new(ValidatorServiceMetrics::new(&Registry::new()));
        let digest = TransactionDigest::random();

        {
            let mut g = make_guard(inflight.clone(), cache.clone(), metrics.clone());
            assert!(matches!(g.try_acquire(digest), AcquireOutcome::Acquired));
            assert_eq!(inflight.lock().len(), 1);
        } // guard dropped here -> remove from set + demote to TTL cache

        assert_eq!(
            inflight.lock().len(),
            0,
            "acquired digest must be removed from the in-flight set on drop"
        );
        assert_eq!(metrics.inflight_transactions.get(), 0);

        // Make moka's read view deterministic before asserting the demoted entry.
        cache.run_pending_tasks();

        // A fresh request now sees the digest as recently processed (tail window).
        let mut g_after = make_guard(inflight.clone(), cache.clone(), metrics.clone());
        assert!(matches!(
            g_after.try_acquire(digest),
            AcquireOutcome::RecentlyProcessed { .. }
        ));
    }
}