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
2026-06-01 11:44:31 +01:00
parent 11320e3f6a
commit 14a85901b4
10 changed files with 239 additions and 25 deletions
@@ -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",
@@ -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"
@@ -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"
@@ -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<StressTestResult> {
+            // 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<StressTestResult, _> = 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<StressTestResult> 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"
+            );
+        }
+    }
 }
@@ -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
@@ -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";
 }
@@ -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(),
        })
    }

@@ -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 {
@@ -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<StressTestResult> = 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<StressTestResult> = 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:#}");
                    }
                }
            }
@@ -274,6 +274,14 @@ impl From<TestRun> 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)]