From b54625ede8d8ed4c500fe2fc54bccd16ab494358 Mon Sep 17 00:00:00 2001 From: Andrew Schran Date: Wed, 1 Jul 2026 15:50:03 -0400 Subject: [PATCH 1/2] sui-node: add discovery address prober for trusted-peer reachability metrics --- Cargo.lock | 10 +- Cargo.toml | 8 +- crates/sui-config/src/lib.rs | 2 +- crates/sui-config/src/node.rs | 80 ++ crates/sui-core/src/consensus_manager/mod.rs | 9 + crates/sui-e2e-tests/tests/discovery_tests.rs | 262 +++- crates/sui-network/src/discovery/mod.rs | 41 + crates/sui-node/Cargo.toml | 2 + crates/sui-node/src/address_prober.rs | 1081 +++++++++++++++++ crates/sui-node/src/admin.rs | 20 + crates/sui-node/src/lib.rs | 139 ++- .../src/node_config_builder.rs | 2 + 12 files changed, 1623 insertions(+), 33 deletions(-) create mode 100644 crates/sui-node/src/address_prober.rs diff --git a/Cargo.lock b/Cargo.lock index 654d89fa8ba8..df45131a2c10 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -877,7 +877,7 @@ dependencies = [ [[package]] name = "anemo" version = "0.0.0" -source = "git+https://github.com/mystenlabs/anemo.git?rev=b1bda3f8555c5594a4966e850170cc11ecfcf557#b1bda3f8555c5594a4966e850170cc11ecfcf557" +source = "git+https://github.com/mystenlabs/anemo.git?rev=68adc31d4ea1f4573f08c2c75317fcb205b823de#68adc31d4ea1f4573f08c2c75317fcb205b823de" dependencies = [ "anyhow", "async-trait", @@ -925,7 +925,7 @@ dependencies = [ [[package]] name = "anemo-build" version = "0.0.0" -source = "git+https://github.com/mystenlabs/anemo.git?rev=b1bda3f8555c5594a4966e850170cc11ecfcf557#b1bda3f8555c5594a4966e850170cc11ecfcf557" +source = "git+https://github.com/mystenlabs/anemo.git?rev=68adc31d4ea1f4573f08c2c75317fcb205b823de#68adc31d4ea1f4573f08c2c75317fcb205b823de" dependencies = [ "prettyplease", "proc-macro2", @@ -936,7 +936,7 @@ dependencies = [ [[package]] name = "anemo-cli" version = "0.0.0" -source = "git+https://github.com/mystenlabs/anemo.git?rev=b1bda3f8555c5594a4966e850170cc11ecfcf557#b1bda3f8555c5594a4966e850170cc11ecfcf557" +source = "git+https://github.com/mystenlabs/anemo.git?rev=68adc31d4ea1f4573f08c2c75317fcb205b823de#68adc31d4ea1f4573f08c2c75317fcb205b823de" dependencies = [ "anemo", "anemo-tower", @@ -952,7 +952,7 @@ dependencies = [ [[package]] name = "anemo-tower" version = "0.0.0" -source = "git+https://github.com/mystenlabs/anemo.git?rev=b1bda3f8555c5594a4966e850170cc11ecfcf557#b1bda3f8555c5594a4966e850170cc11ecfcf557" +source = "git+https://github.com/mystenlabs/anemo.git?rev=68adc31d4ea1f4573f08c2c75317fcb205b823de#68adc31d4ea1f4573f08c2c75317fcb205b823de" dependencies = [ "anemo", "bytes", @@ -16292,6 +16292,7 @@ dependencies = [ "bin-version", "bytes", "clap", + "consensus-config", "consensus-core", "fastcrypto", "fastcrypto-zkp", @@ -16328,6 +16329,7 @@ dependencies = [ "telemetry-subscribers", "tikv-jemallocator", "tokio", + "tonic-rustls", "tower 0.5.3", "tower-http 0.5.2", "tracing", diff --git a/Cargo.toml b/Cargo.toml index 24d8b611cdac..eee2a7bce31b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -640,10 +640,10 @@ p384 = { version = "0.13.0", default-features = false, features = [ ciborium = "0.2" # anemo dependencies -anemo = { git = "https://github.com/mystenlabs/anemo.git", rev = "b1bda3f8555c5594a4966e850170cc11ecfcf557" } -anemo-build = { git = "https://github.com/mystenlabs/anemo.git", rev = "b1bda3f8555c5594a4966e850170cc11ecfcf557" } -anemo-cli = { git = "https://github.com/mystenlabs/anemo.git", rev = "b1bda3f8555c5594a4966e850170cc11ecfcf557" } -anemo-tower = { git = "https://github.com/mystenlabs/anemo.git", rev = "b1bda3f8555c5594a4966e850170cc11ecfcf557" } +anemo = { git = "https://github.com/mystenlabs/anemo.git", rev = "68adc31d4ea1f4573f08c2c75317fcb205b823de" } +anemo-build = { git = "https://github.com/mystenlabs/anemo.git", rev = "68adc31d4ea1f4573f08c2c75317fcb205b823de" } +anemo-cli = { git = "https://github.com/mystenlabs/anemo.git", rev = "68adc31d4ea1f4573f08c2c75317fcb205b823de" } +anemo-tower = { git = "https://github.com/mystenlabs/anemo.git", rev = "68adc31d4ea1f4573f08c2c75317fcb205b823de" } # core-types from the new sdk for use in gRPC sui-sdk-types = { git = "https://github.com/MystenLabs/sui-rust-sdk.git", rev = "43c5bc13202ae398b1519a3eead1f40df8ca277b", features = [ "hash", "serde" ] } diff --git a/crates/sui-config/src/lib.rs b/crates/sui-config/src/lib.rs index f3637407f72d..798173c91b44 100644 --- a/crates/sui-config/src/lib.rs +++ b/crates/sui-config/src/lib.rs @@ -23,7 +23,7 @@ pub mod transaction_deny_config; pub mod validator_client_monitor_config; pub mod verifier_signing_config; -pub use node::{ConsensusConfig, ExecutionCacheConfig, NodeConfig}; +pub use node::{AddressProberConfig, ConsensusConfig, ExecutionCacheConfig, NodeConfig}; pub use rpc_config::{RpcConfig, RpcIndexInitConfig, RpcTlsConfig}; use sui_types::multiaddr::Multiaddr; use tracing::debug; diff --git a/crates/sui-config/src/node.rs b/crates/sui-config/src/node.rs index 63e1b7bd6a8b..0aa1c7270465 100644 --- a/crates/sui-config/src/node.rs +++ b/crates/sui-config/src/node.rs @@ -280,6 +280,10 @@ pub struct NodeConfig { /// When set, enables per-commit binary logs of congestion tracker state. #[serde(skip_serializing_if = "Option::is_none")] pub congestion_log: Option, + + /// Configuration for the trusted peer address prober. + #[serde(skip_serializing_if = "Option::is_none")] + pub address_prober: Option, } #[derive(Clone, Debug, Deserialize, Serialize)] @@ -361,6 +365,82 @@ pub struct ForkRecoveryConfig { pub fork_crash_behavior: ForkCrashBehavior, } +/// Configuration for the address prober: a background task on validators that periodically +/// checks whether trusted peers' advertised P2P and consensus addresses are connectable +/// and reports the results as Prometheus metrics. +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +#[serde(rename_all = "kebab-case")] +pub struct AddressProberConfig { + /// Whether the prober runs. + /// + /// If unspecified, this defaults to `true`. + pub enabled: Option, + + /// How often to re-probe an address that was reachable on its last probe. + /// + /// If unspecified, this defaults to 1 hour. + pub good_interval: Option, + + /// How often to re-probe an address that failed its last probe — should be frequently enough + /// to confirm a sustained failure and to promptly notice a fix. + /// + /// If unspecified, this defaults to 1 minute. + pub failed_interval: Option, + + /// Number of consecutive failed probes before a peer/endpoint/source's connectability gauge + /// flips to 0 (smooths out transient blips). + /// + /// If unspecified, this defaults to `3`. + pub failure_threshold: Option, + + /// Maximum number of address probes in flight at once. + /// + /// If unspecified, this defaults to `16`. + pub concurrency: Option, + + /// Per-probe timeout for the consensus connect (the P2P probe uses anemo's connect timeout). + /// + /// If unspecified, this defaults to 10 seconds. + pub consensus_probe_timeout: Option, +} + +impl AddressProberConfig { + pub fn enabled(&self) -> bool { + self.enabled.unwrap_or(true) + } + + pub fn good_interval(&self) -> Duration { + self.good_interval.unwrap_or(Duration::from_secs(60 * 60)) + } + + pub fn failed_interval(&self) -> Duration { + self.failed_interval.unwrap_or(Duration::from_secs(60)) + } + + pub fn failure_threshold(&self) -> u32 { + self.failure_threshold.unwrap_or(3) + } + + pub fn concurrency(&self) -> usize { + self.concurrency.unwrap_or(16) + } + + pub fn consensus_probe_timeout(&self) -> Duration { + self.consensus_probe_timeout + .unwrap_or(Duration::from_secs(10)) + } + + pub fn validate(&self) -> anyhow::Result<()> { + anyhow::ensure!( + self.failed_interval() <= self.good_interval(), + "address prober failed_interval ({:?}) must be <= good_interval ({:?})", + self.failed_interval(), + self.good_interval(), + ); + Ok(()) + } +} + #[derive(Clone, Debug, Default, Deserialize, Serialize)] #[serde(rename_all = "kebab-case")] pub struct ExecutionTimeObserverConfig { diff --git a/crates/sui-core/src/consensus_manager/mod.rs b/crates/sui-core/src/consensus_manager/mod.rs index 2d37064fc207..f8ad621f7d02 100644 --- a/crates/sui-core/src/consensus_manager/mod.rs +++ b/crates/sui-core/src/consensus_manager/mod.rs @@ -464,6 +464,15 @@ impl ConsensusManager { self.authority.load().as_ref().map(|a| a.0.store()) } + pub fn address_overrides_snapshot( + &self, + ) -> BTreeMap< + ConsensusNetworkPublicKey, + BTreeMap>, + > { + self.address_overrides.lock().map.clone() + } + fn get_store_path(&self, epoch: EpochId) -> PathBuf { let mut store_path = self.storage_base_path.clone(); store_path.push(format!("{}", epoch)); diff --git a/crates/sui-e2e-tests/tests/discovery_tests.rs b/crates/sui-e2e-tests/tests/discovery_tests.rs index 1c5ec3b8a921..c9952cf9a87b 100644 --- a/crates/sui-e2e-tests/tests/discovery_tests.rs +++ b/crates/sui-e2e-tests/tests/discovery_tests.rs @@ -25,7 +25,7 @@ mod test { async fn test_network_resilience_with_incorrect_discovery_addresses() { let test_cluster = TestClusterBuilder::new() .with_num_validators(4) - .with_epoch_duration_ms(30000) + .with_epoch_duration_ms(600_000) .build() .await; @@ -49,6 +49,12 @@ mod test { .discovery .get_or_insert_with(Default::default); disc.use_get_known_peers_v3 = Some(true); + // Use a short failed-probe interval so the prober cycles quickly within the wait below + // (the production default is 1 minute). + config.address_prober = Some(sui_config::AddressProberConfig { + failed_interval: Some(Duration::from_secs(2)), + ..Default::default() + }); } for name in &validator_names[1..] { let node = test_cluster.swarm.node(name).unwrap(); @@ -118,6 +124,83 @@ mod test { ); } + // Verify the address prober flags exactly the bad-address validators. The differential we + // expect, from the good validator's prober, for every bad-address validator: its on-chain + // (chain) P2P address probes as reachable, but its gossiped (discovery) address — the bad + // one — never does. We assert on the cumulative `attempts` counters rather than the smoothed + // `connectable` boolean, so the assertion doesn't depend on exactly when the smoothing window + // crosses its failure threshold. + info!("Waiting for address prober cycles..."); + sleep(Duration::from_secs(60)).await; + + let good_metrics = test_cluster + .swarm + .node(&validator_names[0]) + .unwrap() + .get_node_handle() + .unwrap() + .with(|n| n.address_prober_metrics_for_testing()); + + for bad_peer_id in &bad_peer_ids { + let peer = bad_peer_id.to_string(); + assert!( + good_metrics.attempts_value_for_testing(&peer, "p2p", "chain", "reachable") > 0, + "good validator should reach bad validator {peer} via its chain P2P address" + ); + assert_eq!( + good_metrics.attempts_value_for_testing(&peer, "p2p", "discovery", "reachable"), + 0, + "bad validator {peer}'s discovery P2P address must never probe as reachable" + ); + let discovery_failures = + good_metrics.attempts_value_for_testing(&peer, "p2p", "discovery", "timeout") + + good_metrics.attempts_value_for_testing( + &peer, + "p2p", + "discovery", + "unreachable", + ); + assert!( + discovery_failures > 0, + "bad validator {peer}'s discovery P2P address should fail probing" + ); + } + + // The good validator must NOT be flagged: from a bad validator's prober, the good + // validator's gossiped (discovery) address is the correct, reachable one and never fails. + let bad_validator_metrics = test_cluster + .swarm + .node(&validator_names[1]) + .unwrap() + .get_node_handle() + .unwrap() + .with(|n| n.address_prober_metrics_for_testing()); + let good_peer = good_peer_id.to_string(); + assert!( + bad_validator_metrics.attempts_value_for_testing( + &good_peer, + "p2p", + "discovery", + "reachable" + ) > 0, + "good validator's discovery P2P address should probe as reachable (not flagged)" + ); + assert_eq!( + bad_validator_metrics.attempts_value_for_testing( + &good_peer, + "p2p", + "discovery", + "timeout" + ) + bad_validator_metrics.attempts_value_for_testing( + &good_peer, + "p2p", + "discovery", + "unreachable" + ), + 0, + "good validator's discovery P2P address should never fail probing" + ); + // Publish the "basics" example package (needed for randomness tx). info!("Publishing basics package..."); let package_id = { @@ -248,6 +331,183 @@ mod test { } } + /// A validator that advertises a bad *consensus* address via discovery is flagged by the prober + /// on its consensus `discovery` override, while its on-chain (`chain`) consensus address stays + /// reachable. This exercises the consensus-probe path end-to-end. + #[sim_test] + async fn test_prober_detects_bad_consensus_address() { + let test_cluster = TestClusterBuilder::new() + .with_num_validators(4) + .with_epoch_duration_ms(600_000) + .build() + .await; + + let validator_names: Vec<_> = test_cluster.get_validator_pubkeys(); + + test_cluster.stop_all_validators().await; + let bad_consensus_addr: sui_types::multiaddr::Multiaddr = + "/ip4/1.1.1.1/udp/9998".parse().unwrap(); + for name in &validator_names { + let node = test_cluster.swarm.node(name).unwrap(); + let mut config = node.config(); + config.p2p_config.seed_peers.clear(); + config + .p2p_config + .discovery + .get_or_insert_with(Default::default) + .use_get_known_peers_v3 = Some(true); + // Use a short failed-probe interval so the bad consensus address flips the + // connectability gauge to 0 quickly (the production default is 1 minute). + config.address_prober = Some(sui_config::AddressProberConfig { + failed_interval: Some(Duration::from_secs(2)), + ..Default::default() + }); + } + // Validator 1 advertises a bad consensus external address; it propagates as a consensus + // `Discovery` override on the other validators. + { + let node = test_cluster.swarm.node(&validator_names[1]).unwrap(); + let mut config = node.config(); + if let Some(consensus_config) = config.consensus_config.as_mut() { + consensus_config.external_address = Some(bad_consensus_addr.clone()); + } + } + test_cluster.start_all_validators().await; + + // Let discovery propagate the override and the prober run several cycles (the gauge flips + // after 3 consecutive failures at the 2s failed-probe interval set above). + info!("Waiting for consensus override propagation and prober cycles..."); + sleep(Duration::from_secs(30)).await; + + // The prober labels consensus endpoints by the hex of the peer's network public key. + let bad_consensus_label = test_cluster + .swarm + .node(&validator_names[1]) + .unwrap() + .get_node_handle() + .unwrap() + .with(|n| { + sui_node::address_prober::consensus_peer_label_for_testing( + n.get_config().network_key_pair().public().0.to_bytes(), + ) + }); + + // From validator 0's prober. + let metrics = test_cluster + .swarm + .node(&validator_names[0]) + .unwrap() + .get_node_handle() + .unwrap() + .with(|n| n.address_prober_metrics_for_testing()); + + // The on-chain consensus address is reachable. + assert!( + metrics.attempts_value_for_testing( + &bad_consensus_label, + "consensus", + "chain", + "reachable" + ) > 0, + "validator 1's chain consensus address should be reachable" + ); + // The gossiped (discovery) consensus override — the bad one — is never reachable and fails. + assert_eq!( + metrics.attempts_value_for_testing( + &bad_consensus_label, + "consensus", + "discovery", + "reachable" + ), + 0, + "validator 1's bad discovery consensus address must never be reachable" + ); + assert!( + metrics.total_attempts_for_testing(&bad_consensus_label, "consensus", "discovery") > 0, + "validator 1's discovery consensus override should have been probed" + ); + // With a stable (long) epoch the smoothed differential holds. + assert_eq!( + metrics.connectable_for_testing(&bad_consensus_label, "consensus", "chain"), + 1, + "chain consensus address should be marked connectable" + ); + assert_eq!( + metrics.connectable_for_testing(&bad_consensus_label, "consensus", "discovery"), + 0, + "bad discovery consensus address should be flagged unconnectable" + ); + } + + /// An all-correct cluster (V3 enabled, no bad addresses) — the prober flags nothing. Guards + /// against false positives / alert fatigue. + #[sim_test] + async fn test_prober_no_false_positives_when_all_correct() { + let test_cluster = TestClusterBuilder::new() + .with_num_validators(4) + .with_epoch_duration_ms(600_000) + .build() + .await; + + let validator_names: Vec<_> = test_cluster.get_validator_pubkeys(); + + test_cluster.stop_all_validators().await; + for name in &validator_names { + let node = test_cluster.swarm.node(name).unwrap(); + let mut config = node.config(); + config.p2p_config.seed_peers.clear(); + config + .p2p_config + .discovery + .get_or_insert_with(Default::default) + .use_get_known_peers_v3 = Some(true); + } + test_cluster.start_all_validators().await; + + // Let discovery propagate and the prober run several cycles. + info!("Waiting for prober cycles in all-correct cluster..."); + sleep(Duration::from_secs(80)).await; + + let validator_peer_ids: Vec<_> = validator_names + .iter() + .map(|name| { + let node = test_cluster.swarm.node(name).unwrap(); + anemo::PeerId(node.config().network_key_pair().public().0.to_bytes()) + }) + .collect(); + + // From validator 0's prober, no probed P2P address (any source, any other validator) failed, + // and nothing is flagged unconnectable. + let metrics = test_cluster + .swarm + .node(&validator_names[0]) + .unwrap() + .get_node_handle() + .unwrap() + .with(|n| n.address_prober_metrics_for_testing()); + + for peer_id in &validator_peer_ids[1..] { + let peer = peer_id.to_string(); + for source in ["chain", "discovery"] { + let total = metrics.total_attempts_for_testing(&peer, "p2p", source); + if total == 0 { + continue; // this source was not advertised for this peer + } + let failures = + total - metrics.attempts_value_for_testing(&peer, "p2p", source, "reachable"); + assert_eq!( + failures, 0, + "no probe failures expected for {peer}/{source} in an all-correct cluster" + ); + assert_ne!( + metrics.connectable_for_testing(&peer, "p2p", source), + 0, + "no peer should be flagged unconnectable in an all-correct cluster ({peer}/{source})" + ); + } + } + } + async fn execute_remove_validator_tx(test_cluster: &TestCluster, handle: &SuiNodeHandle) { let address = handle.with(|node| node.get_config().sui_address()); let gas = test_cluster diff --git a/crates/sui-network/src/discovery/mod.rs b/crates/sui-network/src/discovery/mod.rs index 4f3544d6d043..a39aab90b42e 100644 --- a/crates/sui-network/src/discovery/mod.rs +++ b/crates/sui-network/src/discovery/mod.rs @@ -55,6 +55,11 @@ pub use generated::{ }; pub use server::{GetKnownPeersRequestV3, GetKnownPeersResponseV2, GetKnownPeersResponseV3}; +/// Per-source P2P addresses for trusted peers, keyed by peer then by address source. +/// Returned by [`Sender::trusted_peer_p2p_addresses`]. +pub type TrustedPeerP2pAddresses = + BTreeMap>>; + /// Message types for the discovery system mailbox. #[derive(Debug)] pub enum DiscoveryMessage { @@ -73,6 +78,11 @@ pub enum DiscoveryMessage { TrustedPeersUpdated, /// A peer has been reported as continuously failing, triggering disconnect and cooldown. PeerFailureReport { peer_id: PeerId }, + /// Request a snapshot of per-source P2P addresses, for trusted peers with at least one + /// recorded address. + GetTrustedPeerP2pAddresses { + reply: oneshot::Sender, + }, } /// A Handle to the Discovery subsystem. The Discovery system will be shut down once all Handles @@ -117,6 +127,22 @@ impl Sender { .sender .try_send(DiscoveryMessage::PeerFailureReport { peer_id }); } + + /// Snapshot of per-source P2P addresses for trusted peers only (configured/seed/allowlisted + /// peers and on-chain validators). Used by the address prober to enumerate probe candidates; + /// returns an empty map if the discovery event loop has shut down. + pub async fn trusted_peer_p2p_addresses(&self) -> TrustedPeerP2pAddresses { + let (s, r) = oneshot::channel(); + if self + .sender + .send(DiscoveryMessage::GetTrustedPeerP2pAddresses { reply: s }) + .await + .is_err() + { + return TrustedPeerP2pAddresses::default(); + } + r.await.unwrap_or_default() + } } use self::metrics::Metrics; @@ -443,9 +469,24 @@ impl DiscoveryEventLoop { DiscoveryMessage::PeerFailureReport { peer_id } => { self.handle_peer_failure_report(peer_id); } + DiscoveryMessage::GetTrustedPeerP2pAddresses { reply } => { + let _ = reply.send(self.trusted_peer_p2p_addresses()); + } } } + fn trusted_peer_p2p_addresses(&self) -> TrustedPeerP2pAddresses { + let state = self.state.read().unwrap(); + state + .peer_addresses + .iter() + .filter(|(peer_id, _)| { + is_trusted_peer(peer_id, &self.configured_peers, &self.chain_peers) + }) + .map(|(peer_id, sources)| (*peer_id, sources.clone())) + .collect() + } + fn handle_peer_failure_report(&mut self, peer_id: PeerId) { if self.is_trusted_peer(&peer_id) { info!(?peer_id, "ignoring failure report for trusted peer"); diff --git a/crates/sui-node/Cargo.toml b/crates/sui-node/Cargo.toml index 772c6892e279..67afecac540b 100644 --- a/crates/sui-node/Cargo.toml +++ b/crates/sui-node/Cargo.toml @@ -36,6 +36,8 @@ url.workspace = true humantime.workspace = true consensus-core.workspace = true +consensus-config.workspace = true +tonic-rustls.workspace = true sui-tls.workspace = true sui-macros.workspace = true sui-config.workspace = true diff --git a/crates/sui-node/src/address_prober.rs b/crates/sui-node/src/address_prober.rs new file mode 100644 index 000000000000..91dd40cffc10 --- /dev/null +++ b/crates/sui-node/src/address_prober.rs @@ -0,0 +1,1081 @@ +// Copyright (c) Mysten Labs, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! The adddress prober periodically checks addresses of trusted peers for connectability, reported +//! via Prometheus metrics. +//! +//! - P2P: `anemo::Network::probe_address`, which verifies reachability + the peer's `PeerId` without +//! joining the peer set or disturbing any existing connection. +//! - Consensus: a throwaway tonic+rustls `connect()` replicating the real consensus client TLS +//! (expected network key, `consensus_epoch_{epoch}` server name, our network key as the client +//! cert). Only a current committee member can complete this handshake. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use anemo::{Network, PeerId}; +use consensus_config::{ + Authority as ConsensusAuthority, Committee as ConsensusCommittee, + NetworkKeyPair as ConsensusNetworkKeyPair, NetworkPublicKey as ConsensusNetworkPublicKey, +}; +use fastcrypto::encoding::{Encoding, Hex}; +use futures::future::{BoxFuture, FutureExt, join_all}; +use futures::stream::{FuturesUnordered, StreamExt}; +use mysten_metrics::spawn_monitored_task; +use mysten_network::Multiaddr; +use prometheus::{ + IntCounterVec, IntGaugeVec, Registry, register_int_counter_vec_with_registry, + register_int_gauge_vec_with_registry, +}; +use serde::Serialize; +use sui_config::AddressProberConfig; +use sui_core::consensus_manager::ConsensusManager; +use sui_network::discovery::{Sender as DiscoverySender, TrustedPeerP2pAddresses}; +use sui_network::endpoint_manager::AddressSource; +use tokio::sync::{Semaphore, mpsc, oneshot}; +use tokio::time::Instant; +use tracing::{debug, info}; + +const MAILBOX_CAPACITY: usize = 128; // updates are rare (once per epoch) + +/// Which transport an address belongs to. The `&'static str` is the Prometheus `endpoint_type` +/// label. +#[derive(Clone, Copy, PartialEq, Eq)] +enum EndpointType { + P2p, + Consensus, +} + +impl EndpointType { + fn as_str(self) -> &'static str { + match self { + EndpointType::P2p => "p2p", + EndpointType::Consensus => "consensus", + } + } +} + +/// Outcome of probing a single address, unified across the P2P and consensus paths. The string form +/// is the Prometheus `result` label on the attempts counter. +#[derive(Clone, Copy, PartialEq, Eq)] +enum ProbeResult { + Reachable, + Unreachable, + WrongIdentity, + BadAddress, + Timeout, + // Note: Update ProbeResult::ALL if adding new variants. +} + +impl ProbeResult { + /// All variants, so the `result`-labelled attempts series can be enumerated — e.g. to drop a + /// churned-out peer's series, or in test helpers. + const ALL: [ProbeResult; 5] = [ + ProbeResult::Reachable, + ProbeResult::Unreachable, + ProbeResult::WrongIdentity, + ProbeResult::BadAddress, + ProbeResult::Timeout, + ]; + + fn is_reachable(self) -> bool { + matches!(self, ProbeResult::Reachable) + } + + fn as_str(self) -> &'static str { + match self { + ProbeResult::Reachable => "reachable", + ProbeResult::Unreachable => "unreachable", + ProbeResult::WrongIdentity => "wrong_identity", + ProbeResult::BadAddress => "bad_address", + ProbeResult::Timeout => "timeout", + } + } +} + +impl From for ProbeResult { + fn from(outcome: anemo::ProbeOutcome) -> Self { + match outcome { + anemo::ProbeOutcome::Reachable => ProbeResult::Reachable, + anemo::ProbeOutcome::Unreachable => ProbeResult::Unreachable, + anemo::ProbeOutcome::WrongIdentity => ProbeResult::WrongIdentity, + anemo::ProbeOutcome::BadAddress => ProbeResult::BadAddress, + anemo::ProbeOutcome::Timeout => ProbeResult::Timeout, + } + } +} + +/// One concrete address to probe, tagged with how to probe it. +#[derive(Clone)] +enum AddressTarget { + P2p { + peer_id: PeerId, + address: anemo::types::Address, + }, + Consensus { + target_key: ConsensusNetworkPublicKey, + address: Multiaddr, + }, +} + +impl AddressTarget { + fn display(&self) -> String { + match self { + AddressTarget::P2p { address, .. } => address.to_string(), + AddressTarget::Consensus { address, .. } => address.to_string(), + } + } +} + +/// Operator-facing identity of a peer that is a current committee validator, resolved from the +/// consensus committee `Authority`. `None` for trusted non-validator peers (seeds / configured +/// fullnodes) that aren't in the committee. +#[derive(Clone)] +struct AuthorityInfo { + /// On-chain authority name (protocol public key), hex-encoded — matches the Sui-side + /// `AuthorityName`. + authority_name: String, + /// The validator's advertised hostname from the committee. + hostname: String, +} + +/// All addresses advertised for one `(peer, endpoint_type, source)` triple. +struct ProbeGroup { + peer_label: String, + endpoint_type: EndpointType, + source: AddressSource, + /// Validator identity (name + hostname), if this peer is in the current committee. + authority: Option, + addresses: Vec, +} + +/// Identifies a `(peer, endpoint_type, source)` triple in the prober's per-group state. +type GroupKey = (String, &'static str, &'static str); + +/// The most recent probe result for a single concrete address within a group. Retained only for the +/// admin report (the metrics deliberately omit the address to bound cardinality). +struct AddressOutcome { + address: String, + result: ProbeResult, +} + +/// A trusted peer's `(peer, endpoint_type, source)` triple tracked across probe cycles. +struct Group { + peer_label: String, + endpoint_type: EndpointType, + source: AddressSource, + /// Validator identity (name + hostname), if this peer is in the current committee. + authority: Option, + /// Addresses to probe, refreshed from discovery/committee each scan. + targets: Vec, + /// True while this group's probe is in flight, so a scan doesn't re-spawn it. + probing: bool, + last_probed: Option, + consecutive_failures: u32, + /// Smoothed connectability (mirrors the `discovery_probe_connectable` gauge): `true` until + /// `failure_threshold` consecutive failures flip it to `false`. + connectable: bool, + last_success_unix_secs: Option, + outcomes: Vec, +} + +impl Group { + fn new(candidate: ProbeGroup) -> Self { + Self { + peer_label: candidate.peer_label, + endpoint_type: candidate.endpoint_type, + source: candidate.source, + authority: candidate.authority, + targets: candidate.addresses, + probing: false, + last_probed: None, + consecutive_failures: 0, + connectable: false, + last_success_unix_secs: None, + outcomes: Vec::new(), + } + } + + /// When this group is next due to be probed: a never-probed group is due now, otherwise its last + /// probe plus the good/failed interval selected by its recent history. + fn next_due( + &self, + good_interval: Duration, + failed_interval: Duration, + now: Instant, + ) -> Instant { + match self.last_probed { + None => now, + Some(last_probed) => { + let interval = if self.consecutive_failures == 0 { + good_interval + } else { + failed_interval + }; + last_probed + interval + } + } + } +} + +/// Epoch-scoped inputs to the prober. +pub struct ProberEpochContext { + pub epoch: u64, + pub consensus_committee: ConsensusCommittee, + pub consensus_manager: Arc, +} + +enum ProberMessage { + /// This node is a current validator for `epoch`; probe its committee's consensus endpoints + the + /// trusted peers' P2P endpoints. + UpdateEpoch { + epoch: u64, + consensus_committee: ConsensusCommittee, + consensus_manager: Arc, + }, + /// This node is no longer a current validator; the prober idles until the next `UpdateEpoch`. + LeaveCommittee, + /// Admin snapshot request: reply with the prober's latest results (see [`Handle::probe_report`]). + GetReport { reply: oneshot::Sender }, +} + +/// A point-in-time snapshot of the prober's latest results, served by the admin endpoint. +#[derive(Clone, Debug, Serialize)] +pub struct ProbeReport { + pub epoch: Option, + pub groups: Vec, +} + +/// Latest probe outcome for one `(peer, endpoint_type, source)` group. +#[derive(Clone, Debug, Serialize)] +pub struct ProbeGroupReport { + /// `peer_id` hex (P2P) or consensus network public key hex (consensus). + pub peer: String, + /// On-chain authority name (protocol public key) hex, if this peer is a current committee + /// validator; `None` for trusted non-validator peers. + pub authority_name: Option, + /// The validator's committee hostname, if this peer is a current committee validator. + pub hostname: Option, + pub endpoint_type: String, + pub address_source: String, + /// Smoothed connectability (matches the `discovery_probe_connectable` gauge). + pub connectable: bool, + pub consecutive_failures: u32, + /// How long ago the group was last probed, in seconds. + pub seconds_since_last_probe: u64, + /// Unix timestamp (seconds) of the last successful probe, if ever reachable. + pub last_success_unix_seconds: Option, + pub addresses: Vec, +} + +/// Latest probe result for a single concrete address. +#[derive(Clone, Debug, Serialize)] +pub struct ProbeAddressReport { + pub address: String, + /// One of `reachable`, `unreachable`, `wrong_identity`, `bad_address`, `timeout`. + pub result: String, +} + +pub struct AddressProberMetrics { + connectable: IntGaugeVec, + last_success_timestamp_seconds: IntGaugeVec, + attempts_total: IntCounterVec, +} + +impl AddressProberMetrics { + pub fn new(registry: &Registry) -> Arc { + Arc::new(Self { + connectable: register_int_gauge_vec_with_registry!( + "discovery_probe_connectable", + "1 if a trusted peer's advertised address for this endpoint/source is connectable, \ + 0 after N consecutive failed probe cycles", + &["peer_id", "endpoint_type", "address_source"], + registry + ) + .unwrap(), + last_success_timestamp_seconds: register_int_gauge_vec_with_registry!( + "discovery_probe_last_success_timestamp_seconds", + "Unix timestamp (seconds) of the last successful probe for this peer/endpoint/source", + &["peer_id", "endpoint_type", "address_source"], + registry + ) + .unwrap(), + attempts_total: register_int_counter_vec_with_registry!( + "discovery_probe_attempts_total", + "Total address probe attempts by peer/endpoint/source and result", + &["peer_id", "endpoint_type", "address_source", "result"], + registry + ) + .unwrap(), + }) + } +} + +#[cfg(any(test, msim))] +impl AddressProberMetrics { + /// Current value of the smoothed connectability gauge for a triple (creates the series at 0 if + /// it has never been written, so distinguish via [`Self::attempts_value_for_testing`]). + pub fn connectable_for_testing( + &self, + peer_label: &str, + endpoint_type: &str, + source: &str, + ) -> i64 { + self.connectable + .with_label_values(&[peer_label, endpoint_type, source]) + .get() + } + + /// Number of probe attempts recorded for a triple with the given result. + pub fn attempts_value_for_testing( + &self, + peer_label: &str, + endpoint_type: &str, + source: &str, + result: &str, + ) -> u64 { + self.attempts_total + .with_label_values(&[peer_label, endpoint_type, source, result]) + .get() + } + + /// Total probe attempts recorded for a triple across all results. + pub fn total_attempts_for_testing( + &self, + peer_label: &str, + endpoint_type: &str, + source: &str, + ) -> u64 { + ProbeResult::ALL + .into_iter() + .map(|result| { + self.attempts_value_for_testing(peer_label, endpoint_type, source, result.as_str()) + }) + .sum() + } +} + +/// Handle to the address prober. Dropping all clones closes the mailbox and the event loop +/// shuts down. Holds the metrics so tests can read them. +#[derive(Clone)] +pub struct Handle { + sender: mpsc::Sender, + // Retained only so tests can read the prober's metrics. + #[cfg(any(test, msim))] + metrics: Arc, +} + +impl Handle { + /// Activates the prober for an epoch. Call if this node is a current validator. + pub fn update_epoch( + &self, + epoch: u64, + consensus_committee: ConsensusCommittee, + consensus_manager: Arc, + ) { + self.sender + .try_send(ProberMessage::UpdateEpoch { + epoch, + consensus_committee, + consensus_manager, + }) + .expect("address prober mailbox should not overflow or be closed"); + } + + /// Deactivates the prober. + pub fn leave_committee(&self) { + self.sender + .try_send(ProberMessage::LeaveCommittee) + .expect("address prober mailbox should not overflow or be closed"); + } + + /// Snapshot the prober's latest results (full addresses + per-address outcomes). + /// Returns `None` if the event loop has shut down or dropped the reply. + pub async fn probe_report(&self) -> Option { + let (reply, response) = oneshot::channel(); + if self + .sender + .send(ProberMessage::GetReport { reply }) + .await + .is_err() + { + return None; + } + response.await.ok() + } + + #[cfg(any(test, msim))] + pub fn metrics_for_testing(&self) -> Arc { + self.metrics.clone() + } +} + +pub struct Builder { + config: AddressProberConfig, + metrics: Option>, +} + +impl Default for Builder { + fn default() -> Self { + Self::new() + } +} + +impl Builder { + pub fn new() -> Self { + Self { + config: AddressProberConfig::default(), + metrics: None, + } + } + + pub fn config(mut self, config: AddressProberConfig) -> Self { + self.config = config; + self + } + + pub fn with_metrics(mut self, registry: &Registry) -> Self { + self.metrics = Some(AddressProberMetrics::new(registry)); + self + } + + pub fn build(self) -> UnstartedAddressProber { + let metrics = self + .metrics + .unwrap_or_else(|| AddressProberMetrics::new(&Registry::new())); + let (sender, mailbox) = mpsc::channel(MAILBOX_CAPACITY); + let handle = Handle { + sender, + #[cfg(any(test, msim))] + metrics: metrics.clone(), + }; + UnstartedAddressProber { + config: self.config, + metrics, + handle, + mailbox, + } + } +} + +/// A built-but-not-started prober: holds the runtime-independent state until [`start`] injects the +/// network/discovery/keypair and spawns the event loop. +/// +/// [`start`]: UnstartedAddressProber::start +pub struct UnstartedAddressProber { + config: AddressProberConfig, + metrics: Arc, + handle: Handle, + mailbox: mpsc::Receiver, +} + +impl UnstartedAddressProber { + /// Spawns the prober loop. + pub fn start( + self, + network: Network, + discovery: DiscoverySender, + own_consensus_keypair: ConsensusNetworkKeyPair, + ) -> Handle { + let event_loop = AddressProberEventLoop::new( + self.config, + self.metrics, + self.mailbox, + network, + discovery, + own_consensus_keypair, + ); + spawn_monitored_task!(event_loop.start()); + self.handle + } +} + +struct AddressProberEventLoop { + // Resolved config knobs (`Copy`); the per-probe futures capture these directly. + good_interval: Duration, + failed_interval: Duration, + failure_threshold: u32, + consensus_probe_timeout: Duration, + // Node-lifetime inputs. + network: Network, + discovery: DiscoverySender, + /// This node's consensus network keypair, used as the client cert for consensus probes. + own_consensus_keypair: ConsensusNetworkKeyPair, + own_peer_id: PeerId, + own_consensus_key: ConsensusNetworkPublicKey, + metrics: Arc, + mailbox: mpsc::Receiver, + inflight_probe_limit: Arc, + /// In-flight probes; each yields `(group key, per-address outcomes)` when it completes. + tasks: FuturesUnordered)>>, + groups: HashMap, + /// Current epoch's probe inputs, or `None` when this node is not a current validator. + epoch_state: Option>, +} + +impl AddressProberEventLoop { + fn new( + config: AddressProberConfig, + metrics: Arc, + mailbox: mpsc::Receiver, + network: Network, + discovery: DiscoverySender, + own_consensus_keypair: ConsensusNetworkKeyPair, + ) -> Self { + let own_peer_id = network.peer_id(); + let own_consensus_key = own_consensus_keypair.public(); + Self { + good_interval: config.good_interval(), + failed_interval: config.failed_interval(), + failure_threshold: config.failure_threshold(), + consensus_probe_timeout: config.consensus_probe_timeout(), + network, + discovery, + own_consensus_keypair, + own_peer_id, + own_consensus_key, + metrics, + mailbox, + inflight_probe_limit: Arc::new(Semaphore::new(config.concurrency())), + tasks: FuturesUnordered::new(), + groups: HashMap::new(), + epoch_state: None, + } + } + + async fn start(mut self) { + info!( + good_interval_secs = self.good_interval.as_secs(), + failed_interval_secs = self.failed_interval.as_secs(), + "starting discovery address prober", + ); + + // A single resettable timer fires when the next group is due (see `next_deadline`). + let mut timer = Box::pin(tokio::time::sleep_until(Instant::now())); + loop { + tokio::select! { + _ = &mut timer => self.scan().await, + Some((key, outcomes)) = self.tasks.next() => self.handle_probe_result(key, outcomes), + maybe_message = self.mailbox.recv() => match maybe_message { + // Once all `Handle`s have been dropped this yields `None`, so we shut down. + Some(message) => self.handle_message(message), + None => break, + }, + } + timer.as_mut().reset(self.next_deadline()); + } + + info!("discovery address prober ended"); + } + + fn handle_message(&mut self, message: ProberMessage) { + match message { + ProberMessage::UpdateEpoch { + epoch, + consensus_committee, + consensus_manager, + } => { + self.epoch_state = Some(Arc::new(ProberEpochContext { + epoch, + consensus_committee, + consensus_manager, + })); + } + ProberMessage::LeaveCommittee => { + self.epoch_state = None; + // We no longer probe anyone; drop all tracked groups and their metric series so a + // demoted node doesn't keep exporting stale per-peer metrics. + for key in self.groups.keys().cloned().collect::>() { + self.remove_group_metrics(&key); + } + self.groups.clear(); + } + ProberMessage::GetReport { reply } => { + let _ = reply.send(self.build_report()); + } + } + } + + /// When to next scan for due groups: the earliest per-group due time, capped by `failed_interval` + /// so newly-advertised peers/addresses are still discovered promptly even when everything known + /// is healthy. Groups with an in-flight probe are excluded — they reschedule when they complete. + fn next_deadline(&self) -> Instant { + let now = Instant::now(); + let cap = now + self.failed_interval; + if self.epoch_state.is_none() { + return cap; + } + self.groups + .values() + .filter(|group| !group.probing) + .map(|group| group.next_due(self.good_interval, self.failed_interval, now)) + .min() + .map_or(cap, |deadline| deadline.min(cap)) + .max(now) + } + + /// Rebuild the current candidate set from discovery + the committee, merge it into `self.groups`, + /// then spawn a probe for every group that is now due and not already being probed. The discovery + /// snapshot fetch is the only await; the probes run off-loop. + async fn scan(&mut self) { + let Some(context) = self.epoch_state.clone() else { + return; + }; + let epoch = context.epoch; + let trusted_p2p_addresses = self.discovery.trusted_peer_p2p_addresses().await; + let candidates = build_groups( + trusted_p2p_addresses, + &context.consensus_manager, + &context.consensus_committee, + &self.own_peer_id, + &self.own_consensus_key, + ); + self.refresh_groups(candidates); + + let now = Instant::now(); + let due: Vec = self + .groups + .iter() + .filter(|(_, group)| { + !group.probing + && group.next_due(self.good_interval, self.failed_interval, now) <= now + }) + .map(|(key, _)| key.clone()) + .collect(); + for key in due { + self.spawn_probe(epoch, &key); + } + } + + /// Merge a freshly-built candidate set into `self.groups`: refresh addresses/identity for known + /// groups, start tracking new ones, and drop groups no longer advertised — unless a probe is + /// still in flight for them, in which case they survive until that probe is recorded. + fn refresh_groups(&mut self, candidates: Vec) { + let current: HashSet = candidates.iter().map(group_key).collect(); + // Drop groups no longer advertised (unless a probe is still in flight for them) and clear + // their metric series, so a peer removed from the trusted set stops being exported. + let removed: Vec = self + .groups + .iter() + .filter(|(key, group)| !current.contains(*key) && !group.probing) + .map(|(key, _)| key.clone()) + .collect(); + for key in removed { + self.remove_group_metrics(&key); + self.groups.remove(&key); + } + for candidate in candidates { + let key = group_key(&candidate); + match self.groups.get_mut(&key) { + Some(group) => { + group.targets = candidate.addresses; + group.authority = candidate.authority; + } + None => { + self.groups.insert(key, Group::new(candidate)); + } + } + } + } + + /// Drop the Prometheus series for a group that is no longer tracked, so churned-out peers don't + /// linger in the exported metrics. Removing a nonexistent series is a no-op (ignored error). + fn remove_group_metrics(&self, key: &GroupKey) { + let (peer, endpoint_type, source) = (key.0.as_str(), key.1, key.2); + let labels = [peer, endpoint_type, source]; + let _ = self.metrics.connectable.remove_label_values(&labels); + let _ = self + .metrics + .last_success_timestamp_seconds + .remove_label_values(&labels); + // `attempts_total` also carries the `result` label, so one series exists per outcome. + for result in ProbeResult::ALL { + let _ = self.metrics.attempts_total.remove_label_values(&[ + peer, + endpoint_type, + source, + result.as_str(), + ]); + } + } + + /// Mark a group in-flight and spawn its probe onto `self.tasks`. + fn spawn_probe(&mut self, epoch: u64, key: &GroupKey) { + let targets = { + let group = self + .groups + .get_mut(key) + .expect("a due group is present in the map"); + group.probing = true; + group.targets.clone() + }; + let key = key.clone(); + let network = self.network.clone(); + let own_consensus_keypair = self.own_consensus_keypair.clone(); + let consensus_probe_timeout = self.consensus_probe_timeout; + let inflight_probe_limit = self.inflight_probe_limit.clone(); + self.tasks.push( + async move { + let outcomes = join_all(targets.into_iter().map(|target| { + let network = network.clone(); + let own_consensus_keypair = own_consensus_keypair.clone(); + let inflight_probe_limit = inflight_probe_limit.clone(); + async move { + let _permit = inflight_probe_limit + .acquire_owned() + .await + .expect("prober semaphore is never closed"); + let address = target.display(); + let result = probe_one( + &target, + &network, + &own_consensus_keypair, + epoch, + consensus_probe_timeout, + ) + .await; + AddressOutcome { address, result } + } + })) + .await; + (key, outcomes) + } + .boxed(), + ); + } + + fn handle_probe_result(&mut self, key: GroupKey, outcomes: Vec) { + // The group may have been dropped from the map if it churned out while probing; if so, the + // result is stale — discard it. + let Some(group) = self.groups.get_mut(&key) else { + return; + }; + group.probing = false; + + let now = Instant::now(); + let connectable = outcomes.iter().any(|outcome| outcome.result.is_reachable()); + let peer_label = group.peer_label.clone(); + let endpoint_type = group.endpoint_type.as_str(); + let source = address_source_str(group.source); + let labels = [peer_label.as_str(), endpoint_type, source]; + + for outcome in &outcomes { + self.metrics + .attempts_total + .with_label_values(&[ + peer_label.as_str(), + endpoint_type, + source, + outcome.result.as_str(), + ]) + .inc(); + debug!( + peer = %peer_label, + endpoint_type, + source, + address = %outcome.address, + result = outcome.result.as_str(), + "probed address" + ); + } + + // Update the smoothed connectability gauge: set on any reachable probe, cleared only after + // `failure_threshold` consecutive failures so transient blips don't flap the gauge. + if connectable { + self.metrics.connectable.with_label_values(&labels).set(1); + let timestamp = now_unix_seconds(); + self.metrics + .last_success_timestamp_seconds + .with_label_values(&labels) + .set(timestamp); + group.connectable = true; + group.last_success_unix_secs = Some(timestamp); + group.consecutive_failures = 0; + } else { + group.consecutive_failures += 1; + if group.consecutive_failures >= self.failure_threshold { + self.metrics.connectable.with_label_values(&labels).set(0); + group.connectable = false; + } + } + group.last_probed = Some(now); + group.outcomes = outcomes; + } + + /// Snapshot the latest probe results for manual inspection. + fn build_report(&self) -> ProbeReport { + let now = Instant::now(); + let mut groups: Vec = self + .groups + .values() + .filter_map(|group| { + let last_probed = group.last_probed?; + Some(ProbeGroupReport { + peer: group.peer_label.clone(), + authority_name: group + .authority + .as_ref() + .map(|authority| authority.authority_name.clone()), + hostname: group + .authority + .as_ref() + .map(|authority| authority.hostname.clone()), + endpoint_type: group.endpoint_type.as_str().to_string(), + address_source: address_source_str(group.source).to_string(), + connectable: group.connectable, + consecutive_failures: group.consecutive_failures, + seconds_since_last_probe: now.duration_since(last_probed).as_secs(), + last_success_unix_seconds: group.last_success_unix_secs, + addresses: group + .outcomes + .iter() + .map(|outcome| ProbeAddressReport { + address: outcome.address.clone(), + result: outcome.result.as_str().to_string(), + }) + .collect(), + }) + }) + .collect(); + + // Surface problems first: most-failed groups on top, then a stable label ordering. + groups.sort_by(|a, b| { + b.consecutive_failures + .cmp(&a.consecutive_failures) + .then_with(|| a.endpoint_type.cmp(&b.endpoint_type)) + .then_with(|| a.peer.cmp(&b.peer)) + .then_with(|| a.address_source.cmp(&b.address_source)) + }); + + ProbeReport { + epoch: self.epoch_state.as_ref().map(|context| context.epoch), + groups, + } + } +} + +/// Build the set of probe groups for this cycle: discovery's per-source P2P addresses, the +/// consensus override addresses, and the consensus on-chain (`Chain`) baseline from the committee. +fn build_groups( + trusted_p2p_addresses: TrustedPeerP2pAddresses, + consensus_manager: &ConsensusManager, + committee: &ConsensusCommittee, + own_peer_id: &PeerId, + own_consensus_key: &ConsensusNetworkPublicKey, +) -> Vec { + let mut groups = Vec::new(); + + // Index committee validators by their 32-byte network key so each group can be tagged with the + // validator's name + hostname. The anemo `PeerId` is these same bytes (both are the validator's + // `narwhal_network_pubkey`), so this resolves P2P peers as well as consensus ones; trusted + // non-validator peers (seeds / fullnodes) simply don't match and stay unnamed. + let authority_by_network_key: HashMap<[u8; 32], &ConsensusAuthority> = committee + .authorities() + .map(|(_, authority)| (authority.network_key.to_bytes(), authority)) + .collect(); + + // (a) P2P: every source for every trusted peer. + for (peer_id, sources) in trusted_p2p_addresses { + if &peer_id == own_peer_id { + continue; + } + let authority = authority_by_network_key + .get(&peer_id.0) + .copied() + .map(authority_info); + for (source, addresses) in sources { + if addresses.is_empty() { + continue; + } + groups.push(ProbeGroup { + peer_label: peer_id.to_string(), + endpoint_type: EndpointType::P2p, + source, + authority: authority.clone(), + addresses: addresses + .into_iter() + .map(|address| AddressTarget::P2p { peer_id, address }) + .collect(), + }); + } + } + + // (b) Consensus overrides (Discovery/Admin), per source. + for (network_key, sources) in consensus_manager.address_overrides_snapshot() { + if &network_key == own_consensus_key { + continue; + } + let authority = authority_by_network_key + .get(&network_key.to_bytes()) + .copied() + .map(authority_info); + for (source, addresses) in sources { + if addresses.is_empty() { + continue; + } + groups.push(ProbeGroup { + peer_label: consensus_peer_label(&network_key), + endpoint_type: EndpointType::Consensus, + source, + authority: authority.clone(), + addresses: addresses + .into_iter() + .map(|address| AddressTarget::Consensus { + target_key: network_key.clone(), + address, + }) + .collect(), + }); + } + } + + // (c) Consensus on-chain baseline (the committee address), labeled `Chain`. + for (_index, authority) in committee.authorities() { + if &authority.network_key == own_consensus_key { + continue; + } + groups.push(ProbeGroup { + peer_label: consensus_peer_label(&authority.network_key), + endpoint_type: EndpointType::Consensus, + source: AddressSource::Chain, + authority: Some(authority_info(authority)), + addresses: vec![AddressTarget::Consensus { + target_key: authority.network_key.clone(), + address: authority.address.clone(), + }], + }); + } + + groups +} + +/// Operator-facing identity (name + hostname) for a committee validator. +fn authority_info(authority: &ConsensusAuthority) -> AuthorityInfo { + AuthorityInfo { + authority_name: Hex::encode(authority.authority_name.to_bytes()), + hostname: authority.hostname.clone(), + } +} + +async fn probe_one( + target: &AddressTarget, + network: &Network, + own_consensus_keypair: &ConsensusNetworkKeyPair, + epoch: u64, + consensus_probe_timeout: Duration, +) -> ProbeResult { + match target { + AddressTarget::P2p { peer_id, address } => network + .probe_address(address.clone(), *peer_id) + .await + .into(), + AddressTarget::Consensus { + target_key, + address, + } => { + probe_consensus_address( + own_consensus_keypair, + target_key, + epoch, + address, + consensus_probe_timeout, + ) + .await + } + } +} + +/// Replicates the real consensus client's mutual-TLS setup (`tonic_network::get_channel`) in a +/// throwaway endpoint — expected peer network key, `consensus_epoch_{epoch}` server name, and our +/// own network key as the client cert — then eagerly connects with a bounded timeout and drops the +/// connection. Does not use the shared channel pool (that caches one channel per peer and would +/// defeat per-source probing). +async fn probe_consensus_address( + own_consensus_keypair: &ConsensusNetworkKeyPair, + target_key: &ConsensusNetworkPublicKey, + epoch: u64, + address: &Multiaddr, + timeout: Duration, +) -> ProbeResult { + let Some(host_port) = consensus_host_port(address) else { + return ProbeResult::BadAddress; + }; + let uri = format!("https://{host_port}"); + + // Matches `consensus/core/src/network/tonic_tls.rs::certificate_server_name`. + let server_name = format!("consensus_epoch_{epoch}"); + let client_tls_config = sui_tls::create_rustls_client_config( + target_key.clone().into_inner(), + server_name, + Some(own_consensus_keypair.clone().private_key().into_inner()), + ); + + let endpoint = match tonic_rustls::Channel::from_shared(uri) { + Ok(endpoint) => endpoint.connect_timeout(timeout), + Err(_) => return ProbeResult::BadAddress, + }; + let endpoint = match endpoint.tls_config(client_tls_config) { + Ok(endpoint) => endpoint, + Err(_) => return ProbeResult::BadAddress, + }; + + match tokio::time::timeout(timeout, endpoint.connect()).await { + Ok(Ok(_channel)) => ProbeResult::Reachable, + // A failed connect covers both unreachable endpoints and TLS failures (e.g. wrong key or + // wrong epoch); tonic does not let us cleanly distinguish them here. + Ok(Err(_)) => ProbeResult::Unreachable, + Err(_) => ProbeResult::Timeout, + } +} + +/// host:port for the tonic URI, bracketing IPv6 literals. Mirrors +/// `consensus/core/src/network/mod.rs::to_host_port_str` for `/ip{4,6}|dns/.../udp/{port}`. +fn consensus_host_port(addr: &Multiaddr) -> Option { + let host = addr.hostname()?; + let port = addr.port()?; + if host.contains(':') { + Some(format!("[{host}]:{port}")) + } else { + Some(format!("{host}:{port}")) + } +} + +/// Key identifying a group in the per-group scheduling/smoothing state. +fn group_key(group: &ProbeGroup) -> GroupKey { + ( + group.peer_label.clone(), + group.endpoint_type.as_str(), + address_source_str(group.source), + ) +} + +/// Metric `peer_id` label for a consensus endpoint: the hex-encoded network public key. (Consensus +/// peers are keyed by network key, not by anemo `PeerId`.) +fn consensus_peer_label(key: &ConsensusNetworkPublicKey) -> String { + Hex::encode(key.to_bytes()) +} + +/// Test helper: compute the consensus `peer_id` metric label from raw network public key bytes, +/// matching [`consensus_peer_label`]. +#[cfg(any(test, msim))] +pub fn consensus_peer_label_for_testing(network_key_bytes: [u8; 32]) -> String { + Hex::encode(network_key_bytes) +} + +fn address_source_str(source: AddressSource) -> &'static str { + match source { + AddressSource::Admin => "admin", + AddressSource::Config => "config", + AddressSource::Discovery => "discovery", + AddressSource::Seed => "seed", + AddressSource::Chain => "chain", + } +} + +fn now_unix_seconds() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} diff --git a/crates/sui-node/src/admin.rs b/crates/sui-node/src/admin.rs index 34a7455a9c85..3379cf5524fc 100644 --- a/crates/sui-node/src/admin.rs +++ b/crates/sui-node/src/admin.rs @@ -87,6 +87,10 @@ use tracing::info; // // $ curl -X POST 'http://127.0.0.1:1337/update-endpoint?endpoint_type=p2p&id=&addresses=,' // $ curl -X POST 'http://127.0.0.1:1337/update-endpoint?endpoint_type=consensus&id=&addresses=,' +// +// Dump the address prober's latest results (full addresses + per-address outcomes) as JSON. +// +// $ curl 'http://127.0.0.1:1337/address-prober-report' const NO_TRACING_HANDLE: &str = "tracing handle not available"; const LOGGING_ROUTE: &str = "/logging"; @@ -104,6 +108,7 @@ const GET_TX_COST_ROUTE: &str = "/get-tx-cost"; const DUMP_CONSENSUS_TX_COST_ESTIMATES_ROUTE: &str = "/dump-consensus-tx-cost-estimates"; const TRAFFIC_CONTROL: &str = "/traffic-control"; const UPDATE_ENDPOINT: &str = "/update-endpoint"; +const ADDRESS_PROBER_REPORT: &str = "/address-prober-report"; const DB_SHELL_LS: &str = "/db-shell/ls"; const DB_SHELL_READ: &str = "/db-shell/read"; const DB_SHELL_DELETE: &str = "/db-shell/delete"; @@ -160,6 +165,7 @@ pub async fn run_admin_server( ) .route(TRAFFIC_CONTROL, post(traffic_control)) .route(UPDATE_ENDPOINT, post(update_endpoint)) + .route(ADDRESS_PROBER_REPORT, get(address_prober_report)) .route(DB_SHELL_LS, get(handle_ls)) .route(DB_SHELL_READ, get(handle_read)) .route(DB_SHELL_DELETE, delete(handle_delete)) @@ -662,3 +668,17 @@ async fn update_endpoint( ), ) } + +async fn address_prober_report(State(state): State>) -> (StatusCode, String) { + let Some(report) = state.node.address_prober_report().await else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "address prober is not running (node is not a validator, or the prober is disabled)\n" + .to_string(), + ); + }; + match serde_json::to_string_pretty(&report) { + Ok(json) => (StatusCode::OK, format!("{json}\n")), + Err(err) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()), + } +} diff --git a/crates/sui-node/src/lib.rs b/crates/sui-node/src/lib.rs index 98b515593781..09c75e1424bc 100644 --- a/crates/sui-node/src/lib.rs +++ b/crates/sui-node/src/lib.rs @@ -160,6 +160,7 @@ use typed_store::rocks::default_db_options; use crate::metrics::{GrpcMetrics, SuiNodeMetrics}; +pub mod address_prober; pub mod admin; pub mod db_shell; mod handle; @@ -175,6 +176,7 @@ pub struct ValidatorComponents { sui_tx_validator_metrics: Arc, admission_queue: Option, } + pub struct P2pComponents { p2p_network: Network, known_peers: HashMap, @@ -276,6 +278,9 @@ pub struct SuiNode { /// EndpointManager for updating peer network addresses. endpoint_manager: EndpointManager, + /// Handle to the discovery-shared address prober (`None` when disabled). + address_prober: Option, + backpressure_manager: Arc, _db_checkpoint_handle: Option>, @@ -966,6 +971,34 @@ impl SuiNode { None }; + if let Some(prober_config) = &config.address_prober { + prober_config.validate()?; + } + let address_prober = if Self::address_prober_enabled(&config) { + let handle = address_prober::Builder::new() + .config(config.address_prober.clone().unwrap_or_default()) + .with_metrics(&prometheus_registry) + .build() + .start( + p2p_network.clone(), + discovery_handle.sender(), + consensus_config::NetworkKeyPair::new(config.network_key_pair().copy()), + ); + // Seed the current epoch if we are starting as a validator. + if node_role.is_validator() + && let Some(components) = &validator_components + { + handle.update_epoch( + epoch_store.epoch(), + epoch_store.epoch_start_state().get_consensus_committee(), + components.consensus_manager.clone(), + ); + } + Some(handle) + } else { + None + }; + // setup shutdown channel let (shutdown_channel, _) = broadcast::channel::>(1); @@ -988,6 +1021,7 @@ impl SuiNode { end_of_epoch_channel, endpoint_manager, backpressure_manager, + address_prober, _db_checkpoint_handle: db_checkpoint_handle, @@ -1434,6 +1468,37 @@ impl SuiNode { .await } + fn address_prober_enabled(config: &NodeConfig) -> bool { + let prober_enabled = config + .address_prober + .as_ref() + .map(|c| c.enabled()) + .unwrap_or(true); + let v3_enabled = config + .p2p_config + .discovery + .as_ref() + .is_some_and(|d| d.use_get_known_peers_v3()); + prober_enabled && v3_enabled + } + + fn update_address_prober_epoch( + &self, + epoch_store: &AuthorityPerEpochStore, + consensus_manager: &Arc, + ) { + if !epoch_store.is_validator() { + return; + } + if let Some(handle) = &self.address_prober { + handle.update_epoch( + epoch_store.epoch(), + epoch_store.epoch_start_state().get_consensus_committee(), + consensus_manager.clone(), + ); + } + } + async fn start_epoch_specific_validator_components( config: &NodeConfig, state: Arc, @@ -1736,6 +1801,16 @@ impl SuiNode { &self._connection_monitor_handle } + #[cfg(any(test, msim))] + pub fn address_prober_metrics_for_testing( + &self, + ) -> std::sync::Arc { + self.address_prober + .as_ref() + .expect("address prober should be running in tests") + .metrics_for_testing() + } + pub fn node_role(&self) -> NodeRole { self.state.load_epoch_store_one_call_per_task().node_role() } @@ -1978,6 +2053,10 @@ impl SuiNode { consensus_manager.shutdown().await; info!("Consensus has shut down."); + if let Some(handle) = &self.address_prober { + handle.leave_committee(); + } + info!("Epoch store finished reconfiguration."); // No other components should be holding a strong reference to state hasher @@ -1996,30 +2075,33 @@ impl SuiNode { if new_role.runs_consensus() { info!("Restarting consensus as {new_role}"); - Some( - Self::start_epoch_specific_validator_components( - &self.config, - self.state.clone(), - consensus_adapter, - self.checkpoint_store.clone(), - new_epoch_store.clone(), - self.state_sync_handle.clone(), - self.randomness_handle.clone(), - self.randomness_receiver_handle.clone(), - consensus_manager, - consensus_store_pruner, - weak_hasher, - self.backpressure_manager.clone(), - validator_server_handle, - validator_overload_monitor_handle, - checkpoint_metrics, - self.metrics.clone(), - sui_tx_validator_metrics, - admission_queue, - new_role, - ) - .await?, + let components = Self::start_epoch_specific_validator_components( + &self.config, + self.state.clone(), + consensus_adapter, + self.checkpoint_store.clone(), + new_epoch_store.clone(), + self.state_sync_handle.clone(), + self.randomness_handle.clone(), + self.randomness_receiver_handle.clone(), + consensus_manager, + consensus_store_pruner, + weak_hasher, + self.backpressure_manager.clone(), + validator_server_handle, + validator_overload_monitor_handle, + checkpoint_metrics, + self.metrics.clone(), + sui_tx_validator_metrics, + admission_queue, + new_role, ) + .await?; + self.update_address_prober_epoch( + &new_epoch_store, + &components.consensus_manager, + ); + Some(components) } else { info!( "This node has new role {new_role} and no longer runs consensus after reconfiguration" @@ -2074,6 +2156,10 @@ impl SuiNode { .set_consensus_address_updater(components.consensus_manager.clone()); } + self.update_address_prober_epoch( + &new_epoch_store, + &components.consensus_manager, + ); Some(components) } else { None @@ -2184,6 +2270,13 @@ impl SuiNode { &self.endpoint_manager } + pub async fn address_prober_report(&self) -> Option { + match &self.address_prober { + Some(handle) => handle.probe_report().await, + None => None, + } + } + /// Get a short prefix of a digest for metric labels fn get_digest_prefix(digest: impl std::fmt::Display) -> String { let digest_str = digest.to_string(); diff --git a/crates/sui-swarm-config/src/node_config_builder.rs b/crates/sui-swarm-config/src/node_config_builder.rs index 20707a8fe696..0ad43dc31575 100644 --- a/crates/sui-swarm-config/src/node_config_builder.rs +++ b/crates/sui-swarm-config/src/node_config_builder.rs @@ -222,6 +222,7 @@ impl ValidatorConfigBuilder { NodeConfig { recent_submission_dedup_window_ms: None, + address_prober: None, protocol_key_pair: AuthorityKeyPairWithPath::new(validator.key_pair), network_key_pair: KeyPairWithPath::new(SuiKeyPair::Ed25519(validator.network_key_pair)), account_key_pair: KeyPairWithPath::new(validator.account_key_pair), @@ -600,6 +601,7 @@ impl FullnodeConfigBuilder { NodeConfig { recent_submission_dedup_window_ms: None, + address_prober: None, protocol_key_pair: AuthorityKeyPairWithPath::new(validator_config.key_pair), account_key_pair: KeyPairWithPath::new(validator_config.account_key_pair), worker_key_pair: KeyPairWithPath::new(SuiKeyPair::Ed25519( From 64c746954436aba14806966d724ea7118c6428f0 Mon Sep 17 00:00:00 2001 From: Andrew Schran Date: Tue, 14 Jul 2026 15:55:51 -0400 Subject: [PATCH 2/2] fail-fast on bad config --- crates/sui-node/src/lib.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/sui-node/src/lib.rs b/crates/sui-node/src/lib.rs index 09c75e1424bc..5ff4cf283f18 100644 --- a/crates/sui-node/src/lib.rs +++ b/crates/sui-node/src/lib.rs @@ -473,6 +473,11 @@ impl SuiNode { registry_service: RegistryService, server_version: ServerVersion, ) -> Result> { + // Fail fast on config errors before starting any node components. + if let Some(prober_config) = &config.address_prober { + prober_config.validate()?; + } + NodeConfigMetrics::new(®istry_service.default_registry()).record_metrics(&config); let mut config = config.clone(); if config.supported_protocol_versions.is_none() { @@ -971,9 +976,6 @@ impl SuiNode { None }; - if let Some(prober_config) = &config.address_prober { - prober_config.validate()?; - } let address_prober = if Self::address_prober_enabled(&config) { let handle = address_prober::Builder::new() .config(config.address_prober.clone().unwrap_or_default())