//! One HTTP client constructor, so no request waits forever.
//!
//! Every request atgc makes goes to a host named by somebody else — a handle
//! that is a domain name, a PDS from a DID document, a knot from a repo
//! record. A host that answers with an error is easy; a host that answers
//! with nothing is the problem, and `reqwest`'s defaults have no deadline
//! for it at any layer.
//!
//! The failure that produced this module: an account handle whose A record
//! pointed at a parked IP that drops SYNs rather than refusing them. The
//! well-known lookup for it inherited the kernel's SYN retry schedule, so
//! `atgc auth login <handle>` sat silent for 30 seconds before continuing —
//! and the length of that wait was a property of the local network stack, not
//! of anything atgc chose. On a host configured for more retries it is
//! minutes.
//!
//! The three bounds here are deliberately different in kind:
//!
//! - [`CONNECT_TIMEOUT`] caps DNS, TCP and TLS. Nothing legitimate spends
//!   long there, so it can be short.
//! - [`READ_TIMEOUT`] caps the gap *between* bytes of a response, not the
//!   whole exchange. A total `timeout` would be simpler and wrong: this
//!   client also carries `pr create`'s patch blob up to the PDS, and capping
//!   a transfer by its total duration puts a ceiling on payload size that
//!   nobody would think to look for. Time without progress is the thing worth
//!   refusing.
//! - [`MAX_BODY`] caps how much of a response atgc will hold, whether it is
//!   read as it arrives ([`bytes_bounded`]) or checked after a transport
//!   atgc does not own has already buffered it ([`refuse_if_oversize`]). The
//!   bullet
//!   above is why that is a third bound and not a consequence of the second:
//!   a host answering at a steady trickle never leaves a gap between bytes,
//!   so it sits inside both deadlines for as long as it likes, and every
//!   JSON read in the tree was a bare `resp.text()` with nothing else in its
//!   way.
//!
//! # One constructor is also one instance
//!
//! The bounds above are the reason there is a single place to build a client.
//! What that place handed back, until it did not, was a *new* client every
//! time — twenty call sites, one per request, so nothing was ever reused.
//! `atgc pr list --limit 5` measured 99 requests in 18.5 s against this
//! repository, 45 of them to the same appview host, and each of those 45 paid
//! its own DNS lookup, TCP connect, TLS handshake and root-store parse before
//! it could ask for anything. `reqwest::Client` is an `Arc` around a
//! connection pool: sharing one is what makes the second request to a host
//! cheap, and building a second client is what throws the pool away.
//!
//! So [`client`] hands out clones of one instance, built on first use. Two
//! consequences worth knowing:
//!
//! - The bounds are read when that first client is built, not at every call.
//!   `main` calls [`init`] before dispatching any command, so the first build
//!   is always after the flag and the environment have been read; the
//!   fallback to the defaults covers a test that reaches a client without
//!   going through `main`. Both of those were already true of [`init`]
//!   itself, which is a `OnceLock` set once per process.
//! - There are two instances, not one, because there are two policies:
//!   [`client`] follows redirects and [`no_redirect`] does not, for the three
//!   callers that read a `Location` header rather than following it.
//!   [`builder`] stays the one constructor carrying the bounds, and both
//!   instances are built from it.
//!
//! Nothing here is per-host. One pooled client serves every host atgc talks
//! to, which is what the test suite's many loopback mock services need and
//! what the `MAX_BODY` refusals above already assume: the bounds are a
//! property of atgc's patience, not of who is being asked.

use anyhow::Context;
use std::sync::OnceLock;
use std::time::Duration;

/// DNS, TCP and TLS, together. A host that has not completed a handshake in
/// five seconds is either unreachable or so slow that saying so beats waiting.
///
/// The default, not the value — see [`connect_timeout`]. Short on purpose:
/// most of what atgc does is driven by a script or an agent that would rather
/// be told quickly, and the one thing a caller can always do with "the host
/// did not answer" is try again.
pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);

/// The longest gap between bytes of a response before it counts as stalled.
/// Generous, because it also covers a PDS thinking about a blob it has just
/// been handed, which is real work rather than a stall.
///
/// The default, not the value — see [`read_timeout`].
pub const READ_TIMEOUT: Duration = Duration::from_secs(30);

/// The most of one response body atgc will read, whoever sent it.
///
/// # The hosts this is aimed at
///
/// A pull request is a record in its author's repository, so its at-uri
/// names their DID; a `did:web` DID resolves to
/// `https://<a hostname the author chose>/.well-known/did.json`, and a
/// `did:plc` document's `serviceEndpoint` is a URL that same author wrote.
/// `atgc pr view at://did:web:evil.example/sh.tangled.repo.pull/abc` reads a
/// document from a machine the pull's author picked and then a record from
/// whatever host that document named — two whole-body reads before anything
/// has been decided about the pull, both of them `resp.text()` until this
/// existed. `pr diff`, `pr list` and `repo clone <owner>/<name>` reach the
/// same way for any repo whose owner or whose pull author is hostile.
///
/// # One number, not one per caller
///
/// The largest legitimate response measured against the real services is
/// 430 KiB — tangled.org's HTML page for this project's own pull #202, which
/// [`crate::clients::tangled::web::pulls`] reads for the single at-uri in
/// it. A full `listRecords` page, 100 of this account's pull records with
/// their bodies, is 287 KiB. A `did:plc` document is 559 bytes. Eight
/// mebibytes clears the largest of those nineteen times over, and it is
/// already what `pr diff` allows a patch blob, so atgc has one answer to
/// "the most of a stranger's response we will hold" rather than two.
///
/// Per-caller caps were the alternative and are the worse trade. Holding a
/// DID document to 64 KiB would be snug and right today; what it buys over
/// this is a few megabytes of transient memory in the hostile case, and what
/// it costs is a number per call site sized to the biggest response anybody
/// had yet seen — which is the shape of cap
/// [`crate::clients::atproto::pds::MAX_PAGES`] records expiring inside a
/// year, on an account that outgrew it eleven days after its first pull. The
/// property worth having is that there is a bound, not that it sits snugly.
///
/// # Not adjustable
///
/// [`init`] lets the two deadlines be raised because they are a judgement
/// about somebody else's network, which atgc cannot know. This is a
/// judgement about how much of this machine's memory a stranger may spend,
/// and the party keenest to raise it is the one it is aimed at. No
/// legitimate response is within two orders of magnitude of it.
pub const MAX_BODY: u64 = 8 * 1024 * 1024;

