clawhdf5-remote: request timeouts scale with the body

timeout_global (60 s) covered a whole request, and a request can carry
8 MiB (max_request): below about 140 KB/s every block run timed out, was
retried from scratch and failed, so a slow link could not read remote
files at all.

HttpOptions::timeout (now 30 s) bounds connecting and receiving the
response headers; the body gets timeout + its size at the new
HttpOptions::min_speed (16 KiB/s by default: 94 s for a 1 MiB block).
A slow but moving link is not cut off; a stalled one still fails.
(ureq has no idle timeout; its body timeout is a total budget.)

The test server can throttle bodies and stall mid-body. Test: a 256 KiB
block at 256 KiB/s reads with a 300 ms timeout (it failed before), and a
body stalled for 20 s fails in under 5 s.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

crates/clawhdf5-remote/src/http.rs

@@ -24,6 +24,11 @@
 //! never from `https` to `http`, and the custom
 //! [`HttpOptions::headers`] are not sent to another origin.
 //!
+//! Timeouts scale with the request: [`HttpOptions::timeout`] to connect
+//! and to get the response headers, and for the body that plus its size at
+//! [`HttpOptions::min_speed`] — a slow link is not cut off mid-block, a
+//! stalled connection still is.
+//!
 //! Transient failures — connection errors, timeouts, `408`/`429`/`5xx`, and
 //! a body shorter or longer than its `Content-Range` — are retried with
 //! exponential backoff. Responses are requested with
