Draft 09 conformance & 413 fix

cli/src/aim_report.rs

@@ -77,7 +77,13 @@ impl CloudflareAimResults {
             down_loaded_latency_ms,
             up_loaded_latency_ms,
             packet_loss,
-            responsiveness: download_result.rpm,
+            // This payload has no way to express "not measured", and it is built
+            // before the report is rendered, so it cannot fail the run the way
+            // `RpmReport` does. Hardening the telemetry path is deliberately out
+            // of scope here; in practice `None` now only occurs when no interval
+            // produced a sample at all, which is a broken run rather than the
+            // routine unconverged case that used to report 0.
+            responsiveness: download_result.rpm.unwrap_or(0.0),
             origin: config_url.unwrap_or_else(|| "https://rpm.speed.cloudflare.com".to_string()),
         }
     }

cli/src/args/rpm.rs

@@ -3,7 +3,53 @@
 
 //! Arguments for running responsiveness tests.
 
-use clap::Args;
+use clap::{Args, ValueEnum};
+use nq_rpm::{ConnectionErrorPolicy, DEFAULT_UPLOAD_BYTES_PER_REQUEST};
+
+/// Smallest accepted `--upload-max-request-bytes`.
+///
+/// Below roughly this size a request completes almost immediately, so the load
+/// generator spends its time opening streams instead of moving bytes. That
+/// generates very little actual load while hammering the server with requests,
+/// which is both a useless measurement and unfriendly to the endpoint.
+pub const MIN_UPLOAD_BYTES_PER_REQUEST: usize = 1024 * 1024;
+
+/// Below this, warn that request overhead is becoming significant.
+pub const SMALL_UPLOAD_BYTES_PER_REQUEST: usize = 16 * 1024 * 1024;
+
+fn parse_upload_bytes_per_request(raw: &str) -> Result<usize, String> {
+    let bytes: usize = raw
+        .parse()
+        .map_err(|_| format!("`{raw}` is not a whole number of bytes"))?;
+
+    if bytes < MIN_UPLOAD_BYTES_PER_REQUEST {
+        return Err(format!(
+            "{bytes} is too small; the minimum is {MIN_UPLOAD_BYTES_PER_REQUEST} (1 MiB). \
+             Requests this small complete instantly, so the load generator would spend the \
+             test opening streams rather than saturating the link"
+        ));
+    }
+
+    Ok(bytes)
+}
+
+/// CLI spelling of [`ConnectionErrorPolicy`].
+#[derive(Debug, Clone, Copy, ValueEnum)]
+pub enum ConnectionErrorPolicyArg {
+    /// Retire the failed connection and keep measuring.
+    Retire,
+    /// Abort the test on the first failed connection.
+    Abort,
+}
+
+impl From<ConnectionErrorPolicyArg> for ConnectionErrorPolicy {
+    fn from(arg: ConnectionErrorPolicyArg) -> Self {
+        match arg {
+            ConnectionErrorPolicyArg::Retire => ConnectionErrorPolicy::Retire,
+            ConnectionErrorPolicyArg::Abort => ConnectionErrorPolicy::Abort,
+        }
+    }
+}
 
 #[derive(Debug, Args)]
 pub struct RpmArgs {
@@ -42,6 +88,19 @@ pub struct RpmArgs {
         default_value = "https://h3.speed.cloudflare.com/__up"
     )]
     pub upload_url: String,
+    /// Skip TLS certificate verification. Only for testing against servers with
+    /// self-signed certificates (e.g. a local `wrangler dev`). Never use against
+    /// production endpoints.
+    #[clap(long = "insecure", default_value = "false")]
+    pub insecure: bool,
+    /// What to do when a load-generating connection terminates with an error
+    /// (for example an upload rejected with HTTP 413).
+    ///
+    /// `retire` drops the failed connection, lets the ramp replace it, and
+    /// reports how many failed. `abort` stops the test on the first failure,
+    /// which is what draft-ietf-ippm-responsiveness-09 §5.4 literally describes.
+    #[clap(long = "on-connection-error", default_value = "retire")]
+    pub on_connection_error: ConnectionErrorPolicyArg,
     /// The number of intervals to use when calculating the moving average.
     #[clap(long = "mad", default_value = "4")]
     pub moving_average_distance: usize,
@@ -58,6 +117,22 @@ pub struct RpmArgs {
     /// saturate the network.
     #[clap(long = "max-load", default_value = "16")]
     pub max_loaded_connections: usize,
+    /// Maximum bytes sent in any single upload request.
+    ///
+    /// Upload load is generated as a sequence of requests of this size on each
+    /// connection, re-issued as they complete, rather than one enormous request.
+    /// Servers may cap request body size and reject anything larger with HTTP
+    /// 413. Such caps are per-request, so staying under one keeps the link
+    /// loaded indefinitely.
+    ///
+    /// Lower this if uploads are being rejected. It has no effect on links too
+    /// slow to send this much within the test duration.
+    #[clap(
+        long = "upload-max-request-bytes",
+        default_value_t = DEFAULT_UPLOAD_BYTES_PER_REQUEST,
+        value_parser = parse_upload_bytes_per_request,
+    )]
+    pub upload_bytes_per_request: usize,
     /// The duration between interval updates in milliseconds (ms).
     #[clap(long = "interval-duration", default_value = "1000")]
     pub interval_duration_ms: u64,
