Mixnode stress testing (#6575)

* Squashing the mix stress testing branch (#6575) reduced chain watcher per block log severity update network monitors contract semver to 1.0.0 fix build issues fix mixnet client dropping initial packet on egress reconnection adjusted logs for network monitor agent changed default testing interval to 2h refresh NM contract information explicit return type for batch submission for mixnet listener task to get scheduled before beginning connectivity test make sure to always use canonical ip for network monitor noise keys feat: NMv3: make agents decide egress port (#6746) add config v12->v13 config migration for nym nodes fix formatting in wallet types simplified client config creation remove other swagger redirect removed swagger redirect on /swagger/ route log version info on startup add workflows, contract address, and dockerfile bugfix: use correct endpoints when setting up orchestrator (#6733) clippy adjust DEFAULT_MIN_STRESS_TESTED_NODES ratio expose route with new performance metrics fixes and additional docs use stress testing scores stub for usage of stress testing scores stub traits added new fields to nym-api config controlling usage of stress test data guard against duplicate packets prevent usage of chain_authorisation_check_max_attempts with value of 0 make sure duplicate results cant be inserted into the db submit test results from orchestrator on an interval docs and fixes nym-api side of handling result submission stubs for submitting results NM orchestrator verifying nym-api result submission permissions NM orchestrator to update announced key on startup allow NM orchestrator to announce its identity key to the contract stubs within nym-api for accepting NMv3 results added additional metrics docs bugfixes + making sure to only assign mixnode testruns fixed node refresher to only retrieve mixnodes and add additional metrics topology metrics defined basic prometheus metrics authorised endpoint for returning prometheus data create initial stub for prometheus metrics post rebasing fixes adjusted routes missing implementation for storage getters a lot of new stubs and db accessors stubs for results endpoints update utoipa tags for agent rountes shared auth between metrics and results moved stale results eviction into the interval.tick branch refactor and comments create background process to evict stale data include sphinx packet delay as part of the stats fix mock construction add median to the calculated latency distribution remove unused imports cleanup performing testrun and submitting the results assigning testruns to requesting agents basic stub for http server for the NMv3 orchestrator chore: rename existing 'NetworkMonitorAgent' to 'NodeStressTester' make sure to use canonical ips within the noise config fixed contract tests cargo fmt additional comments and unit tests contract and nym-node support of NM agents being run on the same host basic unit tests refactoring make agents retrieve mix port assignment from the orchestrator provide sensible defaults to CLI arguments stub the initial structure for the agent chore: remove redundant import missed tick behaviour removed redundant mutex removed redundant try_get_client reuse existing constant for default nymnode port add node refresher for periodic scraping of bonded nym-node details - NodeRefresher periodically queries the mixnet contract for all bonded nodes and probes each node's HTTP API for host information, sphinx keys, noise keys, and key rotation IDs - Extract NymNodeApiClientRetriever into nym-node-requests with port probing, identity verification, and host information signature checking - Add clone_query_client on NyxdClient so the refresher can hold its own query client without locking the signing client - Batch upsert for nym_node rows (single transaction instead of per-row) - Reuse the new helpers in nym-api's node_describe_cache ensure assignment of testrun begins an IMMEDIATE tx construction of the orchestrator struct initial set of cli args make sure to not assign testable nodes too often very initial database structure and cli fixed construction of RoutableNetworkMonitors remove redundant constructor for NoiseNode forbid 0-nonsense config values add type safety for test route construction moved lioness and arrayref to workspace deps fixed dockerfile build always use canonical addresses in RoutableNetworkMonitors fixed old contract formatting issues removed redundant into() call network monitor agent fixes additional logs config unit tests more docs standalone stress testing invocation further refactoring and changes refactor testing loop and return valid test result upon completion initial sending/receiving test loop generating reusable sphinx headers additional structure for receiving ingress packets initial scaffolding for NMv3 agent added validation of x25519 noise key removed unstable call to 'is_multiple_of' remove calls to from_octets as they're unavailable in pre 1.91 additional docs/comments propagating noise information about NM for mixnet routing pass full socket address of the agent into the contract storage feat: store noise keys alongside ip addresses within the contract removed redundant comment ensure NM packets can only go to NM PR review comments added additional docs allow NM to replay packets + fix replay prometheus metrics propagate information about nm agent to connection handler updated nym-node config migration feat: introduced nym-node websocket subscription for keeping updated list of NM agents allow admin to also revoke monitor agents remove agents upon orchestrator removal fixed schema generation and regenerated the contract schema removed rustc restriction on contracts-common added client methods for interacting with the contract added unit tests for contract methods implemented logic of the network monitors contract create initial structure for network monitors contract start mix stress testing topic branch * make nym-node default to the new blockstream rpc/ws node cluster * reduced mixnet-client log severity * set network monitors contract address for mainnet
2026-05-22 09:43:20 +01:00
parent e5cd9fd69e
commit 46c67440bb
215 changed files with 19059 additions and 1858 deletions
@@ -0,0 +1,56 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use nym_topology::{EpochRewardedSet, NymTopology, NymTopologyMetadata};
+use nym_validator_client::nym_nodes::SemiSkimmedNodeV1;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+pub(crate) mod refresher;
+pub(crate) mod topology_provider;
+
+#[derive(Clone)]
+pub(crate) struct CachedNetwork {
+    inner: Arc<RwLock<CachedNetworkInner>>,
+}
+
+impl CachedNetwork {
+    fn new_empty() -> Self {
+        CachedNetwork {
+            inner: Arc::new(RwLock::new(CachedNetworkInner {
+                rewarded_set: Default::default(),
+                topology_metadata: Default::default(),
+                network_nodes: vec![],
+            })),
+        }
+    }
+
+    async fn network_topology(&self, min_mix_performance: u8) -> NymTopology {
+        let network_guard = self.inner.read().await;
+
+        NymTopology::new(
+            network_guard.topology_metadata,
+            network_guard.rewarded_set.clone(),
+            Vec::new(),
+        )
+        .with_additional_nodes(
+            network_guard
+                .network_nodes
+                .iter()
+                .map(|node| &node.basic)
+                .filter(|node| {
+                    if node.supported_roles.mixnode {
+                        node.performance.round_to_integer() >= min_mix_performance
+                    } else {
+                        true
+                    }
+                }),
+        )
+    }
+}
+
+struct CachedNetworkInner {
+    rewarded_set: EpochRewardedSet,
+    topology_metadata: NymTopologyMetadata,
+    network_nodes: Vec<SemiSkimmedNodeV1>,
+}
@@ -0,0 +1,337 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+//! Background task that periodically refreshes network topology and routing information.
+//!
+//! # Responsibilities
+//!
+//! - Fetches the current Nym node list from nym-api
+//! - Resolves pending (unknown) IP addresses from ingress mixnet packets
+//! - Updates routing filter with allowed/denied node lists
+//! - Maintains Noise protocol key mappings
+//! - Ensures minimally routable topology at startup
+//!
+//! # Refresh Strategy
+//!
+//! Two independent refresh cycles run in parallel:
+//! 1. **Full refresh** (typically every 60s): Complete network state from nym-api
+//! 2. **Pending check** (typically every 5s): Quick resolution of recently seen unknown IPs
+//!
+//! The pending check uses an optimized nym-api endpoint when available, falling back to
+//! full refresh if the endpoint is not supported.
+
+use crate::error::NymNodeError;
+use crate::node::lp::directory::LpNodes;
+use crate::node::nym_apis_client::NymApisClient;
+use crate::node::routing_filter::network_filter::NetworkRoutingFilter;
+use crate::node::shared_network::CachedNetwork;
+use nym_node_metrics::prometheus_wrapper::{PROMETHEUS_METRICS, PrometheusMetric};
+use nym_noise::config::{NoiseNetworkView, NoiseNode};
+use nym_task::ShutdownToken;
+use nym_topology::provider_trait::ToTopologyMetadata;
+use nym_validator_client::ValidatorClientError;
+use std::collections::{HashMap, HashSet};
+use std::time::Duration;
+use tokio::time::{Instant, interval, sleep};
+use tracing::{debug, trace};
+
+pub struct NetworkRefresher {
+    config: NetworkRefresherConfig,
+    client: NymApisClient,
+    shutdown_token: ShutdownToken,
+
+    network: CachedNetwork,
+    routing_filter: NetworkRoutingFilter,
+    noise_view: NoiseNetworkView,
+    lp_nodes: LpNodes,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct NetworkRefresherConfig {
+    full_refresh_interval: Duration,
+    pending_check_interval: Duration,
+    max_startup_waiting_period: Duration,
+    min_mix_performance: u8,
+}
+
+impl NetworkRefresherConfig {
+    pub fn new(
+        full_refresh_interval: Duration,
+        pending_check_interval: Duration,
+        max_startup_waiting_period: Duration,
+        min_mix_performance: u8,
+    ) -> Self {
+        Self {
+            full_refresh_interval,
+            pending_check_interval,
+            max_startup_waiting_period,
+            min_mix_performance,
+        }
+    }
+}
+
+impl NetworkRefresher {
+    pub(crate) async fn initialise_new(
+        config: NetworkRefresherConfig,
+        client: NymApisClient,
+        routing_filter: NetworkRoutingFilter,
+        noise_view: NoiseNetworkView,
+        shutdown_token: ShutdownToken,
+    ) -> Result<Self, NymNodeError> {
+        let mut this = NetworkRefresher {
+            config,
+            client,
+            shutdown_token,
+            network: CachedNetwork::new_empty(),
+            routing_filter,
+            noise_view,
+            lp_nodes: Default::default(),
+        };
+
+        this.obtain_initial_network(
+            config.max_startup_waiting_period,
+            config.min_mix_performance,
+        )
+        .await?;
+        Ok(this)
+    }
+
+    /// Attempt to resolve pending (unknown) IP addresses that were recently seen in packet traffic.
+    ///
+    /// # Algorithm
+    ///
+    /// 1. Collect all pending IPs that need resolution (lock required)
+    /// 2. Short-circuit if all pending IPs are already in allowed/denied sets (race condition check)
+    /// 3. Try optimised nym-api query: `query_nym_nodes_addresses(ips)` for bulk lookup
+    ///    - If supported: Get immediate results, update allowed/denied sets, clear pending queue
+    ///    - If not supported (404): Fall back to full network refresh
+    ///
+    /// # Performance
+    ///
+    /// The optimised query avoids fetching the entire network topology (~1000 nodes) when we only
+    /// need to check a handful of IPs. This is crucial for minimising latency between when a new
+    /// node joins and when it can successfully route packets.
+    ///
+    /// # Fallback Behaviour
+    ///
+    /// If nym-api doesn't support the optimised endpoint, we do a full refresh. This is acceptable
+    /// because:
+    /// - Full refresh is needed anyway for topology updates
+    /// - The pending queue typically has <10 entries
+    /// - This only affects older nym-api versions
+    async fn inspect_pending(&mut self) {
+        let to_resolve = self.routing_filter.pending.nodes().await;
+
+        // no pending requests to resolve
+        if to_resolve.is_empty() {
+            return;
+        }
+
+        let mut allowed = self.routing_filter.allowed_nodes_copy();
+        let mut denied = self.routing_filter.denied_nodes_copy();
+
+        // short circuit: check if the pending nodes are not already resolved
+        // (it could happen due to lack of full sync between pending lock and arcswap(s))
+        if to_resolve
+            .iter()
+            .all(|p| allowed.contains(p) || denied.contains(p))
+        {
+            return;
+        }
+
+        // 1. attempt to use the new nym-api query to get information just by ips
+        let nodes = to_resolve.into_iter().collect();
+        if let Ok(res) = self.client.query_nym_nodes_addresses(nodes).await {
+            for (ip, maybe_id) in res.existence {
+                if maybe_id.is_some() {
+                    allowed.insert(ip);
+                } else {
+                    denied.insert(ip);
+                }
+            }
+
+            self.routing_filter.resolved.swap_allowed(allowed);
+            self.routing_filter.resolved.swap_denied(denied);
+            self.routing_filter.pending.clear().await;
+            return;
+        }
+
+        // 2. we assume nym-api doesn't support that query yet - we have to do the full refresh
+        self.refresh_network_nodes().await;
+    }
+
+    async fn refresh_network_nodes_inner(&mut self) -> Result<(), ValidatorClientError> {
+        let rewarded_set = self.client.rewarded_set().await?;
+        let res = self.client.current_nymnodes().await?;
+        let nodes = res.nodes;
+        let metadata = res.metadata;
+
+        // collect all known/allowed nodes information
+        let known_nodes = nodes
+            .iter()
+            .flat_map(|n| n.basic.ip_addresses.iter())
+            .copied()
+            .collect::<HashSet<_>>();
+
+        let pending = self.routing_filter.pending.nodes().await;
+        let mut current_denied = self.routing_filter.denied_nodes_copy();
+
+        for allowed in &known_nodes {
+            // if some node has become known, it should be removed from the denied set
+            current_denied.remove(allowed);
+        }
+
+        // any pending node, if not present in the new set of allowed nodes, should be added in the denied set
+        for pending_node in pending {
+            if !known_nodes.contains(&pending_node) {
+                current_denied.insert(pending_node);
+            }
+        }
+
+        self.routing_filter.resolved.swap_allowed(known_nodes);
+        self.routing_filter.resolved.swap_denied(current_denied);
+        self.routing_filter.pending.clear().await;
+
+        let noise_update_permit = self.noise_view.get_update_permit().await;
+        let current_nodes = self.noise_view.all_nodes();
+
+        // update noise Nodes
+        let mut new_noise_nodes = HashMap::new();
+
+        // Rebuild the noise map in two passes to respect the two independent data sources:
+        //
+        // 1. Preserve existing NM agent entries — these come from blockchain events processed
+        //    by `NetworkMonitorAgentsModule` and are NOT available from nym-api. Dropping them
+        //    here would leave agents unable to perform noise handshakes until their
+        //    registration is renewed.
+        // Keys are canonicalised so the map matches what supports_noise/get_noise_key look up.
+        for (ip, node) in current_nodes {
+            if !node.is_nym_node() {
+                new_noise_nodes.insert(ip.to_canonical(), node);
+            }
+        }
+
+        // 2. Replace all nym-node entries with fresh data from nym-api.
+        //    (Stale nym-node keys are safe to overwrite — the source of truth is always nym-api)
+        for node in &nodes {
+            let Some(noise_key) = node.x25519_noise_versioned_key else {
+                continue;
+            };
+            let entry = NoiseNode::new_nym_node(noise_key);
+            for ip_addr in &node.basic.ip_addresses {
+                new_noise_nodes.insert(ip_addr.to_canonical(), entry.clone());
+            }
+        }
+        self.noise_view
+            .swap_view(noise_update_permit, new_noise_nodes);
+        debug!("unimplemented: update LP nodes data - will work very similarly to noise nodes");
+
+        let mut network_guard = self.network.inner.write().await;
+        network_guard.topology_metadata = metadata.to_topology_metadata();
+        network_guard.network_nodes = nodes;
+        network_guard.rewarded_set = rewarded_set;
+
+        Ok(())
+    }
+
+    async fn refresh_network_nodes(&mut self) {
+        let timer =
+            PROMETHEUS_METRICS.start_timer(PrometheusMetric::ProcessTopologyQueryResolutionLatency);
+
+        if self.refresh_network_nodes_inner().await.is_err() {
+            // don't use the histogram observation as some queries didn't complete
+            if let Some(obs) = timer {
+                obs.stop_and_discard();
+            }
+        }
+    }
+
+    /// Block until we obtain a minimally routable network topology at startup.
+    ///
+    /// # Startup Sequence
+    ///
+    /// 1. Query nym-api for full network state (nodes + rewarded set)
+    /// 2. Check if topology has sufficient mixnodes in each layer for routing
+    /// 3. If not routable: wait `STARTUP_REFRESH_INTERVAL` (30s) and retry
+    /// 4. If still not routable after `max_startup_waiting_period`: return error and abort startup
+    ///
+    /// # Why Block Startup?
+    ///
+    /// We MUST have a routable topology before accepting packets, otherwise:
+    /// - Packets would be dropped due to incomplete routing tables
+    /// - The node would appear non-functional to the network
+    /// - Internal service providers would be unable to construct return packets
+    ///
+    /// # Timeout Behavior
+    ///
+    /// If the network remains non-routable for too long, this indicates:
+    /// - Network-wide outage (not enough mixnodes online)
+    /// - nym-api connectivity issues
+    /// - Configuration error (wrong nym-api URL)
+    ///
+    /// In any case, the node should NOT start packet processing.
+    pub(crate) async fn obtain_initial_network(
+        &mut self,
+        max_startup_waiting_period: Duration,
+        min_mix_performance: u8,
+    ) -> Result<(), NymNodeError> {
+        // make it configurable
+        const STARTUP_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
+
+        let start = Instant::now();
+
+        loop {
+            self.refresh_network_nodes_inner()
+                .await
+                .map_err(|source| NymNodeError::InitialTopologyQueryFailure { source })?;
+
+            let topology = self.network.network_topology(min_mix_performance).await;
+            if topology.is_minimally_routable() {
+                return Ok(());
+            }
+
+            if start.elapsed() > max_startup_waiting_period {
+                return Err(NymNodeError::InitialTopologyTimeout);
+            }
+
+            sleep(STARTUP_REFRESH_INTERVAL).await;
+        }
+    }
+
+    pub(crate) fn cached_network(&self) -> CachedNetwork {
+        self.network.clone()
+    }
+
+    pub(crate) fn lp_nodes(&self) -> LpNodes {
+        self.lp_nodes.clone()
+    }
+
+    pub(crate) async fn run(&mut self) {
+        let mut full_refresh_interval = interval(self.config.full_refresh_interval);
+        full_refresh_interval.reset();
+
+        let mut pending_check_interval = interval(self.config.pending_check_interval);
+        pending_check_interval.reset();
+
+        loop {
+            tokio::select! {
+                biased;
+                _ = self.shutdown_token.cancelled() => {
+                    trace!("NetworkRefresher: Received shutdown");
+                    break;
+                }
+                _ = pending_check_interval.tick() => {
+                    self.inspect_pending().await;
+                }
+                _ = full_refresh_interval.tick() => {
+                    self.refresh_network_nodes().await;
+                }
+            }
+        }
+        trace!("NetworkRefresher: Exiting");
+    }
+
+    pub(crate) fn start(mut self) {
+        tokio::spawn(async move { self.run().await });
+    }
+}
@@ -0,0 +1,79 @@
+// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
+use crate::node::shared_network::CachedNetwork;
+use async_trait::async_trait;
+use nym_crypto::asymmetric::ed25519;
+use nym_topology::{EntryDetails, NodeId, NymTopology, Role, RoutingNode, TopologyProvider};
+use std::net::SocketAddr;
+use std::ops::Deref;
+use std::sync::Arc;
+use tracing::debug;
+
+const LOCAL_NODE_ID: NodeId = 1234567890;
+
+pub(crate) struct LocalGatewayNode {
+    pub(crate) active_sphinx_keys: ActiveSphinxKeys,
+    pub(crate) mix_host: SocketAddr,
+    pub(crate) identity_key: ed25519::PublicKey,
+    pub(crate) entry: EntryDetails,
+}
+
+impl LocalGatewayNode {
+    pub(crate) fn to_routing_node(&self) -> RoutingNode {
+        RoutingNode {
+            node_id: LOCAL_NODE_ID,
+            mix_host: self.mix_host,
+            entry: Some(self.entry.clone()),
+            identity_key: self.identity_key,
+            sphinx_key: self.active_sphinx_keys.primary().deref().x25519_pubkey(),
+            supported_roles: nym_topology::SupportedRoles {
+                mixnode: false,
+                mixnet_entry: true,
+                mixnet_exit: true,
+            },
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct CachedTopologyProvider {
+    gateway_node: Arc<LocalGatewayNode>,
+    cached_network: CachedNetwork,
+    min_mix_performance: u8,
+}
+
+impl CachedTopologyProvider {
+    pub(crate) fn new(
+        gateway_node: LocalGatewayNode,
+        cached_network: CachedNetwork,
+        min_mix_performance: u8,
+    ) -> Self {
+        CachedTopologyProvider {
+            gateway_node: Arc::new(gateway_node),
+            cached_network,
+            min_mix_performance,
+        }
+    }
+}
+
+#[async_trait]
+impl TopologyProvider for CachedTopologyProvider {
+    async fn get_new_topology(&mut self) -> Option<NymTopology> {
+        let self_node = self.gateway_node.identity_key;
+
+        let mut topology = self
+            .cached_network
+            .network_topology(self.min_mix_performance)
+            .await;
+
+        if !topology.has_node(self.gateway_node.identity_key) {
+            debug!("{self_node} didn't exist in topology. inserting it.",);
+            topology.insert_node_details(self.gateway_node.to_routing_node());
+        }
+        topology.force_set_active(LOCAL_NODE_ID, Role::EntryGateway);
+
+        Some(topology)
+    }
+}