sui_core/validator_client_monitor/

monitor.rs

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

use crate::authority_aggregator::AuthorityAggregator;
use crate::authority_client::AuthorityAPI;
use crate::validator_client_monitor::stats::ClientObservedStats;
use crate::validator_client_monitor::{
    OperationFeedback, OperationType, TransactionClass, metrics::ValidatorClientMetrics,
};
use arc_swap::ArcSwap;
use parking_lot::RwLock;
use rand::seq::SliceRandom;
use std::collections::HashMap;
use std::{
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};
use sui_config::validator_client_monitor_config::ValidatorClientMonitorConfig;
use sui_types::committee::{Committee, CommitteeTrait as _};
use sui_types::{
    base_types::{AuthorityName, ConciseableName as _},
    messages_grpc::ValidatorHealthRequest,
};
use tokio::{
    sync::Notify,
    task::JoinSet,
    time::{interval, timeout},
};
use tracing::{debug, info, warn};

/// Minimum spacing between health-check rounds run on request, so a burst of
/// staggered-submission rejections costs one extra round rather than one per
/// rejection.
const REQUESTED_HEALTH_CHECK_SPACING: Duration = Duration::from_secs(2);
/// Requested rounds allowed between two scheduled rounds. Requests are honored only
/// while the staggering view is off, so this caps what a validator that rejects
/// while the network is not staggering can cost: two extra rounds per interval.
/// Two rather than one so that a round run on the first rejection, before every
/// validator has processed the activating commit, gets one follow-up for the
/// laggards before the driver falls back to waiting for the scheduled round.
const REQUESTED_HEALTH_CHECKS_PER_INTERVAL: u32 = 2;

/// Monitors validator interactions from the client's perspective.
///
/// This component:
/// - Collects client-side metrics from TransactionDriver operations
/// - Runs periodic health checks on all validators from the client
/// - Maintains client-observed statistics for reliability and latency
/// - Provides intelligent validator selection based on client-observed performance
/// - Handles epoch changes by cleaning up stale validator data
///
/// The monitor runs a background task for health checks and uses
/// moving averages to smooth client-side measurements.
pub struct ValidatorClientMonitor<A: Clone> {
    config: ValidatorClientMonitorConfig,
    metrics: Arc<ValidatorClientMetrics>,
    client_stats: RwLock<ClientObservedStats>,
    authority_aggregator: Arc<ArcSwap<AuthorityAggregator<A>>>,
    cached_latencies: RwLock<HashMap<(AuthorityName, TransactionClass), Duration>>,
    /// Latest staggering state each validator self-reported on its health-check
    /// response, with the local receipt time that drives freshness.
    staggering_reports: RwLock<HashMap<AuthorityName, (bool, Instant)>>,
    /// The stake-weighted view over `staggering_reports`, recomputed when a report
    /// arrives and after every health-check round (so aged-out reports drop off).
    staggering_active: AtomicBool,
    /// Wakes the health-check loop for an immediate round; see
    /// [`Self::request_health_check`].
    health_check_now: Notify,
    /// Whether the round in progress was requested rather than scheduled; only read
    /// to attribute a view flip in the log.
    requested_round_in_progress: AtomicBool,
}

