gwseq_io/source/

url.rs

//! HTTP/HTTPS byte-range reads.
//!
//! `ureq` with `rustls`: pure Rust, no system TLS, nothing for the build to
//! find or configure on the installing machine.
//!
//! # A server that ignores Range
//!
//! The single most important thing here, and the reason this is not just "GET
//! with a header". A server with no range support answers a ranged request
//! `200` with the **whole file**, and every block would then silently be the
//! file's first `block_size` bytes. That is not a corner case: Python's own
//! `http.server` does exactly this. So the response is checked — a `206`, or a
//! `200` whose body genuinely is the range asked for — and anything else is
//! refused by name. Because `open()` reads four magic bytes, the refusal lands
//! there rather than at the first `read_values`.
//!
//! # Retries
//!
//! `ureq` gives none of this; every rule below is this module's own.
//!
//! - Retry only transient failures: timeouts, refused or reset connections,
//!   truncated bodies, unresolved hosts, and HTTP 408 / 425 / 429 / 5xx. A bad
//!   URL, a 404 or a rejected certificate fails on the first attempt.
//! - Backoff doubles per attempt (`delay * 2^n`), capped at `max_delay`, and is
//!   jittered over `[d/2, d]` so parallel readers do not resynchronise on one
//!   server.
//! - A `Retry-After` header carrying a seconds count overrides the computed
//!   delay, still capped.
//! - A retry re-fetches its whole range rather than resuming.

use std::ops::Range;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::OnceLock;
use std::time::Duration;

use bytes::Bytes;

use crate::error::{Error, Result};
use crate::source::ByteSource;

#[derive(Debug, Clone)]
pub struct RetryPolicy {
    /// Retries *after* the first attempt, so a request is issued at most
    /// `max_retries + 1` times. 0 disables retrying.
    pub max_retries: u32,
    pub delay: Duration,
    pub max_delay: Duration,
    pub connect_timeout: Duration,
    /// Backstop against a transfer that trickles. The right rule is a
    /// *throughput* bound — abort after 30 s below 1 KB/s — but `ureq` has no
    /// such setting; a whole-transfer deadline is the closest thing, and it is
    /// deliberately generous so a large but healthy read is never cut off for
    /// its size.
    pub timeout: Duration,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_retries: 3,
            delay: Duration::from_millis(500),
            max_delay: Duration::from_secs(30),
            connect_timeout: Duration::from_secs(10),
            timeout: Duration::from_secs(600),
        }
    }
}

#[derive(Debug)]
pub struct UrlSource {
    url: String,
    agent: ureq::Agent,
    policy: RetryPolicy,
    /// From a `HEAD`. Cached: it costs a round trip, and these files do not
    /// change under a reader.
    len: OnceLock<u64>,
    closed: AtomicBool,
}

/// What one attempt came back with.
#[derive(Debug)]
struct Response {
    status: u16,
    body: Vec<u8>,
    content_length: Option<u64>,
    /// The raw `Content-Range`, for the size of a resource whose server will
    /// not answer `HEAD`. See [`UrlSource::len_from_range`].
    content_range: Option<String>,
    retry_after: Option<Duration>,
}

impl UrlSource {
    pub fn open(url: &str) -> Result<Self> {
        Self::open_with(url, RetryPolicy::default())
    }

    pub fn open_with(url: &str, policy: RetryPolicy) -> Result<Self> {
        let agent: ureq::Agent = ureq::Agent::config_builder()
            .timeout_connect(Some(policy.connect_timeout))
            .timeout_global(Some(policy.timeout))
            // Statuses are inspected here rather than raised by the client: the
            // retry rules need the code, and `Retry-After` only arrives with the
            // response that carries it.
            .http_status_as_error(false)
            .user_agent(concat!("gwseq-io/", env!("CARGO_PKG_VERSION")))
            .build()
            .into();
        Ok(Self {
            url: url.to_string(),
            agent,
            policy,
            len: OnceLock::new(),
            closed: AtomicBool::new(false),
        })
    }