@@ -53,8 +58,15 @@ pub struct HttpOptions {
     pub retries: u32,
     /// Delay before the first retry; doubled for each further one.
     pub backoff: Duration,
-    /// Timeout of one request, from connecting to the end of the body.
+    /// Time allowed to connect, and then to receive the response headers.
+    /// The body gets this plus the time it takes at
+    /// [`min_speed`](Self::min_speed), so a request's budget grows with its
+    /// size: a slow but moving link is not cut off, a stalled one is.
     pub timeout: Duration,
+    /// Slowest transfer rate tolerated, in bytes per second: receiving a
+    /// body of `n` bytes may take `timeout + n / min_speed` (16 KiB/s by
+    /// default: 94 s for a 1 MiB block, 9 min for an 8 MiB request).
+    pub min_speed: u64,
     /// Requests of one `read_ranges` call in flight at once.
     pub max_parallel: usize,
     /// Bytes fetched by the first request, from offset 0 (the superblock and
@@ -88,7 +100,8 @@ impl Default for HttpOptions {
         HttpOptions {
             retries: 3,
             backoff: Duration::from_millis(200),
-            timeout: Duration::from_secs(60),
+            timeout: Duration::from_secs(30),
+            min_speed: 16 << 10,
             max_parallel: 8,
             first_request: crate::cache::DEFAULT_BLOCK_SIZE,
             allow_full_download: false,
@@ -282,7 +295,8 @@ impl HttpStorage {
         let config = ureq::Agent::config_builder()
             .http_status_as_error(false)
             .max_redirects(0)
-            .timeout_global(Some(options.timeout))
+            .timeout_connect(Some(options.timeout))
+            .timeout_recv_response(Some(options.timeout))
             .build();
         let mut storage = HttpStorage {
             agent: ureq::Agent::new_with_config(config),
@@ -374,6 +388,18 @@ impl HttpStorage {
         if let Some((a, b)) = range {
             req = req.header("Range", format!("bytes={a}-{b}"));
         }
+        // The body's budget scales with what it may carry: the range, or
+        // a whole file the server may send instead.
+        let mut body = range.map_or(0, |(a, b)| b.saturating_sub(a).saturating_add(1));
+        if self.options.allow_full_download {
+            body = body.max(self.options.max_full_download);
+        }
+        let secs = body as f64 / self.options.min_speed.max(1) as f64;
+        let body_timeout = self
+            .options
+            .timeout
+            .saturating_add(Duration::try_from_secs_f64(secs).unwrap_or(Duration::MAX));
+        let mut req = req.config().timeout_recv_body(Some(body_timeout)).build();
         match &self.validator {
             Validator::ETag(e) => req = req.header("If-Match", e),
             Validator::LastModified(t) => req = req.header("If-Unmodified-Since", t),

crates/clawhdf5-remote/tests/common/server.rs

@@ -55,6 +55,11 @@ pub struct Shared {
     /// When non-zero, answer every request for a served file with this
     /// status (and an empty body).
     pub force_status: AtomicU32,
+    /// When non-zero, send bodies at about this many bytes per second.
+    pub throttle_bps: AtomicU64,
+    /// When non-zero, stop this many milliseconds after the headers and
+    /// half the body (a stalled connection).
+    pub stall_ms: AtomicU64,
     /// Answer ranges with a `Content-Range` one byte off.
     pub wrong_range: AtomicBool,
     /// Path → `Location`: answered `302 Found` (counted as a request).
@@ -384,13 +389,32 @@ fn serve(conn: TcpStream, s: &Shared) -> std::io::Result<()> {
         } else {
             body
         };
-        let mut response = head.into_bytes();
-        response.extend_from_slice(body);
         // Counted before the client can have the bytes, so a test that
         // resets the counters after a read never sees them arrive late.
         s.bytes.fetch_add(body.len() as u64, Ordering::SeqCst);
-        out.write_all(&response)?;
-        out.flush()?;
+        let bps = s.throttle_bps.load(Ordering::SeqCst);
+        let stall = s.stall_ms.load(Ordering::SeqCst);
+        if bps > 0 || stall > 0 {
+            out.write_all(head.as_bytes())?;
+            let (first, rest) = body.split_at(if stall > 0 { body.len() / 2 } else { 0 });
+            out.write_all(first)?;
+            out.flush()?;
+            if stall > 0 {
+                std::thread::sleep(Duration::from_millis(stall));
+            }
+            for piece in rest.chunks(4096) {
+                out.write_all(piece)?;
+                out.flush()?;
+                if bps > 0 {
+                    std::thread::sleep(Duration::from_micros(4096 * 1_000_000 / bps));
+                }
+            }
+        } else {
+            let mut response = head.into_bytes();
+            response.extend_from_slice(body);
+            out.write_all(&response)?;
+            out.flush()?;
+        }
         if truncate || close {
             let _ = out.shutdown(std::net::Shutdown::Both);
             return Ok(());

crates/clawhdf5-remote/tests/http.rs

@@ -800,3 +800,37 @@ fn a_200_covering_the_requested_range_is_the_whole_file() {
     let e = HttpStorage::open(&server.url("/t.h5"), opts).unwrap_err();
     assert!(matches!(e, RemoteError::RangeNotSupported(_)), "{e}");
 }
+
+/// A slow link is not cut off: the body's time budget grows with its size
+/// (`timeout` + size at `min_speed`), so a 256 KiB block at 256 KiB/s
+/// (1 s) reads with a 300 ms `timeout`. A stalled body still fails, soon.
+#[test]
+fn slow_links_read_and_stalled_ones_fail() {
+    let bytes = multi_block_file();
+    let server = Server::start(vec![("/m.h5".into(), bytes.clone())]);
+    let url = server.url("/m.h5");
+    let mut opts = quick();
+    opts.cache.block_size = 256 << 10;
+    opts.cache.max_request = 256 << 10;
+    opts.http.timeout = Duration::from_millis(300);
+    server
+        .shared
+        .throttle_bps
+        .store(256 << 10, Ordering::SeqCst);
+    let f = open_url_with(&url, &opts).unwrap();
+    assert_eq!(f.root().groups().unwrap(), ["grp"]);
+    assert_eq!(
+        f.dataset("grp/small").unwrap().read_f64().unwrap(),
+        [1.0, 2.0, 3.0]
+    );
+    server.shared.throttle_bps.store(0, Ordering::SeqCst);
+
+    // Stalled mid-body for 20 s: a timeout well before that.
+    server.shared.stall_ms.store(20_000, Ordering::SeqCst);
+    opts.http.retries = 0;
+    opts.http.min_speed = 64 << 20;
+    let t = std::time::Instant::now();
+    let e = open_url_with(&url, &opts).unwrap_err();
+    assert!(matches!(e, Error::Remote(RemoteError::Transport(_))), "{e}");
+    assert!(t.elapsed() < Duration::from_secs(5), "{:?}", t.elapsed());
+}