/// The cap has to clear the largest response anyone has measured off the
/// real services, with room left over. A compile error rather than a test,
/// because it is a claim about two constants and there is nothing to run:
/// an edit that tightens this below what a real appview page costs would
/// break `pr view <number>` on the busiest pulls and nothing else, which is
/// the failure least likely to show up against a small test repo.
const _: () = assert!(
    MAX_BODY > LARGEST_OBSERVED * 8,
    "MAX_BODY must stay comfortably clear of the largest response measured"
);

/// tangled.org's HTML page for this project's pull #202, in bytes, measured
/// with `curl | wc -c` on 2026-08-15. The biggest of everything atgc reads;
/// see [`MAX_BODY`] for the rest of the measurements.
const LARGEST_OBSERVED: u64 = 440_648;

/// Whichever bounds this invocation is using, decided once by [`init`].
static BOUNDS: OnceLock<Bounds> = OnceLock::new();

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct Bounds {
    connect: Duration,
    read: Duration,
}

/// Settle both bounds for this invocation: the flag if it was passed, then
/// the environment, then the default.
///
/// # Why these are adjustable at all
///
/// The five seconds above is a judgement about what a *reachable* host takes,
/// and it is right for the network it was measured on and wrong for others. A
/// laptop tethered to a phone, a PDS behind a slow relay, a CI runner in a
/// different hemisphere from the knot: each of those completes a handshake in
/// a time that is fine and that this refuses. When it does, the failure reads
/// as "could not reach the PDS", which is indistinguishable from the host
/// actually being down and sends the reader looking in the wrong place.
///
/// The default stays short because the common case is a script or an agent
/// that would rather be told quickly. What changes is that somebody who knows
/// their link is slow can say so, instead of the number being ours alone.
///
/// # What is not adjustable
///
/// That there *is* a bound. Both of these may be raised as far as anyone
/// likes, and neither may be turned off: an unbounded connect is the bug this
/// module was written for — a handle whose A record pointed at an IP that
/// dropped SYNs, waited out the kernel's own retry schedule, which is minutes
/// on some hosts and is nobody's choice. A caller who wants to wait ten
/// minutes can ask for six hundred seconds and get exactly that.
pub fn init(connect: Option<u64>, read: Option<u64>) -> anyhow::Result<()> {
    let bounds = Bounds {
        connect: decide(
            "--connect-timeout",
            "ATGC_CONNECT_TIMEOUT",
            connect,
            CONNECT_TIMEOUT,
        )?,
        read: decide("--read-timeout", "ATGC_READ_TIMEOUT", read, READ_TIMEOUT)?,
    };
    if bounds
        != (Bounds {
            connect: CONNECT_TIMEOUT,
            read: READ_TIMEOUT,
        })
    {
        crate::logging::debug::log(format!(
            "http bounds: connect {}s, read {}s",
            bounds.connect.as_secs(),
            bounds.read.as_secs()
        ));
    }
    let _ = BOUNDS.set(bounds);
    Ok(())
}

/// One bound, from the flag, the environment, or the default — in that order.
///
/// Kept pure and taking its inputs as arguments so the precedence and the
/// refusals can be tested without setting an environment variable, which
/// `cargo test`'s shared process makes unsafe (see [`crate::docs::testing`]).
fn decide(
    flag_name: &str,
    env_name: &str,
    flag: Option<u64>,
    default: Duration,
) -> anyhow::Result<Duration> {
    if let Some(seconds) = flag {
        return seconds_to_duration(flag_name, seconds);
    }
    let Some(raw) = std::env::var(env_name)
        .ok()
        .filter(|v| !v.trim().is_empty())
    else {
        return Ok(default);
    };
    let seconds: u64 = raw.trim().parse().map_err(|_| {
        crate::exit::fail(
            crate::exit::Exit::Usage,
            format!("{env_name}={raw:?} is not a whole number of seconds"),
        )
    })?;
    seconds_to_duration(env_name, seconds)
}

/// Zero is refused rather than read as "no limit". See [`init`] on why the
/// bound itself is not optional.
fn seconds_to_duration(source: &str, seconds: u64) -> anyhow::Result<Duration> {
    if seconds == 0 {
        return Err(crate::exit::fail(
            crate::exit::Exit::Usage,
            format!("{source} must be at least 1 second"),
        ));
    }
    Ok(Duration::from_secs(seconds))
}

fn bounds() -> Bounds {
    // Not `init`ed only in a test that reaches a client without going through
    // `main`; the defaults are the right answer there.
    *BOUNDS.get().unwrap_or(&Bounds {
        connect: CONNECT_TIMEOUT,
        read: READ_TIMEOUT,
    })
}

/// The connect bound this invocation is using.
pub fn connect_timeout() -> Duration {
    bounds().connect
}

/// The read bound this invocation is using.
pub fn read_timeout() -> Duration {
    bounds().read
}

/// The one redirect-following client, built on first use.
static CLIENT: OnceLock<reqwest::Client> = OnceLock::new();

/// The one client that hands redirects back to the caller. See [`no_redirect`].
static NO_REDIRECT: OnceLock<reqwest::Client> = OnceLock::new();

/// How many clients this process has built. At most two, ever, and it is
/// counted because that is the claim worth pinning: `reqwest::Client` keeps
/// its `Arc` to itself, so a test has no pointer to compare and "the same
/// configuration twice" is exactly what a regression back to a client per
/// call would also print. See `one_client_serves_every_caller`.
static BUILDS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

/// Build a client through [`builder`], counting it. The only place in the
/// tree a `reqwest::Client` is constructed.
fn build(
    configure: impl FnOnce(reqwest::ClientBuilder) -> reqwest::ClientBuilder,
) -> reqwest::Client {
    BUILDS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    configure(builder()).build().unwrap_or_default()
}

/// The client, with both bounds applied — one instance, shared.
///
/// A clone of a `reqwest::Client` is an `Arc` bump that shares the original's
/// connection pool, so every caller of this reuses connections with every
/// other one. See this module's doc for what that is worth and for when the
/// bounds get read.
///
/// Falls back to `reqwest`'s default client if the builder fails, which it
/// only can when the TLS backend cannot be initialized. Refusing to build a
/// client at all would turn a broken system TLS store into an unhelpful error
/// from whichever command happened to run first; the default client fails the
/// same way at the same place, just with reqwest's own message.
pub fn client() -> reqwest::Client {
    CLIENT.get_or_init(|| build(|b| b)).clone()
}