    fn check_open(&self) -> Result<()> {
        if self.closed.load(Ordering::Relaxed) {
            return Err(Error::Closed {
                path: self.url.clone(),
            });
        }
        Ok(())
    }

    fn http_error(&self, status: Option<u16>, message: impl Into<String>) -> Error {
        Error::Http {
            url: self.url.clone(),
            status,
            message: message.into(),
        }
    }

    /// One attempt, with an optional `Range` header. Transport failures come
    /// back as `Err`; an HTTP status of any value comes back as `Ok`.
    ///
    /// `cap` bounds the body. Nothing here reads a body of unknown size: a
    /// range asks for a known number of bytes and `http_get_text` names its
    /// own limit, so a server — hostile, or merely broken — cannot make this
    /// grow until the process dies. Three responses have their body skipped
    /// entirely rather than capped: a `HEAD`, which has none; an error status,
    /// whose body nothing reads; and a ranged request the server answered with
    /// something other than the range, which `check_range_response` is about
    /// to refuse. That last one is the point — the module's headline case, a
    /// server that ignores `Range` and sends the whole resource, used to cost
    /// a transfer of the file *per block* before the refusal.
    fn attempt(
        &self,
        range: Option<Range<u64>>,
        head: bool,
        cap: u64,
    ) -> std::result::Result<Response, ureq::Error> {
        let mut request = if head {
            self.agent.head(&self.url)
        } else {
            self.agent.get(&self.url)
        };
        if let Some(range) = &range {
            request = request.header("Range", format!("bytes={}-{}", range.start, range.end - 1));
        }
        let mut response = request.call()?;

        let status = response.status().as_u16();
        let header = |name: &str| {
            response
                .headers()
                .get(name)
                .and_then(|v| v.to_str().ok())
                .map(str::to_string)
        };
        let content_length = header("content-length").and_then(|v| v.parse().ok());
        let content_range = header("content-range");
        // Only the seconds form; the HTTP-date form is legal and rare, and
        // guessing at a clock skew is worse than falling back to the backoff.
        let retry_after = header("retry-after")
            .and_then(|v| v.trim().parse::<u64>().ok())
            .map(Duration::from_secs);

        let unusable_range = range
            .as_ref()
            .is_some_and(|r| !range_response_is_usable(status, content_length, r));
        let body = if head || status >= 400 || unusable_range {
            Vec::new()
        } else {
            // `Read::take`, not ureq's own `limit`: that one errors past the
            // cap, and this module deliberately tolerates a server answering
            // with more than was asked for — the extra is simply not part of
            // the range. Nothing drains what is left, so the transfer stops
            // where the read does.
            use std::io::Read as _;
            let mut buf = Vec::new();
            response
                .body_mut()
                .with_config()
                .reader()
                .take(cap)
                .read_to_end(&mut buf)?;
            buf
        };
        Ok(Response {
            status,
            body,
            content_length,
            content_range,
            retry_after,
        })
    }

    /// The retry loop around [`Self::attempt`].
    fn perform(
        &self,
        what: &str,
        range: Option<Range<u64>>,
        head: bool,
        cap: u64,
    ) -> Result<Response> {
        self.check_open()?;
        let mut attempt = 0u32;
        loop {
            let outcome = self.attempt(range.clone(), head, cap);
            let (transient, retry_after, failure) = match &outcome {
                Ok(response) if response.status < 400 => {
                    return Ok(outcome.expect("checked Ok above"))
                }
                Ok(response) => (
                    is_retryable_status(response.status),
                    response.retry_after,
                    self.http_error(
                        Some(response.status),
                        format!("{what} failed with HTTP {}", response.status),
                    ),
                ),
                Err(error) => (
                    is_retryable_transport(error),
                    None,
                    self.http_error(None, format!("{what} failed: {error}")),
                ),
            };

            if attempt >= self.policy.max_retries || !transient {
                return Err(failure);
            }
            std::thread::sleep(self.backoff(attempt, retry_after));
            attempt += 1;
        }
    }

