sui_core/transaction_driver/

mod.rs

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

mod effects_certifier;
mod error;
mod metrics;
mod reconfig_observer;
mod request_retrier;
mod transaction_submitter;

/// Exports
pub use error::TransactionDriverError;
pub use metrics::*;
pub use reconfig_observer::{OnsiteReconfigObserver, ReconfigObserver};

use std::{
    net::SocketAddr,
    sync::Arc,
    time::{Duration, Instant},
};

use arc_swap::ArcSwap;
use effects_certifier::*;
use mysten_common::backoff::ExponentialBackoff;
use mysten_metrics::{monitored_future, spawn_logged_monitored_task};
use parking_lot::Mutex;
use rand::Rng;
use sui_config::NodeConfig;
use sui_types::{
    base_types::AuthorityName,
    committee::EpochId,
    error::{ErrorCategory, UserInputError},
    messages_grpc::{SubmitTxRequest, SubmitTxResult, TxType},
    transaction::TransactionDataAPI as _,
};
use tokio::{
    task::JoinSet,
    time::{interval, sleep},
};
use tracing::instrument;
use transaction_submitter::*;

use crate::{
    authority_aggregator::AuthorityAggregator,
    authority_client::AuthorityAPI,
    validator_client_monitor::{
        OperationFeedback, OperationType, ValidatorClientMetrics, ValidatorClientMonitor,
    },
};

/// Trait for components that can update their AuthorityAggregator during reconfiguration.
/// Used by ReconfigObserver to notify components of epoch changes.
pub trait AuthorityAggregatorUpdatable<A: Clone>: Send + Sync + 'static {
    fn epoch(&self) -> EpochId;
    fn authority_aggregator(&self) -> Arc<AuthorityAggregator<A>>;
    fn update_authority_aggregator(&self, new_authorities: Arc<AuthorityAggregator<A>>);
}

/// Options for submitting a transaction.
#[derive(Clone, Default, Debug)]
pub struct SubmitTransactionOptions {
    /// When forwarding transactions on behalf of a client, this is the client's address
    /// specified for ddos protection.
    pub forwarded_client_addr: Option<SocketAddr>,

    /// When submitting a transaction, only the validators in the allowed validator list can be used to submit the transaction to.
    /// When the allowed validator list is empty, any validator can be used.
    pub allowed_validators: Vec<String>,

    /// When submitting a transaction, the validators in the blocked validator list cannot be used to submit the transaction to.
    /// When the blocked validator list is empty, no restrictions are applied.
    pub blocked_validators: Vec<String>,
}

#[derive(Clone, Debug)]
pub struct QuorumTransactionResponse {
    pub effects: sui_types::transaction_driver_types::FinalizedEffects,

    pub events: Option<sui_types::effects::TransactionEvents>,
    // Input objects will only be populated in the happy path
    pub input_objects: Option<Vec<sui_types::object::Object>>,
    // Output objects will only be populated in the happy path
    pub output_objects: Option<Vec<sui_types::object::Object>>,
    pub auxiliary_data: Option<Vec<u8>>,
}

pub struct TransactionDriver<A: Clone> {
    authority_aggregator: Arc<ArcSwap<AuthorityAggregator<A>>>,
    state: Mutex<State>,
    metrics: Arc<TransactionDriverMetrics>,
    submitter: TransactionSubmitter,
    certifier: EffectsCertifier,
    client_monitor: Arc<ValidatorClientMonitor<A>>,
}

