diff --git a/cli/src/aim_report.rs b/cli/src/aim_report.rs
index fa8105a..7781b72 100644
--- a/cli/src/aim_report.rs
+++ b/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()),
         }
     }
diff --git a/cli/src/args/rpm.rs b/cli/src/args/rpm.rs
index f200914..2138b0a 100644
--- a/cli/src/args/rpm.rs
+++ b/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,
diff --git a/cli/src/args/up_down.rs b/cli/src/args/up_down.rs
index 2ad9f4f..96b039f 100644
--- a/cli/src/args/up_down.rs
+++ b/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,
 }
diff --git a/cli/src/report.rs b/cli/src/report.rs
index d7aba1c..8c1696a 100644
--- a/cli/src/report.rs
+++ b/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}");
+    }
+}
diff --git a/cli/src/rpm.rs b/cli/src/rpm.rs
index 5fb77b8..bb08a4d 100644
--- a/cli/src/rpm.rs
+++ b/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,
diff --git a/cli/src/up_down.rs b/cli/src/up_down.rs
index ec2bdba..ecf9162 100644
--- a/cli/src/up_down.rs
+++ b/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);
 
diff --git a/crates/nq-core/src/body/counting_body.rs b/crates/nq-core/src/body/counting_body.rs
index 6e71381..5d4ff09 100644
--- a/crates/nq-core/src/body/counting_body.rs
+++ b/crates/nq-core/src/body/counting_body.rs
@@ -25,6 +25,19 @@ pub enum BodyEvent {
         /// When the body finished.
         at: Timestamp,
     },
+    /// The transfer terminated early with an error and will produce no further
+    /// bytes.
+    ///
+    /// Emitted either by the [`CountingBody`] itself when the wrapped body
+    /// yields an error, or by the client when the request fails or the server
+    /// rejects it (e.g. an HTTP 413 on an upload). Consumers must treat this as
+    /// terminal: the transfer did *not* complete.
+    Failed {
+        /// When the failure was observed.
+        at: Timestamp,
+        /// Human-readable cause, e.g. `"unexpected status 413 Payload Too Large"`.
+        reason: String,
+    },
 }
 
 pin_project_lite::pin_project! {
@@ -73,6 +86,17 @@ impl<B> CountingBody<B> {
             events_rx,
         )
     }
+
+    /// A handle for reporting a failure the body itself cannot observe, such as
+    /// an upload rejected by the server with a non-success status.
+    ///
+    /// The returned sender must be dropped as soon as it is no longer needed.
+    /// [`CountingBody`] is otherwise the sole owner of the sender, and
+    /// consumers rely on the channel closing when the body is dropped to detect
+    /// a transfer that died without reporting anything.
+    pub fn sender(&self) -> mpsc::UnboundedSender<BodyEvent> {
+        self.events_tx.clone()
+    }
 }
 
 impl<B> Body for CountingBody<B>
@@ -164,7 +188,20 @@ where
                 Poll::Ready(None)
             }
             Poll::Ready(Some(Err(e))) => {
+                let now = this.time.now();
                 error!(error=?e, "body errored");
+
+                // Report the failure so consumers retire this transfer instead
+                // of leaving it looking permanently in-flight. Only emitted
+                // once, and never after a `Finished`.
+                if !*this.sent_finished {
+                    let _ = this.events_tx.send(BodyEvent::Failed {
+                        at: now,
+                        reason: format!("body error: {e:?}"),
+                    });
+                    *this.sent_finished = true;
+                }
+
                 Poll::Ready(Some(Err(e)))
             }
             Poll::Pending => {
@@ -210,4 +247,56 @@ mod tests {
         }
         assert!(got_finished, "never received Finished");
     }
+
+    /// A body that yields one data frame and then errors, standing in for a
+    /// transfer killed mid-flight (reset stream, dropped connection).
+    struct ErroringBody {
+        sent: bool,
+    }
+
+    impl Body for ErroringBody {
+        type Data = Bytes;
+        type Error = &'static str;
+
+        fn poll_frame(
+            mut self: std::pin::Pin<&mut Self>,
+            _cx: &mut std::task::Context<'_>,
+        ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
+            if !self.sent {
+                self.sent = true;
+                return Poll::Ready(Some(Ok(hyper::body::Frame::data(Bytes::from_static(
+                    b"hello",
+                )))));
+            }
+            Poll::Ready(Some(Err("stream reset")))
+        }
+    }
+
+    // A body that dies mid-transfer must report `Failed`, otherwise consumers
+    // cannot distinguish it from one that is still running.
+    #[tokio::test]
+    async fn errored_body_emits_failed() {
+        let time: Arc<dyn Time> = Arc::new(TokioTime::new());
+        let (body, mut events) = CountingBody::new(ErroringBody { sent: false }, Duration::ZERO, time);
+
+        // Drive the body until it errors.
+        let _ = body.collect().await;
+        events.close();
+
+        let mut failed = None;
+        let mut got_finished = false;
+        while let Some(ev) = events.recv().await {
+            match ev {
+                BodyEvent::Failed { reason, .. } => failed = Some(reason),
+                BodyEvent::Finished { .. } => got_finished = true,
+                BodyEvent::ByteCount { .. } => {}
+            }
+        }
+
+        assert!(failed.is_some(), "never received Failed");
+        assert!(
+            !got_finished,
+            "a failed body must not also report Finished"
+        );
+    }
 }