    /// How long to wait before the retry following attempt `n` (0-based).
    ///
    /// A `Retry-After` from the server wins; otherwise the delay doubles per
    /// attempt. Both are capped, and the exponential form is jittered over
    /// `[d/2, d]` so parallel readers hitting one server do not resynchronise.
    fn backoff(&self, attempt: u32, retry_after: Option<Duration>) -> Duration {
        if let Some(after) = retry_after {
            return after.min(self.policy.max_delay);
        }
        let delay = self
            .policy
            .delay
            .saturating_mul(1u32 << attempt.min(16))
            .min(self.policy.max_delay);
        let half = delay.as_nanos() as u64 / 2;
        Duration::from_nanos(half + jitter(half + 1))
    }

    /// The resource's size from a one-byte ranged GET, for a server that will
    /// not answer `HEAD`.
    ///
    /// `Content-Range: bytes 0-0/12345` carries the total after the slash. A
    /// `*` there means the server does not know it, which is no more use than
    /// the missing `Content-Length` that led here.
    fn len_from_range(&self) -> Result<u64> {
        let response = self.perform("determining the size of", Some(0..1), false, 1)?;
        let total = response
            .content_range
            .as_deref()
            .and_then(parse_content_range_total)
            // A server with no ranges at all answers this `200` with the whole
            // resource, and its `Content-Length` is then the total — the one
            // thing worth taking from a response the reads themselves will
            // refuse. On a `206` the same header is the length of the range,
            // which is 1 and says nothing, so this arm is only for the `200`.
            .or(if response.status == 200 {
                response.content_length
            } else {
                None
            })
            .ok_or_else(|| {
                self.http_error(
                    Some(response.status),
                    "no content length, and no total in the content range either",
                )
            })?;
        Ok(total)
    }

    /// Refuse a server that answered a ranged request with something other than
    /// the range asked for.
    fn check_range_response(&self, response: &Response, range: &Range<u64>) -> Result<()> {
        if range_response_is_usable(response.status, response.content_length, range) {
            return Ok(());
        }
        Err(self.http_error(
            Some(response.status),
            format!(
                "the server ignored the Range header and answered bytes {}-{} with the whole \
                 resource, so the bytes it sent are not the ones asked for",
                range.start,
                range.end - 1
            ),
        ))
    }
}

/// The total size out of a `Content-Range: bytes 0-0/12345`.
///
/// `None` for the `*` form, which is the server saying it does not know, and
/// for anything that does not parse — a header this reader cannot understand is
/// not a number worth guessing at.
fn parse_content_range_total(header: &str) -> Option<u64> {
    header.rsplit_once('/')?.1.trim().parse().ok()
}

/// Whether a response to a ranged request carries the bytes that were asked for.
///
/// A `206` does by definition. A `200` only when the whole resource *is* the
/// range asked for — an offset of 0, and a request reaching the end of the
/// file; a `200` with no content length says nothing that can be trusted.
///
/// One rule, read twice: [`UrlSource::attempt`] uses it to decide whether the
/// body is worth transferring at all, and [`UrlSource::check_range_response`]
/// to turn the same answer into the error the caller sees.
fn range_response_is_usable(status: u16, content_length: Option<u64>, range: &Range<u64>) -> bool {
    if status == 206 {
        return true;
    }
    status == 200 && range.start == 0 && content_length.is_some_and(|length| range.end >= length)
}

/// Retryable statuses: 408, 425, 429 and anything 5xx.
fn is_retryable_status(status: u16) -> bool {
    matches!(status, 408 | 425 | 429) || status >= 500
}