impl<A> TransactionDriver<A>
where
    A: AuthorityAPI + Send + Sync + 'static + Clone,
{
    // TODO: accept a TransactionDriverConfig to set default allowed & blocked validators.
    pub fn new(
        authority_aggregator: Arc<AuthorityAggregator<A>>,
        reconfig_observer: Arc<dyn ReconfigObserver<A> + Sync + Send>,
        metrics: Arc<TransactionDriverMetrics>,
        node_config: Option<&NodeConfig>,
        client_metrics: Arc<ValidatorClientMetrics>,
    ) -> Arc<Self> {
        if std::env::var("TRANSACTION_DRIVER").is_ok() {
            tracing::warn!(
                "Transaction Driver is the only supported driver for transaction submission. Setting TRANSACTION_DRIVER is a no-op."
            );
        }

        let shared_swap = Arc::new(ArcSwap::new(authority_aggregator));

        // Extract validator client monitor config from NodeConfig or use default
        let monitor_config = node_config
            .and_then(|nc| nc.validator_client_monitor_config.clone())
            .unwrap_or_default();
        let client_monitor =
            ValidatorClientMonitor::new(monitor_config, client_metrics, shared_swap.clone());

        let driver = Arc::new(Self {
            authority_aggregator: shared_swap,
            state: Mutex::new(State::new()),
            metrics: metrics.clone(),
            submitter: TransactionSubmitter::new(metrics.clone()),
            certifier: EffectsCertifier::new(metrics),
            client_monitor,
        });

        let driver_clone = driver.clone();

        spawn_logged_monitored_task!(Self::run_latency_checks(driver_clone));

        driver.enable_reconfig(reconfig_observer);
        driver
    }

    /// Returns the authority aggregator wrapper which upgrades on epoch changes.
    pub fn authority_aggregator(&self) -> &Arc<ArcSwap<AuthorityAggregator<A>>> {
        &self.authority_aggregator
    }

    pub fn select_preferred_validators(&self, delta: f64) -> Vec<AuthorityName> {
        let authority_aggregator = self.authority_aggregator.load();
        self.client_monitor
            .select_shuffled_preferred_validators(&authority_aggregator.committee, delta)
    }

    /// Drives transaction to finalization.
    ///
    /// Internally, retries the attempt to finalize a transaction until:
    /// - The transaction is finalized.
    /// - The transaction observes a non-retriable error.
    /// - Timeout is reached.
    #[instrument(level = "error", skip_all, fields(tx_digest = ?request.transaction.as_ref().map(|t| t.digest()), ping = %request.ping_type.is_some()))]
    pub async fn drive_transaction(
        &self,
        request: SubmitTxRequest,
        options: SubmitTransactionOptions,
        timeout_duration: Option<Duration>,
    ) -> Result<QuorumTransactionResponse, TransactionDriverError> {
        const MAX_DRIVE_TRANSACTION_RETRY_DELAY: Duration = Duration::from_secs(10);

        let tx_data = request.transaction.as_ref().map(|t| t.transaction_data());
        // gas_price=0 for gasless; use 1 for baseline (RGP-equivalent) priority
        let amplification_factor =
            if request.ping_type.is_some() || tx_data.is_some_and(|d| d.is_gasless_transaction()) {
                1
            } else {
                let tx_data = tx_data.unwrap();
                let gas_price = tx_data.gas_price();
                let reference_gas_price = self.authority_aggregator.load().reference_gas_price;
                let amplification_factor = gas_price / reference_gas_price.max(1);
                if amplification_factor == 0 {
                    return Err(TransactionDriverError::ValidationFailed {
                        error: UserInputError::GasPriceUnderRGP {
                            gas_price,
                            reference_gas_price,
                        }
                        .to_string(),
                    });
                }
                amplification_factor
            };

        let tx_type = request.tx_type();
        let ping_label = if request.ping_type.is_some() {
            "true"
        } else {
            "false"
        };
        let timer = Instant::now();

        self.metrics
            .total_transactions_submitted
            .with_label_values(&[tx_type.as_str(), ping_label])
            .inc();

        let mut backoff = ExponentialBackoff::new(
            Duration::from_millis(100),
            MAX_DRIVE_TRANSACTION_RETRY_DELAY,
        );
        let mut attempts = 0;
        let mut latest_retriable_error = None;

        let retry_loop = async {
            loop {
                // TODO(fastpath): Check local state before submitting transaction
                match self
                    .drive_transaction_once(amplification_factor, request.clone(), &options)
                    .await
                {
                    Ok(resp) => {
                        let settlement_finality_latency = timer.elapsed().as_secs_f64();
                        self.metrics
                            .settlement_finality_latency
                            .with_label_values(&[tx_type.as_str(), ping_label])
                            .observe(settlement_finality_latency);
                        let is_out_of_expected_range = settlement_finality_latency >= 8.0
                            || settlement_finality_latency <= 0.1;
                        tracing::debug!(
                            ?tx_type,
                            ?is_out_of_expected_range,
                            "Settlement finality latency: {:.3} seconds",
                            settlement_finality_latency
                        );
                        // Record the number of retries for successful transaction
                        self.metrics
                            .transaction_retries
                            .with_label_values(&["success", tx_type.as_str(), ping_label])
                            .observe(attempts as f64);
                        return Ok(resp);
                    }
                    Err(e) => {
                        self.metrics
                            .drive_transaction_errors
                            .with_label_values(&[
                                e.categorize().into(),
                                tx_type.as_str(),
                                ping_label,
                            ])
                            .inc();
                        if !e.is_submission_retriable() {
                            // Record the number of retries for failed transaction
                            self.metrics
                                .transaction_retries
                                .with_label_values(&["failure", tx_type.as_str(), ping_label])
                                .observe(attempts as f64);
                            if request.transaction.is_some() {
                                tracing::info!(
                                    "User transaction failed to finalize (attempt {}), with non-retriable error: {} ({})",
                                    attempts,
                                    e,
                                    Into::<&str>::into(e.categorize())
                                );
                            }
                            return Err(e);
                        }
                        if request.transaction.is_some() {
                            tracing::info!(
                                "User transaction failed to finalize (attempt {}): {} ({}). Retrying ...",
                                attempts,
                                e,
                                Into::<&str>::into(e.categorize())
                            );
                        }
                        // Buffer the latest retriable error to be returned in case of timeout
                        latest_retriable_error = Some(e);
                    }
                }

                let overload = if let Some(e) = &latest_retriable_error {
                    e.categorize() == ErrorCategory::ValidatorOverloaded
                } else {
                    false
                };
                let delay = if overload {
                    // Increase delay during overload.
                    const OVERLOAD_ADDITIONAL_DELAY: Duration = Duration::from_secs(10);
                    backoff.next().unwrap() + OVERLOAD_ADDITIONAL_DELAY
                } else {
                    backoff.next().unwrap()
                };

                tracing::debug!("Retrying after {:.3}s", delay.as_secs_f32());
                sleep(delay).await;

                attempts += 1;
            }
        };

        match timeout_duration {
            Some(duration) => {
                tokio::time::timeout(duration, retry_loop)
                    .await
                    .unwrap_or_else(|_| {
                        // Timeout occurred, return with latest retriable error if available
                        let e = TransactionDriverError::TimeoutWithLastRetriableError {
                            last_error: latest_retriable_error.map(Box::new),
                            attempts,
                            timeout: duration,
                        };
                        if request.transaction.is_some() {
                            tracing::info!(
                                "User transaction timed out after {} attempts. Last error: {}",
                                attempts,
                                e
                            );
                        }
                        Err(e)
                    })
            }
            None => retry_loop.await,
        }
    }

    #[instrument(level = "error", skip_all, err(level = "debug"))]
    async fn drive_transaction_once(
        &self,
        amplification_factor: u64,
        request: SubmitTxRequest,
        options: &SubmitTransactionOptions,
    ) -> Result<QuorumTransactionResponse, TransactionDriverError> {
        let auth_agg = self.authority_aggregator.load();
        let start_time = Instant::now();
        let tx_type = request.tx_type();
        let tx_digest = request.tx_digest();
        let ping_type = request.ping_type;

        let (name, submit_txn_result) = self
            .submitter
            .submit_transaction(
                &auth_agg,
                &self.client_monitor,
                tx_type,
                amplification_factor,
                request,
                options,
            )
            .await?;
        if let SubmitTxResult::Rejected { error } = &submit_txn_result {
            return Err(TransactionDriverError::ClientInternal {
                error: format!(
                    "SubmitTxResult::Rejected should have been returned as an error in submit_transaction(): {}",
                    error
                ),
            });
        }

        // Wait for quorum effects using EffectsCertifier
        let result = self
            .certifier
            .get_certified_finalized_effects(
                &auth_agg,
                &self.client_monitor,
                tx_digest,
                tx_type,
                name,
                submit_txn_result,
                options,
            )
            .await;

        if result.is_ok() {
            self.client_monitor
                .record_interaction_result(OperationFeedback {
                    authority_name: name,
                    display_name: auth_agg.get_display_name(&name),
                    operation: if tx_type == TxType::SingleWriter {
                        OperationType::SingleWriterFinality
                    } else {
                        OperationType::SharedObjectFinality
                    },
                    ping_type,
                    result: Ok(start_time.elapsed()),
                });
        }
        result
    }

    // Runs a background task to send ping transactions to all validators to perform latency checks for the consensus path.
    async fn run_latency_checks(self: Arc<Self>) {
        const INTERVAL_BETWEEN_RUNS: Duration = Duration::from_secs(15);
        const MAX_JITTER: Duration = Duration::from_secs(10);
        const PING_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);

        let mut interval = interval(INTERVAL_BETWEEN_RUNS);
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

        loop {
            interval.tick().await;

            // Only run latency checks for shared object transactions since single writer
            // transactions no longer use a separate fast path and go through consensus.
            let auth_agg = self.authority_aggregator.load().clone();
            let validators = auth_agg.committee.names().cloned().collect::<Vec<_>>();

            self.metrics.latency_check_runs.inc();

            let mut tasks = JoinSet::new();

            for name in validators {
                let display_name = auth_agg.get_display_name(&name);
                let delay_ms = rand::thread_rng().gen_range(0..MAX_JITTER.as_millis()) as u64;
                let self_clone = self.clone();

                let task = async move {
                    // Add some random delay to the task to avoid all tasks running at the same time
                    if delay_ms > 0 {
                        tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
                    }
                    let start_time = Instant::now();

                    // Send a consensus ping transaction to the validator
                    match self_clone
                        .drive_transaction(
                            SubmitTxRequest::new_ping(),
                            SubmitTransactionOptions {
                                allowed_validators: vec![display_name.clone()],
                                ..Default::default()
                            },
                            Some(PING_REQUEST_TIMEOUT),
                        )
                        .await
                    {
                        Ok(_) => {
                            tracing::debug!(
                                "Ping transaction to validator {} completed end to end in {} seconds",
                                display_name,
                                start_time.elapsed().as_secs_f64()
                            );
                        }
                        Err(err) => {
                            tracing::debug!(
                                "Failed to get certified finalized effects for ping transaction to validator {}: {}",
                                display_name,
                                err
                            );
                        }
                    }
                };

                tasks.spawn(task);
            }

            while let Some(result) = tasks.join_next().await {
                if let Err(e) = result {
                    tracing::debug!("Error while driving ping transaction: {}", e);
                }
            }
        }
    }

    fn enable_reconfig(
        self: &Arc<Self>,
        reconfig_observer: Arc<dyn ReconfigObserver<A> + Sync + Send>,
    ) {
        let driver = self.clone();
        self.state.lock().tasks.spawn(monitored_future!(async move {
            let mut reconfig_observer = reconfig_observer.clone_boxed();
            reconfig_observer.run(driver).await;
        }));
    }
}

impl<A> AuthorityAggregatorUpdatable<A> for TransactionDriver<A>
where
    A: AuthorityAPI + Send + Sync + 'static + Clone,
{
    fn epoch(&self) -> EpochId {
        self.authority_aggregator.load().committee.epoch
    }

    fn authority_aggregator(&self) -> Arc<AuthorityAggregator<A>> {
        self.authority_aggregator.load_full()
    }

    fn update_authority_aggregator(&self, new_authorities: Arc<AuthorityAggregator<A>>) {
        tracing::info!(
            "Transaction Driver updating AuthorityAggregator with committee {}",
            new_authorities.committee
        );

        self.authority_aggregator.store(new_authorities);
    }
}

// Inner state of TransactionDriver.
struct State {
    tasks: JoinSet<()>,
}

impl State {
    fn new() -> Self {
        Self {
            tasks: JoinSet::new(),
        }
    }
}