impl<A> ValidatorClientMonitor<A>
where
    A: AuthorityAPI + Send + Sync + 'static + Clone,
{
    pub fn new(
        config: ValidatorClientMonitorConfig,
        metrics: Arc<ValidatorClientMetrics>,
        authority_aggregator: Arc<ArcSwap<AuthorityAggregator<A>>>,
    ) -> Arc<Self> {
        info!(
            "Validator client monitor starting with config: {:?}",
            config
        );

        let monitor = Arc::new(Self {
            config: config.clone(),
            metrics,
            client_stats: RwLock::new(ClientObservedStats::new(config)),
            authority_aggregator,
            cached_latencies: RwLock::new(HashMap::new()),
            staggering_reports: RwLock::new(HashMap::new()),
            staggering_active: AtomicBool::new(false),
            health_check_now: Notify::new(),
            requested_round_in_progress: AtomicBool::new(false),
        });

        let monitor_clone = monitor.clone();
        tokio::spawn(async move {
            monitor_clone.run_health_checks().await;
        });

        monitor
    }

    #[cfg(test)]
    pub fn new_for_test(authority_aggregator: Arc<AuthorityAggregator<A>>) -> Arc<Self> {
        use prometheus::Registry;

        Self::new(
            ValidatorClientMonitorConfig::default(),
            Arc::new(ValidatorClientMetrics::new(&Registry::default())),
            Arc::new(ArcSwap::new(authority_aggregator)),
        )
    }

    /// Background task that runs periodic health checks on all validators.
    ///
    /// Sends health check requests to all validators in parallel and records
    /// the results (success/failure and latency). Timeouts are treated as
    /// failures without recording latency to avoid polluting latency statistics.
    async fn run_health_checks(self: Arc<Self>) {
        let mut interval = interval(self.config.health_check_interval);
        let mut last_round_started = tokio::time::Instant::now() - REQUESTED_HEALTH_CHECK_SPACING;
        let mut requested_rounds_left = REQUESTED_HEALTH_CHECKS_PER_INTERVAL;

        loop {
            let requested = tokio::select! {
                _ = interval.tick() => {
                    requested_rounds_left = REQUESTED_HEALTH_CHECKS_PER_INTERVAL;
                    false
                }
                _ = self.health_check_now.notified() => {
                    // A request only has something to learn while the view is off; once it
                    // is on, rejections are expected and the scheduled cadence is enough.
                    if self.staggering_active() || requested_rounds_left == 0 {
                        continue;
                    }
                    requested_rounds_left -= 1;
                    tokio::time::sleep_until(last_round_started + REQUESTED_HEALTH_CHECK_SPACING)
                        .await;
                    // The requested round takes the place of the next scheduled one.
                    interval.reset();
                    true
                }
            };
            last_round_started = tokio::time::Instant::now();
            self.requested_round_in_progress
                .store(requested, Ordering::Relaxed);

            let authority_agg = self.authority_aggregator.load();

            let current_validators: Vec<_> = authority_agg.committee.names().cloned().collect();
            self.client_stats
                .write()
                .retain_validators(&current_validators);
            self.staggering_reports
                .write()
                .retain(|name, _| current_validators.contains(name));

            let mut tasks = JoinSet::new();

            for (name, safe_client) in authority_agg.authority_clients.iter() {
                let name = *name;
                let display_name = authority_agg.get_display_name(&name);
                let client = safe_client.clone();
                let timeout_duration = self.config.health_check_timeout;
                let monitor = self.clone();

                tasks.spawn(async move {
                    let start = Instant::now();
                    match timeout(
                        timeout_duration,
                        client.validator_health(ValidatorHealthRequest {}),
                    )
                    .await
                    {
                        Ok(Ok(response)) => {
                            monitor.record_staggering_report(name, response.staggering_active);
                            let latency = start.elapsed();
                            monitor.record_interaction_result(OperationFeedback {
                                authority_name: name,
                                display_name: display_name.clone(),
                                operation: OperationType::HealthCheck,
                                tx_class: None,
                                ping_type: None,
                                result: Ok(latency),
                            });
                        }
                        Ok(Err(_)) => {
                            let _latency = start.elapsed();
                            monitor.record_interaction_result(OperationFeedback {
                                authority_name: name,
                                display_name: display_name.clone(),
                                operation: OperationType::HealthCheck,
                                tx_class: None,
                                ping_type: None,
                                result: Err(()),
                            });
                        }
                        Err(_) => {
                            monitor.record_interaction_result(OperationFeedback {
                                authority_name: name,
                                display_name,
                                operation: OperationType::HealthCheck,
                                tx_class: None,
                                ping_type: None,
                                result: Err(()),
                            });
                        }
                    }
                });
            }

            while let Some(result) = tasks.join_next().await {
                if let Err(e) = result {
                    warn!("Health check task failed: {}", e);
                }
            }

            self.update_cached_latencies(&authority_agg);
            // Ages out reports from validators that stopped answering.
            self.refresh_staggering_view("end of round");
            self.requested_round_in_progress
                .store(false, Ordering::Relaxed);
        }
    }
}

impl<A: Clone> ValidatorClientMonitor<A> {
    /// Calculate and cache latencies for all validators.
    ///
    /// This method is called periodically after health checks complete to update
    /// the cached validator latencies. Those are the end to end latencies as calculated for each validator
    /// taking into account the reliability of the validator.
    fn update_cached_latencies(&self, authority_agg: &AuthorityAggregator<A>) {
        let committee = &authority_agg.committee;
        let mut cached_latencies = self.cached_latencies.write();

        for tx_class in [TransactionClass::Restricted, TransactionClass::Unrestricted] {
            let latencies_map = self
                .client_stats
                .read()
                .get_all_validator_stats(committee, tx_class);

            for (validator, latency) in latencies_map.iter() {
                debug!(
                    "Validator {}, class {:?}, latency {}",
                    validator,
                    tx_class,
                    latency.as_secs_f64()
                );
                let display_name = authority_agg.get_display_name(validator);
                self.metrics
                    .performance
                    .with_label_values(&[display_name.as_str(), tx_class.as_str()])
                    .set(latency.as_secs_f64());
                cached_latencies.insert((*validator, tx_class), *latency);
            }
        }
    }