@@ -79,10 +154,13 @@ impl Default for RpmArgs {
                 .to_string(),
             small_download_url: "https://h3.speed.cloudflare.com/__down?bytes=10".to_string(),
             upload_url: "https://h3.speed.cloudflare.com/__up".to_string(),
+            insecure: false,
+            on_connection_error: ConnectionErrorPolicyArg::Retire,
             moving_average_distance: 4,
             std_tolerance: 0.05,
             trimmed_mean_percent: 0.95,
             max_loaded_connections: 16,
+            upload_bytes_per_request: DEFAULT_UPLOAD_BYTES_PER_REQUEST,
             interval_duration_ms: 1000, // 1s
             test_duration_ms: 12_000,   // 12s
             disable_aim_scores: false,

cli/src/args/up_down.rs

@@ -18,6 +18,11 @@ pub struct DownloadArgs {
     pub(crate) conn_type: ConnType,
     #[clap(short = 'H', long = "header")]
     pub(crate) headers: Vec<String>,
+    /// Skip TLS certificate verification. Only for testing against servers with
+    /// self-signed certificates (e.g. a local proxy). Never use against
+    /// production endpoints.
+    #[clap(long = "insecure", default_value = "false")]
+    pub(crate) insecure: bool,
 }
 
 /// Upload data (POST) to an endpoint,  reporting latency measurements and total
@@ -40,4 +45,9 @@ pub struct UploadArgs {
     /// Headers to add to the request.
     #[clap(short = 'H', long = "header")]
     pub(crate) headers: Vec<String>,
+    /// Skip TLS certificate verification. Only for testing against servers with
+    /// self-signed certificates (e.g. a local proxy). Never use against
+    /// production endpoints.
+    #[clap(long = "insecure", default_value = "false")]
+    pub(crate) insecure: bool,
 }

cli/src/report.rs

@@ -51,6 +51,15 @@ struct RpmReport {
     throughput: usize,
     loaded_latency_ms: f64,
     rpm: usize,
+    /// Load-generating connections that terminated early with an error.
+    /// Non-zero means the link was not fully loaded for part of the run, so
+    /// this result is degraded and should not be compared with a clean one.
+    #[serde(skip_serializing_if = "is_zero")]
+    failed_connections: usize,
+}
+
+fn is_zero(n: &usize) -> bool {
+    *n == 0
 }
 
 impl RpmReport {
@@ -62,7 +71,56 @@ impl RpmReport {
                 .quantile(0.5)
                 .map(pretty_ms)
                 .context("no loaded latency measurements")?,
-            rpm: result.rpm as usize,
+            // Absent RPM is a failure to measure, reported the same way as an
+            // absent throughput just above rather than as a zero.
+            rpm: result
+                .rpm
+                .context("no rpm measurements: no interval produced a responsiveness sample")?
+                as usize,
+            failed_connections: result.failed_connections,
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nq_core::Timestamp;
+
+    /// A result with throughput and loaded latency present, so that the only
+    /// thing under test is how `rpm` is handled.
+    fn result_with_rpm(rpm: Option<f64>) -> ResponsivenessResult {
+        let mut result = ResponsivenessResult {
+            rpm,
+            ..Default::default()
+        };
+        let at = Timestamp::now();
+        result.average_goodput_series.add(at, 104_000_000.0);
+        result.self_probe_latencies.add(at, 0.231);
+        result
+    }
+
+    #[test]
+    fn absent_rpm_fails_the_report_instead_of_serializing_zero() {
+        // The whole point of making `rpm` an Option: a run that measured no
+        // responsiveness must not publish a plausible-looking number. Previously
+        // this path emitted `"rpm": 0`, which is indistinguishable from a real
+        // measurement of a badly bufferbloated link.
+        let err = match RpmReport::from_rpm_result(&result_with_rpm(None)) {
+            Ok(_) => panic!("absent rpm must not produce a report"),
+            Err(err) => err,
+        };
+        assert!(
+            format!("{err:#}").contains("no rpm measurements"),
+            "unexpected error: {err:#}"
+        );
+    }
+
+    #[test]
+    fn present_rpm_is_reported_unchanged() {
+        let report = RpmReport::from_rpm_result(&result_with_rpm(Some(347.9)))
+            .expect("a measured rpm should report");
+        let json = serde_json::to_string(&report).expect("serializing report");
+        assert!(json.contains("\"rpm\":347"), "got {json}");
+    }
+}

cli/src/rpm.rs

@@ -14,19 +14,54 @@ use nq_tokio_network::TokioNetwork;
 use serde::{Deserialize, Serialize};
 use tokio::time::timeout;
 use tokio_util::sync::CancellationToken;
-use tracing::{debug, error, info};
+use tracing::{debug, error, info, warn};
 
 use crate::aim_report::CloudflareAimResults;
-use crate::args::rpm::RpmArgs;
+use crate::args::rpm::{RpmArgs, SMALL_UPLOAD_BYTES_PER_REQUEST};
 use crate::report::Report;
 use crate::util::pretty_secs_to_ms;
 
+/// Warn loudly when a leg lost load-generating connections.
+///
+/// A failed load-generating connection means the link was not fully loaded for
+/// part of the run, so the RPM score for that leg is measured under weaker
+/// working conditions than intended and reads too high. That is far more
+/// dangerous than an outright error, because the result still looks like a
+/// perfectly ordinary number -- so it needs saying out loud rather than only
+/// appearing as a JSON field.
+fn warn_on_degraded_result(leg: &str, failed_connections: usize, upload_bytes_per_request: usize) {
+    if failed_connections == 0 {
+        return;
+    }
+
+    warn!(
+        "{leg}: {failed_connections} load-generating connection(s) failed, so the link was not \
+         fully loaded for part of the test -- treat this {leg} RPM score as unreliable (it will \
+         read higher than the truth)"
+    );
+
+    if leg == "upload" {
+        warn!(
+            "if these were HTTP 413 rejections, the server caps request bodies below the current \
+             --upload-max-request-bytes ({upload_bytes_per_request}); try a lower value"
+        );
+    }
+}
+
 /// Run a responsiveness test.
 pub async fn run(cli_config: RpmArgs) -> anyhow::Result<()> {
     info!("running responsiveness test");
 
     let scoped_headers = crate::access::cf_access_scoped_headers()?;
 
+    if cli_config.insecure {
+        warn!("TLS certificate verification disabled (--insecure); do not use against production");
+        nq_core::set_insecure_tls(true);
+    }
+
+    // Copied out before `cli_config` is partially moved building the URL list.
+    let upload_bytes_per_request = cli_config.upload_bytes_per_request;
+
     let rpm_urls = match cli_config.config.clone() {
         Some(endpoint) => {
             info!("fetching configuration from {endpoint}");
@@ -82,9 +117,20 @@ pub async fn run(cli_config: RpmArgs) -> anyhow::Result<()> {
         max_loaded_connections: cli_config.max_loaded_connections,
         conn_type: ConnectionType::H2,
         determine_load_only: false,
+        upload_bytes_per_request: cli_config.upload_bytes_per_request,
+        on_connection_error: cli_config.on_connection_error.into(),
         scoped_headers,
     };
 
+    if cli_config.upload_bytes_per_request < SMALL_UPLOAD_BYTES_PER_REQUEST {
+        warn!(
+            "--upload-max-request-bytes is {} ({} MiB); request overhead becomes significant \
+             at this size and the upload leg may under-report capacity",
+            cli_config.upload_bytes_per_request,
+            cli_config.upload_bytes_per_request / (1024 * 1024),
+        );
+    }
+
     info!("running download test");
     let download_result = run_test(&config, true).await?;
     debug!("download result={download_result:?}");
@@ -93,6 +139,17 @@ pub async fn run(cli_config: RpmArgs) -> anyhow::Result<()> {
     let upload_result = run_test(&config, false).await?;
     debug!("upload result={upload_result:?}");
 
+    warn_on_degraded_result(
+        "download",
+        download_result.failed_connections,
+        upload_bytes_per_request,
+    );
+    warn_on_degraded_result(
+        "upload",
+        upload_result.failed_connections,
+        upload_bytes_per_request,
+    );
+
     let aim_results = CloudflareAimResults::from_rpm_results(
         &rtt_result,
         &download_result,

cli/src/up_down.rs

@@ -9,7 +9,7 @@ use nq_core::client::{wait_for_finish, ThroughputClient};
 use nq_core::{Network, Time, TokioTime};
 use nq_tokio_network::TokioNetwork;
 use tokio_util::sync::CancellationToken;
-use tracing::info;
+use tracing::{info, warn};
 
 use crate::args::up_down::{DownloadArgs, UploadArgs};
 use crate::util::pretty_secs;
@@ -24,6 +24,10 @@ pub async fn download(args: DownloadArgs) -> anyhow::Result<()> {
         Arc::new(TokioNetwork::new(Arc::clone(&time), shutdown.clone())) as Arc<dyn Network>;
 
     let conn_type = args.conn_type.into();
+    if args.insecure {
+        warn!("TLS certificate verification disabled (--insecure); do not use against production");
+        nq_core::set_insecure_tls(true);
+    }
     info!("downloading: {}", args.url);
 
     let client = ThroughputClient::download()
@@ -90,6 +94,10 @@ pub async fn upload(args: UploadArgs) -> anyhow::Result<()> {
         Arc::new(TokioNetwork::new(Arc::clone(&time), shutdown.clone())) as Arc<dyn Network>;
 
     let conn_type = args.conn_type.into();
+    if args.insecure {
+        warn!("TLS certificate verification disabled (--insecure); do not use against production");
+        nq_core::set_insecure_tls(true);
+    }
     let bytes = args.bytes.unwrap_or(10_000_000);
     info!("uploading {bytes} bytes to: {}", args.url);