From 14a85901b4752ff221f2a4ab786384c279e022d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Mon, 1 Jun 2026 11:44:31 +0100 Subject: [PATCH] Bugfix/cherry pick/waterloo stres testing floats (#6841) * add additional information upon stress testing data submission failure * split stress testing result submission into batches of maximum size * enable 'float_roundtrip' serde_json feature to ensure consistent float serialisation --- Cargo.lock | 4 +- Cargo.toml | 2 +- nym-api/Cargo.toml | 2 +- .../src/models/network_monitor.rs | 164 ++++++++++++++++++ .../Cargo.toml | 2 +- .../src/cli/env.rs | 3 + .../src/cli/run_orchestrator.rs | 7 +- .../src/orchestrator/config.rs | 3 + .../src/orchestrator/result_submitter.rs | 69 ++++++-- .../src/storage/models.rs | 8 + 10 files changed, 239 insertions(+), 25 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6e2f149a0b..ce22ac9bcf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5819,7 +5819,7 @@ dependencies = [ [[package]] name = "nym-api" -version = "1.1.80" +version = "1.1.80-fixed-floats" dependencies = [ "anyhow", "async-trait", @@ -7604,7 +7604,7 @@ dependencies = [ [[package]] name = "nym-network-monitor-orchestrator" -version = "1.0.3" +version = "1.0.5" dependencies = [ "anyhow", "axum", diff --git a/Cargo.toml b/Cargo.toml index 72e8d59a1f..b5798320e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -353,7 +353,7 @@ semver = "1.0.26" serde = "1.0.219" serde_bytes = "0.11.17" serde_derive = "1.0" -serde_json = "1.0.140" +serde_json = { version = "1.0.140", features = ["float_roundtrip"] } serde_json_path = "0.7.2" serde_repr = "0.1" serde_with = "3.9.0" diff --git a/nym-api/Cargo.toml b/nym-api/Cargo.toml index 29dc4d67b9..2316433f77 100644 --- a/nym-api/Cargo.toml +++ b/nym-api/Cargo.toml @@ -3,7 +3,7 @@ [package] name = "nym-api" -version = "1.1.80" +version = "1.1.80-fixed-floats" authors.workspace = true edition = "2021" license = "GPL-3.0" diff --git a/nym-api/nym-api-requests/src/models/network_monitor.rs b/nym-api/nym-api-requests/src/models/network_monitor.rs index 3075a76e88..824a0a2165 100644 --- a/nym-api/nym-api-requests/src/models/network_monitor.rs +++ b/nym-api/nym-api-requests/src/models/network_monitor.rs @@ -181,4 +181,168 @@ pub mod v3 { /// as an authorised network monitor permitted to submit stress testing results. pub authorised: bool, } + + #[cfg(test)] + mod tests { + use super::*; + use crate::signable::SignableMessageBody; + use nym_test_utils::helpers::deterministic_rng; + use time::macros::datetime; + + fn dummy_results() -> Vec { + // Order-distinguishable entries: if deserialisation ever permuted the array, the + // re-serialised body would no longer match the signed bytes, and `verify_signature` + // would return false. `testrun_id` is the order witness. + vec![ + StressTestResult { + testrun_id: 1, + node_id: 42, + is_mixnode: true, + test_timestamp: datetime!(2026-06-01 12:34:56.123456789 UTC), + test_performance: 0.6666666666666666, + was_reachable: true, + }, + StressTestResult { + testrun_id: 2, + node_id: 7, + is_mixnode: true, + test_timestamp: datetime!(2026-06-01 12:34:56 UTC), + test_performance: 0.0, + was_reachable: false, + }, + StressTestResult { + testrun_id: 3, + node_id: u32::MAX, + is_mixnode: true, + test_timestamp: datetime!(2026-06-01 12:34:56.999999999 UTC), + test_performance: 1.0, + was_reachable: true, + }, + ] + } + + // Integrity check on the wire is `serde_json::to_vec(deserialize(serde_json::to_vec(body))) + // == serde_json::to_vec(body)`. If JSON serialisation isn't a fixed point, every batch + // submission would fail nym-api's signature verification. Cover the timestamp shapes the + // orchestrator actually produces, including the `+1ns` bump from the monotonicity safeguard. + #[test] + fn signed_batch_submission_roundtrips_through_json() { + let mut rng = deterministic_rng(); + let keys = ed25519::KeyPair::new(&mut rng); + + let timestamps = [ + datetime!(2026-06-01 12:34:56 UTC), + datetime!(2026-06-01 12:34:56.000000001 UTC), + datetime!(2026-06-01 12:34:56.999999999 UTC), + datetime!(2026-06-01 12:34:56.123456789 UTC), + OffsetDateTime::now_utc(), + OffsetDateTime::now_utc() + time::Duration::NANOSECOND, + ]; + + for timestamp in timestamps { + let body = StressTestBatchSubmissionContent { + signer: *keys.public_key(), + timestamp, + results: dummy_results(), + }; + let signed = body.clone().sign(keys.private_key()); + + let bytes = serde_json::to_vec(&signed).unwrap(); + let deserialised: StressTestBatchSubmission = + serde_json::from_slice(&bytes).unwrap(); + + // The handler verifies against `body.body.signer` — match that exactly. + assert!( + deserialised.verify_signature(&deserialised.body.signer), + "signature failed to verify after JSON round-trip for timestamp {timestamp}", + ); + assert_eq!(deserialised.body.timestamp, timestamp); + } + } + + // Every f64 that the orchestrator's `received as f64 / sent as f64` formula can produce + // (storage/models.rs) must round-trip byte-exactly through JSON. Exhaustively cover the + // range and exercise sent values that produce non-terminating fractions (1/3, 1/7, ...). + #[test] + fn computed_test_performance_values_roundtrip() { + for sent in 1u64..=200 { + for received in 0u64..=(sent * 2) { + let perf = received as f64 / sent as f64; + let s = serde_json::to_string(&perf).unwrap(); + let perf2: f64 = serde_json::from_str(&s).unwrap(); + let s2 = serde_json::to_string(&perf2).unwrap(); + assert_eq!( + s, s2, + "f64 round-trip mismatch for {received}/{sent} = {perf}: {s} -> {s2}", + ); + } + } + } + + // serde_json serialises non-finite f64 as `null`. Confirm what the deserialiser does with + // `null` for a struct field typed as f64 - if it succeeds with a default value (rather than + // erroring), a NaN/Infinity test_performance could silently break signature verification + // because the re-serialised body would no longer have `null` at that position. + #[test] + fn non_finite_test_performance_breaks_loudly_not_silently() { + let nan_result = StressTestResult { + testrun_id: 1, + node_id: 1, + is_mixnode: true, + test_timestamp: datetime!(2026-06-01 12:34:56 UTC), + test_performance: f64::NAN, + was_reachable: true, + }; + let json = serde_json::to_string(&nan_result).unwrap(); + // NaN serialises as `null` - this is the dangerous shape + assert!( + json.contains(r#""test_performance":null"#), + "expected NaN to serialise as null: {json}", + ); + // ...and `null` MUST fail to deserialise rather than silently becoming 0.0 / default; + // if this ever changes, NaN would silently corrupt signature verification. + let deserialised: Result = serde_json::from_str(&json); + assert!( + deserialised.is_err(), + "deserialising null into f64 unexpectedly succeeded - signature verification \ + would silently fail for any submission containing a non-finite test_performance", + ); + } + + // Specifically pin the two hypotheses we want to rule out: + // 1. Vec serialisation/deserialisation preserves order. + // 2. The body bytes serialised standalone (= what gets signed) are byte-identical to + // the body sub-object bytes embedded in the outer SignedMessage JSON (= what the + // server sees after parsing). Re-serialising the deserialised body must reproduce + // the signed bytes verbatim, otherwise no signature could ever verify. + #[test] + fn batch_body_serialisation_is_a_byte_exact_fixed_point() { + let mut rng = deterministic_rng(); + let keys = ed25519::KeyPair::new(&mut rng); + + let body = StressTestBatchSubmissionContent { + signer: *keys.public_key(), + timestamp: datetime!(2026-06-01 12:34:56.123456789 UTC), + results: dummy_results(), + }; + + let signed_bytes = body.plaintext(); + let body_str = std::str::from_utf8(&signed_bytes).unwrap(); + + // (1) array order preserved on the wire + let pos1 = body_str.find(r#""testrun_id":1"#).unwrap(); + let pos2 = body_str.find(r#""testrun_id":2"#).unwrap(); + let pos3 = body_str.find(r#""testrun_id":3"#).unwrap(); + assert!(pos1 < pos2 && pos2 < pos3, "JSON: {body_str}"); + + // (2) round-trip is byte-exact + let deserialised: StressTestBatchSubmissionContent = + serde_json::from_slice(&signed_bytes).unwrap(); + let resigned_bytes = deserialised.plaintext(); + assert_eq!( + signed_bytes, resigned_bytes, + "deserialise-then-re-serialise was not a fixed point" + ); + } + } } diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/Cargo.toml b/nym-network-monitor-v3/nym-network-monitor-orchestrator/Cargo.toml index e0b2516158..175721766a 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/Cargo.toml +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "nym-network-monitor-orchestrator" description = "Orchestrator for performing Nym network stress testing" -version = "1.0.3" +version = "1.0.5" authors.workspace = true edition.workspace = true license.workspace = true 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 d37292bc8f..6ee20dc7d2 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 @@ -37,4 +37,7 @@ pub mod vars { "NYM_NETWORK_MONITOR_CHAIN_AUTH_CHECK_RETRY_DELAY"; pub const NYM_NETWORK_MONITOR_RESULT_SUBMISSION_INTERVAL_ARG: &str = "NYM_NETWORK_MONITOR_RESULT_SUBMISSION_INTERVAL"; + + pub const NYM_NETWORK_MONITOR_RESULT_SUBMISSION_BATCH_SIZE_ARG: &str = + "NYM_NETWORK_MONITOR_RESULT_SUBMISSION_BATCH_SIZE"; } 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 fa85d2aa62..597b65e6bc 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 @@ -9,7 +9,7 @@ use nym_crypto::asymmetric::ed25519; use nym_validator_client::nyxd::bip39; use std::mem; use std::net::SocketAddr; -use std::num::NonZeroU32; +use std::num::{NonZeroU32, NonZeroUsize}; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; @@ -109,6 +109,10 @@ pub(crate) struct Args { /// batch submission (e.g. `15m`, `1h`). #[clap(long, env = NYM_NETWORK_MONITOR_RESULT_SUBMISSION_INTERVAL_ARG, value_parser = humantime::parse_duration, default_value = "15m")] result_submission_interval: Duration, + + /// Maximum number of stress testing results to submit in a single POST request + #[clap(long, env = NYM_NETWORK_MONITOR_RESULT_SUBMISSION_BATCH_SIZE_ARG, default_value = "50")] + result_submission_batch_size: NonZeroUsize, } impl Args { @@ -145,6 +149,7 @@ impl Args { chain_authorisation_check_max_attempts: self.chain_authorisation_check_max_attempts, chain_authorisation_check_retry_delay: self.chain_authorisation_check_retry_delay, result_submission_interval: self.result_submission_interval, + result_submission_batch_size: self.result_submission_batch_size.get(), }) } 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 cad15f3692..5f23be6a9c 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 @@ -71,6 +71,9 @@ pub(crate) struct Config { /// How often the orchestrator flushes accumulated test results to the nym-api as a signed /// batch submission (e.g. `15m`, `1h`). pub(crate) result_submission_interval: Duration, + + /// Maximum number of stress testing results to submit in a single POST request + pub(crate) result_submission_batch_size: usize, } impl Config { diff --git a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/result_submitter.rs b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/result_submitter.rs index fbb97de233..60d8156f39 100644 --- a/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/result_submitter.rs +++ b/nym-network-monitor-v3/nym-network-monitor-orchestrator/src/orchestrator/result_submitter.rs @@ -12,8 +12,9 @@ use nym_validator_client::nym_api::NymApiClientExt; use nym_validator_client::signable::SignableMessageBody; use std::sync::Arc; use std::time::Duration; +use time::OffsetDateTime; use tokio::time::{Instant, MissedTickBehavior, interval_at}; -use tracing::{debug, info}; +use tracing::info; /// Background task that periodically drains freshly-completed test run results from the local /// storage, wraps them into a signed [`StressTestBatchSubmission`][batch], and POSTs the batch to @@ -38,6 +39,9 @@ pub(crate) struct ResultSubmitter { /// Cadence at which [`Self::run`] attempts a submission sweep. submission_interval: Duration, + /// Maximum number of stress testing results to submit in a single POST request + result_submission_batch_size: usize, + shutdown_token: ShutdownToken, } @@ -54,6 +58,7 @@ impl ResultSubmitter { storage, identity_keys, submission_interval: config.result_submission_interval, + result_submission_batch_size: config.result_submission_batch_size, shutdown_token, } } @@ -74,7 +79,7 @@ impl ResultSubmitter { /// /// [batch]: nym_api_requests::models::network_monitor::StressTestBatchSubmission async fn submit_pending_results(&self) -> anyhow::Result<()> { - info!("submitting stress-test results to nym-api"); + info!("attempting to submit stress-test results to nym-api"); let last_submitted = self.storage.get_last_submitted_testrun_id().await?; // `None` means "never submitted" - treat as 0, which pulls everything currently in the // table (testrun.id is AUTOINCREMENT, so always >= 1). @@ -82,29 +87,55 @@ impl ResultSubmitter { let pending = self.storage.get_testruns_after(after_id).await?; if pending.is_empty() { - debug!("stress-test result submission sweep: no new results"); + info!("stress-test result submission sweep: no new results"); return Ok(()); } - // `get_testruns_after` returns rows ordered by id ASC, so the last row carries the - // highest id and is what we advance the watermark to once the batch is accepted. - #[allow(clippy::expect_used)] - let max_id = pending.last().expect("pending is non-empty").id; - let batch_size = pending.len(); - - let results: Vec = pending.into_iter().map(Into::into).collect(); + info!("{} pending stress test results to submit", pending.len()); let signer = *self.identity_keys.public_key(); - let body = StressTestBatchSubmissionContent::new(signer, results); - let signed = body.sign(self.identity_keys.private_key()); + // nym-api requires each submission's timestamp to be strictly greater than the previous one + // for a given signer (replay protection). Within a single sweep, two consecutive chunks + // could otherwise share a `now_utc()` reading if the host clock has too-coarse resolution + // or steps backwards, which would get the second chunk rejected. Track the last timestamp + // we used and bump by a nanosecond if `now_utc()` hasn't advanced past it. + let mut last_timestamp = OffsetDateTime::now_utc(); - self.client - .submit_stress_testing_results(&signed) - .await - .context("failed to POST stress-test batch submission to nym-api")?; + for chunk in pending.chunks(self.result_submission_batch_size) { + // `get_testruns_after` returns rows ordered by id ASC, so the last row carries the + // highest id and is what we advance the watermark to once the batch is accepted. + #[allow(clippy::expect_used)] + let max_id = chunk.last().expect("chunk is non-empty").id; + let batch_size = chunk.len(); + + let results: Vec = chunk.iter().map(Into::into).collect(); + + let now = OffsetDateTime::now_utc(); + let timestamp = if now > last_timestamp { + now + } else { + last_timestamp + time::Duration::NANOSECOND + }; + last_timestamp = timestamp; + + let body = StressTestBatchSubmissionContent { + signer, + timestamp, + results, + }; + let signed = body.sign(self.identity_keys.private_key()); + + self.client + .submit_stress_testing_results(&signed) + .await + .context("failed to POST stress-test batch submission to nym-api")?; + + self.storage.set_last_submitted_testrun_id(max_id).await?; + info!( + "submitted {batch_size} stress-test results batch to nym-api (testrun ids up to {max_id})" + ); + } - self.storage.set_last_submitted_testrun_id(max_id).await?; - info!("submitted {batch_size} stress-test results to nym-api (testrun ids up to {max_id})"); Ok(()) } @@ -130,7 +161,7 @@ impl ResultSubmitter { // Submission errors shouldn't kill the task - local storage retains the // pending rows until the retention window expires, so the next tick will // retry and eventually catch up once the nym-api is reachable again. - tracing::error!("failed to submit stress-test results: {err}"); + tracing::error!("failed to submit stress-test results: {err:#}"); } } } 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 9da074588a..831dd43f48 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 @@ -274,6 +274,14 @@ impl From for StressTestResult { } } +impl From<&TestRun> for StressTestResult { + fn from(run: &TestRun) -> Self { + // the data is small enough that cloning is negligible + // (since we're going to be converting at most couple dozen a minute rather than a few billion...) + run.clone().into() + } +} + /// The data required to insert or update a row in `nym_node`. Does not carry `last_testrun` /// since that is managed separately via [`StorageManager::set_node_last_testrun`]. #[derive(Debug, Clone, sqlx::FromRow)]