/// The same, with automatic redirects turned off — one instance, shared.
///
/// For the three callers that want the `Location` header rather than what it
/// points at: `resolve`'s two probes, which follow git's clone redirects by
/// hand, and `pr`'s label lookup, which reads a repo's owner out of one.
///
/// The `unwrap_or_default` above is safe to repeat here for the reason it is
/// safe there: `ClientBuilder::build` fails only when the TLS backend cannot
/// be initialized, and `reqwest`'s default client panics on exactly that. So
/// the fallback is never reached in a way that would quietly hand these
/// callers a client that *does* follow redirects and leaves them reading a
/// header that is no longer there.
pub fn no_redirect() -> reqwest::Client {
    NO_REDIRECT
        .get_or_init(|| build(|b| b.redirect(reqwest::redirect::Policy::none())))
        .clone()
}

/// The bounds, for building a client — the one constructor both instances
/// above go through.
pub fn builder() -> reqwest::ClientBuilder {
    reqwest::Client::builder()
        .connect_timeout(connect_timeout())
        .read_timeout(read_timeout())
}

/// `reqwest::get`, through [`client`].
///
/// Exists so that call sites reading one small JSON document stay one line.
/// Unlike `reqwest::get`, which builds a client per call, this shares the
/// pool with everything else — so a path that looks like a one-shot read but
/// runs in a loop costs one connection rather than one per turn.
pub async fn get(url: impl reqwest::IntoUrl) -> reqwest::Result<reqwest::Response> {
    client().get(url).send().await
}

/// A response body read whole as text, refused once it passes [`MAX_BODY`].
///
/// The replacement for `resp.text().await` everywhere atgc reads a host it
/// does not control. `source` names the read in the failure and has to
/// identify the host: the point of the refusal is that one *particular*
/// machine sent more than atgc will hold, and a message that does not say
/// which one cannot be acted on. Most callers pass the URL they already
/// built for their `>> GET` log line.
///
/// Decoded as UTF-8 with what is not replaced, where `reqwest`'s own `text`
/// would consult the `Content-Type` charset first. Nothing atgc reads is
/// affected — JSON is UTF-8 by definition of the format, and the only HTML
/// in the tree is tangled.org's, which is UTF-8 — and keeping the charset
/// behaviour would mean decoding the bytes here, after the bound, with a
/// second copy of `encoding_rs`'s job.
pub async fn text_bounded(resp: reqwest::Response, source: &str) -> anyhow::Result<String> {
    let bytes = bytes_bounded(resp, MAX_BODY, source).await?;
    Ok(String::from_utf8_lossy(&bytes).into_owned())
}

/// The same discipline over bytes, with the ceiling named by the caller.
///
/// Checked twice, and the second check is the one that binds. A
/// `Content-Length` refuses before a single byte is downloaded, which is
/// cheap and only as trustworthy as the sender: chunked transfer-encoding
/// and HTTP/2 can each omit it, and nothing obliges it to be true when it is
/// there. So every chunk is counted as it arrives and reading stops the
/// instant the running total passes `max`, which means a hostile sender pays
/// for at most one over-limit chunk rather than for however much more it
/// cares to send.
///
/// `max` is a parameter for exactly one caller.
/// [`crate::clients::atproto::pds::blob_bounded`] carries `pr diff`'s
/// `--max-bytes`, which is a patch's ceiling and a thing a person may
/// legitimately want larger — unlike [`MAX_BODY`], which is a transport
/// bound and nobody's to raise. Everything else passes [`MAX_BODY`]: through
/// [`text_bounded`], or directly from [`crate::clients::xrpc`], which needs
/// the bytes rather than the text because `atgc api` can be pointed at a
/// method whose answer is not text at all.
pub async fn bytes_bounded(
    mut resp: reqwest::Response,
    max: u64,
    source: &str,
) -> anyhow::Result<Vec<u8>> {
    if let Some(len) = resp.content_length()
        && len > max
    {
        return Err(oversize(format!(
            "{source} declared a {len}-byte body, over atgc's {max}-byte limit"
        )));
    }
    let mut buf = Vec::new();
    let mut seen = RunningTotal::new(max);
    while let Some(chunk) = resp
        .chunk()
        .await
        .with_context(|| format!("could not read the response from {source}"))?
    {
        if seen.push(chunk.len()) {
            return Err(oversize(format!(
                "{source} streamed past atgc's {max}-byte limit on a response body"
            )));
        }
        buf.extend_from_slice(&chunk);
    }
    Ok(buf)
}

/// The same limit against a body that has already been read, for the one
/// caller that cannot bound it as it arrives.
///
/// **This is not the bound [`bytes_bounded`] is, and it is worth being exact
/// about the difference.** [`crate::clients::xrpc`]'s authenticated arm sends
/// through jacquard's `HttpClient`, whose entire contract is
/// `http::Response<Vec<u8>>`: the body is buffered before atgc is handed it,
/// so the memory a hostile host wanted is already spent by the time this can
/// fire. Bounding it properly means bounding it inside that transport, where
/// the trait fixes the error type at `reqwest::Error` and there is nothing to
/// build a refusal out of — a gap every authenticated write in atgc shares,
/// recorded in TODO.md rather than papered over here.
///
/// What this still buys is the other half of the bound: a body atgc would
/// have refused from any other host is not handed on to a caller, or written
/// to `atgc api`'s stdout, and the two arms of one command answer the same
/// way instead of the limit having an invisible exception in it.
pub fn refuse_if_oversize(body: &[u8], source: &str) -> anyhow::Result<()> {
    if body.len() as u64 > MAX_BODY {
        return Err(oversize(format!(
            "{source} answered with {} bytes, over atgc's {MAX_BODY}-byte limit",
            body.len()
        )));
    }
    Ok(())
}

/// One refusal, however it was reached, and nothing of the body handed back.
///
/// Two spellings arrive here because a reader wants to know which check
/// fired — a declared length is the sender's claim about itself, a running
/// total is what actually turned up — and both name the host and the limit.
/// What neither does is return the bytes read so far: a truncated JSON
/// document that happens to parse is the failure this is meant to prevent,
/// not a degraded success, and the one call site that could have been
/// tempted (a PDS's error envelope, read for its `message`) propagates this
/// instead of falling back to the status alone.
///
/// [`crate::exit::Exit::Unreachable`] rather than an unclassified `1`, and
/// not a new code. The other two bounds in this module already exit `6`,
/// because [`crate::exit::classify`] reads a `reqwest` timeout off the error
/// chain and says so. All three are one event as far as a caller is
/// concerned — the exchange was refused by atgc's own limits, and nothing
/// about the command line will change that — so a third bound answering
/// with a different status would only make the module inconsistent with
/// itself.
fn oversize(message: String) -> anyhow::Error {
    crate::exit::fail(
        crate::exit::Exit::Unreachable,
        format!(
            "{message}\n\
             that limit is not adjustable: it is what stops a host atgc does not \
             control from deciding how much memory atgc uses"
        ),
    )
}