    /// Record client-observed interaction result with a validator.
    ///
    /// Records operation results including success/failure status and latency
    /// from the client's perspective. Updates both Prometheus metrics and
    /// internal client statistics. This is the primary interface for the
    /// TransactionDriver to report client-observed validator interactions.
    /// TODO: Consider adding a byzantine flag to the feedback.
    pub fn record_interaction_result(&self, feedback: OperationFeedback) {
        let operation_str = match feedback.operation {
            OperationType::Submit => "submit",
            OperationType::Effects => "effects",
            OperationType::HealthCheck => "health_check",
            OperationType::SingleWriterFinality => "single_writer_finality",
            OperationType::SharedObjectFinality => "shared_object_finality",
        };
        let ping_label = if feedback.ping_type.is_some() {
            "true"
        } else {
            "false"
        };

        match feedback.result {
            Ok(latency) => {
                self.metrics
                    .observed_latency
                    .with_label_values(&[feedback.display_name.as_str(), operation_str, ping_label])
                    .observe(latency.as_secs_f64());
                self.metrics
                    .operation_success
                    .with_label_values(&[feedback.display_name.as_str(), operation_str, ping_label])
                    .inc();
            }
            Err(()) => {
                self.metrics
                    .operation_failure
                    .with_label_values(&[feedback.display_name.as_str(), operation_str, ping_label])
                    .inc();
            }
        }

        let mut client_stats = self.client_stats.write();
        client_stats.record_interaction_result(feedback);
    }

    /// Whether any latency has been observed yet. Until then
    /// `select_shuffled_preferred_validators` returns an arbitrary shuffle rather than a real
    /// preference.
    pub fn has_observed_latencies(&self) -> bool {
        !self.cached_latencies.read().is_empty()
    }

    /// Reports older than this no longer count toward [`Self::staggering_active`].
    /// Two health-check intervals, so a single missed health check does not flap the
    /// view.
    fn report_freshness(&self) -> Duration {
        self.config.health_check_interval * 2
    }

    /// Record a validator's self-reported staggering state from its health-check
    /// response. `None` (a validator predating the field) carries no information and
    /// is ignored. Health checks to one validator are sequential, so reports from it
    /// arrive in order and the latest simply wins.
    pub fn record_staggering_report(&self, validator: AuthorityName, active: Option<bool>) {
        let Some(active) = active else {
            return;
        };
        self.staggering_reports
            .write()
            .insert(validator, (active, Instant::now()));
        self.refresh_staggering_view(&format!("report from {}", validator.concise()));
    }

    /// Run a health-check round now rather than at the next interval tick, spaced by
    /// `REQUESTED_HEALTH_CHECK_SPACING` from the previous round and capped at
    /// `REQUESTED_HEALTH_CHECKS_PER_INTERVAL` while the staggering view is off (a
    /// request while it is on is ignored). Validators report their staggering state
    /// on health responses, so this is how the driver learns of an activation within
    /// a round trip of its first staggered-submission rejection instead of within a
    /// health interval. Requests coalesce while a round is pending.
    pub fn request_health_check(&self) {
        self.health_check_now.notify_one();
    }

    /// Whether staggered submission is currently considered active on the validator
    /// network, as last computed by `refresh_staggering_view`: validators
    /// reporting active within the freshness window must total at least f+1 by
    /// stake, so at least one of them is honest — and honest validators converge on
    /// the same mode, so one honest report reflects the network state. A fresh
    /// report counts immediately (flipping on is cheap to believe: the cost of a
    /// wrong belief in either direction is only bounded latency) and decays by aging
    /// out of the window.
    pub fn staggering_active(&self) -> bool {
        self.staggering_active.load(Ordering::Relaxed)
    }

