diff --git a/crates/clawhdf5-remote/src/http.rs b/crates/clawhdf5-remote/src/http.rs index 67b5612..221ca6c 100644 --- a/crates/clawhdf5-remote/src/http.rs +++ b/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), diff --git a/crates/clawhdf5-remote/tests/common/server.rs b/crates/clawhdf5-remote/tests/common/server.rs index 1708084..ec7c8ae 100644 --- a/crates/clawhdf5-remote/tests/common/server.rs +++ b/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(()); diff --git a/crates/clawhdf5-remote/tests/http.rs b/crates/clawhdf5-remote/tests/http.rs index 679f977..2bf6da1 100644 --- a/crates/clawhdf5-remote/tests/http.rs +++ b/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()); +}