/// The transport failures worth retrying.
///
/// Refused and reset connections, unresolved hosts, timeouts, truncated bodies
/// — everything that says "try again", and nothing that says "this will not
/// work": a bad URI, a rejected certificate, a redirect loop.
fn is_retryable_transport(error: &ureq::Error) -> bool {
    use std::io::ErrorKind::*;
    match error {
        ureq::Error::Timeout(_) | ureq::Error::ConnectionFailed | ureq::Error::HostNotFound => true,
        ureq::Error::Io(io) => matches!(
            io.kind(),
            TimedOut
                | ConnectionRefused
                | ConnectionReset
                | ConnectionAborted
                | NotConnected
                | BrokenPipe
                | UnexpectedEof
                | Interrupted
                | WouldBlock
        ),
        _ => false,
    }
}

/// A jitter in `[0, bound)`, from the clock. Not cryptographic and not meant to
/// be: the point is only that two readers do not wake together.
fn jitter(bound: u64) -> u64 {
    if bound == 0 {
        return 0;
    }
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.subsec_nanos() as u64)
        .unwrap_or(0);
    // Mixed with the thread id, so two threads sleeping in the same nanosecond
    // still separate.
    let id = {
        use std::hash::{Hash, Hasher};
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        std::thread::current().id().hash(&mut hasher);
        hasher.finish()
    };
    (nanos ^ id) % bound
}

impl ByteSource for UrlSource {
    fn path(&self) -> &str {
        &self.url
    }

    fn len(&self) -> Result<u64> {
        if let Some(length) = self.len.get() {
            return Ok(*length);
        }
        let length = match self.perform("determining the size of", None, true, 0) {
            Ok(response) => response.content_length,
            // A server that refuses HEAD outright — 405, or 501 from one that
            // implements only GET — is common enough that failing here would
            // make the whole file unreadable over a plain object store. The
            // fallback is the usual one: ask for a single byte and read the
            // size out of the `Content-Range` the 206 carries.
            Err(error) => {
                let refused_head = matches!(
                    &error,
                    Error::Http {
                        status: Some(405 | 501),
                        ..
                    }
                );
                if !refused_head {
                    return Err(error);
                }
                None
            }
        };
        let length = match length {
            Some(length) => length,
            None => self.len_from_range()?,
        };
        Ok(*self.len.get_or_init(|| length))
    }

    fn read_at(&self, offset: u64, len: usize) -> Result<Bytes> {
        self.check_open()?;
        if len == 0 {
            return Ok(Bytes::new());
        }
        // HTTP has no empty range — a server answers one with 416, a hard error
        // not worth retrying — so a read starting at or past the end is
        // resolved here rather than sent, as a local file resolves it.
        let file_size = self.len()?;
        if offset >= file_size {
            return Ok(Bytes::new());
        }

        let range = offset..offset + len as u64;
        let response = self.perform("reading", Some(range.clone()), false, len as u64)?;
        self.check_range_response(&response, &range)?;

        let mut body = response.body;
        // A server may answer with more than was asked for; the extra is not
        // part of the range and is dropped.
        body.truncate(len);
        Ok(Bytes::from(body))
    }

    /// Left as a no-op. Coalescing the R-tree's leaves into multi-range
    /// requests is worth measuring before it is worth writing.
    fn prefetch(&self, _ranges: &[Range<u64>]) {}

    fn close(&self) {
        self.closed.store(true, Ordering::Relaxed);
    }
}

/// The most a [`http_get_text`] body may be, in bytes.
///
/// Its one caller is the UCSC chromosome-sizes API, whose largest answers are
/// an assembly of a few hundred thousand scaffolds — a few megabytes of text.
/// The cap is here because that call has no range to bound it, so without one
/// the size of the response is whatever the far end decides to send.
const MAX_TEXT_BODY: u64 = 64 << 20;