diff --git a/crates/nq-core/src/body/mod.rs b/crates/nq-core/src/body/mod.rs
index e85e4d4..fb68d39 100644
--- a/crates/nq-core/src/body/mod.rs
+++ b/crates/nq-core/src/body/mod.rs
@@ -30,9 +30,16 @@ pub use self::{
 
 /// A body that is currently being sent or received.
 pub struct InflightBody {
+    /// When the request that produced this body was started.
     pub start: Timestamp,
+    /// The connection the body is being transferred on. Shared so further
+    /// requests can be sent on the same connection.
     pub connection: Arc<RwLock<EstablishedConnection>>,
+    /// Connection setup timing, when this body created the connection.
     pub timing: Option<ConnectionTiming>,
+    /// Byte-count and termination events for this body. The channel closing is
+    /// itself meaningful: it signals the body was dropped.
     pub events: mpsc::UnboundedReceiver<BodyEvent>,
+    /// Headers associated with the transfer.
     pub headers: HeaderMap<HeaderValue>,
 }
diff --git a/crates/nq-core/src/client.rs b/crates/nq-core/src/client.rs
index 519be61..a0d4f5f 100644
--- a/crates/nq-core/src/client.rs
+++ b/crates/nq-core/src/client.rs
@@ -9,7 +9,7 @@
 use std::{convert::Infallible, net::ToSocketAddrs, sync::Arc, time::Duration};
 use tokio::sync::RwLock;
 
-use anyhow::Context;
+use anyhow::{Context, bail};
 use http::{HeaderMap, HeaderValue, Uri};
 use http_body_util::BodyExt;
 use hyper::body::{Body, Bytes, Incoming};
@@ -138,6 +138,11 @@ impl ThroughputClient {
 
         let (tx, rx) = oneshot_result();
         let mut events = None;
+        // Lets the request task report a failure the upload body cannot see
+        // itself -- notably a non-success response status. Must be dropped once
+        // the request finishes so the body remains the sole sender and its
+        // channel still closes when the transfer dies.
+        let mut upload_events_tx = None;
 
         let body: NqBody = match self.direction {
             Direction::Up(size) => {
@@ -147,6 +152,7 @@ impl ThroughputClient {
                 let (body, events_rx) =
                     CountingBody::new(dummy_body, Duration::from_millis(50), Arc::clone(&time));
                 events = Some(events_rx);
+                upload_events_tx = Some(body.sender());
 
                 headers.insert("Content-Type", HeaderValue::from_static("text/plain"));
 
@@ -170,6 +176,7 @@ impl ThroughputClient {
         *request.headers_mut() = headers.clone();
         tracing::debug!("created request: {request:?}");
 
+        let failure_time = Arc::clone(&time);
         tokio::spawn(
             async move {
                 if let Err(error) = self
@@ -186,8 +193,22 @@ impl ThroughputClient {
                     )
                     .await
                 {
-                    debug!("error sending ThroughputClient request: {error:#}");
+                    error!("error sending ThroughputClient request: {error:#}");
+
+                    // An upload's failure (rejected status, reset stream, dead
+                    // connection) is invisible to its request body, which just
+                    // stops being polled. Report it explicitly so the transfer
+                    // is retired rather than appearing to run forever.
+                    if let Some(sender) = &upload_events_tx {
+                        let _ = sender.send(BodyEvent::Failed {
+                            at: failure_time.now(),
+                            reason: format!("{error:#}"),
+                        });
+                    }
                 }
+                // Drop the sender clone so the body is again the only owner of
+                // the events channel.
+                drop(upload_events_tx);
             }
             .in_current_span(),
         );
@@ -210,7 +231,7 @@ impl ThroughputClient {
     ) -> Result<Result<(), anyhow::Error>, anyhow::Error> {
         let start = time.now();
         let connection = self
-            .get_or_create_connection(&network, host, host_with_port, start)
+            .get_or_create_connection(&network, &time, host, host_with_port)
             .await?;
         let conn_timing = {
             let conn = connection.read().await;
@@ -273,12 +294,29 @@ impl ThroughputClient {
                     .into_parts();
                 info!("upload response parts: {:?}", parts);
 
+                // A rejected upload (e.g. HTTP 413 when the body exceeds the
+                // server's buffering cap) is a perfectly well-formed HTTP
+                // response, so nothing below would otherwise notice: the load
+                // would be counted as healthy while transferring nothing.
+                if !parts.status.is_success() {
+                    bail!("upload rejected with status {}", parts.status);
+                }
+
                 incoming.boxed()
             }
             Direction::Down => {
                 let (parts, incoming) = response_fut.await?.into_parts();
                 info!("download response parts: {:?}", parts);
 
+                // The response is awaited before the caller is handed its
+                // `InflightBody`, so a bad status can be reported through the
+                // oneshot and the load never starts.
+                if !parts.status.is_success() {
+                    let reason = format!("download rejected with status {}", parts.status);
+                    let _ = tx.send(Err(anyhow::anyhow!("{reason}")));
+                    bail!(reason);
+                }
+
                 let (counting_body, events) =
                     CountingBody::new(incoming, Duration::from_millis(100), Arc::clone(&time));
 
@@ -305,9 +343,9 @@ impl ThroughputClient {
     async fn get_or_create_connection(
         &mut self,
         network: &Arc<dyn Network>,
+        time: &Arc<dyn Time>,
         host: String,
         host_with_port: String,
-        start: Timestamp,
     ) -> Result<Arc<RwLock<EstablishedConnection>>, anyhow::Error> {
         let connection = if let Some(connection) = self.connection.take() {
             connection
@@ -321,8 +359,13 @@ impl ThroughputClient {
 
             debug!("addrs: {addrs:?}");
 
+            // Start the connection timing *after* DNS resolution so that
+            // tcp_f (draft-ietf-ippm-responsiveness-09 §5.3) measures the TCP
+            // handshake alone, without folding in the DNS lookup time.
+            let connect_start = time.now();
+
             network
-                .new_connection(start, addrs[0], host, conn_type)
+                .new_connection(connect_start, addrs[0], host, conn_type)
                 .await
                 .context("creating new connection")?
         } else {
@@ -503,6 +546,9 @@ pub async fn wait_for_finish(
                     finished_at: at,
                 });
             }
+            BodyEvent::Failed { reason, .. } => {
+                return Err(anyhow::anyhow!("body failed after {body_total} bytes: {reason}"));
+            }
         }
     }
 
diff --git a/crates/nq-core/src/connection/http.rs b/crates/nq-core/src/connection/http.rs
index 664ae35..adc4f93 100644
--- a/crates/nq-core/src/connection/http.rs
+++ b/crates/nq-core/src/connection/http.rs
@@ -5,9 +5,10 @@ use std::fmt::Debug;
 use std::future::Future;
 use std::net::SocketAddr;
 use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, Ordering};
 
 use anyhow::bail;
-use boring::ssl::{SslConnector, SslMethod, SslVerifyMode};
+use boring::ssl::{SslConnector, SslMethod, SslVerifyMode, SslVersion};
 use boring::x509::X509;
 use boring::x509::store::X509StoreBuilder;
 use http::header::HOST;
@@ -17,7 +18,7 @@ use hyper::client::conn::{http1, http2};
 use hyper_util::rt::TokioIo;
 use tokio::select;
 use tokio_util::sync::CancellationToken;
-use tracing::{Instrument, debug, error, info};
+use tracing::{Instrument, debug, error, info, warn};
 
 use crate::body::NqBody;
 use crate::util::ByteStream;
@@ -25,6 +26,23 @@ use crate::{ConnectionTiming, ConnectionType, ResponseFuture, Time};
 
 pub type TlsStream = tokio_boring::SslStream<Box<dyn ByteStream>>;
 
+/// Process-wide switch to skip TLS certificate verification. Off by default.
+///
+/// This exists only to allow testing against servers presenting self-signed
+/// certificates (e.g. a local `wrangler dev` speed-test server). It must never
+/// be enabled against production endpoints.
+static INSECURE_TLS: AtomicBool = AtomicBool::new(false);
+
+/// Enable or disable skipping TLS certificate verification globally.
+pub fn set_insecure_tls(insecure: bool) {
+    INSECURE_TLS.store(insecure, Ordering::Relaxed);
+}
+
+/// Whether TLS certificate verification is currently being skipped.
+pub fn insecure_tls() -> bool {
+    INSECURE_TLS.load(Ordering::Relaxed)
+}
+
 /// An [`EstablishedConnection`] contains the connection's timing and a handle
 /// to send HTTP requests with.
 #[derive(Debug)]
@@ -77,7 +95,12 @@ pub async fn tls_connection(
         }
     }
     builder.set_verify_cert_store(store_builder.build())?;
-    builder.set_verify(SslVerifyMode::PEER);
+    if insecure_tls() {
+        debug!("TLS certificate verification disabled (insecure mode)");
+        builder.set_verify(SslVerifyMode::NONE);
+    } else {
+        builder.set_verify(SslVerifyMode::PEER);
+    }
 
     let alpn: &[u8] = match conn_type {
         ConnectionType::H1 { use_tls: false } => {
@@ -97,7 +120,17 @@ pub async fn tls_connection(
 
     timing.set_secure(time.now());
 
-    debug!("created tls connection");
+    // Normalize the TLS handshake time to the number of round-trips the
+    // negotiated version takes (draft-ietf-ippm-responsiveness-09 §5.3).
+    // TLS 1.3 completes in 1 round-trip, TLS 1.2 in 2. Default to 1.
+    let tls_round_trips = match ssl_stream.ssl().version2() {
+        Some(SslVersion::TLS1_3) => 1,
+        Some(SslVersion::TLS1_2) => 2,
+        _ => 1,
+    };
+    timing.set_tls_round_trips(tls_round_trips);
+
+    debug!(tls_round_trips, "created tls connection");
 
     Ok(ssl_stream)
 }
@@ -191,29 +224,69 @@ impl SendRequest {
         &mut self,
         mut req: Request<NqBody>,
     ) -> Pin<Box<dyn Future<Output = hyper::Result<Response<Incoming>>> + Send>> {
-        // inject the host header it it's missing and this is an HTTP/1.1 req.
-        self.insert_host_if_missing(&mut req);
-
         match self {
             SendRequest::H1 {
                 dispatch: send_request,
-            } => Box::pin(send_request.send_request(req)),
+            } => {
+                // HTTP/1.1 to an origin server requires origin-form request
+                // targets (`GET /path`) plus a `Host` header carrying the
+                // authority. Building the request from an absolute URI leaves
+                // it in proxy/absolute-form (`GET http://host/path`), which
+                // origin servers (e.g. workerd) reject. Normalize here.
+                Self::normalize_h1_request(&mut req);
+                Box::pin(send_request.send_request(req))
+            }
             SendRequest::H2 {
                 dispatch: send_request,
-            } => Box::pin(send_request.send_request(req)),
+            } => {
+                // HTTP/2 uses the :authority pseudo-header derived from the
+                // absolute URI, so leave the request untouched.
+                Box::pin(send_request.send_request(req))
+            }
         }
     }
 
-    fn insert_host_if_missing(&mut self, req: &mut Request<NqBody>) {
-        if !matches!(self, SendRequest::H1 { .. }) && !req.headers().contains_key(HOST) {
-            return;
+    /// Rewrite an HTTP/1.1 request into origin-form with a proper `Host` header.
+    fn normalize_h1_request(req: &mut Request<NqBody>) {
+        // Set `Host` from the full authority (host and, if present, port).
+        if !req.headers().contains_key(HOST) {
+            if let Some(authority) = req.uri().authority().cloned() {
+                if let Ok(host) = HeaderValue::from_str(authority.as_str()) {
+                    req.headers_mut().insert(HOST, host);
+                } else {
+                    // HTTP/1.1 requires a Host header; without one the origin
+                    // answers 400. An authority is already restricted to
+                    // characters a header value accepts, so this is not
+                    // expected to be reachable -- but a silent drop is
+                    // near-impossible to diagnose from the far end.
+                    warn!(
+                        %authority,
+                        "could not build a Host header from the URI authority; \
+                         sending the request without one"
+                    );
+                }
+            }
         }
 
-        let Some(Ok(host)) = req.uri().host().map(HeaderValue::from_str) else {
-            return;
-        };
-
-        req.headers_mut().insert(HOST, host);
+        // Collapse the request target to origin-form (path + query only).
+        let path_and_query = req
+            .uri()
+            .path_and_query()
+            .map(|pq| pq.as_str().to_owned())
+            .unwrap_or_else(|| "/".to_owned());
+
+        match path_and_query.parse::<http::Uri>() {
+            Ok(uri) => *req.uri_mut() = uri,
+            // Leaves the request in absolute-form, which origin servers
+            // reject. The string comes from an already-validated
+            // PathAndQuery so this is not expected to be reachable, but the
+            // failure would otherwise be invisible.
+            Err(error) => warn!(
+                path_and_query,
+                %error,
+                "failed to parse origin-form URI; sending absolute-form"
+            ),
+        }
     }
 }
 
diff --git a/crates/nq-core/src/connection/mod.rs b/crates/nq-core/src/connection/mod.rs
index ce31ae3..56c8328 100644
--- a/crates/nq-core/src/connection/mod.rs
+++ b/crates/nq-core/src/connection/mod.rs
@@ -8,7 +8,7 @@ use std::time::Duration;
 
 use crate::Timestamp;
 
-pub use self::http::EstablishedConnection;
+pub use self::http::{EstablishedConnection, insecure_tls, set_insecure_tls};
 pub use self::map::ConnectionManager;
 
 /// The L7 type of a connection.
@@ -76,6 +76,14 @@ pub struct ConnectionTiming {
 
     // Duration of the DNS lookup
     dns_time: Duration,
+
+    /// Number of round-trips the TLS handshake took until the connection was
+    /// ready to transmit data (TLS 1.3 -> 1, TLS 1.2 -> 2). Defaults to 1.
+    ///
+    /// Used to normalize the TLS handshake time per draft-ietf-ippm-
+    /// responsiveness-09 §5.3 ("the TLS establishment time needs to be
+    /// normalized to the number of round-trips").
+    tls_round_trips: u32,
 }
 
 impl ConnectionTiming {
@@ -88,6 +96,7 @@ impl ConnectionTiming {
             time_secure: Duration::ZERO,
             time_application: Duration::ZERO,
             dns_time: Duration::ZERO,
+            tls_round_trips: 1,
         }
     }
 
@@ -145,4 +154,97 @@ impl ConnectionTiming {
     pub fn time_application(&self) -> Duration {
         self.time_application
     }
+
+    /// Sets the number of round-trips the TLS handshake took.
+    pub fn set_tls_round_trips(&mut self, round_trips: u32) {
+        self.tls_round_trips = round_trips.max(1);
+    }
+
+    /// Returns the number of round-trips the TLS handshake took (>= 1).
+    pub fn tls_round_trips(&self) -> u32 {
+        self.tls_round_trips.max(1)
+    }
+
+    /// The duration of the TCP handshake alone (excluding DNS resolution),
+    /// i.e. `tcp_f` in draft-ietf-ippm-responsiveness-09 §5.3.
+    ///
+    /// This is the interval between the transport starting to connect and the
+    /// connection being established. When the connection timing starts after
+    /// DNS resolution (as it does for the responsiveness probes), `time_lookup`
+    /// is zero and this is simply `time_connect`.
+    pub fn tcp_handshake(&self) -> Duration {
+        self.time_connect.saturating_sub(self.time_lookup)
+    }
+
+    /// The duration of the TLS handshake alone (excluding the preceding TCP
+    /// handshake), i.e. the un-normalized `tls_f` in
+    /// draft-ietf-ippm-responsiveness-09 §5.3.
+    ///
+    /// For QUIC/H3 connections `time_secure` is zero (TLS is folded into the
+    /// transport handshake), so this saturates to zero rather than underflowing.
+    /// Divide by [`Self::tls_round_trips`] to obtain the normalized value.
+    pub fn tls_handshake(&self) -> Duration {
+        self.time_secure.saturating_sub(self.time_connect)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Build a timing whose phases complete at the given millisecond offsets
+    /// from `start`.
+    fn timing_at(lookup_ms: u64, connect_ms: u64, secure_ms: u64, application_ms: u64) -> ConnectionTiming {
+        let start = Timestamp::now();
+        let mut t = ConnectionTiming::new(start);
+        t.set_lookup(start + Duration::from_millis(lookup_ms));
+        t.set_connect(start + Duration::from_millis(connect_ms));
+        t.set_secure(start + Duration::from_millis(secure_ms));
+        t.set_application(start + Duration::from_millis(application_ms));
+        t
+    }
+
+    #[test]
+    fn independent_phase_deltas() {
+        // Post-DNS baseline (lookup = 0): connect at 30ms, secure at 60ms,
+        // application at 62ms. Each network phase is ~30ms (1 RTT).
+        let t = timing_at(0, 30, 60, 62);
+        assert_eq!(t.tcp_handshake(), Duration::from_millis(30));
+        assert_eq!(t.tls_handshake(), Duration::from_millis(30));
+    }
+
+    #[test]
+    fn tcp_handshake_excludes_dns_lookup() {
+        // If the timing baseline included DNS (lookup at 10ms, connect at 40ms),
+        // the TCP handshake is connect - lookup = 30ms, not 40ms.
+        let t = timing_at(10, 40, 70, 72);
+        assert_eq!(t.tcp_handshake(), Duration::from_millis(30));
+        assert_eq!(t.tls_handshake(), Duration::from_millis(30));
+    }
+
+    #[test]
+    fn tls_handshake_saturates_for_quic_like_zero_secure() {
+        // QUIC/H3: time_secure stays 0 while time_connect is set. Must not
+        // underflow.
+        let start = Timestamp::now();
+        let mut t = ConnectionTiming::new(start);
+        t.set_connect(start + Duration::from_millis(30));
+        // secure left at zero
+        assert_eq!(t.tls_handshake(), Duration::ZERO);
+    }
+
+    #[test]
+    fn tls_round_trips_defaults_to_one_and_clamps() {
+        let t = ConnectionTiming::new(Timestamp::now());
+        assert_eq!(t.tls_round_trips(), 1);
+
+        let mut t = t;
+        t.set_tls_round_trips(2);
+        assert_eq!(t.tls_round_trips(), 2);
+
+        // A zero round-trip count would produce a divide-by-zero downstream;
+        // it is clamped to 1.
+        t.set_tls_round_trips(0);
+        assert_eq!(t.tls_round_trips(), 1);
+    }
 }
diff --git a/crates/nq-core/src/lib.rs b/crates/nq-core/src/lib.rs
index 55a615b..ed0d31a 100644
--- a/crates/nq-core/src/lib.rs
+++ b/crates/nq-core/src/lib.rs
@@ -19,8 +19,11 @@ mod upgraded;
 mod util;
 
 pub use crate::{
-    body::{BodyEvent, CountingBody, NqBody},
-    connection::{ConnectionManager, ConnectionTiming, ConnectionType, EstablishedConnection},
+    body::{BodyEvent, CountingBody, InflightBody, NqBody},
+    connection::{
+        ConnectionManager, ConnectionTiming, ConnectionType, EstablishedConnection, insecure_tls,
+        set_insecure_tls,
+    },
     network::Network,
     scoped_headers::ScopedHeaders,
     time::{Time, Timestamp, TokioTime},
diff --git a/crates/nq-load-generator/src/lib.rs b/crates/nq-load-generator/src/lib.rs
index 07fcbc1..47720ff 100644
--- a/crates/nq-load-generator/src/lib.rs
+++ b/crates/nq-load-generator/src/lib.rs
@@ -4,17 +4,19 @@
 use std::{collections::HashMap, sync::Arc};
 
 use anyhow::Context;
-use http::{HeaderMap, HeaderName, HeaderValue};
+use http::{HeaderMap, HeaderName, HeaderValue, Uri};
 use nq_core::client::{Direction, ThroughputClient};
 use nq_core::{
-    BodyEvent, ConnectionType, EstablishedConnection, Network, OneshotResult, ScopedHeaders, Time,
-    Timestamp, oneshot_result,
+    BodyEvent, ConnectionType, EstablishedConnection, InflightBody, Network, OneshotResult,
+    ScopedHeaders, Time, Timestamp, oneshot_result,
 };
 use nq_stats::CounterSeries;
 use rand::seq::SliceRandom;
 use serde::Deserialize;
 use tokio::sync::RwLock;
+use tokio::sync::mpsc;
 use tokio::sync::mpsc::UnboundedReceiver;
+use tokio::sync::mpsc::error::TryRecvError;
 use tokio_util::sync::CancellationToken;
 use tracing::Instrument;
 
@@ -27,7 +29,6 @@ pub struct LoadConfig {
     pub scoped_headers: Option<ScopedHeaders>,
     pub download_url: url::Url,
     pub upload_url: url::Url,
-    pub upload_size: usize,
 }
 
 pub struct LoadGenerator {
@@ -67,6 +68,11 @@ impl LoadGenerator {
     ) -> anyhow::Result<OneshotResult<LoadedConnection>> {
         let (tx, rx) = oneshot_result();
 
+        let uri: Uri = match direction {
+            Direction::Up(_) => self.config.upload_url.as_str().parse()?,
+            Direction::Down => self.config.download_url.as_str().parse()?,
+        };
+
         let client = match direction {
             Direction::Down => ThroughputClient::download(),
             Direction::Up(size) => ThroughputClient::upload(size),
@@ -78,17 +84,30 @@ impl LoadGenerator {
             .scoped_headers(self.scoped_headers.clone());
 
         let response_fut = client.send(
-            match direction {
-                Direction::Up(_) => self.config.upload_url.as_str().parse()?,
-                Direction::Down => self.config.download_url.as_str().parse()?,
-            },
-            network,
-            time,
-            shutdown,
+            uri.clone(),
+            Arc::clone(&network),
+            Arc::clone(&time),
+            shutdown.clone(),
         )?;
 
         tracing::debug!("got loaded connection response future");
 
+        // An upload load is an open-ended *sequence* of bounded requests rather
+        // than one request, so its events are produced by a driver task instead
+        // of coming straight off a single body. See [`UploadReissue`].
+        let reissue = match direction {
+            Direction::Up(bound) => Some(UploadReissue {
+                bound,
+                uri,
+                headers: self.headers.clone(),
+                scoped_headers: self.scoped_headers.clone(),
+                network,
+                time,
+                shutdown,
+            }),
+            Direction::Down => None,
+        };
+
         tokio::spawn(
             async move {
                 let inflight_body = response_fut
@@ -97,13 +116,29 @@ impl LoadGenerator {
 
                 tracing::debug!("sending loaded connection");
 
+                let Some(reissue) = reissue else {
+                    let _ = tx.send(Ok(LoadedConnection {
+                        connection: inflight_body.connection,
+                        events_rx: inflight_body.events,
+                        state: LoadState::default(),
+                    }));
+
+                    return Ok(());
+                };
+
+                let (events_tx, events_rx) = mpsc::unbounded_channel();
+                let connection = Arc::clone(&inflight_body.connection);
+
                 let _ = tx.send(Ok(LoadedConnection {
-                    connection: inflight_body.connection,
-                    events_rx: inflight_body.events,
-                    total_bytes_series: CounterSeries::new(),
-                    finished_at: None,
+                    connection: Arc::clone(&connection),
+                    events_rx,
+                    state: LoadState::default(),
                 }));
 
+                reissue
+                    .run(connection, inflight_body.events, events_tx)
+                    .await;
+
                 Ok::<_, anyhow::Error>(())
             }
             .in_current_span(),
@@ -133,43 +168,597 @@ impl LoadGenerator {
         }
     }
 
+    /// Connections still transferring: neither completed nor terminated early.
+    ///
+    /// Excluding failed connections is what lets the ramp replace them and
+    /// keeps self probes off dead connections.
     pub fn ongoing_loads(&self) -> impl Iterator<Item = &LoadedConnection> {
-        self.loads.iter().filter(|load| load.finished_at.is_none())
+        self.loads.iter().filter(|load| load.is_ongoing())
     }
 
     pub fn count_loads(&self) -> usize {
         self.ongoing_loads().count()
     }
 
+    /// Number of load-generating connections that terminated early with an
+    /// error.
+    pub fn count_failed_loads(&self) -> usize {
+        self.loads.iter().filter(|load| load.has_failed()).count()
+    }
+
     pub fn into_connections(self) -> Vec<LoadedConnection> {
         self.loads
     }
 }
 
+/// Drives an upload load-generating connection as an open-ended sequence of
+/// bounded POSTs, all sent on the same established connection.
+///
+/// A single unbounded POST cannot be used against a server that caps how much
+/// request body it will buffer: it is rejected with HTTP 413 once it exceeds
+/// the cap, which kills the load part-way through the test. The RPM score then
+/// reflects a network that is barely loaded, so it comes out flatteringly high
+/// rather than simply failing.
+///
+/// Two properties of such caps make this approach work: they apply per-request
+/// rather than per-connection, and a 413 does not close the HTTP/2 connection.
+/// So an unbounded number of bounded requests can ride one connection and keep
+/// the link continuously loaded without tripping the cap.
+struct UploadReissue {
+    /// Maximum bytes sent in any single request.
+    bound: usize,
+    uri: Uri,
+    headers: HeaderMap<HeaderValue>,
+    scoped_headers: Option<ScopedHeaders>,
+    network: Arc<dyn Network>,
+    time: Arc<dyn Time>,
+    shutdown: CancellationToken,
+}
+
+/// Why the request currently being relayed stopped producing events.
+#[derive(Debug, PartialEq, Eq)]
+enum RequestEnd {
+    /// The body sent every byte it was asked for.
+    Finished,
+    /// The channel closed before the body finished, i.e. the transfer died.
+    Died,
+}
+
+impl UploadReissue {
+    /// Relay `first`'s events, then keep issuing further bounded requests on
+    /// `connection` for as long as the consumer keeps listening.
+    async fn run(
+        self,
+        connection: Arc<RwLock<EstablishedConnection>>,
+        first: UnboundedReceiver<BodyEvent>,
+        events_tx: mpsc::UnboundedSender<BodyEvent>,
+    ) {
+        let mut current = first;
+        let mut relay = CumulativeRelay::default();
+        let mut requests = 1usize;
+
+        loop {
+            let ended = loop {
+                let event = tokio::select! {
+                    // Test teardown. Returning silently is correct: the consumer
+                    // tells teardown apart from a failure via
+                    // `LoadedConnection::stop`, which sets `stopping` before it
+                    // observes the channel closing.
+                    _ = self.shutdown.cancelled() => return,
+                    event = current.recv() => event,
+                };
+
+                let Some(event) = event else {
+                    break RequestEnd::Died;
+                };
+
+                match relay.on_event(event) {
+                    RelayAction::Forward(event) => {
+                        // A closed channel means `stop()` was called. Returning
+                        // drops `current`, closing the in-flight body's event
+                        // channel, which is what truncates it -- the same
+                        // mechanism a single-request load uses.
+                        if events_tx.send(event).is_err() {
+                            return;
+                        }
+                    }
+                    RelayAction::RequestFinished => break RequestEnd::Finished,
+                    RelayAction::Fail(event) => {
+                        let _ = events_tx.send(event);
+                        return;
+                    }
+                }
+            };
+
+            if ended == RequestEnd::Died {
+                let _ = events_tx.send(BodyEvent::Failed {
+                    at: self.time.now(),
+                    reason: format!(
+                        "upload terminated early after {} request(s), {} bytes",
+                        requests,
+                        relay.total()
+                    ),
+                });
+                return;
+            }
+
+            if events_tx.is_closed() {
+                return;
+            }
+
+            // Start the replacement before dealing with the finished request, so
+            // the connection is refilled as early as possible. `Finished` fires
+            // when the body hands its last frame to hyper, which still has that
+            // data buffered -- so the new request's frames queue behind the tail
+            // of the old one and the socket never goes idle.
+            let next = match self.issue(&connection) {
+                Ok(next) => next,
+                Err(error) => {
+                    let _ = events_tx.send(BodyEvent::Failed {
+                        at: self.time.now(),
+                        reason: format!("could not start upload request {requests}: {error:#}"),
+                    });
+                    return;
+                }
+            };
+
+            let next = match next.await {
+                Ok(inflight) => inflight.events,
+                Err(error) => {
+                    let _ = events_tx.send(BodyEvent::Failed {
+                        at: self.time.now(),
+                        reason: format!("upload request {requests} failed to start: {error:#}"),
+                    });
+                    return;
+                }
+            };
+
+            requests += 1;
+            tracing::debug!(
+                requests,
+                total_bytes = relay.total(),
+                "re-issued bounded upload request"
+            );
+
+            // Because `Finished` precedes the response, the status of the
+            // request just completed is still unknown. Keep draining its channel
+            // in the background so a late rejection still retires this load.
+            let finished = std::mem::replace(&mut current, next);
+            tokio::spawn(watch_tail(finished, events_tx.clone()).in_current_span());
+        }
+    }
+
+    fn issue(
+        &self,
+        connection: &Arc<RwLock<EstablishedConnection>>,
+    ) -> anyhow::Result<OneshotResult<InflightBody>> {
+        ThroughputClient::upload(self.bound)
+            .with_connection(Arc::clone(connection))
+            .headers(self.headers.clone())
+            .scoped_headers(self.scoped_headers.clone())
+            .send(
+                self.uri.clone(),
+                Arc::clone(&self.network),
+                Arc::clone(&self.time),
+                self.shutdown.clone(),
+            )
+    }
+}
+
+/// Drain a completed request's event channel, forwarding only a terminal
+/// failure.
+///
+/// [`UploadReissue::run`] moves to the next request as soon as the previous body
+/// is fully handed to hyper, which happens before its response status is known.
+/// A rejection therefore arrives after the driver has stopped reading that
+/// channel; without this it would be dropped, leaving the load looking healthy
+/// while the server refuses every request.
+async fn watch_tail(
+    mut events: UnboundedReceiver<BodyEvent>,
+    events_tx: mpsc::UnboundedSender<BodyEvent>,
+) {
+    while let Some(event) = events.recv().await {
+        if matches!(event, BodyEvent::Failed { .. }) {
+            let _ = events_tx.send(event);
+            return;
+        }
+    }
+}
+
+/// Translates the per-request [`BodyEvent`] streams of a re-issued upload into
+/// one continuous stream for the consumer.
+///
+/// Every request's `CountingBody` counts from zero, but [`CounterSeries`] treats
+/// its samples as a cumulative counter and derives goodput from `end - start`.
+/// Forwarding a per-request total would make that difference *negative* at every
+/// request boundary, silently corrupting goodput and the saturation detection
+/// built on top of it. Totals are therefore rebased onto a running sum here.
+#[derive(Debug, Default)]
+struct CumulativeRelay {
+    /// Bytes accounted for by requests that have already completed.
+    base: usize,
+    /// Most recent total reported by the in-flight request.
+    last: usize,
+}
+
+/// What [`UploadReissue::run`] should do with a translated event.
+#[derive(Debug)]
+enum RelayAction {
+    /// Pass this event on to the consumer.
+    Forward(BodyEvent),
+    /// The current request completed; start another. Deliberately forwards
+    /// nothing: a `Finished` would set `finished_at` and retire a load that is
+    /// in fact still running.
+    RequestFinished,
+    /// Terminal failure. Forward it and stop.
+    Fail(BodyEvent),
+}
+
+impl CumulativeRelay {
+    fn on_event(&mut self, event: BodyEvent) -> RelayAction {
+        match event {
+            BodyEvent::ByteCount { at, total } => {
+                self.last = total;
+                RelayAction::Forward(BodyEvent::ByteCount {
+                    at,
+                    total: self.base + total,
+                })
+            }
+            BodyEvent::Finished { .. } => {
+                self.base += self.last;
+                self.last = 0;
+                RelayAction::RequestFinished
+            }
+            BodyEvent::Failed { at, reason } => RelayAction::Fail(BodyEvent::Failed { at, reason }),
+        }
+    }
+
+    /// Total bytes sent across every request so far.
+    fn total(&self) -> usize {
+        self.base + self.last
+    }
+}
+
+/// The observable state of a load-generating transfer.
+///
+/// Split out from [`LoadedConnection`] so the termination logic can be tested
+/// without constructing a real connection.
+#[derive(Debug, Default)]
+struct LoadState {
+    total_bytes_series: CounterSeries,
+    finished_at: Option<Timestamp>,
+    /// Set when the body's event channel closed *without* a `Finished` event,
+    /// i.e. the transfer died mid-flight.
+    failed: bool,
+    /// Why the transfer failed, when a `Failed` event supplied a reason. A
+    /// bare channel closure gives no reason, so this stays `None`.
+    failure_reason: Option<String>,
+    /// Set by [`LoadedConnection::stop`] so the channel closure it causes is
+    /// not misreported as a failure.
+    stopping: bool,
+}
+
+impl LoadState {
+    fn apply(&mut self, event: BodyEvent) {
+        match event {
+            BodyEvent::ByteCount { at, total } => self.total_bytes_series.add(at, total as f64),
+            BodyEvent::Finished { at } => self.finished_at = Some(at),
+            BodyEvent::Failed { reason, .. } => {
+                self.failed = true;
+                self.failure_reason = Some(reason);
+            }
+        }
+    }
+
+    /// Handle the body's event channel closing.
+    ///
+    /// `CountingBody` owns the only sender, so a closed channel means the body
+    /// was dropped. If that happened before a `Finished` event — and we are not
+    /// deliberately tearing the load down — the transfer terminated early
+    /// (stream reset, connection error, rejected request, ...).
+    ///
+    /// Without this, such a connection keeps `finished_at == None` forever and
+    /// lingers in `ongoing_loads()` as a zombie: it contributes no further
+    /// bytes to goodput yet still occupies a slot in the connection ramp.
+    fn on_disconnected(&mut self) {
+        if self.finished_at.is_none() && !self.stopping {
+            self.failed = true;
+        }
+    }
+
+    /// Whether the transfer is still running (neither completed nor failed).
+    ///
+    /// `finished_at == None` is the normal healthy state for an upload:
+    /// [`UploadReissue`] replaces each bounded request as it completes and
+    /// swallows the per-request `Finished`, so an upload load never reports
+    /// completion. That is exactly why a failure needs its own signal.
+    fn is_ongoing(&self) -> bool {
+        self.finished_at.is_none() && !self.failed
+    }
+
+    /// Drain all currently-available body events, and notice if the channel has
+    /// closed.
+    ///
+    /// `try_recv` yields any buffered events before reporting `Disconnected`,
+    /// so a body that emitted `Finished` and was then dropped is correctly seen
+    /// as completed rather than failed.
+    fn drain(&mut self, events_rx: &mut UnboundedReceiver<BodyEvent>) {
+        loop {
+            match events_rx.try_recv() {
+                Ok(event) => self.apply(event),
+                Err(TryRecvError::Empty) => break,
+                Err(TryRecvError::Disconnected) => {
+                    self.on_disconnected();
+                    break;
+                }
+            }
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct LoadedConnection {
     connection: Arc<RwLock<EstablishedConnection>>,
     events_rx: UnboundedReceiver<BodyEvent>,
-    total_bytes_series: CounterSeries,
-    finished_at: Option<Timestamp>,
+    state: LoadState,
 }
 
 impl LoadedConnection {
     pub fn update(&mut self) {
-        while let Ok(event) = self.events_rx.try_recv() {
-            match event {
-                BodyEvent::ByteCount { at, total } => self.total_bytes_series.add(at, total as f64),
-                BodyEvent::Finished { at } => self.finished_at = Some(at),
-            }
-        }
+        self.state.drain(&mut self.events_rx);
     }
 
     pub fn total_bytes_series(&self) -> &CounterSeries {
-        &self.total_bytes_series
+        &self.state.total_bytes_series
+    }
+
+    /// Whether this connection is still transferring.
+    pub fn is_ongoing(&self) -> bool {
+        self.state.is_ongoing()
+    }
+
+    /// Whether this connection terminated early with an error.
+    pub fn has_failed(&self) -> bool {
+        self.state.failed
+    }
+
+    /// Why this connection failed, if a reason was reported.
+    pub fn failure_reason(&self) -> Option<&str> {
+        self.state.failure_reason.as_deref()
     }
 
     pub fn stop(&mut self) {
+        self.state.stopping = true;
         self.events_rx.close();
         self.update();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+    use tokio::sync::mpsc;
+
+    fn channel() -> (
+        mpsc::UnboundedSender<BodyEvent>,
+        mpsc::UnboundedReceiver<BodyEvent>,
+    ) {
+        mpsc::unbounded_channel()
+    }
+
+    /// Feed a `ByteCount` through the relay and return the total it forwarded.
+    fn forward_bytes(relay: &mut CumulativeRelay, at: Timestamp, total: usize) -> usize {
+        match relay.on_event(BodyEvent::ByteCount { at, total }) {
+            RelayAction::Forward(BodyEvent::ByteCount { total, .. }) => total,
+            other => panic!("a ByteCount must be forwarded, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn totals_accumulate_across_request_boundaries() {
+        let at = Timestamp::now();
+        let mut relay = CumulativeRelay::default();
+
+        assert_eq!(forward_bytes(&mut relay, at, 40), 40);
+        assert_eq!(forward_bytes(&mut relay, at, 100), 100);
+        relay.on_event(BodyEvent::Finished { at });
+
+        // The next request counts from zero again; the consumer must not see
+        // that reset.
+        assert_eq!(forward_bytes(&mut relay, at, 0), 100);
+        assert_eq!(forward_bytes(&mut relay, at, 30), 130);
+        relay.on_event(BodyEvent::Finished { at });
+
+        assert_eq!(forward_bytes(&mut relay, at, 5), 135);
+        assert_eq!(relay.total(), 135);
+    }
+
+    #[test]
+    fn request_finished_is_never_forwarded() {
+        // A forwarded `Finished` would set `finished_at`, so `is_ongoing()` would
+        // go false and the ramp would retire a connection that is still running.
+        let at = Timestamp::now();
+        let mut relay = CumulativeRelay::default();
+
+        assert!(matches!(
+            relay.on_event(BodyEvent::Finished { at }),
+            RelayAction::RequestFinished
+        ));
+    }
+
+    #[test]
+    fn failure_is_terminal_and_forwarded() {
+        let at = Timestamp::now();
+        let mut relay = CumulativeRelay::default();
+
+        let action = relay.on_event(BodyEvent::Failed {
+            at,
+            reason: "upload rejected with status 413 Payload Too Large".to_owned(),
+        });
+
+        match action {
+            RelayAction::Fail(BodyEvent::Failed { reason, .. }) => {
+                assert!(reason.contains("413"));
+            }
+            other => panic!("a Failed must be forwarded as terminal, got {other:?}"),
+        }
+    }
+
+    // The regression that matters most. `CounterSeries::interval_sum` is
+    // `end - start`, so if a request boundary ever let a total reset to zero
+    // reach the series, goodput for that window would go *negative* -- which
+    // would silently corrupt the saturation detection that decides when the
+    // test has reached working conditions.
+    #[test]
+    fn relayed_totals_never_produce_negative_goodput() {
+        let start = Timestamp::now();
+        let step = Duration::from_millis(50);
+
+        let mut relay = CumulativeRelay::default();
+        let mut series = CounterSeries::new();
+        let mut at = start;
+
+        // Three consecutive 100-byte requests, each reporting in 25-byte steps.
+        for _ in 0..3 {
+            for total in [0usize, 25, 50, 75, 100] {
+                at = at + step;
+                let forwarded = forward_bytes(&mut relay, at, total);
+                series.add(at, forwarded as f64);
+            }
+            at = at + step;
+            relay.on_event(BodyEvent::Finished { at });
+        }
+
+        assert_eq!(relay.total(), 300, "three 100-byte requests");
+
+        let mut window = start;
+        while window < at {
+            let next = window + step;
+            let bytes = series.interval_sum(window, next);
+            assert!(
+                bytes >= 0.0,
+                "negative goodput ({bytes}) in one window -- a request boundary leaked a reset"
+            );
+            window = next;
+        }
+
+        assert_eq!(
+            series.interval_sum(start, at),
+            300.0,
+            "the whole run must account for every byte exactly once"
+        );
+    }
+
+    #[test]
+    fn open_channel_leaves_transfer_ongoing() {
+        let (tx, mut rx) = channel();
+        tx.send(BodyEvent::ByteCount {
+            at: Timestamp::now(),
+            total: 1024,
+        })
+        .unwrap();
+
+        let mut state = LoadState::default();
+        state.drain(&mut rx);
+
+        assert!(state.is_ongoing());
+        assert!(!state.failed);
+        // Keep the sender alive: an open channel must not look like a failure.
+        drop(tx);
+    }
+
+    #[test]
+    fn disconnect_without_finished_marks_failed() {
+        let (tx, mut rx) = channel();
+        tx.send(BodyEvent::ByteCount {
+            at: Timestamp::now(),
+            total: 10 * 1024 * 1024,
+        })
+        .unwrap();
+        // The body was dropped mid-transfer (e.g. the server rejected the
+        // upload with 413), closing the channel without a `Finished` event.
+        drop(tx);
+
+        let mut state = LoadState::default();
+        state.drain(&mut rx);
+
+        assert!(state.failed, "early termination must be flagged");
+        assert!(!state.is_ongoing(), "a failed load must not stay ongoing");
+    }
+
+    #[test]
+    fn finished_then_disconnect_is_not_a_failure() {
+        let (tx, mut rx) = channel();
+        let at = Timestamp::now();
+        tx.send(BodyEvent::ByteCount { at, total: 512 }).unwrap();
+        tx.send(BodyEvent::Finished { at }).unwrap();
+        // Normal completion: the body is dropped right after finishing. The
+        // buffered events must be drained before `Disconnected` is observed.
+        drop(tx);
+
+        let mut state = LoadState::default();
+        state.drain(&mut rx);
+
+        assert!(!state.failed, "a completed transfer must not be a failure");
+        assert_eq!(state.finished_at, Some(at));
+        assert!(!state.is_ongoing(), "a completed load is no longer ongoing");
+    }
+
+    #[test]
+    fn teardown_disconnect_is_not_a_failure() {
+        // `stop()` closes the receiver itself; that must not be mistaken for
+        // the connection dying, otherwise every run would end "with failures".
+        let (tx, mut rx) = channel();
+        drop(tx);
+
+        let mut state = LoadState::default();
+        state.stopping = true;
+        state.drain(&mut rx);
+
+        assert!(!state.failed, "teardown must not be flagged as a failure");
+    }
+
+    #[test]
+    fn explicit_failed_event_retires_the_load_with_a_reason() {
+        // e.g. an upload rejected with 413: the client reports the failure the
+        // body itself cannot see.
+        let (tx, mut rx) = channel();
+        let at = Timestamp::now();
+        tx.send(BodyEvent::ByteCount { at, total: 1024 }).unwrap();
+        tx.send(BodyEvent::Failed {
+            at,
+            reason: "upload rejected with status 413 Payload Too Large".to_owned(),
+        })
+        .unwrap();
+
+        let mut state = LoadState::default();
+        state.drain(&mut rx);
+
+        assert!(state.failed);
+        assert!(!state.is_ongoing());
+        assert_eq!(
+            state.failure_reason.as_deref(),
+            Some("upload rejected with status 413 Payload Too Large")
+        );
+        // Sender still alive: the failure must be recognised from the event
+        // alone, without relying on the channel closing.
+        drop(tx);
+    }
+
+    #[test]
+    fn bytes_seen_before_failure_are_retained() {
+        // A failed connection still transferred real bytes; goodput accounting
+        // must keep them.
+        let (tx, mut rx) = channel();
+        let at = Timestamp::now();
+        tx.send(BodyEvent::ByteCount { at, total: 4096 }).unwrap();
+        drop(tx);
+
+        let mut state = LoadState::default();
+        state.drain(&mut rx);
+
+        assert!(state.failed);
+        assert_eq!(state.total_bytes_series.sum(), 4096.0);
+    }
+}
diff --git a/crates/nq-packetloss/src/lib.rs b/crates/nq-packetloss/src/lib.rs
index 7559ad0..78c6c9e 100644
--- a/crates/nq-packetloss/src/lib.rs
+++ b/crates/nq-packetloss/src/lib.rs
@@ -68,7 +68,6 @@ impl PacketLossConfig {
             scoped_headers: self.scoped_headers.clone(),
             download_url: self.download_url.clone(),
             upload_url: self.upload_url.clone(),
-            upload_size: 4_000_000_000, // 4 GB
         }
     }
 }
diff --git a/crates/nq-rpm/src/lib.rs b/crates/nq-rpm/src/lib.rs
index ba8ca3d..abe33b9 100644
--- a/crates/nq-rpm/src/lib.rs
+++ b/crates/nq-rpm/src/lib.rs
@@ -5,23 +5,41 @@ use std::{
     collections::HashMap,
     fmt::{Debug, Display},
     future::Future,
-    ops::Div,
     sync::Arc,
     time::Duration,
 };
 
 use humansize::{DECIMAL, format_size};
 use nq_core::{
-    ConnectionType, Network, ScopedHeaders, Time, Timestamp,
+    ConnectionTiming, ConnectionType, Network, ScopedHeaders, Time, Timestamp,
     client::{Direction, ThroughputClient, wait_for_finish},
 };
 use nq_load_generator::{LoadConfig, LoadGenerator, LoadedConnection};
 use nq_stats::{TimeSeries, instant_minus_intervals};
 use tokio::{select, sync::mpsc};
 use tokio_util::sync::CancellationToken;
-use tracing::{Instrument, debug, error, info};
+use tracing::{Instrument, debug, error, info, warn};
 use url::Url;
 
+/// What to do when a load-generating connection terminates with an error.
+///
+/// draft-ietf-ippm-responsiveness-09 §5.4 says "if at any point one of these
+/// connections terminates with an error, the test should be aborted". That
+/// "should" is lowercase, so it is advisory rather than a BCP 14 requirement,
+/// and aborting outright is often not the most useful behaviour: a server that
+/// rejects oversized uploads (HTTP 413) would abort every run. The default
+/// therefore retires the failed connection and lets the ramp replace it, while
+/// still recording and reporting the failure.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum ConnectionErrorPolicy {
+    /// Retire the failed connection, keep measuring, and report the failure
+    /// count. Aborts only if load can no longer be sustained at all.
+    #[default]
+    Retire,
+    /// Abort the test on the first failure, as the draft literally describes.
+    Abort,
+}
+
 #[derive(Debug, Clone)]
 pub struct ResponsivenessConfig {
     pub large_download_url: Url,
@@ -35,6 +53,20 @@ pub struct ResponsivenessConfig {
     pub max_loaded_connections: usize,
     pub conn_type: ConnectionType,
     pub determine_load_only: bool,
+    /// Maximum bytes sent in any single upload load-generating request.
+    ///
+    /// Upload load is generated as a sequence of requests of this size on each
+    /// connection, rather than one enormous request, because servers may cap
+    /// request body size and reject anything larger with HTTP 413. Such caps
+    /// apply per-request, so staying under one here keeps the link loaded
+    /// indefinitely without ever tripping it.
+    ///
+    /// Must be below the smallest such cap on the path, with margin. It has no
+    /// effect on connections too slow to send this many bytes within the test
+    /// duration, since their first request never completes either way.
+    pub upload_bytes_per_request: usize,
+    /// What to do when a load-generating connection terminates with an error.
+    pub on_connection_error: ConnectionErrorPolicy,
     /// Headers attached only to requests whose host matches the scope's
     /// allowlist.
     pub scoped_headers: Option<ScopedHeaders>,
@@ -47,11 +79,18 @@ impl ResponsivenessConfig {
             scoped_headers: self.scoped_headers.clone(),
             download_url: self.large_download_url.clone(),
             upload_url: self.upload_url.clone(),
-            upload_size: 4_000_000_000, // 4 GB
         }
     }
 }
 
+/// Default bytes per upload load-generating request.
+///
+/// 100 MB sits well under the request body caps servers commonly impose, with
+/// margin for the stricter ones. On links too slow to send that much within the
+/// test duration the first request never completes anyway, so for them this is
+/// indistinguishable from an unbounded request.
+pub const DEFAULT_UPLOAD_BYTES_PER_REQUEST: usize = 100_000_000;
+
 impl Default for ResponsivenessConfig {
     fn default() -> Self {
         Self {
@@ -70,6 +109,8 @@ impl Default for ResponsivenessConfig {
             max_loaded_connections: 16,
             conn_type: ConnectionType::H2,
             determine_load_only: false,
+            upload_bytes_per_request: DEFAULT_UPLOAD_BYTES_PER_REQUEST,
+            on_connection_error: ConnectionErrorPolicy::default(),
             scoped_headers: None,
         }
     }
@@ -86,14 +127,32 @@ pub struct Responsiveness {
     goodput_saturated: bool,
     rpm_saturated: bool,
     direction: Direction,
-    rpm: f64,
+    /// The value to report, set once responsiveness saturation is declared.
+    /// `None` until then; see [`Self::last_rpm`] for the unconverged case.
+    rpm: Option<f64>,
+    /// RPM at the most recent interval that actually produced a measurement.
+    ///
+    /// This is the value reported when the test hits its time limit without
+    /// declaring saturation, which for the upload leg is the norm rather than
+    /// the exception. Each sample is already a trimmed mean over the moving
+    /// average window (see [`compute_responsiveness`]), so it is reported as-is
+    /// and must not be averaged a second time.
+    last_rpm: Option<f64>,
     capacity: f64,
+    /// Load-generating connections that terminated early with an error.
+    failed_connections: usize,
+    /// Consecutive intervals that ended with no live load-generating
+    /// connection while failures were occurring.
+    starved_intervals: usize,
 }
 
 impl Responsiveness {
     pub fn new(config: ResponsivenessConfig, download: bool) -> anyhow::Result<Self> {
         let load_generator = LoadGenerator::new(config.load_config())?;
 
+        // Read before `config` is moved into the struct below.
+        let upload_bytes_per_request = config.upload_bytes_per_request;
+
         Ok(Self {
             start: Timestamp::now(),
             config,
@@ -102,14 +161,20 @@ impl Responsiveness {
             self_probe_results: Default::default(),
             average_goodput_series: TimeSeries::new(),
             rpm_series: TimeSeries::new(),
+            failed_connections: 0,
+            starved_intervals: 0,
             goodput_saturated: false,
             rpm_saturated: false,
+            // For uploads this is the size of each individual request, which the
+            // load generator re-issues on the same connection for the duration of
+            // the test -- not a total to be reached.
             direction: if download {
                 Direction::Down
             } else {
-                Direction::Up(std::cmp::min(32u64 * 1024 * 1024 * 1024, usize::MAX as u64) as usize)
+                Direction::Up(upload_bytes_per_request)
             },
-            rpm: 0.0,
+            rpm: None,
+            last_rpm: None,
             capacity: 0.0,
         })
     }
@@ -202,12 +267,27 @@ impl Responsiveness {
         }
 
         let now = env.time.now();
-        if self.rpm == 0.0 {
-            self.rpm = self
-                .rpm_series
-                .interval_average(now - Duration::from_secs(2), now)
-                .unwrap_or(0.0);
-        }
+
+        // The loop above exited without responsiveness ever stabilizing, which
+        // happens whenever the time limit is reached first -- the normal outcome
+        // for the upload leg. draft-ietf-ippm-responsiveness-09 §5.4 says to
+        // report the current result in that case rather than nothing, and
+        // "current_responsiveness" means the value at the final interval, not an
+        // average of recent ones: each sample is already a trimmed mean across
+        // the moving average window.
+        //
+        // This deliberately does NOT read a wall-clock window. It used to be
+        // `interval_average(now - 2s, now)`, but samples are stamped with the
+        // computed `start + interval_duration * interval` rather than the time
+        // they were taken, and `on_interval(i)` runs about one interval after the
+        // instant it stamps. The newest sample therefore sat almost exactly 2s
+        // behind `now`, so whether it fell inside the window came down to how
+        // promptly the loop happened to exit. When the exit was ~1s late the
+        // window matched nothing and `unwrap_or(0.0)` reported 0 RPM for an
+        // otherwise healthy run. Measured over 14 local runs: 13 exited within
+        // a millisecond and squeaked in, one exited 0.999s later and reported
+        // zero.
+        self.rpm = select_reported_rpm(self.rpm, self.last_rpm);
 
         // stop all on-going loads.
         let mut loads = self.load_generator.into_connections();
@@ -219,6 +299,7 @@ impl Responsiveness {
             foreign_loaded_latencies: self.foreign_probe_results.http,
             self_probe_latencies: self.self_probe_results.http,
             loaded_connections: loads,
+            failed_connections: self.failed_connections,
             duration: now.duration_since(self.start),
             average_goodput_series: self.average_goodput_series,
         })
@@ -263,6 +344,8 @@ impl Responsiveness {
             self.config.interval_duration,
         );
 
+        self.enforce_connection_error_policy()?;
+
         // always start a load generating connection
         // TODO: only if goodput is not saturated?
         if self.load_generator.count_loads() < self.config.max_loaded_connections
@@ -290,21 +373,41 @@ impl Responsiveness {
             self.goodput_saturated = true;
         }
 
+        // `None` means this window held no probe measurements at all.
         let current_rpm = compute_responsiveness(
             &self.foreign_probe_results,
             &self.self_probe_results,
             start_data_interval,
             end_data_interval,
             self.config.trimmed_mean_percent,
-        )
-        .unwrap_or(0.0);
+        );
 
-        if current_rpm.is_nan() {
-            panic!("NaN rpm!");
+        // An interval with no probes still contributes a 0.0 sample. That is
+        // arguably wrong on its face, but it is load-bearing and must not be
+        // "cleaned up" in isolation: a 0.0 sitting among values near 340 is a
+        // large outlier that inflates `interval_std`, and that inflation is the
+        // only thing currently preventing responsiveness from being declared
+        // stable during the ramp. Saturation is not gated on goodput saturation
+        // (see the conformance audit), so an early declaration latches
+        // `self.rpm` at a high ramp value and keeps it.
+        //
+        // Measured: dropping these samples made the upload leg latch at interval
+        // 1-2 while throughput_saturated was still false, reporting 593-709 RPM
+        // against a true value near 331 -- roughly double. Removing them
+        // requires gating saturation on goodput first, which is a separate
+        // change with its own validation.
+        //
+        // No NaN check is needed here: `compute_responsiveness` only yields
+        // `Some` for finite values, and 0.0 is finite.
+        let current_rpm_or_zero = current_rpm.unwrap_or(0.0);
+        self.rpm_series.add(end_data_interval, current_rpm_or_zero);
+
+        // Only genuine measurements are eligible to be reported at the end, so a
+        // probe-less final interval cannot surface as "0 RPM".
+        if let Some(current_rpm) = current_rpm {
+            self.last_rpm = Some(current_rpm);
         }
 
-        self.rpm_series.add(end_data_interval, current_rpm);
-
         let std_rpm = self
             .rpm_series
             .interval_std(start_data_interval, end_data_interval);
@@ -312,8 +415,11 @@ impl Responsiveness {
         let is_rpm_saturated = if let Some(std_rpm) = std_rpm {
             // RPM is saturated if the std of the last MAD RPMs is
             // within tolerance % of the current_rpm.
-            if std_rpm < current_rpm * self.config.std_tolerance {
-                self.rpm = current_rpm;
+            //
+            // When `current_rpm_or_zero` is 0.0 this is `std_rpm < 0.0`, which is
+            // never true, so a probe-less interval can never latch 0 RPM.
+            if std_rpm < current_rpm_or_zero * self.config.std_tolerance {
+                self.rpm = Some(current_rpm_or_zero);
                 self.rpm_saturated = true;
                 true
             } else {
@@ -328,7 +434,8 @@ impl Responsiveness {
             current_goodput,
             std_goodput,
             goodput_saturated,
-            current_rpm,
+            current_rpm_or_zero,
+            current_rpm.is_some(),
             std_rpm,
             is_rpm_saturated,
         );
@@ -345,6 +452,7 @@ impl Responsiveness {
         std_goodput: f64,
         goodput_saturated: bool,
         current_rpm: f64,
+        rpm_measured: bool,
         std_rpm: Option<f64>,
         is_rpm_saturated: bool,
     ) {
@@ -354,6 +462,9 @@ impl Responsiveness {
             .long_units(false)
             .decimal_places(2);
 
+        // Logs the value the algorithm actually used, including the substituted
+        // 0.0, so the log matches the arithmetic. The substitution itself is
+        // surfaced separately below rather than being silent.
         info!(
             interval,
             loads = self.load_generator.count_loads(),
@@ -364,6 +475,14 @@ impl Responsiveness {
             "interval finished"
         );
 
+        if !rpm_measured {
+            warn!(
+                interval,
+                "no probe measurements in this interval's window; recorded 0 RPM, \
+                 which inflates the stability std for the next MAD intervals"
+            );
+        }
+
         info!(
             interval,
             throughput_std = format_size(std_goodput as usize, custom_options),
@@ -401,6 +520,63 @@ impl Responsiveness {
         8.0 * bytes_seen / total_time
     }
 
+    /// Apply [`ConnectionErrorPolicy`] to load-generating connections that
+    /// terminated early.
+    ///
+    /// Implements draft-ietf-ippm-responsiveness-09 §5.4's guidance that the
+    /// test should be aborted when a connection terminates with an error. See
+    /// [`ConnectionErrorPolicy`] for why the default is more forgiving than the
+    /// literal wording.
+    fn enforce_connection_error_policy(&mut self) -> anyhow::Result<()> {
+        let failed = self.load_generator.count_failed_loads();
+        let newly_failed = failed.saturating_sub(self.failed_connections);
+        self.failed_connections = failed;
+
+        if newly_failed > 0 {
+            let reason = self
+                .load_generator
+                .connections()
+                .filter_map(|c| c.failure_reason())
+                .last()
+                .unwrap_or("connection terminated early")
+                .to_owned();
+
+            warn!(
+                newly_failed,
+                total_failed = failed,
+                reason = %reason,
+                "load-generating connection(s) terminated with an error"
+            );
+
+            if self.config.on_connection_error == ConnectionErrorPolicy::Abort {
+                anyhow::bail!(
+                    "aborting test: {failed} load-generating connection(s) terminated with an \
+                     error (most recent: {reason})"
+                );
+            }
+        }
+
+        // Retiring failed connections only helps if the ramp can replace them.
+        // If an interval ends with nothing left transferring while failures are
+        // happening, no load is being generated and any responsiveness figure
+        // would be measured off an idle link -- so refuse to report one.
+        if failed > 0 && self.load_generator.count_loads() == 0 {
+            self.starved_intervals += 1;
+
+            if self.starved_intervals >= 2 {
+                anyhow::bail!(
+                    "aborting test: no load-generating connections could be sustained \
+                     ({failed} terminated with an error); the link was never saturated so a \
+                     responsiveness result would be meaningless"
+                );
+            }
+        } else {
+            self.starved_intervals = 0;
+        }
+
+        Ok(())
+    }
+
     /// A GET/POST to an endpoint which sends/receives a large number of bytes
     /// as quickly as possible. The intent of these connections is to saturate
     /// a single connection's flow.
@@ -470,14 +646,15 @@ impl Responsiveness {
                     anyhow::bail!("a new connection with timing should have been created");
                 };
 
+                let (tcp, tls, http) =
+                    foreign_probe_phases(&connection_timing, finished_result.finished_at);
+
                 if event_tx
                     .send(Event::ForeignProbe(ForeignProbeResult {
                         start: connection_timing.start(),
-                        time_connect: connection_timing.time_connect(),
-                        time_secure: connection_timing.time_secure(),
-                        time_body: finished_result
-                            .finished_at
-                            .duration_since(connection_timing.start()),
+                        tcp,
+                        tls,
+                        http,
                     }))
                     .await
                     .is_err()
@@ -575,11 +752,11 @@ pub struct ForeignProbeResults {
 impl ForeignProbeResults {
     pub fn add(&mut self, result: ForeignProbeResult) {
         self.connect
-            .add(result.start, result.time_connect.as_secs_f64() * 1000.0);
+            .add(result.start, result.tcp.as_secs_f64() * 1000.0);
         self.secure
-            .add(result.start, result.time_secure.as_secs_f64() * 1000.0);
+            .add(result.start, result.tls.as_secs_f64() * 1000.0);
         self.http
-            .add(result.start, result.time_body.as_secs_f64() * 1000.0);
+            .add(result.start, result.http.as_secs_f64() * 1000.0);
     }
 
     pub fn connect(&self) -> &TimeSeries {
@@ -611,11 +788,33 @@ impl SelfProbeResults {
     }
 }
 
-/// The responsiveness is then calculated as the weighted mean:
+/// Responsiveness per draft-ietf-ippm-responsiveness-09 §5.3.1.1 (TLS-enabled
+/// case): convert each side to RPM first, then take the arithmetic mean of the
+/// two RPMs.
+///
+///   Foreign_Responsiveness = 60000 / ((TM(tcp_f) + TM(tls_f) + TM(http_f)) / 3)
+///   Loaded_Responsiveness  = 60000 / TM(http_l)
+///   Responsiveness         = (Foreign_Responsiveness + Loaded_Responsiveness) / 2
+///
+/// https://datatracker.ietf.org/doc/html/draft-ietf-ippm-responsiveness-09#section-5.3.1.1
+/// Pick the RPM to report for a leg that has finished.
+///
+/// `saturated` holds a value only once responsiveness saturation has been
+/// declared, which draft-ietf-ippm-responsiveness-09 §5.4 wants reported as the
+/// final result. Otherwise the test hit its time limit, and the draft directs us
+/// to report the current result instead -- the most recent interval that
+/// produced a measurement.
+///
+/// `None` from both means no interval ever measured anything, which is missing
+/// data and must not be flattened into a number by callers.
 ///
-/// Responsiveness = 60000 /
-/// (1/6*(TM(tcp_f) + TM(tls_f) + TM(http_f)) + 1/2*TM(http_s))
-/// https://datatracker.ietf.org/doc/html/draft-ietf-ippm-responsiveness-03#section-4.3.1-4
+/// Deliberately takes no clock and no time window; see the call site in
+/// [`Responsiveness::run_test`] for the wall-clock window this replaced and why
+/// it could report zero.
+fn select_reported_rpm(saturated: Option<f64>, last_interval: Option<f64>) -> Option<f64> {
+    saturated.or(last_interval)
+}
+
 fn compute_responsiveness(
     foreign_results: &ForeignProbeResults,
     self_results: &SelfProbeResults,
@@ -628,19 +827,61 @@ fn compute_responsiveness(
     let tcp_f = tm(foreign_results.connect())?;
     let tls_f = tm(foreign_results.secure())?;
     let http_f = tm(foreign_results.http())?;
-    let http_s = tm(self_results.http())?;
+    let http_l = tm(self_results.http())?;
+
+    // Mean foreign round-trip time and loaded round-trip time, in milliseconds.
+    let foreign_rtt = (tcp_f + tls_f + http_f) / 3.0;
+    let loaded_rtt = http_l;
+
+    // Guard against non-positive RTTs, which would produce a non-finite RPM.
+    if foreign_rtt <= 0.0 || loaded_rtt <= 0.0 {
+        return None;
+    }
+
+    let foreign_rpm = 60_000.0 / foreign_rtt;
+    let loaded_rpm = 60_000.0 / loaded_rtt;
 
-    let foreign_sum = tcp_f + tls_f + http_f;
+    let responsiveness = (foreign_rpm + loaded_rpm) / 2.0;
 
-    Some(60_000.0 / (foreign_sum.div(6.0) + http_s.div(2.0)))
+    responsiveness.is_finite().then_some(responsiveness)
 }
 
 #[derive(Debug)]
 pub struct ForeignProbeResult {
+    /// Timestamp used to place the probe within the measurement window.
     start: Timestamp,
-    time_connect: Duration,
-    time_secure: Duration,
-    time_body: Duration,
+    /// TCP handshake duration (`tcp_f`).
+    tcp: Duration,
+    /// TLS handshake duration, normalized to the number of TLS round-trips
+    /// (`tls_f`).
+    tls: Duration,
+    /// HTTP request-issued to full-response-received duration (`http_f`).
+    http: Duration,
+}
+
+/// Computes the three independent foreign-probe phases per
+/// draft-ietf-ippm-responsiveness-09 §5.3:
+///
+/// * `tcp_f`  — the TCP handshake duration (DNS excluded).
+/// * `tls_f`  — the TLS handshake duration, normalized to the number of TLS
+///   round-trips the negotiated version uses.
+/// * `http_f` — the elapsed time between issuing the GET request and receiving
+///   the entire response, derived as `finished_at - (start + time_application)`,
+///   i.e. the interval after the connection is ready to transmit data.
+///
+/// These are deliberately non-overlapping: the earlier draft-03-style code
+/// measured every phase cumulatively from the connection start, which
+/// over-counted the foreign round-trip time (and thus under-reported RPM).
+fn foreign_probe_phases(
+    timing: &ConnectionTiming,
+    finished_at: Timestamp,
+) -> (Duration, Duration, Duration) {
+    let tcp_f = timing.tcp_handshake();
+    let tls_f = timing.tls_handshake() / timing.tls_round_trips();
+    let request_issued = timing.start() + timing.time_application();
+    let http_f = finished_at.duration_since(request_issued);
+
+    (tcp_f, tls_f, http_f)
 }
 
 #[derive(Debug)]
@@ -677,11 +918,21 @@ struct Env {
 pub struct ResponsivenessResult {
     pub duration: Duration,
     pub capacity: f64,
-    pub rpm: f64,
+    /// Round-trips per minute under working conditions.
+    ///
+    /// `None` means the test produced no responsiveness measurement at all --
+    /// not that responsiveness was zero. Consumers must surface that as missing
+    /// data rather than substituting a placeholder, because a plausible-looking
+    /// number is indistinguishable from a real one once it leaves this crate.
+    pub rpm: Option<f64>,
     pub foreign_loaded_latencies: TimeSeries,
     pub self_probe_latencies: TimeSeries,
     pub loaded_connections: Vec<LoadedConnection>,
     pub average_goodput_series: TimeSeries,
+    /// Load-generating connections that terminated early with an error. A
+    /// non-zero value means the link was not fully loaded for part of the run,
+    /// so the result is degraded.
+    pub failed_connections: usize,
 }
 
 impl ResponsivenessResult {
@@ -704,6 +955,228 @@ impl Display for ResponsivenessResult {
             "capacity",
             format_size(self.capacity as usize, custom_options)
         )?;
-        write!(f, "{:>8}: {}", "rpm", self.rpm.round() as usize)
+        match self.rpm {
+            Some(rpm) => write!(f, "{:>8}: {}", "rpm", rpm.round() as usize),
+            None => write!(f, "{:>8}: unavailable", "rpm"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    fn ms(v: f64) -> Duration {
+        Duration::from_secs_f64(v / 1000.0)
+    }
+
+    /// Build foreign/self probe series with `n` identical samples for the given
+    /// per-phase latencies (in milliseconds), returning the results plus a
+    /// [from, to] window that covers all samples.
+    fn series(
+        tcp_ms: f64,
+        tls_ms: f64,
+        http_f_ms: f64,
+        http_l_ms: f64,
+    ) -> (ForeignProbeResults, SelfProbeResults, Timestamp, Timestamp) {
+        let start = Timestamp::now();
+        let mut foreign = ForeignProbeResults::default();
+        let mut selfp = SelfProbeResults::default();
+
+        for i in 0..10u64 {
+            let at = start + Duration::from_millis(i);
+            foreign.add(ForeignProbeResult {
+                start: at,
+                tcp: ms(tcp_ms),
+                tls: ms(tls_ms),
+                http: ms(http_f_ms),
+            });
+            selfp.add(SelfProbeResult {
+                start: at,
+                time_body: ms(http_l_ms),
+            });
+        }
+
+        (foreign, selfp, start, start + Duration::from_millis(100))
+    }
+
+    /// The old draft-03 harmonic combination, kept here only to prove the new
+    /// formula reports a higher (less biased) value.
+    fn draft03(tcp: f64, tls: f64, http_f: f64, http_l: f64) -> f64 {
+        let foreign_sum = tcp + tls + http_f;
+        60_000.0 / (foreign_sum / 6.0 + http_l / 2.0)
+    }
+
+    #[test]
+    fn arithmetic_mean_of_the_two_rpms() {
+        // F = (30+30+30)/3 = 30 -> foreign_rpm = 2000
+        // L = 30            -> loaded_rpm  = 2000
+        // responsiveness    = (2000 + 2000) / 2 = 2000
+        let (f, s, from, to) = series(30.0, 30.0, 30.0, 30.0);
+        let rpm = compute_responsiveness(&f, &s, from, to, 0.95).unwrap();
+        assert!((rpm - 2000.0).abs() < 1e-6, "got {rpm}");
+    }
+
+    #[test]
+    fn equals_draft03_only_when_foreign_equals_loaded() {
+        // When F == L the arithmetic and harmonic means coincide.
+        let (f, s, from, to) = series(30.0, 30.0, 30.0, 30.0);
+        let rpm = compute_responsiveness(&f, &s, from, to, 0.95).unwrap();
+        assert!((rpm - draft03(30.0, 30.0, 30.0, 30.0)).abs() < 1e-6);
+    }
+
+    #[test]
+    fn reports_higher_than_draft03_when_rtts_diverge() {
+        // Foreign RTT (60ms) slower than loaded RTT (20ms): AM > HM.
+        // new: (60000/60 + 60000/20)/2 = (1000 + 3000)/2 = 2000
+        // old: 60000/((180/6) + (20/2)) = 60000/40 = 1500
+        let (f, s, from, to) = series(60.0, 60.0, 60.0, 20.0);
+        let rpm = compute_responsiveness(&f, &s, from, to, 0.95).unwrap();
+        let old = draft03(60.0, 60.0, 60.0, 20.0);
+        assert!((rpm - 2000.0).abs() < 1e-6, "got {rpm}");
+        assert!(rpm > old, "new {rpm} should exceed draft-03 {old}");
+    }
+
+    #[test]
+    fn returns_none_without_samples() {
+        let f = ForeignProbeResults::default();
+        let s = SelfProbeResults::default();
+        let start = Timestamp::now();
+        let to = start + Duration::from_millis(100);
+        assert!(compute_responsiveness(&f, &s, start, to, 0.95).is_none());
+    }
+
+    #[test]
+    fn returns_none_on_zero_rtt() {
+        // Degenerate all-zero latencies must not yield a non-finite RPM.
+        let (f, s, from, to) = series(0.0, 0.0, 0.0, 0.0);
+        assert!(compute_responsiveness(&f, &s, from, to, 0.95).is_none());
+    }
+
+    /// Build a ConnectionTiming with phases at the given ms offsets from a
+    /// post-DNS baseline, plus a TLS round-trip count.
+    fn conn_timing(
+        connect_ms: u64,
+        secure_ms: u64,
+        application_ms: u64,
+        tls_round_trips: u32,
+    ) -> (ConnectionTiming, Timestamp) {
+        let start = Timestamp::now();
+        let mut t = ConnectionTiming::new(start);
+        t.set_connect(start + Duration::from_millis(connect_ms));
+        t.set_secure(start + Duration::from_millis(secure_ms));
+        t.set_application(start + Duration::from_millis(application_ms));
+        t.set_tls_round_trips(tls_round_trips);
+        (t, start)
+    }
+
+    #[test]
+    fn foreign_phases_are_independent_single_rtt_each() {
+        // connect @30, secure @60, application @62, body finished @92.
+        // tcp_f = 30, tls_f = 30 (1 RT), http_f = 92 - 62 = 30.
+        let (t, start) = conn_timing(30, 60, 62, 1);
+        let finished_at = start + Duration::from_millis(92);
+        let (tcp, tls, http) = foreign_probe_phases(&t, finished_at);
+        assert_eq!(tcp, Duration::from_millis(30));
+        assert_eq!(tls, Duration::from_millis(30));
+        assert_eq!(http, Duration::from_millis(30));
+    }
+
+    #[test]
+    fn foreign_tls_phase_normalized_by_round_trips() {
+        // TLS 1.2 (2 round-trips): raw TLS handshake 60ms -> normalized 30ms.
+        // connect @30, secure @90 (60ms TLS), application @92, finished @122.
+        let (t, start) = conn_timing(30, 90, 92, 2);
+        let finished_at = start + Duration::from_millis(122);
+        let (tcp, tls, http) = foreign_probe_phases(&t, finished_at);
+        assert_eq!(tcp, Duration::from_millis(30));
+        assert_eq!(tls, Duration::from_millis(30)); // 60ms / 2
+        assert_eq!(http, Duration::from_millis(30));
+    }
+
+    #[test]
+    fn foreign_phases_differ_from_cumulative_measurement() {
+        // Proves the fix changed behavior: the old code used cumulative
+        // durations (connect-from-start, secure-from-start, finished-from-start).
+        let (t, start) = conn_timing(30, 60, 62, 1);
+        let finished_at = start + Duration::from_millis(92);
+
+        let (tcp, tls, http) = foreign_probe_phases(&t, finished_at);
+        let new_sum = (tcp + tls + http).as_secs_f64() * 1000.0; // 90ms
+
+        // Old (draft-03-style) cumulative sum.
+        let old_tcp = t.time_connect().as_secs_f64() * 1000.0; // 30
+        let old_tls = t.time_secure().as_secs_f64() * 1000.0; // 60
+        let old_http = finished_at.duration_since(t.start()).as_secs_f64() * 1000.0; // 92
+        let old_sum = old_tcp + old_tls + old_http; // 182
+
+        assert!(new_sum < old_sum, "new {new_sum} should be < old {old_sum}");
+        assert!((new_sum - 90.0).abs() < 1e-6);
+        assert!((old_sum - 182.0).abs() < 1e-6);
+    }
+
+    #[test]
+    fn reports_the_saturated_value_when_responsiveness_converged() {
+        // A declared saturation value wins over the last interval's sample.
+        assert_eq!(select_reported_rpm(Some(340.0), Some(999.0)), Some(340.0));
+    }
+
+    #[test]
+    fn reports_the_last_interval_when_the_time_limit_is_reached() {
+        // The unconverged case, which is the norm for the upload leg: report the
+        // most recent measurement rather than nothing.
+        assert_eq!(select_reported_rpm(None, Some(347.9)), Some(347.9));
+    }
+
+    #[test]
+    fn reports_nothing_when_no_interval_ever_measured() {
+        // Must stay absent rather than becoming 0.0: a zero is indistinguishable
+        // from a real measurement once it leaves this crate.
+        assert_eq!(select_reported_rpm(None, None), None);
+    }
+
+    /// Pins the failure mode that made this fix necessary, so that nobody
+    /// reintroduces a wall-clock window to read the final RPM.
+    ///
+    /// RPM samples are stamped with the *computed* interval boundary
+    /// `start + interval_duration * interval`, but `on_interval(i)` runs roughly
+    /// one interval after the instant it stamps. The old code read the result
+    /// back with `interval_average(now - 2s, now)` against the wall clock, so the
+    /// newest sample sat almost exactly on the window's lower edge and whether it
+    /// was included depended purely on how promptly the run loop exited.
+    #[test]
+    fn a_wall_clock_window_can_miss_every_sample() {
+        let start = Timestamp::now();
+        let interval_duration = Duration::from_secs(1);
+
+        // Eleven intervals of samples, stamped the way `on_interval` stamps them.
+        let mut rpm_series = TimeSeries::new();
+        for i in 0..11u32 {
+            rpm_series.add(start + interval_duration * i, 300.0 + f64::from(i));
+        }
+        let newest = start + interval_duration * 10;
+
+        // Prompt exit: the newest sample lands exactly on the lower edge and is
+        // included, which is why this usually appeared to work.
+        let now = newest + Duration::from_secs(2);
+        assert_eq!(
+            rpm_series.interval_average(now - Duration::from_secs(2), now),
+            Some(310.0),
+            "sample exactly on the window edge should be included"
+        );
+
+        // Exit delayed by a single millisecond past that edge: the window now
+        // matches nothing at all, and the old `.unwrap_or(0.0)` turned this into
+        // a reported 0 RPM for a run whose final interval measured 310.
+        let now = newest + Duration::from_secs(2) + Duration::from_millis(1);
+        assert_eq!(
+            rpm_series.interval_average(now - Duration::from_secs(2), now),
+            None,
+            "a 1ms later exit must empty the window -- this is the bug"
+        );
+
+        // The replacement does not consult a clock, so the delay is irrelevant.
+        assert_eq!(select_reported_rpm(None, Some(310.0)), Some(310.0));
     }
 }