/// How many bytes of a streamed response have arrived, checked once per
/// chunk against `max`. The only part of [`bytes_bounded`]'s loop that is
/// not a socket, split out so the abort can be proven to fire at the right
/// byte without opening one — feed it plain chunk lengths and see.
struct RunningTotal {
    seen: u64,
    max: u64,
}

impl RunningTotal {
    fn new(max: u64) -> Self {
        Self { seen: 0, max }
    }

    /// Record one more chunk. Returns whether the running total is now over
    /// the limit; the caller stops reading right there; a hostile sender
    /// pays for at most one over-limit chunk rather than however much more
    /// it cares to keep sending.
    fn push(&mut self, chunk_len: usize) -> bool {
        self.seen += chunk_len as u64;
        self.seen > self.max
    }
}

// ---------------------------------------------------------------------------
// Retrying
// ---------------------------------------------------------------------------

/// How many times one request is sent before atgc gives up, the first attempt
/// included. Three, which is what `pr create`'s image uploads have used since
/// they were the only thing in the tree that retried at all.
pub const ATTEMPTS: u32 = 3;

/// The pause before the second attempt; it doubles for each one after.
/// Short enough that a hiccup costs nobody a noticeable wait, and the whole
/// budget — 500ms then 1s — is a second and a half.
pub const BACKOFF: Duration = Duration::from_millis(500);

/// The most of a `Retry-After` atgc will actually wait.
///
/// Thirty seconds, which is [`READ_TIMEOUT`]: the longest atgc is already
/// willing to spend on one exchange that is making no progress. A header is
/// the server's opinion about its own load and is worth honouring — it is the
/// only number in the exchange that is not a guess — but it is *not* the
/// server's to decide how long a command sits there. Real services ask for
/// minutes, and a `Retry-After: 3600` on a `pr list` inside a script is
/// indistinguishable from a hang.
///
/// Past the clamp the wait is still the clamp rather than a refusal: the
/// server said "not yet", and waiting the longest atgc will wait is a better
/// answer to that than either ignoring it or failing without having tried.
pub const MAX_RETRY_AFTER: Duration = Duration::from_secs(30);

/// What to do about one failed attempt.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Verdict {
    /// A second identical request earns an identical answer.
    Fatal,
    /// Worth another attempt. `Some` carries what the server itself asked
    /// for, already clamped; `None` means fall back to the doubling backoff.
    Again(Option<Duration>),
}

/// Is this status worth sending the same request again?
///
/// **Off the status, never the message.** The retry this replaced matched
/// substrings in error prose, so a 1,404-byte image abandoned its upload on
/// the first hiccup and a 403 whose wording did not spell the number was
/// tried three times — see the module doc of [`crate::exit`] for the same
/// complaint about failure classification.
///
/// 429 is the server asking for exactly this. The four 5xx are the server's
/// own moment: 500 covers a service that panicked on one request, and
/// 502/503/504 are a proxy in front of a service that is restarting or
/// overloaded. Every other 4xx is the request's own fault and does not
/// improve by being repeated — nor does an unlisted 5xx like 501, which is a
/// statement about the method rather than about this moment.
pub fn retryable_status(status: u16) -> bool {
    matches!(status, 429 | 500 | 502 | 503 | 504)
}

/// Is this transport failure worth another attempt?
///
/// Connect and timeout only, which are the two that say "this exchange did
/// not happen" rather than "this exchange went wrong". They are also exactly
/// the two [`crate::exit::classify`] reads as [`crate::exit::Exit::Unreachable`],
/// so what atgc retries and what atgc tells a caller to retry are one list.
///
/// A decode failure, a redirect loop or a body error is not here: the request
/// arrived and something about the answer was wrong, and sending it again
/// produces the same wrong answer more slowly.
pub fn retryable_transport(err: &reqwest::Error) -> bool {
    err.is_connect() || err.is_timeout()
}

/// The verdict for one response, header included.
///
/// `Retry-After` is read on 429 and on 503 and nowhere else. Those are the
/// two statuses RFC 9110 defines it for — "too many requests" and "come back
/// when I am up again" — and a header on a 500 would be a server volunteering
/// a schedule for a fault it has not admitted to.
pub fn response_verdict(resp: &reqwest::Response) -> Verdict {
    let status = resp.status().as_u16();
    if !retryable_status(status) {
        return Verdict::Fatal;
    }
    let asked = matches!(status, 429 | 503)
        .then(|| retry_after(resp.headers(), std::time::SystemTime::now()))
        .flatten();
    Verdict::Again(asked)
}

/// The `Retry-After` header as a duration, clamped, or `None` if it is
/// absent, empty, not one of the two spellings, or already in the past.
///
/// `now` is a parameter because the HTTP-date form is only meaningful
/// against a clock, and a test that had to wait for a real one could only
/// assert something vague. See [`parse_retry_after`] for the two forms.
pub fn retry_after(
    headers: &reqwest::header::HeaderMap,
    now: std::time::SystemTime,
) -> Option<Duration> {
    let raw = headers.get(reqwest::header::RETRY_AFTER)?.to_str().ok()?;
    parse_retry_after(raw, now)
}

/// Both spellings RFC 9110 allows for `Retry-After`, clamped to
/// [`MAX_RETRY_AFTER`].
///
/// - *delta-seconds*: `Retry-After: 120`. A whole number of seconds from now.
/// - *HTTP-date*: `Retry-After: Wed, 21 Oct 2026 07:28:00 GMT`. An instant,
///   so what atgc waits is the gap between it and `now`.
///
/// Both are sent in the wild — Cloudflare and most rate limiters send the
/// first, `503`s from a scheduled maintenance window tend to send the second
/// — so reading only one would be reading none of some services.
///
/// Anything else is `None`, and the caller falls back to its own backoff. A
/// header atgc cannot read is not worth failing over: the request is already
/// going to be retried, and the only question the header answers is *when*.
/// A date in the past is `None` for the same reason it is not zero — "you may
/// retry now" and "there was no header" want the same behaviour, which is the
/// backoff rather than a hot loop.
fn parse_retry_after(raw: &str, now: std::time::SystemTime) -> Option<Duration> {
    let raw = raw.trim();
    if let Ok(seconds) = raw.parse::<u64>() {
        return Some(Duration::from_secs(seconds).min(MAX_RETRY_AFTER));
    }
    let when = http_date(raw)?;
    let gap = when.duration_since(now).ok()?;
    if gap.is_zero() {
        return None;
    }
    Some(gap.min(MAX_RETRY_AFTER))
}