/// A plain GET returning the body as text.
///
/// Not part of [`ByteSource`] — nothing about reading a bigWig needs it. It is
/// here because this is the module that owns the HTTP client, and the UCSC
/// genome API is the one caller.
pub fn http_get_text(url: &str) -> Result<String> {
    let source = UrlSource::open(url)?;
    let response = source.perform("fetching", None, false, MAX_TEXT_BODY)?;
    String::from_utf8(response.body).map_err(|_| Error::Http {
        url: url.to_string(),
        status: Some(response.status),
        message: "response was not valid UTF-8".to_string(),
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn the_statuses_worth_retrying_are_the_documented_ones() {
        for status in [408, 425, 429, 500, 502, 503, 504, 599] {
            assert!(is_retryable_status(status), "{status}");
        }
        for status in [400, 401, 403, 404, 410, 416, 451] {
            assert!(!is_retryable_status(status), "{status}");
        }
    }

    #[test]
    fn backoff_doubles_and_stays_inside_its_jitter_window() {
        let source = UrlSource::open("http://127.0.0.1:1/x").unwrap();
        for attempt in 0..4u32 {
            let nominal = (source.policy.delay.as_secs_f64() * 2f64.powi(attempt as i32))
                .min(source.policy.max_delay.as_secs_f64());
            let got = source.backoff(attempt, None).as_secs_f64();
            assert!(
                got >= nominal / 2.0 - 1e-6 && got <= nominal + 1e-6,
                "attempt {attempt}: {got} outside [{}, {nominal}]",
                nominal / 2.0
            );
        }
    }

    #[test]
    fn backoff_is_capped_however_many_attempts() {
        let source = UrlSource::open("http://127.0.0.1:1/x").unwrap();
        for attempt in [10u32, 20, 40] {
            assert!(source.backoff(attempt, None) <= source.policy.max_delay);
        }
    }

    #[test]
    fn retry_after_wins_over_the_backoff_but_is_still_capped() {
        let source = UrlSource::open("http://127.0.0.1:1/x").unwrap();
        assert_eq!(
            source.backoff(0, Some(Duration::from_secs(7))),
            Duration::from_secs(7)
        );
        assert_eq!(
            source.backoff(0, Some(Duration::from_secs(9999))),
            source.policy.max_delay
        );
    }

    fn response(status: u16, content_length: Option<u64>) -> Response {
        Response {
            status,
            body: Vec::new(),
            content_length,
            content_range: None,
            retry_after: None,
        }
    }

    #[test]
    fn a_range_answered_with_the_whole_file_is_refused() {
        let source = UrlSource::open("http://example.invalid/x").unwrap();
        // The block a reader actually asks for: not from the start, and not
        // the whole file.
        let err = source
            .check_range_response(&response(200, Some(1000)), &(4096..8192))
            .unwrap_err()
            .to_string();
        assert!(err.contains("ignored the Range header"), "{err}");
        assert!(err.contains("bytes 4096-8191"), "{err}");
    }

    #[test]
    fn a_206_is_always_fine_and_a_200_only_when_it_is_the_whole_file() {
        let source = UrlSource::open("http://example.invalid/x").unwrap();
        assert!(source
            .check_range_response(&response(206, Some(100)), &(4096..8192))
            .is_ok());
        // A file of 1000 bytes asked for from 0 with a request reaching past
        // the end: the whole resource *is* the range.
        assert!(source
            .check_range_response(&response(200, Some(1000)), &(0..4096))
            .is_ok());
        // The same file asked for from 0 but only in part is not.
        assert!(source
            .check_range_response(&response(200, Some(1000)), &(0..500))
            .is_err());
        // And a 200 with no content length says nothing that can be trusted.
        assert!(source
            .check_range_response(&response(200, None), &(0..4096))
            .is_err());
    }

    #[test]
    fn a_closed_source_refuses_and_names_the_url() {
        let source = UrlSource::open("http://127.0.0.1:1/x.bigwig").unwrap();
        source.close();
        source.close(); // idempotent
        let err = source.read_at(0, 4).unwrap_err().to_string();
        assert!(err.contains("is closed"), "{err}");
        assert!(err.contains("x.bigwig"), "{err}");
    }

    #[test]
    fn a_zero_length_read_never_leaves_the_process() {
        // Nothing is listening on port 1; a request here would fail.
        let source = UrlSource::open("http://127.0.0.1:1/x.bigwig").unwrap();
        assert!(source.read_at(0, 0).unwrap().is_empty());
    }

    // ---------------------------------------------------------------------------
    // A server to read from
    // ---------------------------------------------------------------------------
    //
    // Everything above this point was reviewed by reading, because there was no
    // way to run it: the suite has no fixture and no network, and the one place a
    // wrong answer here is expensive — a server that ignores `Range` — cannot be
    // reached without a server that ignores `Range`.
    //
    // So here is one. A `TcpListener` on a loopback port the OS picks, a thread per
    // connection, and a handful of canned behaviours. It needs no fixture, no
    // network beyond loopback, and no port anyone else could be holding.

    use std::io::{BufRead, BufReader, Write};
    use std::net::{TcpListener, TcpStream};
    use std::sync::atomic::AtomicU64;
    use std::sync::Arc;

    /// What the test server does with a request.
    #[derive(Clone, Copy, Debug)]
    enum Behaviour {
        /// A correct server: `206` with the bytes asked for, `HEAD` answered.
        Ranged,
        /// Ignores `Range` and answers `200` with the whole resource. Python's own
        /// `http.server` does exactly this, which is why the module has a check for
        /// it at all.
        IgnoresRange,
        /// Refuses `HEAD` with 405 and is otherwise correct.
        NoHead,
        /// Answers every GET `200` with a body far larger than it declares, and
        /// keeps writing. What a cap is for.
        Endless,
        /// The worst of both: no `HEAD`, and no attention paid to `Range`. Its
        /// length is still knowable, and its reads still have to be refused.
        NoHeadNoRange,
    }

    struct TestServer {
        port: u16,
        /// Body bytes the server managed to write before the client stopped
        /// reading. What says whether a transfer really was avoided.
        written: Arc<AtomicU64>,
        stop: Arc<AtomicBool>,
    }

    impl TestServer {
        fn start(behaviour: Behaviour, body: Vec<u8>) -> Self {
            let listener = TcpListener::bind("127.0.0.1:0").expect("a loopback port");
            let port = listener.local_addr().expect("an address").port();
            let written = Arc::new(AtomicU64::new(0));
            let stop = Arc::new(AtomicBool::new(false));
            let (w, s) = (written.clone(), stop.clone());
            std::thread::spawn(move || {
                for stream in listener.incoming() {
                    if s.load(Ordering::Relaxed) {
                        break;
                    }
                    let Ok(stream) = stream else { break };
                    let (w, body) = (w.clone(), body.clone());
                    std::thread::spawn(move || {
                        let _ = serve(stream, behaviour, &body, &w);
                    });
                }
            });
            Self {
                port,
                written,
                stop,
            }
        }

        fn url(&self) -> String {
            format!("http://127.0.0.1:{}/x.bigwig", self.port)
        }

        fn body_bytes_written(&self) -> u64 {
            self.written.load(Ordering::Relaxed)
        }

        /// A source with no retries and short timeouts: a test that is going to
        /// fail should fail now, not in half a minute.
        fn source(&self) -> UrlSource {
            UrlSource::open_with(
                &self.url(),
                RetryPolicy {
                    max_retries: 0,
                    delay: Duration::from_millis(1),
                    max_delay: Duration::from_millis(1),
                    connect_timeout: Duration::from_secs(2),
                    timeout: Duration::from_secs(10),
                },
            )
            .expect("a source")
        }
    }

    impl Drop for TestServer {
        fn drop(&mut self) {
            self.stop.store(true, Ordering::Relaxed);
            // Wake the accept loop so the thread notices and leaves.
            let _ = TcpStream::connect(("127.0.0.1", self.port));
        }
    }

    fn serve(
        mut stream: TcpStream,
        behaviour: Behaviour,
        body: &[u8],
        written: &AtomicU64,
    ) -> std::io::Result<()> {
        let mut reader = BufReader::new(stream.try_clone()?);
        let mut request = String::new();
        reader.read_line(&mut request)?;
        let mut range: Option<(u64, u64)> = None;
        loop {
            let mut line = String::new();
            if reader.read_line(&mut line)? == 0 || line.trim().is_empty() {
                break;
            }
            if let Some(spec) = line.to_ascii_lowercase().strip_prefix("range: bytes=") {
                if let Some((a, b)) = spec.trim().split_once('-') {
                    if let (Ok(a), Ok(b)) = (a.parse::<u64>(), b.parse::<u64>()) {
                        range = Some((a, b));
                    }
                }
            }
        }
        let head = request.starts_with("HEAD ");
        let total = body.len() as u64;

        if head && matches!(behaviour, Behaviour::NoHead | Behaviour::NoHeadNoRange) {
            return stream
                .write_all(b"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n");
        }
        if head {
            return stream.write_all(
                format!(
                    "HTTP/1.1 200 OK\r\nContent-Length: {total}\r\nAccept-Ranges: bytes\r\n\r\n"
                )
                .as_bytes(),
            );
        }

        if matches!(behaviour, Behaviour::Endless) {
            // The size it declares, and then rather more than that.
            stream.write_all(
                format!("HTTP/1.1 200 OK\r\nContent-Length: {total}\r\n\r\n").as_bytes(),
            )?;
            let chunk = vec![b'z'; 64 << 10];
            loop {
                match stream.write(&chunk) {
                    Ok(0) | Err(_) => return Ok(()),
                    Ok(n) => {
                        written.fetch_add(n as u64, Ordering::Relaxed);
                        // Bounded so a test that reads everything fails by
                        // assertion rather than by running forever.
                        if written.load(Ordering::Relaxed) > 256 << 20 {
                            return Ok(());
                        }
                    }
                }
            }
        }

        let ignores = matches!(
            behaviour,
            Behaviour::IgnoresRange | Behaviour::NoHeadNoRange
        );
        match range {
            Some((start, end)) if !ignores => {
                let end = end.min(total.saturating_sub(1));
                let slice = &body[start as usize..=(end as usize)];
                stream.write_all(
                    format!(
                        "HTTP/1.1 206 Partial Content\r\nContent-Length: {}\r\n\
                     Content-Range: bytes {start}-{end}/{total}\r\n\r\n",
                        slice.len()
                    )
                    .as_bytes(),
                )?;
                stream.write_all(slice)?;
                written.fetch_add(slice.len() as u64, Ordering::Relaxed);
                Ok(())
            }
            // No range asked for, or a server that pays no attention to one.
            _ => {
                stream.write_all(
                    format!("HTTP/1.1 200 OK\r\nContent-Length: {total}\r\n\r\n").as_bytes(),
                )?;
                // Written in pieces so a client that stops reading stops the
                // transfer, which is the whole thing under test.
                for piece in body.chunks(64 << 10) {
                    match stream.write(piece) {
                        Ok(0) | Err(_) => return Ok(()),
                        Ok(n) => {
                            written.fetch_add(n as u64, Ordering::Relaxed);
                        }
                    }
                }
                Ok(())
            }
        }
    }

    fn payload(n: usize) -> Vec<u8> {
        (0..n).map(|i| (i % 251) as u8).collect()
    }

    #[test]
    fn a_correct_server_serves_the_bytes_that_were_asked_for() {
        let body = payload(100_000);
        let server = TestServer::start(Behaviour::Ranged, body.clone());
        let source = server.source();
        assert_eq!(source.len().unwrap(), 100_000);
        for (offset, len) in [(0u64, 16usize), (4096, 4096), (99_990, 10)] {
            let got = source.read_at(offset, len).unwrap();
            assert_eq!(&got[..], &body[offset as usize..offset as usize + len]);
        }
        // The length is cached, so a second read costs no second HEAD.
        assert_eq!(source.len().unwrap(), 100_000);
    }

    #[test]
    fn a_server_that_ignores_range_is_refused_without_transferring_the_file() {
        // The module's headline case. The refusal was always there; what was not is
        // that it cost a download of the whole resource *per block* to reach it.
        let body = payload(4 << 20);
        let server = TestServer::start(Behaviour::IgnoresRange, body);
        let source = server.source();

        let err = source.read_at(1 << 20, 4096).unwrap_err();
        let text = err.to_string();
        assert!(text.contains("ignored the Range header"), "{text}");

        // The HEAD that resolved the length sends no body, and the GET was refused
        // on its status line. A few tens of kilobytes may already be in flight when
        // the socket closes; four megabytes means the file was downloaded.
        let written = server.body_bytes_written();
        assert!(
            written < 1 << 20,
            "{written} body bytes crossed the socket for a request that was refused"
        );
    }

    #[test]
    fn a_body_larger_than_the_range_asked_for_is_cut_off() {
        // A server whose body outruns what it declared, which is the shape of an
        // unbounded response: the read stops at the cap rather than growing to
        // whatever the far end feels like sending.
        let server = TestServer::start(Behaviour::Endless, payload(64));
        let source = server.source();
        // 64 bytes declared, so a 64-byte read from 0 is the whole resource and the
        // 200 is accepted — and then the body keeps coming.
        let got = source.read_at(0, 64).unwrap();
        assert_eq!(got.len(), 64, "the read grew past the range it asked for");
        let written = server.body_bytes_written();
        assert!(
            written < 32 << 20,
            "{written} bytes were taken for a 64-byte read"
        );
    }

    #[test]
    fn a_server_that_refuses_head_still_gives_up_its_length() {
        let body = payload(54_321);
        let server = TestServer::start(Behaviour::NoHead, body.clone());
        let source = server.source();
        // 405 on HEAD, so the size comes out of a one-byte range's Content-Range.
        assert_eq!(source.len().unwrap(), 54_321);
        let got = source.read_at(50_000, 321).unwrap();
        assert_eq!(&got[..], &body[50_000..50_321]);
    }

    #[test]
    fn a_server_with_neither_head_nor_ranges_still_gives_up_its_length() {
        // The length comes out of the `Content-Length` of the `200` the
        // one-byte range provoked, since there is no `Content-Range` to read it
        // from. The reads themselves are still refused — knowing how long a
        // file is does not make its bytes arrive in the right order.
        let server = TestServer::start(Behaviour::NoHeadNoRange, payload(4321));
        let source = server.source();
        assert_eq!(source.len().unwrap(), 4321);
        let err = source.read_at(100, 16).unwrap_err().to_string();
        assert!(err.contains("ignored the Range header"), "{err}");
    }

    #[test]
    fn a_read_past_the_end_is_empty_and_a_read_over_it_is_short() {
        let body = payload(1000);
        let server = TestServer::start(Behaviour::Ranged, body.clone());
        let source = server.source();
        assert!(source.read_at(1000, 16).unwrap().is_empty());
        assert!(source.read_at(5000, 16).unwrap().is_empty());
        let got = source.read_at(990, 100).unwrap();
        assert_eq!(&got[..], &body[990..]);
    }

    #[test]
    fn the_content_range_total_is_read_only_when_it_is_a_number() {
        assert_eq!(parse_content_range_total("bytes 0-0/12345"), Some(12345));
        assert_eq!(parse_content_range_total("bytes 0-0/*"), None);
        assert_eq!(parse_content_range_total("nonsense"), None);
        assert_eq!(parse_content_range_total(""), None);
    }
}