    /// Recomputes the view and logs a flip with what caused it, so a run's driver-side
    /// detection can be timed against the validators' own transition logs.
    fn refresh_staggering_view(&self, trigger: &str) {
        let authority_agg = self.authority_aggregator.load();
        let committee = &authority_agg.committee;
        let freshness = self.report_freshness();
        let now = Instant::now();
        let active_stake: u64 = self
            .staggering_reports
            .read()
            .iter()
            .filter(|(_, (active, received_at))| {
                *active && now.saturating_duration_since(*received_at) < freshness
            })
            .map(|(name, _)| committee.weight(name))
            .sum();
        let active = active_stake >= committee.validity_threshold();
        if self.staggering_active.swap(active, Ordering::Relaxed) != active {
            info!(
                "Staggering view flipped {}: {} of {} stake reports staggering active \
                 (validity threshold {}), on {} during a {} health-check round",
                if active { "on" } else { "off" },
                active_stake,
                committee.total_votes(),
                committee.validity_threshold(),
                trigger,
                if self.requested_round_in_progress.load(Ordering::Relaxed) {
                    "requested"
                } else {
                    "scheduled"
                },
            );
        }
        self.metrics
            .staggering_active_stake
            .set(active_stake as i64);
        self.metrics.staggering_active.set(active as i64);
    }

    #[cfg(test)]
    pub fn record_staggering_report_at(&self, validator: AuthorityName, active: bool, at: Instant) {
        self.staggering_reports
            .write()
            .insert(validator, (active, at));
        self.refresh_staggering_view("test report");
    }

    /// Select validators based on client-observed performance for the given transaction type.
    ///
    /// The current committee is passed in to ensure this function has the latest committee information.
    ///
    /// Also the tx type is passed in so that we can select validators based on their respective latencies
    /// for the transaction type.
    ///
    /// Validators with latencies within `delta` of the lowest latency in the given transaction type
    /// are shuffled, to balance the load among the fastest validators.
    ///
    /// Returns a vector containing all validators, where
    /// 1. Fast validators within `delta` of the lowest latency are shuffled.
    /// 2. Remaining slow validators are sorted by latency in ascending order.
    pub fn select_shuffled_preferred_validators(
        &self,
        committee: &Committee,
        delta: f64,
        tx_class: TransactionClass,
    ) -> Vec<AuthorityName> {
        let mut rng = rand::thread_rng();

        let cached_latencies = self.cached_latencies.read();
        if cached_latencies.is_empty() {
            let mut validators: Vec<_> = committee.names().cloned().collect();
            validators.shuffle(&mut rng);
            return validators;
        };

        // Since the cached latencies are updated periodically, it is possible that it was ran on
        // an out-of-date committee.
        let mut validator_with_latencies: Vec<_> = committee
            .names()
            .map(|v| {
                (
                    *v,
                    cached_latencies
                        .get(&(*v, tx_class))
                        .cloned()
                        .unwrap_or(Duration::ZERO),
                )
            })
            .collect();
        if validator_with_latencies.is_empty() {
            return vec![];
        }
        // Shuffle the validators to balance the load among validators with the same latency.
        validator_with_latencies.shuffle(&mut rng);
        // Sort by latency in ascending order. We want to select the validators with the lowest latencies.
        validator_with_latencies.sort_by_key(|(_, latency)| *latency);

        // Shuffle the validators within delta of the lowest latency, for load balancing.
        let lowest_latency = validator_with_latencies[0].1;
        let threshold = lowest_latency.mul_f64(1.0 + delta);
        let k = validator_with_latencies
            .iter()
            .enumerate()
            .find(|(_, (_, latency))| *latency > threshold)
            .map(|(i, _)| i)
            .unwrap_or(validator_with_latencies.len());
        validator_with_latencies[..k].shuffle(&mut rng);
        self.metrics.shuffled_validators.observe(k as f64);

        validator_with_latencies
            .into_iter()
            .map(|(v, _)| v)
            .collect()
    }

    #[cfg(test)]
    pub fn force_update_cached_latencies(&self, authority_agg: &AuthorityAggregator<A>) {
        self.update_cached_latencies(authority_agg);
    }

    #[cfg(test)]
    pub fn get_client_stats_for_test(
        &self,
        validator: &AuthorityName,
    ) -> crate::validator_client_monitor::stats::ValidatorClientStats {
        self.client_stats
            .read()
            .validator_stats
            .get(validator)
            .cloned()
            .unwrap()
    }

    #[cfg(test)]
    pub fn get_client_stats_len(&self) -> usize {
        self.client_stats.read().validator_stats.len()
    }

    #[cfg(test)]
    pub fn has_validator_stats(&self, validator: &AuthorityName) -> bool {
        self.client_stats
            .read()
            .validator_stats
            .contains_key(validator)
    }
}