/// One HTTP-date, as an instant.
///
/// IMF-fixdate — `Sun, 06 Nov 1994 08:49:37 GMT` — is the only form a server
/// is allowed to *send*; RFC 9110 requires recipients to also accept two
/// obsolete forms, and `parse_from_rfc2822` covers the RFC 850 one's
/// two-digit-year relatives where chrono can. What is not accepted falls back
/// to the backoff, which is the same place an absent header lands.
fn http_date(raw: &str) -> Option<std::time::SystemTime> {
    use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
    let stamp = DateTime::parse_from_rfc2822(raw)
        .map(|dt| dt.with_timezone(&Utc))
        .ok()
        .or_else(|| {
            NaiveDateTime::parse_from_str(raw, "%a, %d %b %Y %H:%M:%S GMT")
                .ok()
                .map(|naive| Utc.from_utc_datetime(&naive))
        })?;
    let seconds = u64::try_from(stamp.timestamp()).ok()?;
    Some(std::time::UNIX_EPOCH + Duration::from_secs(seconds))
}

/// How long to wait before attempt number `attempt + 1`.
///
/// The server's own number wins where it sent one, because it is the only
/// figure in the exchange that is not a guess. Otherwise the backoff doubles:
/// 500ms, then 1s.
fn pause_before(attempt: u32, asked: Option<Duration>) -> Duration {
    asked.unwrap_or_else(|| BACKOFF * 2u32.saturating_pow(attempt.saturating_sub(1)))
}

/// Wait, and say so.
///
/// A retry nobody can see is a stall with no explanation: the command sits
/// there for a second and a half and the only account of it is the exit code
/// it did not produce. [`crate::term::say::Level::Step`] is where the rest of
/// "what this command is doing while it does it" lives, so `-q` silences this
/// along with the rest of the progress and `-qq` silences everything.
async fn pause(topic: crate::term::say::Topic, what: &str, attempt: u32, asked: Option<Duration>) {
    let delay = pause_before(attempt, asked);
    let because = match asked {
        Some(_) => " (the server asked)",
        None => "",
    };
    crate::term::say::emit(
        crate::term::say::Level::Step,
        topic,
        &format!(
            "{what} failed on attempt {attempt} of {ATTEMPTS}; \
             retrying in {:.1}s{because}",
            delay.as_secs_f32()
        ),
    );
    crate::logging::debug::log(format!(
        "retry {what}: attempt {attempt} of {ATTEMPTS}, waiting {delay:?}{because}"
    ));
    tokio::time::sleep(delay).await;
}

/// Run `attempt` until it succeeds, is refused for good, or runs out of
/// budget — the one retry loop in the crate.
///
/// Generic over the error because the two transports that need this do not
/// share one: the reads below fail with `reqwest`'s error, and
/// [`crate::cmd::images`]'s blob upload fails with jacquard's `AgentError`,
/// which wraps a status atgc never sees as a response. `classify` is what
/// each one knows about its own errors; everything else — the budget, the
/// schedule, the announcement — is here rather than copied.
///
/// The last attempt's error is returned as it is. Nothing is added to it,
/// because the caller's own `with_context` is what names the host, and a
/// budget this loop chose is not a fact about the failure.
pub async fn retrying<T, E, F, Fut>(
    topic: crate::term::say::Topic,
    what: &str,
    classify: impl Fn(&E) -> Verdict,
    mut attempt: F,
) -> Result<T, E>
where
    F: FnMut(u32) -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    for number in 1..=ATTEMPTS {
        match attempt(number).await {
            Ok(value) => return Ok(value),
            Err(e) => match classify(&e) {
                Verdict::Fatal => return Err(e),
                Verdict::Again(_) if number == ATTEMPTS => return Err(e),
                Verdict::Again(asked) => pause(topic, what, number, asked).await,
            },
        }
    }
    unreachable!("the last attempt either returned or failed");
}

