diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/cli/env.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/cli/env.rs index 6ee20dc7d20..75d7c22faef 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/cli/env.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/cli/env.rs @@ -11,6 +11,16 @@ pub mod vars { "NYM_NETWORK_MONITOR_ORCHESTRATOR_METRICS_AND_RESULTS_TOKEN"; pub const NYM_NETWORK_MONITOR_TEST_INTERVAL_ARG: &str = "NYM_NETWORK_MONITOR_TEST_INTERVAL"; pub const NYM_NETWORK_MONITOR_TEST_TIMEOUT_ARG: &str = "NYM_NETWORK_MONITOR_TEST_TIMEOUT"; + pub const NYM_NETWORK_MONITOR_LIVENESS_ENABLED_ARG: &str = + "NYM_NETWORK_MONITOR_LIVENESS_ENABLED"; + pub const NYM_NETWORK_MONITOR_LIVENESS_TEST_INTERVAL_ARG: &str = + "NYM_NETWORK_MONITOR_LIVENESS_TEST_INTERVAL"; + pub const NYM_NETWORK_MONITOR_LIVENESS_TEST_TIMEOUT_ARG: &str = + "NYM_NETWORK_MONITOR_LIVENESS_TEST_TIMEOUT"; + pub const NYM_NETWORK_MONITOR_LIVENESS_MIXNODE_WAVE_SIZE_ARG: &str = + "NYM_NETWORK_MONITOR_LIVENESS_MIXNODE_WAVE_SIZE"; + pub const NYM_NETWORK_MONITOR_LIVENESS_GATEWAY_WAVE_SIZE_ARG: &str = + "NYM_NETWORK_MONITOR_LIVENESS_GATEWAY_WAVE_SIZE"; pub const NYM_NETWORK_MONITOR_HTTP_SERVER_BIND_ADDRESS_ARG: &str = "NYM_NETWORK_MONITOR_HTTP_SERVER_BIND_ADDRESS"; pub const NYM_NETWORK_MONITOR_NYM_API_ENDPOINT_ARG: &str = diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/cli/run_orchestrator.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/cli/run_orchestrator.rs index 597b65e6bc8..7a4b910b3a5 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/cli/run_orchestrator.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/cli/run_orchestrator.rs @@ -3,7 +3,7 @@ use super::env::vars::*; use crate::orchestrator::NetworkMonitorOrchestrator; -use crate::orchestrator::config::Config; +use crate::orchestrator::config::{Config, LivenessConfig}; use anyhow::{Context, anyhow, bail}; use nym_crypto::asymmetric::ed25519; use nym_validator_client::nyxd::bip39; @@ -32,10 +32,35 @@ pub(crate) struct Args { test_interval: Duration, /// Maximum time a single test run is allowed to run before being considered timed out - /// (e.g. `5m`). + /// (e.g. `5m`). Used as the stress kind's lease budget. #[clap(long, env = NYM_NETWORK_MONITOR_TEST_TIMEOUT_ARG, value_parser = humantime::parse_duration, default_value = "5m")] test_timeout: Duration, + /// Whether liveness testing may be assigned to agents (e.g. `--liveness-enabled false`). + /// Takes an explicit value rather than being a bare flag, so that a deployment can switch + /// liveness off through the environment without a redeploy. + #[clap(long, env = NYM_NETWORK_MONITOR_LIVENESS_ENABLED_ARG, action = clap::ArgAction::Set, default_value_t = true)] + liveness_enabled: bool, + + /// How often each node should be liveness-tested, per role (e.g. `15m`). + #[clap(long, env = NYM_NETWORK_MONITOR_LIVENESS_TEST_INTERVAL_ARG, value_parser = humantime::parse_duration, default_value = "15m")] + liveness_test_interval: Duration, + + /// Maximum time a single liveness wave is allowed to run before its targets are released for + /// reassignment (e.g. `1m`). Bounds ONE concurrent wave, not the sum over its targets, and has + /// to cover the slower of the two probes, which is the gateway one. + #[clap(long, env = NYM_NETWORK_MONITOR_LIVENESS_TEST_TIMEOUT_ARG, value_parser = humantime::parse_duration, default_value = "1m")] + liveness_test_timeout: Duration, + + /// Maximum number of targets handed out in a single mixnode liveness assignment. + #[clap(long, env = NYM_NETWORK_MONITOR_LIVENESS_MIXNODE_WAVE_SIZE_ARG, default_value = "100")] + liveness_mixnode_wave_size: NonZeroUsize, + + /// Maximum number of targets handed out in a single gateway liveness assignment. Lower than + /// the mixnode wave, since each target costs the agent a live client session. + #[clap(long, env = NYM_NETWORK_MONITOR_LIVENESS_GATEWAY_WAVE_SIZE_ARG, default_value = "50")] + liveness_gateway_wave_size: NonZeroUsize, + /// HTTP address to bind the HTTP server to (e.g. `0.0.0.0:8080`). #[clap(long, env = NYM_NETWORK_MONITOR_HTTP_SERVER_BIND_ADDRESS_ARG, default_value = "0.0.0.0:8080")] http_server_bind_address: SocketAddr, @@ -68,7 +93,7 @@ pub(crate) struct Args { node_refresh_rate: Duration, /// Timeout for querying a single node for its detailed information (sphinx key, noise key, - /// etc.). Queries that exceed this budget leave the corresponding fields as `NULL` + /// etc.). A node that exceeds this budget keeps whatever an earlier cycle learned about it /// (e.g. `10s`). #[clap(long, env = NYM_NETWORK_MONITOR_NODE_INFO_QUERY_TIMEOUT_ARG, value_parser = humantime::parse_duration, default_value = "10s")] node_info_query_timeout: Duration, @@ -129,6 +154,13 @@ impl Args { http_server_bind_address: self.http_server_bind_address, test_interval: self.test_interval, test_timeout: self.test_timeout, + liveness: LivenessConfig { + enabled: self.liveness_enabled, + test_interval: self.liveness_test_interval, + test_timeout: self.liveness_test_timeout, + mixnode_wave_size: self.liveness_mixnode_wave_size.get(), + gateway_wave_size: self.liveness_gateway_wave_size.get(), + }, database_path: self.database_path.clone(), node_refresh_rate: self.node_refresh_rate, node_info_query_timeout: self.node_info_query_timeout, @@ -217,3 +249,99 @@ pub(crate) async fn execute(mut args: Args) -> anyhow::Result<()> { orchestrator.run().await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use clap::Parser; + + // `Args` is a subcommand's argument group, so it needs a parser root to be exercised on its own + #[derive(Parser)] + struct TestCli { + #[clap(flatten)] + args: Args, + } + + /// The arguments with no default, which every parse has to supply. The mnemonic is the + /// all-zeros bip39 test vector - it still has to pass checksum validation to parse. + const REQUIRED: &[&str] = &[ + "run-orchestrator", + "--agents-token", + "agents-token", + "--metrics-and-results-token", + "metrics-token", + "--nym-api-endpoint", + "https://nym-api.example.com/api", + "--mnemonic", + "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about", + "--database-path", + "/var/lib/nym-network-monitor/db.sqlite", + "--private-key", + "6HRy7XkUqDPr1JdKPKGdBnDaKvbNJhCTAqrnQNVJEmS7", + ]; + + fn parse(overrides: &[&str]) -> LivenessConfig { + let argv: Vec<&str> = REQUIRED.iter().chain(overrides.iter()).copied().collect(); + TestCli::try_parse_from(argv) + .expect("failed to parse arguments") + .args + .build_orchestrator_config() + .expect("failed to build the config") + .liveness + } + + #[test] + fn liveness_knobs_carry_their_documented_defaults() { + let liveness = parse(&[]); + assert!(liveness.enabled); + assert_eq!(liveness.test_interval, Duration::from_secs(15 * 60)); + assert_eq!(liveness.test_timeout, Duration::from_secs(60)); + + // the two waves are sized independently, the gateway one lower because each of its targets + // costs a live client session rather than a Noise connection + assert_eq!(liveness.mixnode_wave_size, 100); + assert_eq!(liveness.gateway_wave_size, 50); + } + + // every one of these values is provisional, so being able to move it without a code change is + // itself a requirement. the enable flag is the load-bearing case: as a bare presence flag it + // would parse and then be impossible to switch off, which is the one thing it exists to do + #[test] + fn every_liveness_knob_is_overridable() { + let liveness = parse(&[ + "--liveness-enabled", + "false", + "--liveness-test-interval", + "3m", + "--liveness-test-timeout", + "30s", + "--liveness-mixnode-wave-size", + "7", + "--liveness-gateway-wave-size", + "3", + ]); + + assert!(!liveness.enabled); + assert_eq!(liveness.test_interval, Duration::from_secs(3 * 60)); + assert_eq!(liveness.test_timeout, Duration::from_secs(30)); + assert_eq!(liveness.mixnode_wave_size, 7); + assert_eq!(liveness.gateway_wave_size, 3); + } + + // an assignment with no targets is not a valid assignment, so an empty wave is rejected at + // parse time rather than producing one at dispatch. asserted per flag: the two waves are + // separate knobs, so one of them keeping its NonZero parser proves nothing about the other + #[test] + fn a_zero_wave_size_is_rejected() { + for flag in [ + "--liveness-mixnode-wave-size", + "--liveness-gateway-wave-size", + ] { + let argv: Vec<&str> = REQUIRED.iter().copied().chain([flag, "0"]).collect(); + assert!( + TestCli::try_parse_from(argv).is_err(), + "{flag} accepted an empty wave" + ); + } + } +} diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/http/api/v1/agent/mod.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/http/api/v1/agent/mod.rs index deed7a6733a..f15fd603f95 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/http/api/v1/agent/mod.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/http/api/v1/agent/mod.rs @@ -187,7 +187,7 @@ async fn request_testrun( } // 2. attempt to assign a testrun to the agent - let assignment = state.assign_next_mixnode_testrun().await?; + let assignment = state.assign_next_testrun().await?; if assignment.is_none() { PROMETHEUS_METRICS.inc(PrometheusMetric::EmptyTestrunAssignments); } else { diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/http/state.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/http/state.rs index e71e57aadfd..3f55bb28143 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/http/state.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/http/state.rs @@ -2,14 +2,18 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::http::api::v1::error::ApiError; +use crate::orchestrator::config::LivenessConfig; use crate::orchestrator::prometheus::{PROMETHEUS_METRICS, PrometheusMetric}; use crate::storage::NetworkMonitorStorage; -use crate::storage::models::{NewTestRun, TestRunMeasurement}; +use crate::storage::models::{ + AssignedTestrun, NewTestRun, PairingHead, PairingSchedule, TestKind, TestPairing, + TestRunMeasurement, TestedRole, +}; use axum::extract::FromRef; use nym_crypto::asymmetric::{ed25519, x25519}; use nym_network_monitor_orchestrator_requests::models::{ - AgentMixAddresses, MixnetProbeTarget, NymNodeData, NymNodeWithTestRun, PagedResult, Pagination, - TestRunAssignment, TestRunData, TestRunInProgressData, TestRunResult, + AgentMixAddresses, NymNodeData, NymNodeWithTestRun, PagedResult, Pagination, TestRunAssignment, + TestRunData, TestRunInProgressData, TestRunResult, }; use nym_validator_client::DirectSigningHttpRpcValidatorClient; use nym_validator_client::client::NodeId; @@ -18,7 +22,9 @@ use std::collections::HashMap; use std::collections::hash_map::Entry; use std::net::SocketAddr; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; +use strum::{EnumCount, IntoEnumIterator}; use time::OffsetDateTime; use tokio::sync::{Mutex, RwLock}; use tracing::{error, warn}; @@ -258,92 +264,230 @@ pub(crate) struct KnownAgent { pub(crate) announced: bool, } +/// Counts one dispatched assignment against its pairing, and records the wave's width where the +/// pairing has one. A stress assignment has no wave series: its width is fixed at one by the wire +/// type, so a histogram of it would carry no information. +fn emit_assignment_metrics(pairing: TestPairing, wave_size: usize) { + let (assignments, wave) = match (pairing.test_kind, pairing.tested_role) { + (TestKind::Stress, _) => (PrometheusMetric::MixnodeStressAssignments, None), + (TestKind::Liveness, TestedRole::Mixnode) => ( + PrometheusMetric::MixnodeLivenessAssignments, + Some(PrometheusMetric::MixnodeLivenessWaveSize), + ), + (TestKind::Liveness, TestedRole::Gateway) => ( + PrometheusMetric::GatewayLivenessAssignments, + Some(PrometheusMetric::GatewayLivenessWaveSize), + ), + }; + + PROMETHEUS_METRICS.inc(assignments); + if let Some(wave) = wave { + PROMETHEUS_METRICS.observe_histogram(wave, wave_size as f64); + } +} + +/// The orchestrator writes every field a probe target is built from itself, so a decoding failure +/// is corruption or a schema regression rather than anything the request did. Logged here, where +/// there is a request to answer, since the storage layer reports it as a plain error. +fn malformed_target(err: anyhow::Error) -> ApiError { + error!("could not build a probe target out of a stored node row: {err}"); + ApiError::MalformedStoredData +} + /// Coordinates test run assignment and result storage. /// -/// Wraps the underlying [`NetworkMonitorStorage`] and applies the configured -/// `testrun_staleness_age` when deciding which nodes are eligible for testing. +/// Wraps the underlying [`NetworkMonitorStorage`] and holds each kind's cadence and lease, deciding +/// which kind an agent asking for work is handed. #[derive(Clone)] pub(crate) struct TestrunManager { - /// Minimum time that must elapse after a node's last test before it becomes + /// Minimum time that must elapse after a node's last stress test before it becomes /// eligible for another one. Passed to the storage layer as a staleness gate. testrun_staleness_age: Duration, - /// How long a dispatched run holds its node before the lease expires and the slot is freed - /// for reassignment. Materialised onto each `testrun_in_progress` row at dispatch. + /// How long a dispatched stress run holds its node before the lease expires and the slot is + /// freed for reassignment. Materialised onto each `testrun_in_progress` row at dispatch. testrun_lease_budget: Duration, + + /// The liveness kind's own cadence, lease and per-role wave sizes. + liveness: LivenessConfig, + + /// Which kind gets first refusal on the next request. Shared rather than owned per clone: + /// [`AppState`] is cloned per request, so a plain field would hand every request the same kind. + kind_cursor: Arc, } impl TestrunManager { - /// Selects the most stale idle mixnode and atomically marks it as having a test - /// in progress. Returns `None` if no mixnode is currently eligible. - async fn assign_next_mixnode_testrun( + /// Hands out one assignment, rotating which kind is offered the request first. + /// + /// The rotation is over KINDS only, so a future kind joins it as one variant rather than a + /// policy rewrite, and it advances per request so that neither cadence starves the other: stress + /// is un-waved and so needs the majority of assignments, while liveness comes due eight times as + /// often. A kind that is disabled or has nothing due falls through to the next, which is what + /// keeps a drained kind from wasting the request. + async fn assign_next_testrun( &self, storage: &NetworkMonitorStorage, ) -> Result, ApiError> { - let node_to_test = match storage - .assign_next_mixnode_testrun(self.testrun_staleness_age, self.testrun_lease_budget) + let first = self.kind_cursor.fetch_add(1, Ordering::Relaxed) % TestKind::COUNT; + + for kind in TestKind::iter().cycle().skip(first).take(TestKind::COUNT) { + if kind == TestKind::Liveness && !self.liveness.enabled { + continue; + } + + if let Some(assignment) = self.assign_for_kind(storage, kind).await? { + return Ok(Some(assignment)); + } + } + + Ok(None) + } + + /// Dispatches whichever of a kind's pairings is furthest behind, or `None` if none of them has + /// work. + /// + /// The role is deliberately not a policy decision: it falls out of the staleness ordering, so + /// the two liveness roles interleave by need - serving one advances its own staleness position + /// and hands the next turn to the other. + async fn assign_for_kind( + &self, + storage: &NetworkMonitorStorage, + kind: TestKind, + ) -> Result, ApiError> { + let Some(pairing) = self.most_overdue_pairing(storage, kind).await? else { + return Ok(None); + }; + + let targets = match storage + .assign_next_testruns(&self.schedule_for(pairing)) .await { - Ok(node) => node, + Ok(targets) => targets, Err(err) => { error!("testrun assignment storage failure: {err}"); return Err(ApiError::StorageFailure); } }; - let Some(assigned) = node_to_test else { - return Ok(None); - }; - let node_ips = assigned.node.announced_ips(); - let tested_ip = assigned.tested_ip; - let node = assigned.node.inner; + let assignment = self.build_assignment(pairing, &targets)?; - let Ok(identity_key) = node.identity_key.parse() else { - return Err(ApiError::MalformedStoredData); - }; + // counted only once the assignment is built, so the series count work actually handed out + // rather than nodes that were locked and then dropped as malformed + if assignment.is_some() { + emit_assignment_metrics(pairing, targets.len()); + } - let (Some(address), Some(noise_key), Some(sphinx_key), Some(key_rotation)) = ( - node.mixnet_socket_address, - node.noise_key, - node.sphinx_key, - node.key_rotation_id, - ) else { - // this should never happen as the db query should ignore entries where those fields are set to NULL - error!( - "database inconsistency - attempted to assign node {} for stress testing, but we don't have its complete data", - node.node_id - ); - return Err(ApiError::StorageFailure); - }; + Ok(assignment) + } - // the stored socket address only contributes the mix port - the address to test comes from - // the rotation over everything the node announced - let Ok(node_address) = address.parse::() else { - return Err(ApiError::MalformedStoredData); - }; - let node_address = SocketAddr::new(tested_ip, node_address.port()); + /// The pairing of `kind` whose next node has waited longest, or `None` when none of them has an + /// eligible node. A tie leaves the kind's first pairing in place, so a fresh database - where + /// every pairing is equally never-tested - drains deterministically rather than arbitrarily. + async fn most_overdue_pairing( + &self, + storage: &NetworkMonitorStorage, + kind: TestKind, + ) -> Result, ApiError> { + // a kind owning a single pairing has nothing to choose between, and the assignment itself + // reports whether that pairing has work + if let [only] = kind.pairings() { + return Ok(Some(*only)); + } - let Ok(noise_key) = noise_key.parse() else { - return Err(ApiError::MalformedStoredData); - }; + let mut most_overdue: Option<(TestPairing, PairingHead)> = None; + for &pairing in kind.pairings() { + let head = match storage + .peek_pairing_head(pairing, self.staleness_age(kind)) + .await + { + Ok(head) => head, + Err(err) => { + error!("pairing head lookup storage failure: {err}"); + return Err(ApiError::StorageFailure); + } + }; - let Ok(sphinx_key) = sphinx_key.parse() else { - return Err(ApiError::MalformedStoredData); - }; + let Some(head) = head else { + continue; + }; + // strictly more overdue, so an equally overdue pairing does not displace the incumbent + if most_overdue.is_none_or(|(_, incumbent)| head < incumbent) { + most_overdue = Some((pairing, head)); + } + } - // only the stress kind is ever assigned today; the liveness variants stay unconstructed - // until per-kind scheduling lands - Ok(Some(TestRunAssignment::MixnodeStress(Box::new( - MixnetProbeTarget { - node_id: node.node_id as u32, - identity_key, - node_address, - node_ips, - noise_key, - sphinx_key, - key_rotation_id: key_rotation as u32, + Ok(most_overdue.map(|(pairing, _)| pairing)) + } + + /// How long a node rests before `kind` is due against it again. + fn staleness_age(&self, kind: TestKind) -> Duration { + match kind { + TestKind::Stress => self.testrun_staleness_age, + TestKind::Liveness => self.liveness.test_interval, + } + } + + /// The cadence, lease and wave size to dispatch `pairing` with. + fn schedule_for(&self, pairing: TestPairing) -> PairingSchedule { + match pairing.test_kind { + TestKind::Stress => { + PairingSchedule::stress(self.testrun_staleness_age, self.testrun_lease_budget) + } + TestKind::Liveness => PairingSchedule { + pairing, + staleness_age: self.liveness.test_interval, + lease_budget: self.liveness.test_timeout, + wave_size: self.liveness.wave_size(pairing.tested_role), }, - )))) + } + } + + /// Wraps the locked targets in the assignment shape their pairing is carried in. + /// + /// An empty assignment is not a valid assignment - "no work" is an absent assignment on the + /// response - so a wave that ends up empty reads as no work rather than being sent as one. + fn build_assignment( + &self, + pairing: TestPairing, + targets: &[AssignedTestrun], + ) -> Result, ApiError> { + if targets.is_empty() { + return Ok(None); + } + + let assignment = match (pairing.test_kind, pairing.tested_role) { + (TestKind::Stress, _) => { + // the stress variant carries exactly one target, and its schedule asks for exactly + // one. a surplus would mean the two have drifted apart, and the nodes past the first + // are already locked, so they would sit leased without ever reaching an agent + if targets.len() > 1 { + error!( + "a stress assignment selected {} targets - dispatching the first, the rest stay locked until their lease expires", + targets.len() + ); + } + + TestRunAssignment::MixnodeStress(Box::new( + targets[0].mixnet_probe_target().map_err(malformed_target)?, + )) + } + (TestKind::Liveness, TestedRole::Mixnode) => TestRunAssignment::MixnodeLiveness( + targets + .iter() + .map(AssignedTestrun::mixnet_probe_target) + .collect::>() + .map_err(malformed_target)?, + ), + (TestKind::Liveness, TestedRole::Gateway) => TestRunAssignment::GatewayLiveness( + targets + .iter() + .map(AssignedTestrun::gateway_probe_target) + .collect::>() + .map_err(malformed_target)?, + ), + }; + + Ok(Some(assignment)) } /// Persists a completed test run result, with its measurements, under the kind and role the @@ -425,6 +569,7 @@ impl AppState { storage: NetworkMonitorStorage, testrun_staleness_age: Duration, testrun_lease_budget: Duration, + liveness: LivenessConfig, validator_client: Arc>, ) -> Self { AppState { @@ -433,18 +578,18 @@ impl AppState { testrun_manager: TestrunManager { testrun_staleness_age, testrun_lease_budget, + liveness, + kind_cursor: Arc::new(AtomicUsize::new(0)), }, validator_client, } } - /// Selects the most stale idle mixnode and atomically marks it as having a test - /// in progress. Returns `None` if no mixnode is currently eligible. - pub(crate) async fn assign_next_mixnode_testrun( - &self, - ) -> Result, ApiError> { + /// Hands the requesting agent one assignment: whichever kind's turn it is, of whichever of that + /// kind's pairings is furthest behind. `None` when nothing is due. + pub(crate) async fn assign_next_testrun(&self) -> Result, ApiError> { self.testrun_manager - .assign_next_mixnode_testrun(&self.storage) + .assign_next_testrun(&self.storage) .await } @@ -848,3 +993,227 @@ mod tests { } } } + +#[cfg(test)] +mod assignment_tests { + use super::*; + use crate::storage::models::{NewNymNode, NodeType}; + use nym_test_utils::helpers::seeded_rng; + use time::macros::datetime; + + fn liveness_config(enabled: bool) -> LivenessConfig { + LivenessConfig { + enabled, + test_interval: Duration::from_secs(15 * 60), + test_timeout: Duration::from_secs(60), + mixnode_wave_size: 100, + gateway_wave_size: 50, + } + } + + /// A manager carrying the shipped defaults, so the rotation is exercised against the cadences it + /// actually runs with. + fn manager(liveness_enabled: bool) -> TestrunManager { + manager_with(liveness_config(liveness_enabled)) + } + + fn manager_with(liveness: LivenessConfig) -> TestrunManager { + TestrunManager { + testrun_staleness_age: Duration::from_secs(2 * 60 * 60), + testrun_lease_budget: Duration::from_secs(5 * 60), + liveness, + kind_cursor: Arc::new(AtomicUsize::new(0)), + } + } + + /// A fully-described node, with real keys: unlike the storage tests, these rows are decoded into + /// probe targets, so placeholder strings would fail as malformed rather than as untestable. + fn node(node_id: i64, node_type: NodeType, clients_ws_port: Option) -> NewNymNode { + let seed = [node_id as u8; 32]; + let x25519_key = x25519::PublicKey::from(&x25519::PrivateKey::new(&mut seeded_rng(seed))); + let identity_key = *ed25519::KeyPair::new(&mut seeded_rng(seed)).public_key(); + + NewNymNode { + node_id, + identity_key: identity_key.to_base58_string(), + last_seen_bonded: datetime!(2025-06-01 00:00:00 UTC), + mixnet_socket_address: Some("1.2.3.4:1789".to_string()), + announced_ips: Some("1.2.3.4".to_string()), + noise_key: Some(x25519_key.to_base58_string()), + sphinx_key: Some(x25519_key.to_base58_string()), + key_rotation_id: Some(7), + node_type, + clients_ws_port, + } + } + + async fn storage_with(nodes: &[NewNymNode]) -> NetworkMonitorStorage { + let storage = NetworkMonitorStorage::in_memory().await; + storage + .batch_insert_or_update_nym_nodes(nodes) + .await + .unwrap(); + storage + } + + // Neither cadence may starve the other, so the kind an agent is offered rotates per request. + // Two nodes rather than one because a single node is locked by whichever kind takes it first, + // which would hide the rotation behind the per-node mutex. + #[tokio::test] + async fn successive_requests_rotate_the_kind() { + let manager = manager(true); + let storage = storage_with(&[ + node(1, NodeType::Mixnode, None), + node(2, NodeType::Mixnode, None), + ]) + .await; + + let first = manager + .assign_next_testrun(&storage) + .await + .unwrap() + .unwrap(); + let second = manager + .assign_next_testrun(&storage) + .await + .unwrap() + .unwrap(); + + assert!(matches!(first, TestRunAssignment::MixnodeStress(_))); + assert!(matches!(second, TestRunAssignment::MixnodeLiveness(_))); + } + + // The flag exists to stop liveness being handed out at all, so its turn must go to stress + // rather than being spent producing nothing. + #[tokio::test] + async fn a_disabled_liveness_kind_never_takes_a_turn() { + let manager = manager(false); + let storage = storage_with(&[ + node(1, NodeType::Mixnode, None), + node(2, NodeType::Mixnode, None), + ]) + .await; + + for _ in 0..2 { + let assignment = manager + .assign_next_testrun(&storage) + .await + .unwrap() + .unwrap(); + assert!(matches!(assignment, TestRunAssignment::MixnodeStress(_))); + } + + // and with both nodes locked by stress, the request is answered with no work rather than + // with a liveness assignment + assert!( + manager + .assign_next_testrun(&storage) + .await + .unwrap() + .is_none() + ); + } + + // A kind whose turn it is but which has nothing due must not waste the request: here only a + // gateway is bonded, so stress (which probes forwarding) has nothing and the request falls + // through to the gateway liveness pairing. + #[tokio::test] + async fn a_kind_with_nothing_due_falls_through_to_the_next() { + let manager = manager(true); + let storage = storage_with(&[node(1, NodeType::Gateway, Some(9000))]).await; + + let assignment = manager + .assign_next_testrun(&storage) + .await + .unwrap() + .unwrap(); + + let TestRunAssignment::GatewayLiveness(wave) = assignment else { + panic!("a gateway-only population produced {assignment:?}"); + }; + assert_eq!(wave.len(), 1); + assert_eq!(wave[0].mixnet.node_id, 1); + // the port the ingress phase opens its session on comes from the stored row + assert_eq!(wave[0].clients_ws_port, 9000); + } + + /// Deliberately unequal and both far below the shipped values, so a wave that took the wrong + /// role's cap - or the storage default - fails rather than coincidentally passing. + fn narrow_waves() -> LivenessConfig { + LivenessConfig { + mixnode_wave_size: 3, + gateway_wave_size: 1, + ..liveness_config(true) + } + } + + // Each role's wave is cut to ITS cap, not to a shared one. The populations are homogeneous so + // that the pairing under test is the only one with work: with both roles available the tie-break + // would settle it and the gateway cap would never be exercised. + #[tokio::test] + async fn a_mixnode_liveness_wave_is_capped_by_the_mixnode_wave_size() { + let manager = manager_with(narrow_waves()); + let nodes: Vec<_> = (1..=5) + .map(|id| node(id, NodeType::Mixnode, None)) + .collect(); + let storage = storage_with(&nodes).await; + + // spend stress's turn, which takes one node, so the next request is liveness's + assert!( + manager + .assign_next_testrun(&storage) + .await + .unwrap() + .is_some() + ); + + let assignment = manager + .assign_next_testrun(&storage) + .await + .unwrap() + .unwrap(); + + let TestRunAssignment::MixnodeLiveness(wave) = assignment else { + panic!("a mixnode-only population produced {assignment:?}"); + }; + assert_eq!(wave.len(), 3); + } + + #[tokio::test] + async fn a_gateway_liveness_wave_is_capped_by_the_gateway_wave_size() { + let manager = manager_with(narrow_waves()); + let nodes: Vec<_> = (1..=5) + .map(|id| node(id, NodeType::Gateway, Some(9000))) + .collect(); + let storage = storage_with(&nodes).await; + + let assignment = manager + .assign_next_testrun(&storage) + .await + .unwrap() + .unwrap(); + + let TestRunAssignment::GatewayLiveness(wave) = assignment else { + panic!("a gateway-only population produced {assignment:?}"); + }; + assert_eq!(wave.len(), 1); + } + + // The decoy for the fall-through test above: the same bonded gateway, differing only in never + // having reported the websocket port its ingress phase opens a session on. Now NEITHER kind has + // anything to give - stress does not probe gateways - so the request goes away empty rather than + // carrying a target the agent could not use. + #[tokio::test] + async fn a_gateway_that_announces_no_websocket_port_is_not_liveness_tested() { + let manager = manager(true); + let storage = storage_with(&[node(1, NodeType::Gateway, None)]).await; + + assert!( + manager + .assign_next_testrun(&storage) + .await + .unwrap() + .is_none() + ); + } +} diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/config.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/config.rs index 5f23be6a9c9..71eae5d203a 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/config.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/config.rs @@ -1,6 +1,7 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only +use crate::storage::models::TestedRole; use anyhow::Context; use nym_network_defaults::{NymNetworkDetails, env_configured}; use nym_validator_client::nyxd::AccountId; @@ -12,6 +13,49 @@ use std::time::Duration; use tracing::info; use url::Url; +/// The liveness kind's own scheduling knobs. Grouped rather than flattened into [`Config`] because +/// every one of them is per-kind: the stress kind keeps `test_interval` and `test_timeout`, and +/// this is the same set of decisions taken for liveness. +/// +/// Every value is provisional and deployment-tunable by design - no behaviour may depend on a +/// specific one. +#[derive(Debug, Copy, Clone)] +pub(crate) struct LivenessConfig { + /// Whether liveness work may be assigned at all. On by default, so switching it off is a + /// deployment-time decision; an agent that predates wave support cannot deserialise a liveness + /// assignment, so a fleet mid-upgrade is a reason to set it. + pub(crate) enabled: bool, + + /// How often each (node, role) pairing should be liveness-tested (e.g. `15m`). Well below the + /// stress `test_interval`, since liveness is the low-volume probe. + pub(crate) test_interval: Duration, + + /// Lease stamped on a liveness assignment's in-progress rows. Bounds ONE concurrent wave rather + /// than the sum over its targets, because the agent probes the whole wave at once, so it does + /// not scale with the wave size. It does have to cover the SLOWER of the two probes: a gateway + /// wave pays session setup and measures two interfaces where a mixnode wave measures one. + pub(crate) test_timeout: Duration, + + /// Maximum number of targets in a mixnode liveness wave. + pub(crate) mixnode_wave_size: usize, + + /// Maximum number of targets in a gateway liveness wave. Lower than the mixnode wave: since a + /// wave is one concurrent batch, this is what an agent holds open at once, and a gateway target + /// costs a full client session where a mixnode target costs a Noise connection. v1 ran a + /// 50-client window over its whole gateway population per cycle. + pub(crate) gateway_wave_size: usize, +} + +impl LivenessConfig { + /// The wave size that applies to `role`. + pub(crate) fn wave_size(&self, role: TestedRole) -> usize { + match role { + TestedRole::Mixnode => self.mixnode_wave_size, + TestedRole::Gateway => self.gateway_wave_size, + } + } +} + #[derive(Debug, Clone)] pub(crate) struct Config { /// HTTPS RPC URL of a Nyx node (e.g. `https://rpc.nymtech.net`). @@ -28,9 +72,12 @@ pub(crate) struct Config { pub(crate) test_interval: Duration, /// Maximum time a single test run is allowed to run before being considered timed out - /// (e.g. `5m`). + /// (e.g. `5m`). The stress kind's lease budget. pub(crate) test_timeout: Duration, + /// Scheduling knobs of the liveness kind, whose cadence and lease are its own. + pub(crate) liveness: LivenessConfig, + /// Path to the SQLite database file. pub(crate) database_path: PathBuf, @@ -39,7 +86,7 @@ pub(crate) struct Config { pub(crate) node_refresh_rate: Duration, /// Timeout for querying a single node for its detailed information (sphinx key, noise key, - /// etc.). Queries that exceed this budget leave the corresponding fields as `NULL` + /// etc.). A node that exceeds this budget keeps whatever an earlier cycle learned about it /// (e.g. `10s`). pub(crate) node_info_query_timeout: Duration, diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/mod.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/mod.rs index e76dfe0cd9e..515d22543e0 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/mod.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/mod.rs @@ -261,6 +261,7 @@ impl NetworkMonitorOrchestrator { // the lease a dispatched run holds its node for is the same budget the eviction sweep // uses to decide a run has gone silent, now materialised on the row at dispatch self.config.test_timeout, + self.config.liveness, self.client.clone(), ); diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/node_refresher.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/node_refresher.rs index e8cecfdfc2a..6f729561d3f 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/node_refresher.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/node_refresher.rs @@ -4,7 +4,7 @@ use crate::orchestrator::config::Config; use crate::orchestrator::prometheus::{PROMETHEUS_METRICS, PrometheusMetric}; use crate::storage::NetworkMonitorStorage; -use crate::storage::models::{NewNymNode, NodeType}; +use crate::storage::models::{BondedNymNode, NewNymNode, NodeType}; use anyhow::Context; use futures::{StreamExt, stream}; use nym_bin_common::bin_info; @@ -34,7 +34,7 @@ pub(crate) struct NodeRefresher { pub(crate) node_refresh_rate: Duration, /// Timeout for querying a single node for its detailed information (sphinx key, noise key, - /// etc.). Queries that exceed this budget leave the corresponding fields as `NULL` + /// etc.). A node that exceeds this budget keeps whatever an earlier cycle learned about it /// (e.g. `10s`). pub(crate) node_info_query_timeout: Duration, @@ -44,6 +44,34 @@ pub(crate) struct NodeRefresher { pub(crate) shutdown_token: ShutdownToken, } +/// What one node's refresh produced. The two cases are persisted differently, and keeping them +/// apart in the type is what makes "described completely or not at all" checkable rather than a +/// convention: there is no value of this type that carries a half-described node. +enum RefreshedNode { + /// Everything the node's own endpoint reported, all from one reading of it. + Described(NewNymNode), + + /// The node is bonded, but its endpoint did not answer (or answered incompletely), so only that + /// much is known this cycle. + BondOnly(BondedNymNode), +} + +impl RefreshedNode { + fn described(self) -> Option { + match self { + RefreshedNode::Described(node) => Some(node), + RefreshedNode::BondOnly(_) => None, + } + } + + fn bond_only(self) -> Option { + match self { + RefreshedNode::BondOnly(node) => Some(node), + RefreshedNode::Described(_) => None, + } + } +} + /// Information about the node retrieved from the node directly struct SelfDescribedData { /// Mixnet socket address (host:port) at which the node accepts sphinx packets. @@ -65,6 +93,15 @@ struct SelfDescribedData { /// The supported roles of the node in the network. roles: NodeRoles, + + /// Port of the node's PLAIN client websocket listener, which a gateway liveness probe opens its + /// session on. `None` for a node announcing no entry-gateway interface, and for one whose + /// websocket query failed. + /// + /// Its `wss` counterpart is deliberately not read: the only consumer of that fact is the + /// divergence bucket in nym-api, which reads the same interface from its own described-nodes + /// cache, so a copy here would be one nothing in this service looks at. + clients_ws_port: Option, } impl NodeRefresher { @@ -131,13 +168,30 @@ impl NodeRefresher { .mix_port .unwrap_or(DEFAULT_MIX_LISTENING_PORT); - // retrieve information about the node roles so that we can classify the node - // (we're not testing gateways yet, but we still store them for completeness) + // retrieve information about the node roles so that we can classify the node, and so that we + // know whether to ask it about its client websocket interface at all let roles = api_client .get_roles() .await .context("failed to retrieve node roles")?; + // the gateway liveness probe opens a client session, which needs the port that interface + // listens on. asked for separately because it is not one of the announced ports, and only of + // gateway-capable nodes, since a pure mixnode serves no client websocket. a gateway that + // will not answer for it fails the whole describe rather than yielding a node described + // everywhere except here + let clients_ws_port = if roles.gateway_enabled { + Some( + api_client + .get_mixnet_websockets() + .await + .context("failed to retrieve the client websocket interface")? + .ws_port, + ) + } else { + None + }; + Ok(SelfDescribedData { // only contributes the mix port now that the address under test is picked per run mixnet_socket_address: SocketAddr::new(*ip_address, mix_port), @@ -146,11 +200,19 @@ impl NodeRefresher { sphinx_key, key_rotation_id, roles, + clients_ws_port, }) } - async fn get_node_details(&self, bond: NymNodeBond, timeout: Duration) -> NewNymNode { - let mut node_update = NewNymNode::from_bond(&bond); + /// Refreshes one node, either completely or not at all. + /// + /// A node is described as a whole: every field comes from the same reading of its endpoint, so a + /// row can never hold a fresh key beside an address from an earlier cycle. When any part of the + /// describe fails, the outcome carries the bond alone and the node's previously learned fields + /// are left exactly as they were, rather than being overwritten with nulls that would make an + /// otherwise testable node ineligible for every kind until the next successful cycle. + async fn get_node_details(&self, bond: NymNodeBond, timeout: Duration) -> RefreshedNode { + let bonded = BondedNymNode::from_bond(&bond); let node_id = bond.node_id; let self_described = match tokio::time::timeout(timeout, self.get_node_details_inner(bond)) @@ -160,30 +222,35 @@ impl NodeRefresher { debug!( "timed out while attempting to retrieve self-described node details for node {node_id}" ); - return node_update; + return RefreshedNode::BondOnly(bonded); } Ok(Err(err)) => { debug!("failed to retrieve self-described node details for node {node_id}: {err}"); - return node_update; + return RefreshedNode::BondOnly(bonded); } Ok(Ok(info)) => info, }; - node_update.mixnet_socket_address = Some(self_described.mixnet_socket_address.to_string()); - node_update.announced_ips = Some( - self_described - .announced_ips - .iter() - .map(|ip| ip.to_string()) - .collect::>() - .join(","), - ); - node_update.noise_key = Some(self_described.noise_key.to_base58_string()); - node_update.sphinx_key = Some(self_described.sphinx_key.to_base58_string()); - node_update.key_rotation_id = Some(self_described.key_rotation_id as i64); - node_update.node_type = NodeType::from_roles(&self_described.roles); - - node_update + RefreshedNode::Described(NewNymNode { + node_id: bonded.node_id, + identity_key: bonded.identity_key, + last_seen_bonded: bonded.last_seen_bonded, + // only contributes the mix port now that the address under test is picked per run + mixnet_socket_address: Some(self_described.mixnet_socket_address.to_string()), + announced_ips: Some( + self_described + .announced_ips + .iter() + .map(|ip| ip.to_string()) + .collect::>() + .join(","), + ), + noise_key: Some(self_described.noise_key.to_base58_string()), + sphinx_key: Some(self_described.sphinx_key.to_base58_string()), + key_rotation_id: Some(self_described.key_rotation_id as i64), + node_type: NodeType::from_roles(&self_described.roles), + clients_ws_port: self_described.clients_ws_port.map(i64::from), + }) } async fn refresh_bonded_nodes(&self) -> anyhow::Result<()> { @@ -202,13 +269,29 @@ impl NodeRefresher { .collect() .await; + // the two outcomes are persisted differently: a described node replaces everything stored + // about it, while one that could not be described only proves it is still bonded + let (described, bond_only): (Vec<_>, Vec<_>) = refreshed_nodes + .into_iter() + .partition(|node| matches!(node, RefreshedNode::Described(_))); + let described: Vec<_> = described + .into_iter() + .filter_map(RefreshedNode::described) + .collect(); + let bond_only: Vec<_> = bond_only + .into_iter() + .filter_map(RefreshedNode::bond_only) + .collect(); + let mut per_type: HashMap = HashMap::new(); - for node in &refreshed_nodes { + for node in &described { *per_type.entry(node.node_type).or_insert(0) += 1; } let count_of = |t: NodeType| per_type.get(&t).copied().unwrap_or(0); - let unknown = count_of(NodeType::Unknown); - let successful = (refreshed_nodes.len() as i64) - unknown; + // a described node reporting no roles at all is as unusable as one that never answered, so + // both land in the unknown bucket + let unknown = count_of(NodeType::Unknown) + bond_only.len() as i64; + let successful = described.len() as i64 - count_of(NodeType::Unknown); info!("managed to retrieve full node information on {successful} nodes ({unknown} failed)"); PROMETHEUS_METRICS.set( @@ -227,12 +310,14 @@ impl NodeRefresher { PROMETHEUS_METRICS.set(PrometheusMetric::SuccessfulNymNodeDataRetrieval, successful); PROMETHEUS_METRICS.set(PrometheusMetric::FailedNymNodeDataRetrieval, unknown); - // 3. persist every node (including unreachable ones so we keep their - // previously-learned keys around for the next refresh). The testrun - // assignment query filters out non-mixnode / unknown entries. + // 3. persist what each node yielded. A described node has every field replaced; one that + // could not be described has only its bond recorded, keeping whatever was learned about + // it before, since nulling that would drop an otherwise testable node out of every kind + // until a later cycle answered. self.storage - .batch_insert_or_update_nym_nodes(&refreshed_nodes) + .batch_insert_or_update_nym_nodes(&described) .await?; + self.storage.batch_touch_bonded_nodes(&bond_only).await?; // Observe the cycle duration last so it reflects the full refresh path // (contract query + per-node queries + storage write). diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/prometheus.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/prometheus.rs index 9d49a8f47e8..78b56abc400 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/prometheus.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/prometheus.rs @@ -108,6 +108,20 @@ const AVG_PACKET_RTT: &[f64] = &[ 1000., // 1s+ (implicitly) ]; +/// Histogram buckets for the liveness wave-size series. Spans both configured caps (100 mixnode +/// targets, 50 gateway ones) with a dedicated `<= 1` bucket, because a wave that has collapsed to a +/// single target reads very differently from one that is merely short: it means the population due +/// for that pairing has run dry. +const LIVENESS_WAVE_SIZE: &[f64] = &[ + 1., // 1 - 2 + 2., // 2 - 5 + 5., // 5 - 10 + 10., // 10 - 20 + 20., // 20 - 50 + 50., // 50 - 100 (the gateway cap) + 100., // 100+ (implicitly, past the mixnode cap) +]; + /// Every Prometheus series emitted by the orchestrator. Each variant maps to exactly one metric /// and must carry a `help` strum property — this is verified by the `every_variant_has_help_property` /// test. @@ -266,6 +280,54 @@ pub enum PrometheusMetric { help = "The number of submitted stress-test results that nym-api dropped in per-entry validation (non-mixnode entry, or performance score outside [0, 1])" ))] SubmittedResultsRejected, + + // Assignments are counted per (kind, role) pairing rather than per kind: the two liveness roles + // are separate machinery against separate populations, so an operator has to be able to see that + // gateway liveness is flowing without inferring it from a total. + #[strum(props( + help = "The number of stress test runs assigned to agents against a node's mixnode role" + ))] + MixnodeStressAssignments, + + #[strum(props( + help = "The number of liveness waves assigned to agents against a node's mixnode role" + ))] + MixnodeLivenessAssignments, + + #[strum(props( + help = "The number of liveness waves assigned to agents against a node's entry-gateway role" + ))] + GatewayLivenessAssignments, + + #[strum(props( + help = "The number of targets in an assigned mixnode liveness wave. A distribution sitting well below the configured wave size means the population is keeping up with the cadence rather than the wave being the constraint" + ))] + MixnodeLivenessWaveSize, + + #[strum(props( + help = "The number of targets in an assigned gateway liveness wave, whose configured cap is lower than the mixnode one because each target costs the agent a live client session" + ))] + GatewayLivenessWaveSize, + + #[strum(props( + help = "The number of stress test runs currently in progress (rows in testrun_in_progress under the stress kind)" + ))] + StressTestrunsInProgress, + + #[strum(props( + help = "The number of liveness test runs currently in progress (rows in testrun_in_progress under the liveness kind)" + ))] + LivenessTestrunsInProgress, + + #[strum(props( + help = "The number of stress test runs whose lease expired before a result arrived, freeing the node for reassignment" + ))] + StressLeasesExpired, + + #[strum(props( + help = "The number of liveness test runs whose lease expired before a result arrived. A persistently non-zero value means the liveness lease is too short for the wave it has to cover" + ))] + LivenessLeasesExpired, } impl PrometheusMetric { @@ -342,6 +404,19 @@ impl PrometheusMetric { PrometheusMetric::SubmittedResultsAccepted => Metric::new_int_counter(&name, help), PrometheusMetric::SubmittedResultsDuplicate => Metric::new_int_counter(&name, help), PrometheusMetric::SubmittedResultsRejected => Metric::new_int_counter(&name, help), + PrometheusMetric::MixnodeStressAssignments => Metric::new_int_counter(&name, help), + PrometheusMetric::MixnodeLivenessAssignments => Metric::new_int_counter(&name, help), + PrometheusMetric::GatewayLivenessAssignments => Metric::new_int_counter(&name, help), + PrometheusMetric::MixnodeLivenessWaveSize => { + Metric::new_histogram(&name, help, Some(LIVENESS_WAVE_SIZE)) + } + PrometheusMetric::GatewayLivenessWaveSize => { + Metric::new_histogram(&name, help, Some(LIVENESS_WAVE_SIZE)) + } + PrometheusMetric::StressTestrunsInProgress => Metric::new_int_gauge(&name, help), + PrometheusMetric::LivenessTestrunsInProgress => Metric::new_int_gauge(&name, help), + PrometheusMetric::StressLeasesExpired => Metric::new_int_counter(&name, help), + PrometheusMetric::LivenessLeasesExpired => Metric::new_int_counter(&name, help), } } @@ -455,7 +530,7 @@ mod tests { // a sanity check for anyone adding new metrics. if this test fails, // make sure any methods on `PrometheusMetric` enum don't need updating // or require custom Display impl - assert_eq!(34, PrometheusMetric::COUNT) + assert_eq!(43, PrometheusMetric::COUNT) } #[test] diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/stale_results_eviction.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/stale_results_eviction.rs index a587786608c..05d16b6b37b 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/stale_results_eviction.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/stale_results_eviction.rs @@ -100,6 +100,12 @@ impl StaleResultsEviction { Ok(count) => PROMETHEUS_METRICS.set(PrometheusMetric::TestrunsInProgress, count), Err(err) => error!("failed to count in-flight testruns for metric: {err}"), } + + // the per-kind gauges are only ever published here, so their freshness is this sweep's + // cadence rather than the assignment path's + if let Err(err) = self.storage.publish_in_progress_gauges().await { + error!("failed to count in-flight testruns per kind for metrics: {err}"); + } Ok(()) } diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/manager.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/manager.rs index 7a6e91357aa..a4accda526c 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/manager.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/manager.rs @@ -2,9 +2,10 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::storage::models::{ - AssignedTestrun, AssignmentCandidate, CompletedTestRun, InsertedTestRun, - KeyedTestRunMeasurement, NewNymNode, NewTestRun, NymNode, TestKind, TestRun, TestRunInProgress, - TestRunMeasurement, TestedRole, next_ip_to_test, + AssignedTestrun, AssignmentCandidate, AssignmentRequest, BondedNymNode, CompletedTestRun, + InsertedTestRun, KeyedTestRunMeasurement, NewNymNode, NewTestRun, NymNode, PairingHead, + TestKind, TestPairing, TestRun, TestRunInProgress, TestRunMeasurement, TestedRole, + next_ip_to_test, }; use sqlx::{QueryBuilder, SqliteConnection}; use std::collections::HashMap; @@ -20,6 +21,26 @@ pub(crate) struct StorageManager { pub(crate) connection_pool: sqlx::SqlitePool, } +/// The eligibility predicates that depend on the role a run would probe: which node types may be +/// assigned in it, and which stored fields its probe cannot do without. Composed into the candidate +/// query as a literal fragment rather than expressed as role-conditioned SQL, so that each role's +/// filter reads as the plain predicate it is and leaves the node-type index usable. +/// +/// A node classified `mixnode_and_gateway` is eligible in BOTH roles, which is what makes it +/// testable as each, one run per role. +fn role_eligibility(role: TestedRole) -> &'static str { + match role { + // the mixnet listener every probe needs is already required of every candidate + TestedRole::Mixnode => "AND n.node_type IN ('mixnode', 'mixnode_and_gateway')", + + // the gateway probe opens a client session over the announced websocket port, so a node + // that has never reported one is untestable in this role however it is bonded + TestedRole::Gateway => { + "AND n.node_type IN ('gateway', 'mixnode_and_gateway') AND n.clients_ws_port IS NOT NULL" + } + } +} + /// Fetches the measurements of the given runs, grouped by the run they belong to. /// /// Takes a connection rather than the pool so that callers can run it inside the same transaction @@ -306,14 +327,68 @@ impl StorageManager { /// /// The comparison is strict, so a lease expiring exactly at `now` survives until the next /// sweep, matching the result eviction sweep. + /// + /// Reports how many rows each kind lost, because that is the signal that a kind's lease budget + /// is too short for the work it covers, and a total would hide it: liveness leases are minutes + /// shorter than stress ones, so the two expire at very different rates even when both are + /// healthy. Counted before the delete, in the same transaction, since the delete itself reports + /// only a total. pub(crate) async fn clear_expired_testruns_in_progress( &self, now: OffsetDateTime, - ) -> anyhow::Result { - let res = sqlx::query!("DELETE FROM testrun_in_progress WHERE expires_at < ?", now,) - .execute(&self.connection_pool) + ) -> anyhow::Result> { + let mut tx = self.connection_pool.begin_with("BEGIN IMMEDIATE").await?; + + let expiring = sqlx::query_as::<_, (TestKind, i64)>( + "SELECT test_kind, COUNT(*) FROM testrun_in_progress WHERE expires_at < ? GROUP BY test_kind", + ) + .bind(now) + .fetch_all(&mut *tx) + .await?; + + sqlx::query!("DELETE FROM testrun_in_progress WHERE expires_at < ?", now,) + .execute(&mut *tx) .await?; - Ok(res.rows_affected()) + + tx.commit().await?; + Ok(expiring + .into_iter() + .map(|(kind, count)| (kind, count as u64)) + .collect()) + } + + /// Records that these nodes are still bonded WITHOUT touching anything their own endpoint would + /// have supplied. + /// + /// Used for a node whose describe failed this cycle. Overwriting its learned fields with nulls + /// would fail every eligibility predicate at once and drop the node out of all kinds until a + /// later cycle answered, so a failed describe leaves the previous reading in place instead. A + /// node seen for the first time is inserted with those columns empty, which is the one state + /// that genuinely means "never described". + pub(crate) async fn batch_touch_bonded_nodes( + &self, + nodes: &[BondedNymNode], + ) -> anyhow::Result<()> { + let mut tx = self.connection_pool.begin().await?; + + for node in nodes { + sqlx::query!( + r#" + INSERT INTO nym_node (node_id, identity_key, last_seen_bonded) + VALUES (?, ?, ?) + ON CONFLICT (node_id) DO UPDATE SET + last_seen_bonded = excluded.last_seen_bonded + "#, + node.node_id, + node.identity_key, + node.last_seen_bonded, + ) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; + Ok(()) } /// Returns the number of rows currently in `testrun_in_progress` — i.e. the number of @@ -325,42 +400,51 @@ impl StorageManager { Ok(total) } - /// Atomically selects the most stale idle mixnode and marks it as having a test run in - /// progress. - /// - /// Staleness, the rotation pointer and the resulting lock are all read and written for the - /// `(stress, mixnode)` pairing specifically, so no other kind's cadence can disturb this one. - /// Only that pairing is assignable today; per-kind selection replaces the hardcoded pairing - /// with the kind the orchestrator chose. + /// The same count broken down by kind, for the per-kind in-flight gauges. Kinds with no rows are + /// absent from the map rather than present as zero, so a caller publishing gauges has to decide + /// what an absent kind means - it means zero. + pub(crate) async fn count_testruns_in_progress_by_kind( + &self, + ) -> anyhow::Result> { + let counts = sqlx::query_as::<_, (TestKind, i64)>( + "SELECT test_kind, COUNT(*) FROM testrun_in_progress GROUP BY test_kind", + ) + .fetch_all(&self.connection_pool) + .await?; + + Ok(counts.into_iter().collect()) + } + + /// Atomically selects the most stale idle nodes eligible for one (kind, role) pairing and marks + /// each of them as having a test run in progress. /// - /// "Most stale" is defined as: nodes that pairing has never tested come first, followed by - /// those whose last run under it has the oldest timestamp. `last_tested_before` acts as a - /// minimum-staleness gate that never-tested nodes bypass; the caller is expected to pass - /// `now - staleness_age`. + /// Staleness, the rotation pointer and the resulting locks are all read and written for the + /// requested pairing alone, so no other kind's or role's cadence can disturb this one. A stress + /// request asks for one target; a liveness request asks for up to its role's wave size, and the + /// returned targets form one wave. /// - /// `now` and `expires_at` are stamped onto the resulting `testrun_in_progress` row, the latter - /// materialising the lease deadline so the eviction sweep needs no knowledge of kinds. Both are - /// accepted as arguments rather than read from the clock so a caller can use one consistent - /// timestamp across related operations. + /// "Most stale" is defined as: nodes this pairing has never tested come first, followed by those + /// whose last run under it has the oldest timestamp. [`AssignmentRequest::last_tested_before`] + /// acts as a minimum-staleness gate that never-tested nodes bypass. /// - /// Nodes with a row in `testrun_in_progress` are excluded entirely, REGARDLESS of the kind or - /// role that row belongs to: a node under one kind of test must not be measured by another at - /// the same time. Nodes missing `mixnet_socket_address`, `noise_key` or `sphinx_key` are - /// excluded as untestable, and only `mixnode` / `mixnode_and_gateway` nodes are eligible. + /// Eligibility beyond staleness: nodes with a row in `testrun_in_progress` are excluded + /// entirely, REGARDLESS of the kind or role that row belongs to, since a node under one kind of + /// test must not be measured by another at the same time; nodes missing `mixnet_socket_address`, + /// `noise_key` or `sphinx_key` are untestable by any probe; and the node types a role may assign + /// along with the extra fields its probe needs come from [`role_eligibility`]. A node whose + /// in-flight row has just cleared is immediately eligible for another kind, the per-node lock + /// being the whole of the mutual exclusion between kinds. /// - /// Returns `None` if no eligible idle mixnode exists. - pub(crate) async fn assign_next_mixnode_testrun( + /// Returns an empty vector when no eligible idle node exists. A target whose stored addresses + /// cannot be parsed is dropped from the wave rather than failing the assignment. + pub(crate) async fn assign_next_testruns( &self, - now: OffsetDateTime, - last_tested_before: OffsetDateTime, - expires_at: OffsetDateTime, - ) -> anyhow::Result> { - let (test_kind, tested_role) = (TestKind::Stress, TestedRole::Mixnode); - + request: &AssignmentRequest, + ) -> anyhow::Result> { // Starts a write (IMMEDIATE) transaction, to prevent issue when upgrading from a read one to a write one let mut tx = self.connection_pool.begin_with("BEGIN IMMEDIATE").await?; - let candidate = sqlx::query_as::<_, AssignmentCandidate>( + let query = format!( r#" SELECT n.node_id, @@ -383,70 +467,131 @@ impl StorageManager { AND n.mixnet_socket_address IS NOT NULL AND n.noise_key IS NOT NULL AND n.sphinx_key IS NOT NULL - AND n.node_type IN ('mixnode', 'mixnode_and_gateway') + {role_gate} AND (s.last_tested_at IS NULL OR s.last_tested_at < ?) ORDER BY s.last_tested_at ASC NULLS FIRST - LIMIT 1 + LIMIT ? "#, - ) - .bind(test_kind) - .bind(tested_role) - .bind(last_tested_before) - .fetch_optional(&mut *tx) - .await?; + role_gate = role_eligibility(request.pairing.tested_role), + ); + + // bound in the order the placeholders appear above: the pairing being joined, the staleness + // cutoff, then the wave size + let candidates = sqlx::query_as::<_, AssignmentCandidate>(&query) + .bind(request.pairing.test_kind) + .bind(request.pairing.tested_role) + .bind(request.last_tested_before) + .bind(request.wave_size as i64) + .fetch_all(&mut *tx) + .await?; - let Some(candidate) = candidate else { - tx.commit().await?; - return Ok(None); - }; + let mut assigned = Vec::with_capacity(candidates.len()); + for candidate in candidates { + // rotate onto the next announced address of that node, following this pairing's own + // pointer. the eligibility filter guarantees a parseable `mixnet_socket_address`, so + // this can only be `None` for a row whose stored addresses are corrupt, and dropping + // that one target keeps the rest of the wave assignable + let announced = candidate.node.announced_ips(); + let Some(tested_ip) = next_ip_to_test(&announced, candidate.last_tested_ip.as_deref()) + else { + continue; + }; + + // advance the rotation pointer here rather than on result submission, so that runs which + // never report back still move the node onto its next address + let node_id = candidate.node.inner.node_id; + let stored_tested_ip = tested_ip.to_string(); + sqlx::query!( + r#" + INSERT INTO node_test_state (node_id, test_kind, tested_role, last_tested_ip) + VALUES (?, ?, ?, ?) + ON CONFLICT (node_id, test_kind, tested_role) DO UPDATE SET + last_tested_ip = excluded.last_tested_ip + "#, + node_id, + request.pairing.test_kind, + request.pairing.tested_role, + stored_tested_ip, + ) + .execute(&mut *tx) + .await?; - // rotate onto the next announced address of that node, following this pairing's own - // pointer. the eligibility filter guarantees a parseable `mixnet_socket_address`, so this - // can only be `None` for a row whose stored addresses are corrupt - let announced = candidate.node.announced_ips(); - let Some(tested_ip) = next_ip_to_test(&announced, candidate.last_tested_ip.as_deref()) - else { - tx.commit().await?; - return Ok(None); - }; + sqlx::query!( + r#" + INSERT INTO testrun_in_progress (node_id, started_at, expires_at, test_kind, tested_role) + VALUES (?, ?, ?, ?, ?) + "#, + node_id, + request.now, + request.expires_at, + request.pairing.test_kind, + request.pairing.tested_role, + ) + .execute(&mut *tx) + .await?; - // advance the rotation pointer here rather than on result submission, so that runs which - // never report back still move the node onto its next address - let node_id = candidate.node.inner.node_id; - let stored_tested_ip = tested_ip.to_string(); - sqlx::query!( - r#" - INSERT INTO node_test_state (node_id, test_kind, tested_role, last_tested_ip) - VALUES (?, ?, ?, ?) - ON CONFLICT (node_id, test_kind, tested_role) DO UPDATE SET - last_tested_ip = excluded.last_tested_ip - "#, - node_id, - test_kind, - tested_role, - stored_tested_ip, - ) - .execute(&mut *tx) - .await?; + assigned.push(AssignedTestrun { + node: candidate.node, + tested_ip, + }); + } - sqlx::query!( + tx.commit().await?; + Ok(assigned) + } + + /// How overdue the node this pairing would assign next is, or `None` when it has nothing + /// eligible. + /// + /// Exists so that a kind with more than one pairing can pick between them by need: the most + /// overdue head wins, which keeps the two liveness roles interleaving in proportion to how far + /// behind each has fallen instead of by a fixed share. + /// + /// Applies the SAME eligibility and ordering as [`Self::assign_next_testruns`], and reports the + /// staleness position of the very row that assignment would take first. The two queries are + /// written out separately so that each keeps its binds beside its own placeholders; that they + /// agree is pinned by a test, since a peek judging a different population could nominate a + /// pairing whose node the assignment then fails to find. + pub(crate) async fn peek_pairing_head( + &self, + pairing: TestPairing, + last_tested_before: OffsetDateTime, + ) -> anyhow::Result> { + let query = format!( r#" - INSERT INTO testrun_in_progress (node_id, started_at, expires_at, test_kind, tested_role) - VALUES (?, ?, ?, ?, ?) + SELECT s.last_tested_at + FROM nym_node n + LEFT JOIN testrun_in_progress tip ON tip.node_id = n.node_id + LEFT JOIN node_test_state s ON s.node_id = n.node_id + AND s.test_kind = ? + AND s.tested_role = ? + WHERE tip.node_id IS NULL + AND n.mixnet_socket_address IS NOT NULL + AND n.noise_key IS NOT NULL + AND n.sphinx_key IS NOT NULL + {role_gate} + AND (s.last_tested_at IS NULL OR s.last_tested_at < ?) + ORDER BY s.last_tested_at ASC NULLS FIRST + LIMIT 1 "#, - node_id, - now, - expires_at, - test_kind, - tested_role, - ) - .execute(&mut *tx) - .await?; + role_gate = role_eligibility(pairing.tested_role), + ); + + // bound in the order the placeholders appear above: the pairing being joined, then the + // staleness cutoff. the column is a nullable TIMESTAMP, which sqlx cannot infer a Rust type + // for through the query! macro, hence the explicit `Option` here + let head = sqlx::query_scalar::<_, Option>(&query) + .bind(pairing.test_kind) + .bind(pairing.tested_role) + .bind(last_tested_before) + .fetch_optional(&self.connection_pool) + .await?; - tx.commit().await?; - Ok(Some(AssignedTestrun { - node: candidate.node, - tested_ip, + // the outer Option is whether a node is eligible at all, the inner one whether this pairing + // has ever measured it + Ok(head.map(|last_tested_at| match last_tested_at { + Some(last_tested_at) => PairingHead::LastTestedAt(last_tested_at), + None => PairingHead::NeverTested, })) } @@ -750,29 +895,28 @@ mod tests { ExercisedInterface, NewNymNode, NewTestRun, NodeTestState, NodeType, }; use std::net::IpAddr; - use std::path::Path; use time::macros::datetime; async fn setup() -> StorageManager { - let pool = sqlx::SqlitePool::connect("sqlite::memory:") - .await - .expect("failed to create in-memory SQLite pool"); - let migrations_path = Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations"); - sqlx::migrate::Migrator::new(migrations_path.as_path()) + crate::storage::NetworkMonitorStorage::in_memory() .await - .expect("failed to find migrations") - .run(&pool) - .await - .expect("failed to run migrations"); - StorageManager { - connection_pool: pool, - } + .storage_manager } fn node(id: i64, identity_key: &str) -> NewNymNode { node_with_ips(id, identity_key, "1.2.3.4") } + /// A gateway-capable node, optionally announcing the client websocket port that the gateway + /// liveness probe cannot open a session without. + fn gateway_node(id: i64, clients_ws_port: Option) -> NewNymNode { + NewNymNode { + node_type: NodeType::Gateway, + clients_ws_port, + ..node(id, &format!("key_{id}")) + } + } + /// A node announcing `announced_ips` (comma-separated), for exercising the address rotation. fn node_with_ips(id: i64, identity_key: &str, announced_ips: &str) -> NewNymNode { NewNymNode { @@ -882,15 +1026,41 @@ mod tests { datetime!(9999-12-31 23:59:59 UTC) } - /// Assigns at `now`, with an hour-long lease and the given staleness gate. + /// A request for `pairing` at `now`, with an hour-long lease and the given staleness gate. + fn request( + pairing: TestPairing, + now: OffsetDateTime, + last_tested_before: OffsetDateTime, + wave_size: usize, + ) -> AssignmentRequest { + AssignmentRequest { + pairing, + now, + last_tested_before, + expires_at: now + time::Duration::hours(1), + wave_size, + } + } + + /// A `(stress, mixnode)` request, i.e. the one-target wave that pairing always asks for. + fn stress_request( + now: OffsetDateTime, + last_tested_before: OffsetDateTime, + ) -> AssignmentRequest { + request(TestPairing::STRESS_MIXNODE, now, last_tested_before, 1) + } + + /// Assigns the stress pairing at `now`, returning the single target such a request can produce. async fn assign( db: &StorageManager, now: OffsetDateTime, last_tested_before: OffsetDateTime, ) -> Option { - db.assign_next_mixnode_testrun(now, last_tested_before, now + time::Duration::hours(1)) + db.assign_next_testruns(&stress_request(now, last_tested_before)) .await .unwrap() + .into_iter() + .next() } /// Seeds a pairing's rotation pointer, standing in for an assignment of a (kind, role) the @@ -976,6 +1146,76 @@ mod tests { assert_eq!(count, 2); } + /// The bond-only write, i.e. what a node whose describe failed this cycle gets. + async fn touch(db: &StorageManager, node_id: i64, last_seen_bonded: OffsetDateTime) { + db.batch_touch_bonded_nodes(&[BondedNymNode { + node_id, + identity_key: format!("key_{node_id}"), + last_seen_bonded, + }]) + .await + .unwrap() + } + + // A failed describe must not cost the node everything an earlier cycle learned: nulling + // these columns fails every eligibility predicate at once, which would drop a node that is + // merely slow out of EVERY kind until a later cycle answered. + #[tokio::test] + async fn a_bond_only_refresh_keeps_what_was_already_learned() { + let db = setup().await; + let mut described = node(1, "key_1"); + described.clients_ws_port = Some(9000); + described.node_type = NodeType::MixnodeAndGateway; + db.batch_insert_or_update_nym_nodes(&[described]) + .await + .unwrap(); + + touch(&db, 1, datetime!(2025-06-02 00:00:00 UTC)).await; + + let node = db.get_nym_node_by_id(1).await.unwrap().unwrap().inner; + assert_eq!(node.mixnet_socket_address.as_deref(), Some("1.2.3.4:1789")); + assert_eq!(node.noise_key.as_deref(), Some("placeholder_noise_key")); + assert_eq!(node.sphinx_key.as_deref(), Some("placeholder_sphinx_key")); + assert_eq!(node.announced_ips.as_deref(), Some("1.2.3.4")); + assert_eq!(node.key_rotation_id, Some(0)); + assert_eq!(node.clients_ws_port, Some(9000)); + assert_eq!(node.node_type, NodeType::MixnodeAndGateway); + + // the one thing it does record is that the bond is still there + assert_eq!(node.last_seen_bonded, datetime!(2025-06-02 00:00:00 UTC)); + } + + // A node seen for the first time still has to exist as a row, and its empty describe columns + // are the one state that genuinely means "never described" rather than "described once". + #[tokio::test] + async fn a_bond_only_refresh_inserts_an_undescribed_node() { + let db = setup().await; + touch(&db, 7, datetime!(2025-06-02 00:00:00 UTC)).await; + + let node = db.get_nym_node_by_id(7).await.unwrap().unwrap().inner; + assert_eq!(node.identity_key, "key_7"); + assert!(node.mixnet_socket_address.is_none()); + assert!(node.noise_key.is_none()); + assert!(node.sphinx_key.is_none()); + assert!(node.clients_ws_port.is_none()); + assert_eq!(node.node_type, NodeType::Unknown); + } + + // and being described afterwards fills it in, so a node that answers late is not stuck as a + // stub + #[tokio::test] + async fn a_later_describe_replaces_a_bond_only_row() { + let db = setup().await; + touch(&db, 1, datetime!(2025-06-02 00:00:00 UTC)).await; + db.batch_insert_or_update_nym_nodes(&[node(1, "key_1")]) + .await + .unwrap(); + + let node = db.get_nym_node_by_id(1).await.unwrap().unwrap().inner; + assert_eq!(node.noise_key.as_deref(), Some("placeholder_noise_key")); + assert_eq!(node.node_type, NodeType::Mixnode); + } + #[tokio::test] async fn empty_batch_is_noop() { let db = setup().await; @@ -1216,12 +1456,23 @@ mod tests { node_id: i64, started_at: OffsetDateTime, expires_at: OffsetDateTime, + ) { + lease_of(db, node_id, TestKind::Stress, started_at, expires_at).await + } + + /// The same, under a chosen kind. The role plays no part in expiry, so it stays fixed. + async fn lease_of( + db: &StorageManager, + node_id: i64, + test_kind: TestKind, + started_at: OffsetDateTime, + expires_at: OffsetDateTime, ) { db.mark_testrun_in_progress( node_id, started_at, expires_at, - TestKind::Stress, + test_kind, TestedRole::Mixnode, ) .await @@ -1274,10 +1525,35 @@ mod tests { lease(&db, 4, datetime!(2025-06-01 11:55:00 UTC), now).await; let cleared = db.clear_expired_testruns_in_progress(now).await.unwrap(); - assert_eq!(cleared, 1); + assert_eq!(cleared.get(&TestKind::Stress).copied(), Some(1)); assert_eq!(remaining(&db).await, vec![2, 3, 4]); } + // the breakdown is the whole point of counting before the delete: it says WHICH kind's lease + // is too short for the work it covers, which a single total cannot, since the two kinds run + // on leases minutes apart and so expire at different rates even when both are healthy + #[tokio::test] + async fn expiries_are_counted_per_kind() { + let db = setup().await; + for node_id in 1..=3 { + seed_node(&db, node_id).await; + } + let expired_at = datetime!(2025-06-01 11:00:00 UTC); + + lease_of(&db, 1, TestKind::Stress, expired_at, expired_at).await; + lease_of(&db, 2, TestKind::Liveness, expired_at, expired_at).await; + lease_of(&db, 3, TestKind::Liveness, expired_at, expired_at).await; + + let cleared = db + .clear_expired_testruns_in_progress(datetime!(2025-06-01 12:00:00 UTC)) + .await + .unwrap(); + + assert_eq!(cleared.get(&TestKind::Stress).copied(), Some(1)); + assert_eq!(cleared.get(&TestKind::Liveness).copied(), Some(2)); + assert!(remaining(&db).await.is_empty()); + } + #[tokio::test] async fn clears_nothing_when_every_lease_is_live() { let db = setup().await; @@ -1294,9 +1570,28 @@ mod tests { .clear_expired_testruns_in_progress(datetime!(2025-06-01 12:00:00 UTC)) .await .unwrap(); - assert_eq!(cleared, 0); + assert!(cleared.is_empty()); assert_eq!(remaining(&db).await, vec![1]); } + + // the per-kind gauges are published from this count, and a kind absent from the map is + // published as zero - so absence has to mean "none in flight", not "not measured" + #[tokio::test] + async fn in_flight_rows_are_counted_per_kind_and_a_drained_kind_is_absent() { + let db = setup().await; + for node_id in 1..=3 { + seed_node(&db, node_id).await; + } + let started_at = datetime!(2025-06-01 12:00:00 UTC); + let expires_at = datetime!(2025-06-01 13:00:00 UTC); + + lease_of(&db, 1, TestKind::Liveness, started_at, expires_at).await; + lease_of(&db, 2, TestKind::Liveness, started_at, expires_at).await; + + let counts = db.count_testruns_in_progress_by_kind().await.unwrap(); + assert_eq!(counts.get(&TestKind::Liveness).copied(), Some(2)); + assert_eq!(counts.get(&TestKind::Stress), None); + } } mod evict_old_testruns { @@ -1411,9 +1706,100 @@ mod tests { } } - mod assign_next_mixnode_testrun { + mod assign_next_testruns { use super::*; + // The role gate is the branch of the composed query that no stress request reaches, so this + // exercises it against one eligible node and one decoy per predicate it applies. + // One assignment, many targets, each locked and rotated in its own right - the property that + // separates a wave from the single-target assignment this used to be. + #[tokio::test] + async fn a_wave_locks_every_target_it_returns_and_stops_at_the_wave_size() { + let db = setup().await; + for node_id in 1..=3 { + seed_node(&db, node_id).await; + } + + let now = datetime!(2025-06-01 12:00:00 UTC); + let wave = db + .assign_next_testruns(&AssignmentRequest { + expires_at: now + time::Duration::minutes(1), + ..request(TestPairing::LIVENESS_MIXNODE, now, no_staleness_gate(), 2) + }) + .await + .unwrap(); + + assert_eq!(wave.len(), 2); + + for target in &wave { + let node_id = target.node.inner.node_id; + + // a lock per target, each carrying this wave's lease rather than one shared row + let row = db.get_testrun_in_progress(node_id).await.unwrap().unwrap(); + assert_eq!(row.expires_at, now + time::Duration::minutes(1)); + assert_eq!(row.test_kind, TestKind::Liveness); + assert_eq!(row.tested_role, TestedRole::Mixnode); + + // and a rotation pointer per target, under the pairing that was dispatched + let state = work_state(&db, node_id, TestKind::Liveness, TestedRole::Mixnode) + .await + .unwrap(); + assert_eq!(state.last_tested_ip.as_deref(), Some("1.2.3.4")); + } + + // the target the cap left behind is still assignable, i.e. it was passed over rather + // than locked + let remaining = db + .assign_next_testruns(&request( + TestPairing::LIVENESS_MIXNODE, + now, + no_staleness_gate(), + 2, + )) + .await + .unwrap(); + assert_eq!(remaining.len(), 1); + } + + #[tokio::test] + async fn a_gateway_request_takes_only_gateways_announcing_a_websocket_port() { + let db = setup().await; + db.batch_insert_or_update_nym_nodes(&[ + gateway_node(1, Some(9000)), + // gateway-capable, but has never reported the port the session needs + gateway_node(2, None), + // announces a port, but is not gateway-capable + NewNymNode { + clients_ws_port: Some(9000), + ..node(3, "key_3") + }, + ]) + .await + .unwrap(); + + let now = datetime!(2025-06-01 12:00:00 UTC); + let assigned = db + .assign_next_testruns(&request( + TestPairing::LIVENESS_GATEWAY, + now, + no_staleness_gate(), + 2, + )) + .await + .unwrap(); + + let assigned_ids: Vec<_> = assigned + .iter() + .map(|target| target.node.inner.node_id) + .collect(); + assert_eq!(assigned_ids, vec![1]); + + // the lock records the pairing that was dispatched, not the one stress would have used + let row = db.get_testrun_in_progress(1).await.unwrap().unwrap(); + assert_eq!(row.test_kind, TestKind::Liveness); + assert_eq!(row.tested_role, TestedRole::Gateway); + } + #[tokio::test] async fn returns_none_when_no_nodes() { let db = setup().await; @@ -1534,6 +1920,64 @@ mod tests { assert!(result.is_none()); } + // and the other direction, which is the one the liveness kind depends on: a node being + // stress-tested at high rate must not be measured by a liveness probe at the same time, or + // both results describe something other than the node + #[tokio::test] + async fn a_stress_run_in_flight_blocks_a_liveness_assignment() { + let db = setup().await; + seed_node(&db, 1).await; + let now = datetime!(2025-06-01 12:00:00 UTC); + assert!(assign(&db, now, no_staleness_gate()).await.is_some()); + + let wave = db + .assign_next_testruns(&request( + TestPairing::LIVENESS_MIXNODE, + now, + no_staleness_gate(), + 10, + )) + .await + .unwrap(); + assert!(wave.is_empty()); + } + + // The per-node lock is the WHOLE of the exclusion between kinds: there is no cooldown after + // it clears. The staleness gate here would reject the node if the stress run's timestamp + // were consulted for liveness, so this also pins that each pairing reads only its own. + #[tokio::test] + async fn a_node_freed_by_one_kind_is_immediately_assignable_by_another() { + let db = setup().await; + seed_node(&db, 1).await; + let now = datetime!(2025-06-01 12:00:00 UTC); + + // a stress run takes the node, completes, and releases the lock + assert!(assign(&db, now, no_staleness_gate()).await.is_some()); + insert_run( + &db, + &NewTestRun { + test_timestamp: now, + ..minimal_test_run(1) + }, + ) + .await; + assert!(db.get_testrun_in_progress(1).await.unwrap().is_none()); + + // at the very same instant, under a gate an hour in the past + let wave = db + .assign_next_testruns(&request( + TestPairing::LIVENESS_MIXNODE, + now, + now - time::Duration::hours(1), + 10, + )) + .await + .unwrap(); + + assert_eq!(wave.len(), 1); + assert_eq!(wave[0].node.inner.node_id, 1); + } + #[tokio::test] async fn skips_node_tested_too_recently() { let db = setup().await; @@ -1619,6 +2063,145 @@ mod tests { /// The two properties the three-part `(node_id, test_kind, tested_role)` work-state key exists /// to provide, driven through the real assignment and eviction paths. + mod peek_pairing_head { + use super::*; + + /// Everything that makes a node ineligible, so the peek is asked about a population where + /// only one node qualifies and every exclusion is represented. + async fn seed_mixed_population(db: &StorageManager) { + db.batch_insert_or_update_nym_nodes(&[ + // eligible, never tested by any pairing + node(1, "key_1"), + // eligible on paper, but locked by a run of another kind entirely + node(2, "key_2"), + // untestable: a gateway cannot be probed as a mixing hop + NewNymNode { + node_type: NodeType::Gateway, + ..node(3, "key_3") + }, + // untestable: never answered its self-described endpoint + NewNymNode { + noise_key: None, + ..node(4, "key_4") + }, + ]) + .await + .unwrap(); + mark_in_progress(db, 2, datetime!(2025-06-01 11:00:00 UTC)).await; + } + + #[tokio::test] + async fn a_pairing_with_nothing_eligible_has_no_head() { + let db = setup().await; + let head = db + .peek_pairing_head(TestPairing::LIVENESS_MIXNODE, no_staleness_gate()) + .await + .unwrap(); + assert!(head.is_none()); + } + + #[tokio::test] + async fn a_never_tested_node_reads_as_the_most_overdue_head() { + let db = setup().await; + seed_node(&db, 1).await; + + let head = db + .peek_pairing_head(TestPairing::LIVENESS_MIXNODE, no_staleness_gate()) + .await + .unwrap(); + assert_eq!(head, Some(PairingHead::NeverTested)); + + // and it outranks any measured node, which is what makes the most overdue head the + // minimum of the pairings a kind holds + assert!( + PairingHead::NeverTested + < PairingHead::LastTestedAt(datetime!(1970-01-01 00:00:00 UTC)) + ); + } + + #[tokio::test] + async fn a_measured_node_reads_as_its_last_run() { + let db = setup().await; + seed_node(&db, 1).await; + insert_run( + &db, + &NewTestRun { + test_kind: TestKind::Liveness, + test_timestamp: datetime!(2025-06-01 09:00:00 UTC), + ..minimal_test_run(1) + }, + ) + .await; + + let head = db + .peek_pairing_head(TestPairing::LIVENESS_MIXNODE, no_staleness_gate()) + .await + .unwrap(); + assert_eq!( + head, + Some(PairingHead::LastTestedAt(datetime!( + 2025-06-01 09:00:00 UTC + ))) + ); + + // the same node under a pairing that has never measured it still reads as never tested, + // so one pairing's progress cannot answer for another's + let other = db + .peek_pairing_head(TestPairing::STRESS_MIXNODE, no_staleness_gate()) + .await + .unwrap(); + assert_eq!(other, Some(PairingHead::NeverTested)); + } + + // The peek and the assignment are separate hand-written queries, each keeping its binds + // beside its own placeholders. That independence is only safe while they agree on who is + // eligible and in what order, so this asserts the peek describes the very row the + // assignment then takes, against a population exercising every exclusion. + #[tokio::test] + async fn the_peek_describes_the_node_the_assignment_takes() { + let db = setup().await; + seed_mixed_population(&db).await; + + let now = datetime!(2025-06-01 12:00:00 UTC); + let head = db + .peek_pairing_head(TestPairing::LIVENESS_MIXNODE, no_staleness_gate()) + .await + .unwrap(); + + let assigned = db + .assign_next_testruns(&request( + TestPairing::LIVENESS_MIXNODE, + now, + no_staleness_gate(), + 1, + )) + .await + .unwrap(); + + // the peek said there was work, and the assignment took exactly the node it described + assert_eq!(head, Some(PairingHead::NeverTested)); + assert_eq!(assigned.len(), 1); + assert_eq!(assigned[0].node.inner.node_id, 1); + + // with that node locked, both queries agree the pairing is drained + let head = db + .peek_pairing_head(TestPairing::LIVENESS_MIXNODE, no_staleness_gate()) + .await + .unwrap(); + let assigned = db + .assign_next_testruns(&request( + TestPairing::LIVENESS_MIXNODE, + now, + no_staleness_gate(), + 1, + )) + .await + .unwrap(); + assert!(head.is_none()); + assert!(assigned.is_empty()); + } + } + mod per_pairing_work_state { use super::*; diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/mod.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/mod.rs index 45533cd7b03..fa4d424a130 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/mod.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/mod.rs @@ -4,8 +4,9 @@ use crate::orchestrator::prometheus::{PROMETHEUS_METRICS, PrometheusMetric}; use crate::storage::manager::StorageManager; use crate::storage::models::{ - AssignedTestrun, CompletedTestRun, NewNymNode, NewTestRun, NymNode, TestKind, - TestRunInProgress, TestRunMeasurement, + AssignedTestrun, AssignmentRequest, BondedNymNode, CompletedTestRun, NewNymNode, NewTestRun, + NymNode, PairingHead, PairingSchedule, TestKind, TestPairing, TestRunInProgress, + TestRunMeasurement, }; use anyhow::Context; use nym_network_monitor_orchestrator_requests::models::Pagination; @@ -14,6 +15,7 @@ use sqlx::ConnectOptions; use sqlx::sqlite::{SqliteAutoVacuum, SqliteSynchronous}; use std::path::Path; use std::time::Duration; +use strum::IntoEnumIterator; use time::OffsetDateTime; use tracing::log::{LevelFilter, debug}; @@ -32,7 +34,44 @@ pub(crate) struct NetworkMonitorStorage { pub(crate) storage_manager: StorageManager, } +/// The in-flight gauge belonging to a kind. Exhaustive rather than defaulting, so a new kind is a +/// compile error here instead of a silently unpublished series. +fn in_progress_metric(kind: TestKind) -> PrometheusMetric { + match kind { + TestKind::Stress => PrometheusMetric::StressTestrunsInProgress, + TestKind::Liveness => PrometheusMetric::LivenessTestrunsInProgress, + } +} + +/// The expired-lease counter belonging to a kind, exhaustive for the same reason. +fn expired_leases_metric(kind: TestKind) -> PrometheusMetric { + match kind { + TestKind::Stress => PrometheusMetric::StressLeasesExpired, + TestKind::Liveness => PrometheusMetric::LivenessLeasesExpired, + } +} + impl NetworkMonitorStorage { + /// A migrated, empty database held entirely in memory, for tests that need storage without a + /// file on disk. + #[cfg(test)] + pub(crate) async fn in_memory() -> Self { + let connection_pool = sqlx::SqlitePool::connect("sqlite::memory:") + .await + .expect("failed to create in-memory SQLite pool"); + let migrations = Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations"); + sqlx::migrate::Migrator::new(migrations.as_path()) + .await + .expect("failed to find migrations") + .run(&connection_pool) + .await + .expect("failed to run migrations"); + + NetworkMonitorStorage { + storage_manager: StorageManager { connection_pool }, + } + } + /// Opens (or creates) the SQLite database at `database_path`, configures /// WAL journaling and incremental auto-vacuum, and runs the embedded /// migrations. Slow statements (>50ms) are logged at `WARN`. @@ -103,6 +142,15 @@ impl NetworkMonitorStorage { Ok(()) } + /// Records that these nodes are still bonded without touching anything learned from their own + /// endpoints, for nodes whose describe failed this cycle. + pub(crate) async fn batch_touch_bonded_nodes( + &self, + nodes: &[BondedNymNode], + ) -> anyhow::Result<()> { + self.storage_manager.batch_touch_bonded_nodes(nodes).await + } + /// The in-flight row for a node, i.e. what the orchestrator dispatched and is still waiting on. /// Read on submission to learn the kind and role a result must be recorded under, since the /// submission itself reports only the node and the address. @@ -124,54 +172,102 @@ impl NetworkMonitorStorage { /// Releases every in-flight lock whose lease has already expired, on the assumption that those /// runs will never report back. Decrements the `TestrunsInProgress` gauge by the number of rows - /// actually cleared. + /// actually cleared, and counts each kind's expiries against its own series. /// /// Takes no timeout: the deadline lives on each row, stamped at dispatch from the budget of the /// kind being dispatched, so this sweep needs no knowledge of any kind's lease. + /// + /// Returns the total number of locks released. pub(crate) async fn clear_expired_testruns_in_progress(&self) -> anyhow::Result { let cleared = self .storage_manager .clear_expired_testruns_in_progress(OffsetDateTime::now_utc()) .await?; - if cleared > 0 { - PROMETHEUS_METRICS.inc_by(PrometheusMetric::TestrunsInProgress, -(cleared as i64)); + + for (kind, count) in &cleared { + PROMETHEUS_METRICS.inc_by(expired_leases_metric(*kind), *count as i64); } - Ok(cleared) + + let total: u64 = cleared.values().sum(); + if total > 0 { + PROMETHEUS_METRICS.inc_by(PrometheusMetric::TestrunsInProgress, -(total as i64)); + } + Ok(total) } - /// Atomically selects the most stale idle mixnode and marks it as having a test run in - /// progress, with a lease of `lease_budget` from now. + /// Publishes the per-kind in-flight gauges from the authoritative row counts. /// - /// Staleness and the address rotation are evaluated for the `(stress, mixnode)` pairing alone, - /// so no other kind's cadence disturbs this one. "Most stale" means: nodes that pairing has - /// never tested come first, followed by those whose last run under it is oldest. + /// Set from a count rather than maintained by inc/dec like the total gauge: the delta paths + /// (assign, submit, expire) would each have to attribute their change to a kind, and a single + /// missed attribution leaves a per-kind gauge permanently wrong, whereas a recount cannot drift. + /// Every kind is published on every call, so a kind that has drained reads as zero rather than + /// holding its last value. + pub(crate) async fn publish_in_progress_gauges(&self) -> anyhow::Result<()> { + let counts = self + .storage_manager + .count_testruns_in_progress_by_kind() + .await?; + + for kind in TestKind::iter() { + PROMETHEUS_METRICS.set( + in_progress_metric(kind), + counts.get(&kind).copied().unwrap_or_default(), + ); + } + Ok(()) + } + + /// Atomically selects the nodes due for one (kind, role) pairing and marks each as having a test + /// run in progress, leased for `schedule.lease_budget` from now. One target for a stress + /// pairing, up to `schedule.wave_size` for a liveness one. /// - /// `staleness_age` acts as a minimum-staleness gate: a node already tested by this pairing is - /// only eligible if its last run completed more than `staleness_age` ago. Never-tested nodes - /// are always eligible. + /// Resolves the schedule's durations against a single `now`, so every gate applied and every row + /// stamped by one assignment agrees on when it happened. /// - /// Nodes with a row in `testrun_in_progress` are excluded whatever kind or role that row holds. - /// Only nodes classified as `mixnode` or `mixnode_and_gateway` are eligible. + /// "Most stale" means: nodes this pairing has never tested come first, followed by those whose + /// last run under it is oldest. `staleness_age` is a minimum-staleness gate that never-tested + /// nodes bypass. /// - /// Returns `None` if no eligible idle mixnode exists. - pub(crate) async fn assign_next_mixnode_testrun( + /// Nodes with a row in `testrun_in_progress` are excluded whatever kind or role that row holds, + /// and become eligible for any kind again as soon as that row clears. + /// + /// Returns an empty vector if nothing is eligible. + pub(crate) async fn assign_next_testruns( &self, - staleness_age: Duration, - lease_budget: Duration, - ) -> anyhow::Result> { + schedule: &PairingSchedule, + ) -> anyhow::Result> { let now = OffsetDateTime::now_utc(); - let last_tested_before = now - staleness_age; - let expires_at = now + lease_budget; - let assigned = self - .storage_manager - .assign_next_mixnode_testrun(now, last_tested_before, expires_at) - .await?; - if assigned.is_some() { - PROMETHEUS_METRICS.inc(PrometheusMetric::TestrunsInProgress); + let request = AssignmentRequest { + pairing: schedule.pairing, + now, + last_tested_before: now - schedule.staleness_age, + expires_at: now + schedule.lease_budget, + wave_size: schedule.wave_size, + }; + + let assigned = self.storage_manager.assign_next_testruns(&request).await?; + if !assigned.is_empty() { + PROMETHEUS_METRICS.inc_by(PrometheusMetric::TestrunsInProgress, assigned.len() as i64); } Ok(assigned) } + /// How overdue the node one pairing would assign next is, judged against the same staleness gate + /// the assignment would apply, or `None` if that pairing has nothing eligible. + /// + /// Read before dispatching a kind that owns more than one pairing, to settle which of them is + /// furthest behind. + pub(crate) async fn peek_pairing_head( + &self, + pairing: TestPairing, + staleness_age: Duration, + ) -> anyhow::Result> { + let last_tested_before = OffsetDateTime::now_utc() - staleness_age; + self.storage_manager + .peek_pairing_head(pairing, last_tested_before) + .await + } + /// Fetches a single completed test run with its measurements by its row id, or `None` if it /// has been evicted or never existed. pub(crate) async fn get_testrun_by_id( diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/models.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/models.rs index 2d138f645bd..9eddee0f7ca 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/models.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/storage/models.rs @@ -1,7 +1,7 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -use anyhow::Context; +use anyhow::{Context, bail}; use nym_api_requests::models::v3::StressTestResult; use nym_crypto::asymmetric::{ed25519, x25519}; use nym_network_monitor_orchestrator_requests::models::{ @@ -13,12 +13,16 @@ use nym_validator_client::client::NodeId; use nym_validator_client::nyxd::nym_mixnet_contract_common::NymNodeBond; use std::net::{IpAddr, SocketAddr}; use std::time::Duration; +use strum::{EnumCount, EnumIter}; use time::OffsetDateTime; /// What a test run measures. Selects the run's cadence, eligibility rules and expected measurement /// set, so - like its API counterpart - it deliberately has no `Default`: a silently defaulted kind /// would measure the wrong thing rather than fail. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, sqlx::Type)] +/// +/// Every kind exists in order to be assigned, so the scheduler rotates over the variants themselves +/// rather than over a list kept in step with them by hand, and does so in DECLARATION ORDER. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, sqlx::Type, EnumCount, EnumIter)] #[sqlx(type_name = "TEXT", rename_all = "lowercase")] pub(crate) enum TestKind { Stress, @@ -36,6 +40,45 @@ pub(crate) enum TestedRole { Gateway, } +/// A (kind, role) combination the orchestrator can assign, and the key under which each one keeps +/// its own work state in `node_test_state`: its own staleness position and its own address rotation +/// cursor. That independence is the point of the type - a `mixnode_and_gateway` node is due +/// separately as a mixing hop and as a gateway, and neither pairing's run moves the other's clock. +#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)] +pub(crate) struct TestPairing { + pub(crate) test_kind: TestKind, + pub(crate) tested_role: TestedRole, +} + +impl TestPairing { + pub(crate) const STRESS_MIXNODE: TestPairing = TestPairing { + test_kind: TestKind::Stress, + tested_role: TestedRole::Mixnode, + }; + + pub(crate) const LIVENESS_MIXNODE: TestPairing = TestPairing { + test_kind: TestKind::Liveness, + tested_role: TestedRole::Mixnode, + }; + + pub(crate) const LIVENESS_GATEWAY: TestPairing = TestPairing { + test_kind: TestKind::Liveness, + tested_role: TestedRole::Gateway, + }; +} + +impl TestKind { + /// The pairings this kind may assign, in the order that breaks a tie between two equally overdue + /// ones. A kind's roles follow from what its probe measures: forwarding is performed only by a + /// mixing hop, while the liveness probe has a shape for each role. + pub(crate) fn pairings(&self) -> &'static [TestPairing] { + match self { + TestKind::Stress => &[TestPairing::STRESS_MIXNODE], + TestKind::Liveness => &[TestPairing::LIVENESS_MIXNODE, TestPairing::LIVENESS_GATEWAY], + } + } +} + /// Which of the node's packet-handling interfaces a [`TestRunMeasurement`] describes. Names the /// node function exercised rather than a route; the test kind never appears here, since it is a /// property of the run and lives on the parent row. @@ -470,19 +513,21 @@ pub(crate) struct NewNymNode { pub(crate) clients_ws_port: Option, } -impl NewNymNode { +/// What is known about a node from its on-chain bond alone, i.e. without its own endpoint having +/// answered. Written on its own when a refresh could not describe the node, so that the bond is +/// still recorded without disturbing anything learned in an earlier cycle. +pub(crate) struct BondedNymNode { + pub(crate) node_id: i64, + pub(crate) identity_key: String, + pub(crate) last_seen_bonded: OffsetDateTime, +} + +impl BondedNymNode { pub(crate) fn from_bond(bond: &NymNodeBond) -> Self { - NewNymNode { + BondedNymNode { node_id: bond.node_id as i64, identity_key: bond.identity().to_string(), last_seen_bonded: OffsetDateTime::now_utc(), - mixnet_socket_address: None, - announced_ips: None, - noise_key: None, - sphinx_key: None, - key_rotation_id: None, - node_type: NodeType::Unknown, - clients_ws_port: None, } } } @@ -619,6 +664,71 @@ pub(crate) struct AssignmentCandidate { pub(crate) last_tested_ip: Option, } +/// How overdue the node a pairing would assign next is, i.e. the key the role selection within one +/// kind compares. +/// +/// `Ord` comes from the declaration order and then from the timestamp, so `NeverTested` outranks +/// every measured node and an older measurement outranks a newer one - the same ordering the +/// assignment query applies through `NULLS FIRST`, which is what makes the most overdue head the +/// minimum. +#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) enum PairingHead { + NeverTested, + LastTestedAt(OffsetDateTime), +} + +/// What the scheduler settled on for one pairing, expressed in the durations its configuration +/// carries. Resolved into an [`AssignmentRequest`] against a single `now`. +#[derive(Debug, Copy, Clone)] +pub(crate) struct PairingSchedule { + pub(crate) pairing: TestPairing, + + /// Minimum time since this pairing's last run against a node before it is due again. + pub(crate) staleness_age: Duration, + + /// How long a dispatched run holds its node before the lease expires. + pub(crate) lease_budget: Duration, + + /// Upper bound on the targets one assignment may carry: one for a stress run, the role's wave + /// size for a liveness run. + pub(crate) wave_size: usize, +} + +impl PairingSchedule { + /// The stress pairing. Its wave is always ONE target, since a stress assignment carries a single + /// probe target by construction. + pub(crate) fn stress(staleness_age: Duration, lease_budget: Duration) -> Self { + PairingSchedule { + pairing: TestPairing::STRESS_MIXNODE, + staleness_age, + lease_budget, + wave_size: 1, + } + } +} + +/// One pairing's dispatch parameters with every duration already resolved against one `now`, which +/// is what the assignment query binds. Absolute rather than relative so a caller can hold a single +/// timestamp across the gates it applies and the rows it stamps. +#[derive(Debug, Copy, Clone)] +pub(crate) struct AssignmentRequest { + pub(crate) pairing: TestPairing, + + /// Stamped as `started_at` on every in-progress row this assignment writes. + pub(crate) now: OffsetDateTime, + + /// Staleness gate: a node this pairing has tested before is eligible only if that run predates + /// this. Never-tested nodes bypass it. + pub(crate) last_tested_before: OffsetDateTime, + + /// Lease deadline stamped on every in-progress row, so the eviction sweep needs no knowledge of + /// which kind produced the row. + pub(crate) expires_at: OffsetDateTime, + + /// Maximum number of targets to select and lock. + pub(crate) wave_size: usize, +} + /// A node selected for a test run, along with the address that this particular run should target. pub(crate) struct AssignedTestrun { pub(crate) node: NymNode, @@ -627,6 +737,70 @@ pub(crate) struct AssignedTestrun { pub(crate) tested_ip: IpAddr, } +impl AssignedTestrun { + /// The target an agent probes over the node's mixnet listener: the stored keys decoded, and the + /// address this run rotated onto carrying the node's announced mix port. + /// + /// Every field it needs is one the assignment query filters on, so a missing one means + /// corruption or a schema regression rather than an untestable node - the same relationship + /// [`NymNodeData`]'s conversion has to its stored row, and reported the same way. + pub(crate) fn mixnet_probe_target(&self) -> anyhow::Result { + let node = &self.node.inner; + + let identity_key = ed25519::PublicKey::from_base58_string(&node.identity_key) + .context("invalid identity_key")?; + + let (Some(address), Some(noise_key), Some(sphinx_key), Some(key_rotation_id)) = ( + node.mixnet_socket_address.as_deref(), + node.noise_key.as_deref(), + node.sphinx_key.as_deref(), + node.key_rotation_id, + ) else { + bail!( + "node {} was assigned for testing without its complete data", + node.node_id + ) + }; + + // the stored socket address only contributes the mix port - the address under test comes + // from the rotation over everything the node announced + let mix_port = address + .parse::() + .context("invalid mixnet_socket_address")? + .port(); + + Ok(api::MixnetProbeTarget { + node_id: node.node_id as u32, + identity_key, + node_address: SocketAddr::new(self.tested_ip, mix_port), + node_ips: self.node.announced_ips(), + noise_key: x25519::PublicKey::from_base58_string(noise_key) + .context("invalid noise_key")?, + sphinx_key: x25519::PublicKey::from_base58_string(sphinx_key) + .context("invalid sphinx_key")?, + key_rotation_id: key_rotation_id as u32, + }) + } + + /// The gateway probe's target: [`Self::mixnet_probe_target`] for the egress phase, plus the + /// plain client websocket port the ingress phase opens its session on. The gateway role's + /// eligibility requires that port, so its absence here is likewise a stored-data fault. + pub(crate) fn gateway_probe_target(&self) -> anyhow::Result { + let mixnet = self.mixnet_probe_target()?; + let clients_ws_port = self + .node + .inner + .clients_ws_port + .context("missing clients_ws_port")?; + + Ok(api::GatewayProbeTarget { + mixnet, + clients_ws_port: u16::try_from(clients_ws_port) + .context("clients_ws_port outside the port range")?, + }) + } +} + /// Outcome of persisting a completed run: the id the run was stored under, and whether its /// in-flight row was still there to clear. The submission path rejects a result whose lease has /// already expired, so this is normally one - but the sweep can reap the row in the window between diff --git a/openspec/changes/network-monitor-liveness-tests/design.md b/openspec/changes/network-monitor-liveness-tests/design.md index 233d56994db..519599bcd33 100644 --- a/openspec/changes/network-monitor-liveness-tests/design.md +++ b/openspec/changes/network-monitor-liveness-tests/design.md @@ -115,14 +115,16 @@ The run produces two measurements. The score denominator is fixed by the kind at ### Decision 7: Per-kind staleness and rotation, retained per-node mutex, leases materialised on the row -**Choice.** A new `node_test_state (node_id, test_kind, tested_role)` table holds `last_tested_at`, `last_testrun_id` and `last_tested_ip`, replacing `nym_node.last_testrun` and `nym_node.last_tested_ip`. `testrun_in_progress` keeps its `node_id` primary key and gains `expires_at` plus `test_kind` and `tested_role` columns. Eviction becomes `DELETE FROM testrun_in_progress WHERE expires_at < ?`. Liveness eligibility additionally requires that the node's `(stress, mixnode)` `last_tested_at` is older than a cooldown. +**Choice.** A new `node_test_state (node_id, test_kind, tested_role)` table holds `last_tested_at`, `last_testrun_id` and `last_tested_ip`, replacing `nym_node.last_testrun` and `nym_node.last_tested_ip`. `testrun_in_progress` keeps its `node_id` primary key and gains `expires_at` plus `test_kind` and `tested_role` columns. Eviction becomes `DELETE FROM testrun_in_progress WHERE expires_at < ?`. The per-node in-flight row is the ONLY exclusion between kinds: a node is eligible for another kind the moment its row clears. -**Why.** Per-kind state is what stops a 15-minute liveness cadence and a 2-hour stress cadence from fighting over one staleness pointer and one rotation cursor. The key carries the ROLE as well as the kind because the two liveness probes are different measurements of the same node: a `mixnode_and_gateway` node must be eligible for both, and a two-part key would let its mixnode-liveness run advance the very timestamp that gates its gateway-liveness eligibility, so it would alternate roles across cycles instead of being measured in both. Materialising the deadline on the row means the eviction sweep never needs to learn about kinds, so a future expensive kind that runs for minutes needs no eviction change; today's sweep takes a single cutoff derived from a global `test_timeout` and cannot express two budgets. Keeping the in-progress key on `node_id` alone gives the "one test at a time per node, across kinds AND roles" property for free. The cooldown covers the case the mutex cannot: a liveness test handed out the instant a stress test's row clears measures a node whose queues are still draining. +**Why.** Per-kind state is what stops a 15-minute liveness cadence and a 2-hour stress cadence from fighting over one staleness pointer and one rotation cursor. The key carries the ROLE as well as the kind because the two liveness probes are different measurements of the same node: a `mixnode_and_gateway` node must be eligible for both, and a two-part key would let its mixnode-liveness run advance the very timestamp that gates its gateway-liveness eligibility, so it would alternate roles across cycles instead of being measured in both. Materialising the deadline on the row means the eviction sweep never needs to learn about kinds, so a future expensive kind that runs for minutes needs no eviction change; today's sweep takes a single cutoff derived from a global `test_timeout` and cannot express two budgets. Keeping the in-progress key on `node_id` alone gives the "one test at a time per node, across kinds AND roles" property for free. `tested_role` on `testrun_in_progress` is not observability, it is the authoritative source of the role when the result comes back. `testrun` records `tested_role`, and the submission carries only the node and the address, so without it the orchestrator would have to trust the agent's echo for a field it assigned itself. **Alternative considered.** A per-kind in-progress table. Rejected: it would permit simultaneous stress and liveness measurement of one node, which biases both. +**Also considered and DROPPED: a `liveness_after_stress_cooldown`**, withholding a node from liveness until some interval after its last stress run, on the theory that a probe handed out the instant the stress lock clears measures a node whose queues are still draining. Dropped because the load does not support it: this design already argues, when dismissing the same worry on the agent side, that a stress test's 30000 packets at 1000 packets/second is about 16 Mbps for a 2KB packet and saturates nothing, and a node that was not saturated is not draining seconds later either. The one contamination effect actually documented here is replay-deferral batching (Decision 11), which depends on CONCURRENT load that the per-node lock already excludes, and which shows up in latency rather than in the delivery ratio the score is built from. Keeping it would have cost an optional gate through the schedule, the request and the query, whose conditional bind was the most fragile part of the assignment. If a just-stressed node ever does score differently, the evidence for reintroducing it will be in the per-kind results. + The wss announcement is deliberately NOT stored alongside `clients_ws_port`, even though the refresher sees it: the only consumer is the divergence gauge's bucketing in Decision 12, which runs in nym-api and can read the same self-described `mixnet_websockets` interface from that side's described-nodes cache. Storing it here would be a second copy of a fact no orchestrator path reads, since the probe ignores wss entries by construction and the submission carries no such field. **Consequence.** Denormalising `last_tested_at` also fixes an existing defect. Today staleness is read through `JOIN testrun tr ON tr.id = n.last_testrun` with `ON DELETE SET NULL`, so when eviction removes a node's last run the node reads as never-tested and jumps the queue. @@ -222,8 +224,8 @@ Rollback at any step is a config change rather than a revert: liveness assignmen ## Resolved Questions -1. **Liveness packet count and per-target rate.** RESOLVED by decision rather than by measurement: the profile ships with provisional defaults chosen for score granularity, and every value is a configured knob. A per-target count of 100 gives 1% granularity and roughly 2.2% binomial noise at a true 95% delivery, against v1's three packets per route at 33% granularity, so it is a thirtyfold increase in evidence per node and a sensible floor. The aggregate rate budget and wave size start at 500 packets/second and 20 targets. Deliberately NOT measured first: the agent hosts are not the machines these numbers would be measured on, so an agent that cannot sustain the budget is a configuration change. The sweep arithmetic that the knobs must satisfy is `T_send = count x wave / aggregate_rate` and `population / wave <= invocations x (interval / T_wave)`. -2. **May a single agent invocation mix kinds?** RESOLVED, no, by construction: an invocation takes exactly ONE assignment, which is either a single stress target or a single liveness wave, then exits. No stickiness rule is needed. The residual concern is different from the one originally recorded: consecutive invocations on the same HOST, where a liveness wave measures an agent still recovering from a stress test and charges the loss to every node in the wave. Not addressed now, because 30000 packets at 1000pps is around 16 Mbps for a 2KB packet and saturates nothing on a container host, and sockets die with the process. If deployment ever shows contaminated waves, the remedy is a cooldown keyed on the agent's IP (not its socket address, since several agents share a host NIC behind distinct ports), symmetrical to the per-node `liveness_after_stress_cooldown`. +1. **Liveness packet count and per-target rate.** RESOLVED by decision rather than by measurement: the profile ships with provisional defaults chosen for score granularity, and every value is a configured knob. A per-target count of 100 gives 1% granularity and roughly 2.2% binomial noise at a true 95% delivery, against v1's three packets per route at 33% granularity, so it is a thirtyfold increase in evidence per node and a sensible floor. The aggregate rate budget starts at 500 packets/second, and the wave size is sized per role: 100 targets for the mixnode probe and 50 for the gateway probe. Splitting it follows from a wave being one concurrent batch, which makes the wave size the count an agent holds open at once: a gateway target costs a live client session measuring two interfaces, a mixnode target costs a Noise connection measuring one, and v1 already ran its whole gateway population through a 50-client window per cycle. The lease still does not scale with the wave, but it has to cover the slower of the two probes. Deliberately NOT measured first: the agent hosts are not the machines these numbers would be measured on, so an agent that cannot sustain the budget is a configuration change. The sweep arithmetic that the knobs must satisfy is `T_send = count x wave / aggregate_rate` and `population / wave <= invocations x (interval / T_wave)`. +2. **May a single agent invocation mix kinds?** RESOLVED, no, by construction: an invocation takes exactly ONE assignment, which is either a single stress target or a single liveness wave, then exits. No stickiness rule is needed. The residual concern is different from the one originally recorded: consecutive invocations on the same HOST, where a liveness wave measures an agent still recovering from a stress test and charges the loss to every node in the wave. Not addressed now, because 30000 packets at 1000pps is around 16 Mbps for a 2KB packet and saturates nothing on a container host, and sockets die with the process. If deployment ever shows contaminated waves, the remedy is a cooldown keyed on the agent's IP (not its socket address, since several agents share a host NIC behind distinct ports). The same arithmetic is why the per-node cooldown was dropped from Decision 7 rather than kept as its symmetrical counterpart. 3. **Does a wss-configured gateway still bind its plain ws port?** RESOLVED, yes, and more strongly than the question assumed: nym-node binds no TLS listener at all, so an announced wss entry always denotes an externally terminated proxy. See Decision 4. 4. **How ephemeral can the monitor session actually be?** RESOLVED, fully non-persisting is reachable, with the storage-assigned client id becoming optional. See Decision 5. 5. **Does the gateway's session path have access to the derived monitor set?** RESOLVED, it needs its own handle rather than a read of the mixnet structures, because the gateway client handling lives in a different crate from the routing set and noise map. `CommonHandlerState` already carries live shared handles passed down from nym-node (`upgrade_mode`, `active_clients_store`), so the identity set becomes one more field on it plus a builder setter, populated from the same startup load and websocket events. diff --git a/openspec/changes/network-monitor-liveness-tests/proposal.md b/openspec/changes/network-monitor-liveness-tests/proposal.md index 84657cd7b08..5f5dc0ba9af 100644 --- a/openspec/changes/network-monitor-liveness-tests/proposal.md +++ b/openspec/changes/network-monitor-liveness-tests/proposal.md @@ -13,7 +13,7 @@ Network monitor v3 already has the machinery to fix this: chain-authorised agent - **BREAKING (nym-node):** a client websocket session whose registration handshake authenticates an ed25519 identity announced on-chain for an authorised network-monitor agent MUST be treated as an ephemeral monitor session: unmetered, and writing nothing to gateway storage. The agent presents no ecash ticketbook. - **A contract migration adds an optional ed25519 identity key to each agent entry**, announced by the agent and written by the orchestrator in the existing authorisation transaction. This is what lets the session exemption above key on a cryptographically verified identity rather than on a source IP that k8s host ports share and CNI pools recycle, and it is the gate where an identity is already available: the gateway registration handshake proves possession of the client's ed25519 key before any session exists. The field is additive and optional, which un-upgraded nodes ignore, and it needs no data migration because the contract's agent save is an upsert and agents re-announce before every run. - **The agent tests a wave of targets concurrently** rather than one target per invocation, turning the assignment lease bound from the sum of per-target worst cases into the maximum of them, and making a full-network liveness sweep viable at v1's cadence. This requires a shared ingress listener, a multi-target Noise view, and per-target attribution of returned packets. -- **Per-test-kind scheduling in the orchestrator**: per-kind staleness gates, per-kind address rotation cursors, per-kind lease budgets materialised as an `expires_at` on each in-progress row, and a cooldown that keeps a liveness test from measuring a node still recovering from a stress test. The existing single-in-flight-test-per-node mutex is retained unchanged and now spans kinds. +- **Per-test-kind scheduling in the orchestrator**: per-kind staleness gates, per-kind address rotation cursors, and per-kind lease budgets materialised as an `expires_at` on each in-progress row. The existing single-in-flight-test-per-node mutex is retained unchanged, now spans kinds, and is the only exclusion between them. - **Per-kind result submission** to nym-api, with a separate watermark per kind and a per-signer replay high-water mark that cannot be shared between kinds. - **Liveness enters node performance as a third component with weight zero** (shadow mode) alongside the v1 routing score and the v3 stress score, plus a divergence gauge comparing v3 liveness against v1 routing, bucketed by whether the gateway announces a wss entry so that expected divergence is separable from unexpected divergence. diff --git a/openspec/changes/network-monitor-liveness-tests/specs/nym-network-monitor/spec.md b/openspec/changes/network-monitor-liveness-tests/specs/nym-network-monitor/spec.md index 79d2e543286..b8bc8c5a0db 100644 --- a/openspec/changes/network-monitor-liveness-tests/specs/nym-network-monitor/spec.md +++ b/openspec/changes/network-monitor-liveness-tests/specs/nym-network-monitor/spec.md @@ -211,7 +211,11 @@ Rehydrating that cache from the contract requires recovering which pair of on-ch ### Requirement: The node refresher builds the testable-node registry from the mixnet contract and each node's self-description -The node refresher SHALL source the node list from the MIXNET contract (all `NymNodeBond`s), NOT from nym-api. For each bonded node it MUST query that node's self-described HTTP endpoint directly (with host-info verification) to learn EVERY ip address the node announces, its announced mix port, its versioned x25519 noise key, its sphinx key and key-rotation id, and its role-derived `NodeType`. For a node that announces an entry-gateway interface it MUST additionally learn that interface's plain client websocket port. It MUST NOT record whether the node also announces a wss entry: the only consumer of that fact is the divergence gauge's bucketing, which lives in nym-api and reads the same self-described `mixnet_websockets` interface from its own described-nodes cache, so storing it here would be a second copy no orchestrator path reads. Per-node queries MUST be bounded by `node_info_query_timeout` (default 10 seconds) and run with concurrency `number_of_concurrent_node_queries` (default 32); a node that fails to answer leaves the corresponding fields NULL. The refresher MUST persist ALL bonded nodes, including unreachable ones (upserting on `node_id`, updating every field except `identity_key`), so that previously-learned keys are retained when a node is transiently unreachable. +The node refresher SHALL source the node list from the MIXNET contract (all `NymNodeBond`s), NOT from nym-api. For each bonded node it MUST query that node's self-described HTTP endpoint directly (with host-info verification) to learn EVERY ip address the node announces, its announced mix port, its versioned x25519 noise key, its sphinx key and key-rotation id, and its role-derived `NodeType`. For a node that announces an entry-gateway interface it MUST additionally learn that interface's plain client websocket port. It MUST NOT record whether the node also announces a wss entry: the only consumer of that fact is the divergence gauge's bucketing, which lives in nym-api and reads the same self-described `mixnet_websockets` interface from its own described-nodes cache, so storing it here would be a second copy no orchestrator path reads. Per-node queries MUST be bounded by `node_info_query_timeout` (default 10 seconds) and run with concurrency `number_of_concurrent_node_queries` (default 32). + +A node MUST be described COMPLETELY or not at all: every self-described field comes from one reading of the node's endpoint, and a failure of any part of that reading - including the client websocket interface of a gateway-capable node - MUST discard the whole reading rather than storing the fields that did answer. The refresher MUST persist ALL bonded nodes, including unreachable ones, but the two outcomes are written differently: a described node has every field replaced, while a node that could not be described has only its bond recorded, leaving everything an earlier cycle learned about it in place. Nulling those fields instead would fail every eligibility predicate at once and drop a merely slow node out of EVERY kind until a later cycle answered, which at the liveness cadence costs several test slots per incident. Empty self-described columns therefore mean "never described", not "did not answer this time". `identity_key` is never updated, since a `node_id` maps to exactly one identity and is never reassigned. + +The consequence, accepted deliberately, is that a node which stops answering keeps its last reading indefinitely and continues to be assigned against it. That is the intended behaviour: a probe against stale data fails, and for a liveness measurement an unreachable node failing its probe is the measurement, whereas silently not testing it is not. The announced address set MUST be canonicalised (`IpAddr::to_canonical()`), deduplicated and sorted before being stored, because test runs rotate through it by position: a node is free to report its addresses in a different order on every refresh (a resolved hostname typically will), and a duplicate entry would stall the rotation on a subset of the set. The stored `mixnet_socket_address` MUST be derived deterministically from the first address of that sorted set plus the announced mix port, and contributes only that port to the address a given run actually targets. @@ -229,15 +233,27 @@ The announced address set MUST be canonicalised (`IpAddr::to_canonical()`), dedu #### Scenario: An unreachable node is retained with prior data - **WHEN** a bonded node does not answer within `node_info_query_timeout` -- **THEN** the node row is still upserted, leaving newly-unknown fields NULL and keeping any previously stored keys +- **THEN** only its bond is recorded, and every field an earlier cycle learned about it survives untouched, so it stays eligible for testing + +#### Scenario: A node seen for the first time without answering is a stub +- **WHEN** a node the orchestrator has never described does not answer +- **THEN** its row is inserted with the self-described columns empty, which is what marks it as never described rather than as unreachable this cycle + +#### Scenario: A partial reading is discarded +- **WHEN** a gateway-capable node answers for its keys and roles but not for its client websocket interface +- **THEN** the whole reading is discarded and the node keeps its previous data, rather than being stored as described everywhere except that interface ### Requirement: Testruns are assigned lazily from a staleness-ordered node table guarded by an in-flight lock set There SHALL be no in-memory work queue. Work is identified by `(node_id, test_kind)`, and staleness, the address rotation and the eligibility gates are all evaluated PER KIND, so that kinds running at different cadences do not disturb one another. -When an agent requests work, the orchestrator MUST choose the kind, then select targets inside a `BEGIN IMMEDIATE` write transaction that: excludes any node with a `testrun_in_progress` row, REGARDLESS of which kind that row belongs to; requires the fields that kind needs to be non-null; requires the node's type to be one the kind may assign; treats a node as eligible only if that kind has never tested it or last tested it before `now - staleness_age` for that kind; for the `liveness` kind additionally requires that the node's `(stress, mixnode)` pairing last ran before `now - liveness_after_stress_cooldown`; orders by that kind's test timestamp ascending with never-tested first; takes one target for a `stress` assignment or up to `liveness_wave_size` targets for a `liveness` assignment; rotates each selected node onto the next address in its announced set FOR THAT KIND AND ROLE; records that address as that pairing's rotation pointer; and atomically inserts a `testrun_in_progress` row for each, stamped with `started_at`, the kind, the role, and an `expires_at` of `now` plus that kind's lease budget. The response MUST carry the chosen kind and its per-target payload, or an empty assignment when no eligible node exists. +When an agent requests work, the orchestrator MUST choose the kind, then select targets inside a `BEGIN IMMEDIATE` write transaction that: excludes any node with a `testrun_in_progress` row, REGARDLESS of which kind that row belongs to; requires the fields that kind needs to be non-null; requires the node's type to be one the kind may assign; treats a node as eligible only if that kind has never tested it or last tested it before `now - staleness_age` for that kind; orders by that kind's test timestamp ascending with never-tested first; takes one target for a `stress` assignment or up to the chosen role's wave size for a `liveness` assignment; rotates each selected node onto the next address in its announced set FOR THAT KIND AND ROLE; records that address as that pairing's rotation pointer; and atomically inserts a `testrun_in_progress` row for each, stamped with `started_at`, the kind, the role, and an `expires_at` of `now` plus that kind's lease budget. The response MUST carry the chosen kind and its per-target payload, or an empty assignment when no eligible node exists. -Excluding any node that has an open in-progress row of ANY kind is required, not incidental: a node being stress-tested at high rate while a liveness probe measures it would bias both results. +The kind MUST be chosen by a cursor that rotates over the kinds themselves, advancing once per request, so that no kind can starve another: the stress kind is un-waved and therefore needs the majority of assignments, while the liveness kind comes due several times as often, and any fixed share would under-serve one of them. A kind whose enable flag is unset, or which has nothing due, MUST fall through to the next rather than spending the request. The rotation MUST be over kinds ONLY and not over (kind, role) pairings, so that a further kind joins it without a policy change. + +Within the chosen kind, the pairing MUST be the one whose next assignable node is furthest behind, judged on the same staleness ordering the assignment applies, with never-tested outranking every measured node. The role is therefore NOT a scheduling decision: the two liveness roles interleave in proportion to how far behind each has fallen, because serving one advances its own staleness position and hands the next turn to the other. Equally overdue pairings MUST resolve deterministically rather than arbitrarily, so that a freshly migrated database - where every pairing is equally never-tested - drains predictably. + +Excluding any node that has an open in-progress row of ANY kind is required, not incidental: a node being stress-tested at high rate while a liveness probe measures it would bias both results. That lock is also the ONLY exclusion between kinds: a node is eligible again the moment its row clears, with no cooldown afterwards, because a stress test's 30000 packets at 1000 packets/second is around 16 Mbps for a 2KB packet and so leaves nothing draining for a later probe to charge to the node. The rotation MUST take the address following the previously handed-out one for that kind, wrapping around at the end of the set and restarting from the first address when the pointer is unset or no longer announced. It MUST advance when the assignment is handed out rather than when a result arrives, so a run that is abandoned still moves the node onto its next address. A node stored before the announced set was tracked MUST remain testable by falling back to the single address in its `mixnet_socket_address`. @@ -247,6 +263,14 @@ The staleness gate is per NODE AND KIND while the rotation is per ADDRESS, so a - **WHEN** an authorised, announced agent requests a testrun and eligible nodes exist - **THEN** the orchestrator picks a kind and returns the never-tested-or-oldest node for that kind, inserting a `testrun_in_progress` row for it in the same transaction +#### Scenario: Successive requests rotate the kind +- **WHEN** two agents ask for work in turn and both kinds have nodes due +- **THEN** the first is handed one kind and the second the other, so neither cadence is starved by the other's backlog + +#### Scenario: A kind with nothing due gives up its turn +- **WHEN** it is one kind's turn but that kind is disabled, or none of its pairings has an eligible node +- **THEN** the request falls through to the next kind rather than being answered with no work + #### Scenario: A node under one kind of test is not assigned another - **WHEN** a node has an open `testrun_in_progress` row from a stress test - **THEN** it is excluded from liveness assignment until that row is cleared, and vice versa @@ -255,13 +279,13 @@ The staleness gate is per NODE AND KIND while the rotation is per ADDRESS, so a - **WHEN** a liveness test and a stress test are both assigned for one node over time - **THEN** each kind advances its own rotation pointer, so neither skips addresses because of the other -#### Scenario: A recently stress-tested node is not immediately liveness-tested -- **WHEN** a node's stress test completed more recently than `liveness_after_stress_cooldown` -- **THEN** it is not eligible for a liveness assignment yet, so its liveness score is not measured while it is still recovering from load +#### Scenario: A node freed by one kind is immediately available to another +- **WHEN** a node's stress test finishes and its in-flight row clears +- **THEN** it is eligible for a liveness assignment straight away, the per-node lock being the whole of the mutual exclusion between kinds #### Scenario: A liveness assignment carries a wave - **WHEN** an agent is assigned liveness work and many nodes are eligible -- **THEN** up to `liveness_wave_size` targets are returned in one assignment, each with its own in-flight row and lease +- **THEN** up to that role's wave size of targets are returned in one assignment, each with its own in-flight row and lease #### Scenario: No eligible node yields an empty assignment - **WHEN** every node is either in progress or was tested by the chosen kind more recently than its `staleness_age` @@ -482,11 +506,11 @@ While the weight is zero, nym-api MUST expose a DIVERGENCE metric comparing each The orchestrator SHALL be configured with the following defaults: `test_interval` 2 hours, `test_timeout` 5 minutes, `node_refresh_rate` 2 hours, `node_info_query_timeout` 10 seconds, `testrun_eviction_age` 7 days, `result_submission_interval` 15 minutes, `result_submission_batch_size` 50, `number_of_concurrent_node_queries` 32, `chain_authorisation_check_max_attempts` 10, `chain_authorisation_check_retry_delay` 1 minute, and an HTTP bind of `0.0.0.0:8080`; plus required secrets (`agents_token`, `metrics_and_results_token`, the bip39 `mnemonic`, and the base58 ed25519 `private_key`) and required endpoints (`nym_api_endpoint`, `rpc_url`, the mixnet and network-monitors contract addresses, and `database_path`). -Where a knob governs a per-kind behaviour it MUST be expressible per kind. The orchestrator MUST additionally carry, for the liveness kind: a staleness interval (defaulting well below the stress `test_interval`, so that liveness tracks v1's cadence), a lease budget used as the in-progress `expires_at` (which MUST bound one concurrent wave, not the sum over its targets), a wave size, a `liveness_after_stress_cooldown`, and an enable flag allowing liveness assignment to be switched off without redeploying. `test_timeout` remains the stress kind's lease budget. +Where a knob governs a per-kind behaviour it MUST be expressible per kind. The orchestrator MUST additionally carry, for the liveness kind: a staleness interval (defaulting well below the stress `test_interval`, so that liveness tracks v1's cadence) of 15 minutes, a lease budget used as the in-progress `expires_at` (which MUST bound one concurrent wave, not the sum over its targets, and MUST therefore cover the slower of the two probes) of 1 minute, a wave size PER ROLE - 100 for mixnode probes and 50 for gateway probes, since a wave is one concurrent batch and a gateway target costs the agent a live client session where a mixnode target costs a Noise connection - and an enable flag allowing liveness assignment to be switched off without a code change, i.e. by configuration and a restart rather than by a build, defaulting to ON and therefore expressible as an explicit value rather than as a bare presence flag, since a flag that can only be switched on cannot be switched off by a deployment. `test_timeout` remains the stress kind's lease budget. The agent SHALL be configured with the following defaults: `sending_duration` 30 seconds, `waiting_duration` 5 seconds, `packet_delay` 50 milliseconds (which MUST be non-zero), `target_rate` 1000 packets/second, `reuse_header` true, `egress_connection_timeout` 5 seconds, `noise_handshake_timeout` 3 seconds, `sending_batch_size` 50, and a listener bind of `[::]:9000`; plus the required orchestrator URL, orchestrator bearer token, announced ipv4 host address, announced ipv6 host address, shared announced port, and noise-key path. The agent MUST additionally carry a liveness profile: a per-target packet count, an AGGREGATE send-rate budget from which the per-target rate is derived (never the reverse), a straggler wait, and per-target timeouts. All knobs MUST be overridable by CLI flag or environment variable. -The liveness profile's initial values are PROVISIONAL, chosen for score granularity rather than measured against agent hardware: a per-target packet count of 100 (giving 1% granularity against v1's three packets per route), an aggregate budget of 500 packets/second, and a wave size of 20. Because they are provisional, every one of them MUST be tunable in a deployment without a code change, and no behaviour may depend on a specific value: an agent host that cannot sustain the aggregate budget MUST be correctable by configuration alone. +The liveness profile's initial values are PROVISIONAL, chosen for score granularity rather than measured against agent hardware: a per-target packet count of 100 (giving 1% granularity against v1's three packets per route), an aggregate budget of 500 packets/second, and wave sizes of 100 mixnode targets and 50 gateway targets. Because they are provisional, every one of them MUST be tunable in a deployment without a code change, and no behaviour may depend on a specific value: an agent host that cannot sustain the aggregate budget MUST be correctable by configuration alone. The announced pair MUST be validated at configuration time, applying the same rule the orchestrator enforces on announce, so a misconfigured deployment fails immediately rather than on its first announcement. The listener bind default MUST remain dual-stack (`[::]`), since an ipv4-only bind cannot receive the return traffic for a run whose return hop is the agent's ipv6 address, and since one shared listener serves every target of a concurrent wave. @@ -502,7 +526,7 @@ The announced pair MUST be validated at configuration time, applying the same ru - **WHEN** the agent is configured with two announced addresses of the same family, or with an ipv4-mapped ipv6 address - **THEN** configuration construction fails before the agent announces itself -#### Scenario: Liveness can be disabled without a redeploy +#### Scenario: Liveness can be disabled without a new build - **WHEN** the orchestrator's liveness enable flag is unset - **THEN** no liveness assignment is handed out and stress testing continues unaffected diff --git a/openspec/changes/network-monitor-liveness-tests/tasks.md b/openspec/changes/network-monitor-liveness-tests/tasks.md index 24cee95dcd1..699d4563ed3 100644 --- a/openspec/changes/network-monitor-liveness-tests/tasks.md +++ b/openspec/changes/network-monitor-liveness-tests/tasks.md @@ -1,6 +1,6 @@ ## 1. Provisional defaults and the one compatibility check -- [ ] 1.1 Adopt the provisional liveness profile defaults (100 packets per target, a 500 packets/second aggregate budget, a wave size of 20) and make every one of them CLI- and env-overridable, so an agent host that cannot sustain the budget is a configuration change rather than a code change +- [ ] 1.1 Adopt the provisional liveness profile defaults (100 packets per target, a 500 packets/second aggregate budget, wave sizes of 100 mixnode and 50 gateway targets) and make every one of them CLI- and env-overridable, so an agent host that cannot sustain the budget is a configuration change rather than a code change - [ ] 1.2 Size the liveness lease budget, straggler wait and per-target timeouts from those defaults with slack, and make them configurable on the same terms, with no behaviour depending on a specific value - [ ] 1.3 Confirm on a devnet that an un-upgraded node still ingests an `AuthoriseNetworkMonitor` carrying an unknown field, rather than logging a parse failure and skipping it. The `cw_serde` reading says it will; this assumption is load-bearing for every future contract change and gates the orchestrator deploy, not the migration, which emits no message at all @@ -41,12 +41,12 @@ ## 5. Orchestrator scheduling -- [ ] 5.1 Rewrite `assign_next_mixnode_testrun` as a kind-aware assignment: choose the kind, filter by that kind's eligible node types and required non-null fields, apply that kind's staleness age for the chosen role, exclude any node with an in-progress row of any kind or role, apply `liveness_after_stress_cooldown` for liveness, take one target for stress or up to `liveness_wave_size` for liveness, advance each node's rotation pointer for that (kind, role) pairing, and insert one in-progress row per target with its lease, kind and role -- [ ] 5.2 Add the kind-selection policy (which kind an agent is handed when several are due) and the liveness enable flag that switches liveness assignment off without a redeploy -- [ ] 5.3 Extend the node refresher to record the entry-gateway client websocket port -- [ ] 5.4 Add the liveness config knobs (staleness interval, lease budget, wave size, cooldown, enable flag) with the provisional defaults from 1.1 and 1.2, all CLI- and env-overridable -- [ ] 5.5 Add prometheus series for liveness assignments, wave sizes, per-kind in-progress counts, lease expiries, and cooldown skips -- [ ] 5.6 Unit-test that a node with an open stress in-progress row is not assigned liveness and vice versa, that a recently stress-tested node is skipped by the cooldown, and that a wave never exceeds `liveness_wave_size` +- [x] 5.1 Rewrite `assign_next_mixnode_testrun` as a kind-aware assignment: choose the kind, filter by that kind's eligible node types and required non-null fields, apply that kind's staleness age for the chosen role, exclude any node with an in-progress row of any kind or role, take one target for stress or up to the chosen role's liveness wave size, advance each node's rotation pointer for that (kind, role) pairing, and insert one in-progress row per target with its lease, kind and role +- [x] 5.2 Add the kind-selection policy (which kind an agent is handed when several are due) and the liveness enable flag that switches liveness assignment off without a redeploy +- [x] 5.3 Extend the node refresher to record the entry-gateway client websocket port +- [x] 5.4 Add the liveness config knobs (staleness interval, lease budget, per-role wave sizes, enable flag) with the provisional defaults from 1.1 and 1.2, all CLI- and env-overridable +- [x] 5.5 Add prometheus series for liveness assignments, wave sizes, per-kind in-progress counts, and lease expiries +- [x] 5.6 Unit-test that a node with an open stress in-progress row is not assigned liveness and vice versa, that a node freed by one kind is immediately assignable by another, and that a wave never exceeds its role's wave size ## 6. nym-node: final-hop delivery for monitors