/// One GET, retried on the statuses and the transport failures above.
///
/// The read-path counterpart of [`get`], and the only difference a caller
/// sees is in what it does *not* see: a knot that was restarting, a PDS that
/// answered 429 once. Retrying a GET is safe because a GET is idempotent by
/// definition of the method — which is exactly why the writes in the tree
/// (`putRecord`, `applyWrites`, `deleteRecord`, a knot procedure) are not on
/// this and are not going to be.
///
/// # What comes back when the budget runs out
///
/// A transport failure is returned as it is, still carrying the
/// `reqwest::Error` that [`crate::exit::classify`] reads as
/// [`crate::exit::Exit::Unreachable`] — the code that means "try again
/// unchanged", which is precisely what atgc has just spent three attempts
/// finding out is worth doing.
///
/// A status that stayed retryable to the end is *not* handed back as a
/// response for the caller to report as it always did. It becomes a failure
/// tagged with the same code, because a 429 or a 503 after the whole budget
/// is the same event as a host that did not answer, and a caller branching on
/// `6` should not have to learn that this one arrives as a `1`. The server's
/// own `message` is carried into it where the body is the ATProto error
/// envelope both a PDS and Bobbin send, so the diagnosis is not lost with the
/// response.
///
/// Every *other* status — a 404, a 403, a 400 — comes back as `Ok(response)`
/// untouched, because the caller's own reading of it is the diagnosis.
pub async fn get_retrying(
    url: &str,
    topic: crate::term::say::Topic,
    what: &str,
) -> anyhow::Result<reqwest::Response> {
    let mut exhausted: Option<reqwest::Response> = None;
    for number in 1..=ATTEMPTS {
        match get(url).await {
            Ok(resp) => match response_verdict(&resp) {
                Verdict::Fatal => return Ok(resp),
                Verdict::Again(_) if number == ATTEMPTS => {
                    exhausted = Some(resp);
                    break;
                }
                Verdict::Again(asked) => pause(topic, what, number, asked).await,
            },
            Err(e) if !retryable_transport(&e) || number == ATTEMPTS => return Err(e.into()),
            Err(e) => {
                crate::logging::debug::log(format!("<< {what}: {e}"));
                pause(topic, what, number, None).await;
            }
        }
    }
    // An invariant rather than a possibility: every arm of the loop above
    // either returns or sets this before breaking. If it is ever `None` the
    // loop has been changed and this is a defect in atgc — which is what it
    // now says, instead of aborting the process on an `expect`.
    let resp = exhausted.ok_or_else(|| {
        crate::exit::bug(format!(
            "the retry loop for {what} finished without a response or an error"
        ))
    })?;
    let status = resp.status();
    let detail = text_bounded(resp, what)
        .await
        .ok()
        .and_then(|text| serde_json::from_str::<serde_json::Value>(&text).ok())
        .and_then(|body| body["message"].as_str().map(str::to_string));
    let said = match detail {
        Some(message) => format!(": {message}"),
        None => String::new(),
    };
    Err(crate::exit::fail(
        crate::exit::Exit::Unreachable,
        format!("{what} was still answering {status} after {ATTEMPTS} attempts{said}"),
    ))
}

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

    /// Every caller shares one client, so connections are reused; and the
    /// two redirect policies are two clients, so a caller that reads a
    /// `Location` header never gets one that followed it.
    ///
    /// Counted rather than compared: two clients built from the same
    /// [`builder`] print the same `Debug`, so equality there would pass
    /// against exactly the client-per-call this replaced. Once both cells
    /// are filled the count can no longer move, which is what makes the
    /// exact `2` safe next to whatever else this binary's tests are doing.
    #[test]
    fn one_client_serves_every_caller() {
        let following = client();
        let manual = no_redirect();
        for _ in 0..8 {
            let _ = client();
            let _ = no_redirect();
        }
        assert_eq!(
            BUILDS.load(std::sync::atomic::Ordering::Relaxed),
            2,
            "one client per redirect policy, for the life of the process"
        );
        assert_eq!(
            format!("{following:?}"),
            format!("{:?}", client()),
            "a second call is the same client, not a second one"
        );
        assert_ne!(
            format!("{following:?}"),
            format!("{manual:?}"),
            "and the no-redirect client is a different one"
        );
    }

    /// A retry decision is made on the number, not on the prose around it.
    /// The four 5xx worth another attempt, the one 4xx that asks for one,
    /// and nothing else — a 501 is a statement about the method and a 404 is
    /// a statement about the path, and neither improves on the second ask.
    #[test]
    fn only_the_listed_statuses_retry() {
        for retryable in [429, 500, 502, 503, 504] {
            assert!(retryable_status(retryable), "{retryable} retries");
        }
        for hopeless in [400, 401, 403, 404, 409, 413, 418, 501, 505] {
            assert!(!retryable_status(hopeless), "{hopeless} fails fast");
        }
        assert!(!retryable_status(200), "a success is not a retry");
    }

    /// `Retry-After: 120`, the form nearly every rate limiter sends.
    #[test]
    fn retry_after_reads_delta_seconds() {
        let now = std::time::UNIX_EPOCH + Duration::from_secs(1_000_000);
        assert_eq!(
            parse_retry_after("12", now),
            Some(Duration::from_secs(12)),
            "a whole number of seconds is seconds from now"
        );
        assert_eq!(
            parse_retry_after("  12  ", now),
            Some(Duration::from_secs(12)),
            "whitespace around the value is not a parse failure"
        );
        assert_eq!(
            parse_retry_after("0", now),
            Some(Duration::ZERO),
            "zero is a real answer: retry immediately"
        );
    }

    /// The other spelling the spec allows, which a maintenance-window 503
    /// tends to send: an instant, so what atgc waits is the gap to it.
    #[test]
    fn retry_after_reads_an_http_date() {
        // 1994-11-06T08:49:37Z, the RFC's own example, and a `now` ten
        // seconds before it.
        let when = 784_111_777;
        let now = std::time::UNIX_EPOCH + Duration::from_secs(when - 10);
        assert_eq!(
            parse_retry_after("Sun, 06 Nov 1994 08:49:37 GMT", now),
            Some(Duration::from_secs(10)),
            "the gap between now and the date, not the date"
        );
        assert_eq!(
            parse_retry_after("Sun, 06 Nov 1994 08:49:37 +0000", now),
            Some(Duration::from_secs(10)),
            "the same instant spelled with a numeric offset"
        );
    }

    /// A date that has already passed is not a wait of zero and not a
    /// failure: it lands where an absent header lands, on the backoff.
    /// Zero here would be a hot loop against a server that just said no.
    #[test]
    fn a_date_in_the_past_falls_back_to_the_backoff() {
        let when = 784_111_777;
        let now = std::time::UNIX_EPOCH + Duration::from_secs(when + 60);
        assert_eq!(
            parse_retry_after("Sun, 06 Nov 1994 08:49:37 GMT", now),
            None
        );
    }

    /// **An unreadable header is not an error.** The request is being
    /// retried either way; the only thing the header decides is when, so
    /// anything atgc cannot read falls through to its own schedule rather
    /// than failing a command over a string a server chose.
    #[test]
    fn an_unparseable_retry_after_falls_back_to_the_backoff() {
        let now = std::time::UNIX_EPOCH + Duration::from_secs(1_000_000);
        for junk in [
            "",
            "   ",
            "soon",
            "-5",
            "12.5",
            "1e3",
            "Sun, 06 Nov 1994 08:49:37",
            "Thu, 99 Xxx 1994 08:49:37 GMT",
        ] {
            assert_eq!(
                parse_retry_after(junk, now),
                None,
                "{junk:?} is not a Retry-After"
            );
        }
    }

    /// **The server may ask for an hour; it does not get one.** Both
    /// spellings are clamped, because the header is the server's opinion
    /// about its own load and not its decision about how long a CLI sits
    /// there.
    #[test]
    fn a_long_retry_after_is_clamped_in_both_spellings() {
        let now = std::time::UNIX_EPOCH + Duration::from_secs(784_111_777);
        assert_eq!(
            parse_retry_after("3600", now),
            Some(MAX_RETRY_AFTER),
            "an hour of delta-seconds is capped"
        );
        assert_eq!(
            parse_retry_after("Sun, 06 Nov 1994 09:49:37 GMT", now),
            Some(MAX_RETRY_AFTER),
            "an hour away as a date is capped the same way"
        );
        assert!(
            MAX_RETRY_AFTER <= READ_TIMEOUT,
            "the clamp must not be longer than the longest wait atgc already allows"
        );
    }

    /// The schedule when nothing was asked for — 500ms, then 1s — and the
    /// fact that anything asked for wins outright.
    #[test]
    fn the_backoff_doubles_and_a_server_answer_replaces_it() {
        assert_eq!(pause_before(1, None), BACKOFF);
        assert_eq!(pause_before(2, None), BACKOFF * 2);
        assert_eq!(pause_before(3, None), BACKOFF * 4);
        assert_eq!(
            pause_before(1, Some(Duration::from_secs(7))),
            Duration::from_secs(7),
            "the only number in the exchange that is not a guess"
        );
    }

    /// The transport half of the same decision: the two failures that mean
    /// the exchange did not happen. Both are what `exit::classify` already
    /// calls `Unreachable`, so the list atgc retries and the list it tells a
    /// caller to retry stay the same list.
    #[tokio::test]
    async fn a_connect_failure_retries_and_classifies_as_unreachable() {
        // 203.0.113.0/24 is TEST-NET-3 and unroutable, but this never opens
        // a socket: the URL's port is out of range, so reqwest refuses to
        // build the request. Instead, drive the two predicates directly on
        // the one error shape that can be made without a network.
        let err = get("http://203.0.113.1:99999/").await.expect_err("no port");
        assert!(
            !retryable_transport(&err),
            "a builder error is not a transport failure and must not retry"
        );
        let wrapped = anyhow::Error::new(err);
        assert_eq!(crate::exit::classify(&wrapped), crate::exit::Exit::Failure);
    }

    /// **Exhausting the budget is still `6`.** The whole point of the code
    /// is "try again unchanged", and having just spent three attempts
    /// establishing that the host is the problem is the strongest evidence
    /// for it atgc ever has. The message says the status and the count, so a
    /// reader can tell one attempt from three.
    #[test]
    fn running_out_of_budget_exits_six_and_says_how_many_attempts() {
        let err = crate::exit::fail(
            crate::exit::Exit::Unreachable,
            format!(
                "listRecords at pds.example was still answering 429 Too Many Requests \
                 after {ATTEMPTS} attempts: slow down"
            ),
        );
        assert_eq!(crate::exit::classify(&err), crate::exit::Exit::Unreachable);
        let message = err.to_string();
        assert!(message.contains("429"), "{message}");
        assert!(message.contains("after 3 attempts"), "{message}");
        assert!(
            message.contains("slow down"),
            "the server's own words: {message}"
        );
    }

    /// The budget is spent and no more, and a fatal verdict spends none of
    /// it. Driven through `retrying` with a counter instead of a socket:
    /// what is being asserted is the loop, and a request would only make it
    /// slower to observe.
    #[tokio::test]
    async fn the_budget_is_three_attempts_and_a_fatal_verdict_uses_one() {
        let tries = std::cell::Cell::new(0);
        let result: Result<(), &str> = retrying(
            crate::term::say::Topic::Pds,
            "a host that keeps saying 503",
            |_| Verdict::Again(Some(Duration::ZERO)),
            |_| {
                tries.set(tries.get() + 1);
                async { Err("503") }
            },
        )
        .await;
        assert_eq!(result, Err("503"), "the last failure comes back as it is");
        assert_eq!(tries.get(), ATTEMPTS, "three attempts, not four");

        let tries = std::cell::Cell::new(0);
        let result: Result<(), &str> = retrying(
            crate::term::say::Topic::Pds,
            "a host that said 404",
            |_| Verdict::Fatal,
            |_| {
                tries.set(tries.get() + 1);
                async { Err("404") }
            },
        )
        .await;
        assert_eq!(result, Err("404"));
        assert_eq!(tries.get(), 1, "a 4xx is asked exactly once");
    }

    /// A success on the second attempt is a success, and the caller never
    /// learns there was a first.
    #[tokio::test]
    async fn a_retry_that_works_returns_the_value() {
        let tries = std::cell::Cell::new(0);
        let result: Result<u32, &str> = retrying(
            crate::term::say::Topic::Pds,
            "a host having a moment",
            |_| Verdict::Again(Some(Duration::ZERO)),
            |number| {
                tries.set(tries.get() + 1);
                async move { if number == 1 { Err("500") } else { Ok(number) } }
            },
        )
        .await;
        assert_eq!(result, Ok(2));
        assert_eq!(tries.get(), 2);
    }

    /// The header is only read where the spec defines it, which is the two
    /// statuses that mean "come back later". A 500 volunteering a schedule
    /// for a fault it has not admitted to is ignored.
    #[test]
    fn the_header_is_read_on_429_and_503_only() {
        assert!(matches!(
            response_verdict_for(429, Some("7")),
            Verdict::Again(Some(d)) if d == Duration::from_secs(7)
        ));
        assert!(matches!(
            response_verdict_for(503, Some("7")),
            Verdict::Again(Some(d)) if d == Duration::from_secs(7)
        ));
        assert_eq!(response_verdict_for(500, Some("7")), Verdict::Again(None));
        assert_eq!(response_verdict_for(429, None), Verdict::Again(None));
        assert_eq!(response_verdict_for(404, Some("7")), Verdict::Fatal);
    }

    /// The status-and-header half of [`response_verdict`], without a
    /// `reqwest::Response` — which cannot be constructed outside a real
    /// exchange. The two share every line that decides anything.
    fn response_verdict_for(status: u16, header: Option<&str>) -> Verdict {
        if !retryable_status(status) {
            return Verdict::Fatal;
        }
        let mut headers = reqwest::header::HeaderMap::new();
        if let Some(value) = header {
            headers.insert(
                reqwest::header::RETRY_AFTER,
                reqwest::header::HeaderValue::from_str(value).expect("a header value"),
            );
        }
        let asked = matches!(status, 429 | 503)
            .then(|| retry_after(&headers, std::time::UNIX_EPOCH))
            .flatten();
        Verdict::Again(asked)
    }

    /// The bound that matters, against a host that behaves the way the parked
    /// IP did: accepts the packet and never answers. 203.0.113.0/24 is
    /// TEST-NET-3 (RFC 5737) and is not routed, so the connect can only
    /// either time out here or fail faster with a network error — never
    /// succeed, and never reach anyone's real server.
    ///
    /// Asserted as a ceiling rather than an equality: the point is that the
    /// wait is atgc's to choose and no longer the kernel's SYN schedule,
    /// which on this machine ran to 30s.
    ///
    /// Ignored because it opens a socket and spends the timeout doing it, not
    /// because it fails. [`crate::docs::testing`] allows network access in `#[ignore]`d
    /// tests and nowhere else, and a five-second test does not belong in a
    /// suite that finishes in one. Run with `cargo test -- --ignored`.
    ///
    /// There is no offline half to split out, unlike `resolve`'s: the claim
    /// is about what the local network stack does when nobody answers, and
    /// the only way to observe that is to wait for it.
    #[tokio::test]
    #[ignore = "opens a socket and waits out the connect timeout: 5s against unroutable TEST-NET-3"]
    async fn a_silent_host_gives_up_on_our_schedule_not_the_kernels() {
        let started = std::time::Instant::now();
        let result = get("http://203.0.113.1/.well-known/atproto-did").await;
        let waited = started.elapsed();
        assert!(result.is_err(), "TEST-NET-3 cannot answer");
        assert!(
            waited < CONNECT_TIMEOUT + Duration::from_secs(5),
            "waited {waited:?}, which is past the connect timeout"
        );
    }

    /// Flag beats environment beats default, which is the order every other
    /// setting in atgc resolves in — `--account` over `ATGC_ACCOUNT`,
    /// `--no-input` over `ATGC_NO_INPUT`.
    ///
    /// Driven through `decide` with its inputs passed in rather than through
    /// `init`, because setting an environment variable is `unsafe` in edition
    /// 2024 and this crate forbids it. The env arm is exercised by naming a
    /// variable that is not set, which is the branch that has to fall
    /// through; the value-carrying arm is the flag's, and they share
    /// `seconds_to_duration`.
    #[test]
    fn a_flag_wins_and_an_unset_variable_falls_through_to_the_default() {
        let unset = "ATGC_TEST_TIMEOUT_THAT_IS_NOT_SET";
        assert_eq!(
            decide("--connect-timeout", unset, Some(12), CONNECT_TIMEOUT).unwrap(),
            Duration::from_secs(12),
            "the flag should win"
        );
        assert_eq!(
            decide("--connect-timeout", unset, None, CONNECT_TIMEOUT).unwrap(),
            CONNECT_TIMEOUT,
            "nothing set should leave the default alone"
        );
        assert_eq!(
            decide("--read-timeout", unset, None, READ_TIMEOUT).unwrap(),
            READ_TIMEOUT
        );
    }

    /// **The bound may be raised, never removed.** Zero is refused rather
    /// than read as "no limit", because an unbounded connect is the wait this
    /// module exists to stop and it would be a strange way to spell it.
    #[test]
    fn zero_is_refused_and_says_what_to_do_instead() {
        let err = decide("--connect-timeout", "X", Some(0), CONNECT_TIMEOUT)
            .expect_err("zero is not a timeout");
        assert_eq!(
            err.to_string(),
            "--connect-timeout must be at least 1 second"
        );
        assert_eq!(crate::exit::classify(&err), crate::exit::Exit::Usage);
    }

    /// Ten minutes is a fine answer if that is what somebody's link needs —
    /// there is no ceiling, only a floor.
    #[test]
    fn a_long_wait_is_allowed() {
        assert_eq!(
            decide("--connect-timeout", "X", Some(600), CONNECT_TIMEOUT).unwrap(),
            Duration::from_secs(600)
        );
    }

    /// Before `init` runs — a unit test that reaches a client without going
    /// through `main` — the defaults stand rather than panicking on an
    /// unset `OnceLock`.
    #[test]
    fn the_defaults_hold_before_init() {
        assert_eq!(connect_timeout(), CONNECT_TIMEOUT);
        assert_eq!(read_timeout(), READ_TIMEOUT);
    }

    /// A bare `reqwest::Client` has neither bound, and the whole point of the
    /// module is that nothing constructs one of those by hand. Cheap enough
    /// to state directly: the builder is the only thing callers touch.
    #[test]
    fn the_builder_is_what_carries_the_bounds() {
        // `ClientBuilder` exposes no getters, so this checks the one thing it
        // can: that the configured builder still builds. A typo'd duration or
        // a future reqwest that rejects one of these fails here rather than
        // silently at the first request.
        assert!(builder().build().is_ok());
    }

    /// The gap the running total closes: a response with no Content-Length —
    /// chunked transfer-encoding, HTTP/2, or a sender that simply lies —
    /// sails straight past the header check, and `resp.bytes()` would then
    /// buffer however much the sender felt like providing. Driving a real
    /// [`bytes_bounded`] needs a socket, which [`crate::docs::testing`] keeps
    /// out of the unit tests, so this pins the part that decides: the running
    /// total, fed the same chunk lengths `resp.chunk()` hands the real loop,
    /// with no header involved at any point.
    #[test]
    fn a_running_total_trips_the_instant_it_passes_the_limit_with_no_header_in_sight() {
        let mut total = RunningTotal::new(10);
        assert!(!total.push(4), "4 of 10: under");
        assert!(
            !total.push(6),
            "10 of 10: at the limit exactly, not over it"
        );
        assert!(total.push(1), "11 of 10: this is the chunk that trips it");
    }

    /// A single chunk bigger than the whole limit trips on its first call —
    /// what a chunked, header-less stream looks like when a hostile sender
    /// pushes its whole payload as one frame rather than trickling it.
    #[test]
    fn one_oversized_chunk_trips_on_its_own() {
        assert!(RunningTotal::new(10).push(11));
    }

    /// **The refusal carries a status, and it is the one the deadlines
    /// carry.** A caller that already branches on `6` for "atgc's transport
    /// bounds refused this" would otherwise have to learn a second number for
    /// the third bound, and the message would be the only way to tell.
    ///
    /// The wording is pinned too, on the two facts a reader has to act on:
    /// which host, and what the limit was. A refusal that says neither is
    /// indistinguishable from the host being broken.
    #[test]
    fn an_oversize_refusal_names_the_host_and_the_limit_and_exits_six() {
        let err = oversize(format!(
            "https://evil.example/.well-known/did.json streamed past atgc's \
             {MAX_BODY}-byte limit on a response body"
        ));
        assert_eq!(crate::exit::classify(&err), crate::exit::Exit::Unreachable);
        let message = err.to_string();
        assert!(message.contains("evil.example"), "{message}");
        assert!(message.contains("8388608-byte limit"), "{message}");
        assert!(message.contains("not adjustable"), "{message}");
    }
}