hls-runtime 0.12.0

//! `TokioClient` — a tokio + reqwest IO adapter driving
//! [`crate::client::HlsClient`] over real HTTP (issue #717 slice 5).
//!
//! Feature-gated behind `tokio` (NOT default): the sans-IO core in
//! `engine.rs` has zero dependency on tokio, reqwest, a socket, or a clock —
//! this module is a thin async shell that performs the actual HTTP GETs the
//! core's [`crate::client::Action`]s describe (playlist reload, incl. blocking
//! `_HLS_msn`/`_HLS_part`; resource fetch, incl. `Range` byte-ranges) and
//! feeds the responses back into the core, looping until the caller stops
//! polling or the stream ends.
//!
//! # Auth
//!
//! [`TokioClientConfig::auth`] takes a [`broadcast_auth::Credentials`] — the
//! same shared scheme-agnostic model `rtsp-runtime` and `multimux`'s HTTP
//! input adapters (`source::http_auth`) use (issue #663 P3b/P3c). Basic
//! (RFC 7617) and Bearer (RFC 6750) are pre-applied on every request to the
//! playlist URL's origin (scheme + host + port, RFC 6454) via
//! reqwest's own request-builder helpers (`RequestBuilder::basic_auth`/
//! `bearer_auth`) — no challenge round-trip needed. Digest (RFC 7616) is not
//! something reqwest supports natively: the first request for a given
//! resource is sent bare, and only if the server answers `401` with a
//! `WWW-Authenticate` challenge does [`TokioClient`] compute the
//! `Authorization` response via [`broadcast_auth::Authenticator`] and resend
//! once — the resulting authenticator is then cached and applied
//! preemptively to every subsequent request for as long as it keeps being
//! accepted (RFC 7616 §3.3's `nc` advances across those calls), so a live
//! pull doesn't round-trip a fresh challenge on every single fetch.
//!
//! Credentials of any scheme go only to that origin: a playlist naming a
//! resource on another host (a CDN, say) gets that request sent bare, and a
//! `401` from such a host is not answered.
//!
//! # Error recovery
//!
//! - Retry delays come from `backon`'s exponential schedule (minimum
//!   [`TokioClientConfig::retry_backoff`], doubling, capped at
//!   [`TokioClientConfig::max_retry_backoff`]) with optional jitter
//!   ([`TokioClientConfig::jitter`]: each delay `d` becomes a random value in
//!   `[d, 2d)`, clamped to `max_retry_backoff`, so many clients reloading
//!   together do not retry in lock step but no delay ever exceeds the cap).
//! - A **resource** (init/part/segment) fetch that keeps failing is retried
//!   up to [`TokioClientConfig::max_resource_retries`] times (total attempts)
//!   with that schedule, then [`crate::client::HlsClient::on_error`] is
//!   called and the adapter moves on — the sans-IO core un-marks that
//!   resource as "requested", so the *next* playlist reload naturally
//!   re-requests it (see `engine.rs`'s `on_error` docs). One flaky fetch
//!   never stalls the whole client. A `4xx` answer (other than `408`/`429`)
//!   is not retried — e.g. RFC 8216bis §6.2.6's `404` for a stale preload
//!   hint — and goes straight to `on_error`.
//! - A **playlist** reload has no such fallback in the sans-IO core — unlike
//!   a resource, [`crate::client::HlsClient::on_error`] with `None` does not
//!   re-queue anything (there is no "next reload" to fall back to; the
//!   *current* reload IS the mechanism that discovers what to fetch next).
//!   So a playlist fetch is retried indefinitely with capped backoff
//!   ([`TokioClientConfig::retry_backoff`]/[`TokioClientConfig::max_retry_backoff`])
//!   rather than ever giving up — a caller wanting a hard ceiling on how long
//!   [`TokioClient::next_output`] may block should wrap it in
//!   `tokio::time::timeout` itself.
//!
//! - A response the sans-IO core **rejects** (a key of the wrong length such as
//!   an HTML error page served with `200`, a segment that does not decrypt or
//!   demux, a delta playlist that cannot be merged, an unsupported
//!   encryption, a duplicate delivery) is a per-request failure, not a fatal
//!   one: it is counted in [`TokioClientStats::rejected_responses`], kept in
//!   [`TokioClient::last_rejection`], and retried after a capped backoff — the
//!   core makes the next reload re-request what was lost, and a VOD playlist
//!   (which has no further reload) gets a plain playlist reload or an init
//!   retry from the adapter. Only sixteen such rounds in a row with no
//!   [`Output`] between them end the session
//!   ([`TokioError::Client`] / [`TokioError::Stalled`]).
//! - Every response body is read incrementally and abandoned with
//!   [`TokioError::BodyTooLarge`] (never retried) once it exceeds
//!   [`TokioClientConfig::max_body_bytes`].
//!
//! # Cancellation
//!
//! [`TokioClientConfig::cancel`] is a `CancellationToken`: once cancelled,
//! [`TokioClient::next_output`] abandons whatever request or backoff sleep is
//! in flight and returns `Ok(None)` (the same "stop polling" signal as the end
//! of the stream). There is no other shutdown mechanism and no polling sleep.

use std::time::Duration;

use backon::{BackoffBuilder, ExponentialBuilder};
use broadcast_auth::{Authenticator, Credentials, RequestContext};
use headers::HeaderMapExt;
use reqwest::header::{AUTHORIZATION, WWW_AUTHENTICATE};
use reqwest::{Client, StatusCode};
use tokio_util::sync::CancellationToken;

use super::{Action, HlsClient, Output, ResourceId};

/// Default [`TokioClientConfig::max_body_bytes`].
pub const DEFAULT_MAX_BODY_BYTES: usize = 256 * 1024 * 1024;

/// How many rounds in a row may pass without yielding an [`Output`] — each a
/// rejected response, or the core having nothing queued — before
/// [`TokioClient::next_output`] gives up with an error.
const MAX_ROUNDS_WITHOUT_OUTPUT: u32 = 16;

/// Errors from the tokio IO adapter itself — distinct from
/// [`crate::client::Error`], the sans-IO core's own parse/demux error type
/// (wrapped here via
/// [`TokioError::Client`]).
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum TokioError {
    /// The underlying HTTP request failed outright (connect/timeout/TLS/
    /// transport error) — never a non-2xx response, see [`Self::Status`].
    #[error("HTTP request to {url} failed: {source}")]
    Http {
        /// The request URL.
        url: String,
        /// The underlying reqwest error.
        #[source]
        source: reqwest::Error,
    },
    /// The server returned a non-success HTTP status.
    #[error("HTTP {status} fetching {url}")]
    Status {
        /// The request URL.
        url: String,
        /// The response status.
        status: reqwest::StatusCode,
    },
    /// The sans-IO core rejected the fetched playlist/resource — see
    /// [`crate::client::Error`] for the underlying reason.
    #[error(transparent)]
    Client(#[from] super::Error),
    /// A `401`'s `WWW-Authenticate` challenge could not be parsed, or no
    /// `Authorization` response could be computed from it (e.g. an
    /// unsupported Digest `algorithm`/`qop`) — see [`broadcast_auth::Error`].
    #[error("auth challenge/response failed: {0}")]
    Auth(#[from] broadcast_auth::Error),
    /// Building the `Range` header for `url` would overflow `u64`
    /// (`offset + (length - 1)`, RFC 8216bis §4.4.4.9). In normal operation
    /// this never fires: [`crate::client::HlsClient`]'s internal
    /// `resolve_byte_range` already rejects any byte range whose
    /// `offset + length` overflows before it ever reaches this adapter (see
    /// `crate::client::Error::ByteRangeOverflow`) — this variant is
    /// defense-in-depth against a future/foreign caller constructing a
    /// `(u64, u64)` byte range directly, bypassing the sans-IO core.
    #[error("byte range for {url} overflows u64: offset {offset} + length {length}")]
    ByteRangeOverflow {
        /// The request URL the range applies to.
        url: String,
        /// The range's starting offset.
        offset: u64,
        /// The range's length.
        length: u64,
    },
    /// The sans-IO core had nothing queued to do and the stream had not ended.
    /// In normal operation `on_playlist` always queues a reload (or a wait)
    /// before returning, so this signals a core bug rather than a network
    /// condition; it replaces a 10 ms defensive sleep that used to hide it.
    #[error("the HLS core queued no action and the stream has not ended")]
    Stalled,
    /// A response body exceeded [`TokioClientConfig::max_body_bytes`]; the
    /// read was abandoned.
    #[error("response body of {url} exceeds {limit} bytes")]
    BodyTooLarge {
        /// The request URL.
        url: String,
        /// The configured limit.
        limit: usize,
    },
    /// [`TokioClientConfig::cancel`] fired while a request was in flight.
    /// Internal: [`TokioClient::next_output`] converts it to `Ok(None)`.
    #[doc(hidden)]
    #[error("cancelled")]
    Cancelled,
}

/// Tunables for [`TokioClient`]. [`Default`] gives sane values for a
/// well-behaved LL-HLS origin reachable over a real (or loopback) network.
#[derive(Debug, Clone)]
pub struct TokioClientConfig {
    /// Per-request timeout for a plain (non-blocking) playlist GET, or a
    /// resource (init/part/segment) GET.
    pub request_timeout: Duration,
    /// Per-request timeout for a **blocking** Playlist Reload
    /// (`_HLS_msn`/`_HLS_part`, RFC 8216bis §6.2.5.2) — must exceed the
    /// origin's own blocking hold time (e.g. `multimux`'s `LlHlsOutput` caps
    /// at 5s) with headroom, since the origin is expected to hold the
    /// response open until new content exists or its own cap elapses.
    pub blocking_timeout: Duration,
    /// How many times a **resource** fetch is retried (capped exponential
    /// backoff) before the adapter gives up on that specific fetch and moves
    /// on (see the module docs' "Error recovery" section). A playlist reload
    /// is never subject to this cap — it retries indefinitely.
    pub max_resource_retries: u32,
    /// Initial backoff between retry attempts (`backon`'s minimum delay);
    /// doubles per attempt up to [`Self::max_retry_backoff`].
    pub retry_backoff: Duration,
    /// Ceiling the doubled [`Self::retry_backoff`] is capped at.
    pub max_retry_backoff: Duration,
    /// Optional auth attached to every request to the playlist URL's origin
    /// (see the module docs).
    pub auth: Option<Credentials>,
    /// TCP connect timeout (reqwest's `connect_timeout`). Default 10 s.
    pub connect_timeout: Duration,
    /// Add `backon` jitter to every retry delay (each delay `d` becomes a
    /// random value in `[d, 2d)`, clamped to [`Self::max_retry_backoff`], which
    /// stays a hard maximum). Default `true`.
    pub jitter: bool,
    /// Cancelling this token makes [`TokioClient::next_output`] abandon its
    /// in-flight request or backoff sleep and return `Ok(None)`.
    pub cancel: CancellationToken,
    /// Largest response body (playlist, key, init, part or segment) the
    /// adapter will read; a larger one is abandoned mid-read with
    /// [`TokioError::BodyTooLarge`]. Default [`DEFAULT_MAX_BODY_BYTES`].
    pub max_body_bytes: usize,
}

impl TokioClientConfig {
    /// Set [`Self::auth`].
    #[must_use]
    pub fn with_auth(mut self, auth: Credentials) -> Self {
        self.auth = Some(auth);
        self
    }

    /// Set [`Self::cancel`].
    #[must_use]
    pub fn with_cancel(mut self, cancel: CancellationToken) -> Self {
        self.cancel = cancel;
        self
    }

    /// Set [`Self::jitter`].
    #[must_use]
    pub fn with_jitter(mut self, on: bool) -> Self {
        self.jitter = on;
        self
    }
}

impl Default for TokioClientConfig {
    fn default() -> Self {
        Self {
            request_timeout: Duration::from_secs(5),
            blocking_timeout: Duration::from_secs(10),
            max_resource_retries: 3,
            retry_backoff: Duration::from_millis(200),
            max_retry_backoff: Duration::from_secs(2),
            auth: None,
            connect_timeout: Duration::from_secs(10),
            jitter: true,
            cancel: CancellationToken::new(),
            max_body_bytes: DEFAULT_MAX_BODY_BYTES,
        }
    }
}

/// Diagnostic counters for what [`TokioClient`] has actually done —
/// distinguishing "parsed an LL-HLS tag" from "acted on it", the same bar
/// this crate's own acceptance tests hold the adapter to (issue #717's
/// "blocking-reload + preload-hint prefetch actually exercised" acceptance
/// item). Also useful to a real caller for observability.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TokioClientStats {
    /// Playlist GETs performed (blocking + non-blocking).
    pub playlist_fetches: u64,
    /// Of those, how many carried `_HLS_msn`/`_HLS_part` — a Blocking
    /// Playlist Reload (RFC 8216bis §6.2.5.2).
    pub blocking_reloads: u64,
    /// Resource (init/part/segment) GETs performed.
    pub resource_fetches: u64,
    /// Of those, how many were for the exact URL most recently named by the
    /// playlist's `#EXT-X-PRELOAD-HINT` (RFC 8216bis §4.4.5.3) — i.e.
    /// fetched ahead of that resource's own numbered (`#EXT-X-PART`)
    /// appearance, not merely alongside it.
    pub preload_hint_resource_fetches: u64,
    /// Responses the sans-IO core rejected (wrong key length, undecryptable
    /// or undemuxable segment, unmergeable delta, unsupported encryption, ...);
    /// each is a per-request failure that is retried, not a fatal error.
    pub rejected_responses: u64,
}

/// An async shell driving [`crate::client::HlsClient`] over real HTTP.
///
/// ```no_run
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// use hls_runtime::client::tokio_client::TokioClient;
///
/// let mut client = TokioClient::new("http://127.0.0.1:8080/live/media.m3u8")?;
/// while let Some(output) = client.next_output().await? {
///     // Hand `output` (Init / Samples / Discontinuity / EndOfStream) to a decoder.
///     let _ = output;
/// }
/// # Ok(())
/// # }
/// ```
pub struct TokioClient {
    core: HlsClient,
    http: Client,
    config: TokioClientConfig,
    playlist_url: String,
    /// `playlist_url` parsed, for the same-origin check on credentials;
    /// `None` if it does not parse, in which case no credentials are sent.
    playlist_origin_url: Option<reqwest::Url>,
    stats: TokioClientStats,
    last_preload_hint_url: Option<String>,
    ended: bool,
    /// Cached from the most recent Digest `401` challenge/response (see the
    /// module docs' "Auth" section) — `None` until the first Digest
    /// challenge is answered, or when [`TokioClientConfig::auth`] isn't
    /// [`Credentials::Digest`]. Applied preemptively to every subsequent
    /// request once set, advancing `nc` (RFC 7616 §3.3) each time.
    digest_authenticator: Option<Authenticator>,
    last_rejection: Option<super::Error>,
    rounds_without_output: u32,
}

impl TokioClient {
    /// Build a client for the Media Playlist at `playlist_url`, with
    /// [`TokioClientConfig::default`] tunables.
    ///
    /// # Errors
    /// Only if the underlying `reqwest::Client` fails to build (e.g. TLS
    /// backend initialisation failure) — not a network error, since no
    /// request has been made yet.
    pub fn new(playlist_url: impl Into<String>) -> Result<Self, TokioError> {
        Self::with_config(playlist_url, TokioClientConfig::default())
    }

    /// Build a client with explicit [`TokioClientConfig`] tunables.
    ///
    /// # Errors
    /// See [`Self::new`].
    pub fn with_config(
        playlist_url: impl Into<String>,
        config: TokioClientConfig,
    ) -> Result<Self, TokioError> {
        let playlist_url = playlist_url.into();
        let http = Client::builder()
            .connect_timeout(config.connect_timeout)
            .build()
            .map_err(|source| TokioError::Http {
                url: playlist_url.clone(),
                source,
            })?;
        Ok(Self {
            core: HlsClient::new(playlist_url.clone()),
            http,
            config,
            playlist_origin_url: reqwest::Url::parse(&playlist_url).ok(),
            playlist_url,
            stats: TokioClientStats::default(),
            last_preload_hint_url: None,
            ended: false,
            digest_authenticator: None,
            last_rejection: None,
            rounds_without_output: 0,
        })
    }

    /// Diagnostic counters for requests actually made so far — see
    /// [`TokioClientStats`].
    pub fn stats(&self) -> TokioClientStats {
        self.stats
    }

    /// The most recent response the sans-IO core rejected, which the adapter
    /// retried instead of failing the session — see
    /// [`TokioClientStats::rejected_responses`].
    pub fn last_rejection(&self) -> Option<&super::Error> {
        self.last_rejection.as_ref()
    }

    /// Drive the sans-IO core — performing whatever HTTP its next
    /// [`Action`] needs, retrying transient failures per
    /// [`TokioClientConfig`] — until at least one [`Output`] is available.
    ///
    /// Returns `Ok(None)` once, right after [`Output::EndOfStream`] has
    /// already been yielded by a previous call, signalling the caller to
    /// stop polling. A live stream that never sends `#EXT-X-ENDLIST` simply
    /// never returns `Ok(None)` — the caller drives this in a loop (or a
    /// `tokio::select!` alongside its own shutdown signal) for as long as it
    /// wants to keep playing.
    ///
    /// # Errors
    /// [`TokioError::Client`] only when the sans-IO core has rejected
    /// responses for sixteen rounds in a row without any [`Output`] in
    /// between; a single rejected response is retried (see the module docs).
    /// [`TokioError::Stalled`] on the same bound when the core has nothing to
    /// do. Resource fetch failures are retried/recovered internally and never
    /// surface here; a playlist fetch failure retries indefinitely rather than
    /// ever returning an [`TokioError::Http`]/[`TokioError::Status`] — see the
    /// module docs' "Error recovery" section.
    pub async fn next_output(&mut self) -> Result<Option<Output>, TokioError> {
        loop {
            if self.config.cancel.is_cancelled() {
                return Ok(None);
            }
            if let Some(out) = self.core.next_output() {
                if matches!(out, Output::EndOfStream) {
                    self.ended = true;
                }
                self.rounds_without_output = 0;
                return Ok(Some(out));
            }
            if self.ended {
                return Ok(None);
            }

            match self.core.poll() {
                Some(action @ Action::FetchPlaylist { .. }) => {
                    let request_url = action
                        .playlist_request_url()
                        .expect("Action::FetchPlaylist always has a playlist_request_url");
                    let is_blocking = matches!(
                        action,
                        Action::FetchPlaylist {
                            blocking: Some(_),
                            ..
                        }
                    );
                    let timeout = if is_blocking {
                        self.config.blocking_timeout
                    } else {
                        self.config.request_timeout
                    };
                    let bytes = match self.fetch_playlist_resilient(&request_url, timeout).await {
                        Ok(bytes) => bytes,
                        Err(TokioError::Cancelled) => return Ok(None),
                        Err(other) => return Err(other),
                    };
                    self.stats.playlist_fetches += 1;
                    if is_blocking {
                        self.stats.blocking_reloads += 1;
                    }
                    self.note_preload_hint(&bytes);
                    if let Err(rejected) = self.core.on_playlist(&bytes) {
                        self.core.ensure_playlist_reload();
                        if self.reject(rejected).await? {
                            return Ok(None);
                        }
                    }
                }
                Some(Action::FetchResource {
                    id,
                    url,
                    byte_range,
                }) => match self.fetch_resource_bounded(&url, byte_range).await {
                    Ok(bytes) => {
                        self.stats.resource_fetches += 1;
                        if matches!(id, ResourceId::Part { .. })
                            && self.last_preload_hint_url.as_deref() == Some(url.as_str())
                        {
                            self.stats.preload_hint_resource_fetches += 1;
                        }
                        if let Err(rejected) = self.core.on_resource(id, &bytes)
                            && self.reject(rejected).await?
                        {
                            return Ok(None);
                        }
                    }
                    Err(TokioError::Cancelled) => return Ok(None),
                    Err(_source) => {
                        // Retries exhausted: un-mark as requested so the
                        // next playlist reload naturally re-requests it,
                        // rather than stalling the whole client on one bad
                        // fetch.
                        self.core.on_error(Some(id));
                    }
                },
                Some(Action::FetchKey { id, url }) => {
                    match self.fetch_resource_bounded(&url, None).await {
                        Ok(bytes) => {
                            if let Err(rejected) = self.core.on_key(id, &bytes)
                                && self.reject(rejected).await?
                            {
                                return Ok(None);
                            }
                        }
                        Err(TokioError::Cancelled) => return Ok(None),
                        Err(_source) => self.core.on_key_error(id),
                    }
                }
                Some(Action::WaitMs(ms)) => {
                    if self.sleep_or_cancelled(Duration::from_millis(ms)).await {
                        return Ok(None);
                    }
                }
                // `on_playlist` re-queues a reload (or `WaitMs`) for a live
                // stream; an empty queue here is a VOD whose init or media was
                // lost, which a retry of the init / a plain playlist reload
                // recovers, within a bound.
                None => {
                    if self.core.retry_init() || self.core.ensure_playlist_reload() {
                        if self.idle_round().await? {
                            return Ok(None);
                        }
                    } else {
                        return Err(TokioError::Stalled);
                    }
                }
            }
        }
    }

    /// Count a response the core rejected and back off before the next
    /// request. `Ok(true)` means cancelled; `Err` once rejections have
    /// repeated [`MAX_ROUNDS_WITHOUT_OUTPUT`] times with no output.
    async fn reject(&mut self, rejected: super::Error) -> Result<bool, TokioError> {
        self.stats.rejected_responses += 1;
        if self.rounds_without_output >= MAX_ROUNDS_WITHOUT_OUTPUT {
            return Err(TokioError::Client(rejected));
        }
        self.last_rejection = Some(rejected);
        self.back_off().await
    }

    /// Count a round in which the core had nothing queued and back off.
    async fn idle_round(&mut self) -> Result<bool, TokioError> {
        if self.rounds_without_output >= MAX_ROUNDS_WITHOUT_OUTPUT {
            return Err(TokioError::Stalled);
        }
        self.back_off().await
    }

    async fn back_off(&mut self) -> Result<bool, TokioError> {
        let round = usize::try_from(self.rounds_without_output).unwrap_or(usize::MAX);
        self.rounds_without_output = self.rounds_without_output.saturating_add(1);
        let delay = schedule(&self.config, None)
            .nth(round)
            .unwrap_or(self.config.max_retry_backoff);
        Ok(self.sleep_or_cancelled(delay).await)
    }

    /// Sleeps `d` or until [`TokioClientConfig::cancel`] fires; `true` means
    /// cancelled.
    async fn sleep_or_cancelled(&self, d: Duration) -> bool {
        tokio::select! {
            () = tokio::time::sleep(d) => false,
            () = self.config.cancel.cancelled() => true,
        }
    }

    /// Re-parse a just-fetched playlist purely to note its
    /// `#EXT-X-PRELOAD-HINT` URL for [`TokioClientStats`] bookkeeping — the
    /// sans-IO core does its own, authoritative parse independently inside
    /// [`crate::client::HlsClient::on_playlist`]; this is a deliberate, small,
    /// side-channel duplication (stats only, never fed back into scheduling)
    /// rather than growing an API on the core to expose its parsed state.
    fn note_preload_hint(&mut self, playlist_bytes: &[u8]) {
        self.last_preload_hint_url = core::str::from_utf8(playlist_bytes)
            .ok()
            .and_then(|text| broadcast_hls::MediaPlaylist::parse(text).ok())
            .and_then(|pl| pl.low_latency)
            .and_then(|ll| ll.preload_hint_part)
            .map(|hint| super::url::resolve(&self.playlist_url, &hint));
    }

    /// Applies whatever auth can be attached *before* sending — Basic/Bearer
    /// always (they never need a challenge), plus a cached Digest
    /// [`Authenticator`] once one exists (advancing `nc`). A `Digest`
    /// config with no cached authenticator yet is left unauthenticated for
    /// this attempt; [`Self::fetch_bytes`] answers the resulting `401`.
    fn apply_auth_preemptive(
        &mut self,
        req: reqwest::RequestBuilder,
        method: &str,
        uri: &str,
    ) -> reqwest::RequestBuilder {
        if !self.is_playlist_origin(uri) {
            return req;
        }
        match &self.config.auth {
            Some(Credentials::Basic { username, password }) => {
                req.basic_auth(username, Some(password))
            }
            Some(Credentials::Bearer { token }) => req.bearer_auth(token),
            Some(_) => {
                // `Credentials::Digest` (or any future non_exhaustive
                // variant): no preemptive header without a cached
                // authenticator from a prior challenge.
                if let Some(auth) = self.digest_authenticator.as_mut()
                    && let Ok(value) = auth.authorization(&RequestContext::new(method, uri))
                {
                    return req.header(AUTHORIZATION, value);
                }
                req
            }
            None => req,
        }
    }

    /// Answers a `401` response by computing the `Authorization` value from
    /// its `WWW-Authenticate` challenge (via [`broadcast_auth`]) and
    /// resending once — only when [`TokioClientConfig::auth`] is
    /// [`Credentials::Digest`] (Basic/Bearer are already pre-applied and a
    /// `401` for those means wrong credentials, not a missing challenge
    /// round-trip). The freshly built [`Authenticator`] is cached on
    /// success so later requests apply it preemptively.
    async fn retry_after_unauthorized(
        &mut self,
        method: &str,
        uri: &str,
        req: reqwest::RequestBuilder,
        response: reqwest::Response,
    ) -> Result<reqwest::Response, TokioError> {
        if !self.is_playlist_origin(uri) {
            return Ok(response);
        }
        let Some(creds @ Credentials::Digest { .. }) = self.config.auth.clone() else {
            return Ok(response);
        };
        let Some(challenge) = response
            .headers()
            .get(WWW_AUTHENTICATE)
            .and_then(|v| v.to_str().ok())
            .map(str::to_string)
        else {
            return Ok(response);
        };
        let mut authenticator = Authenticator::from_challenge(&challenge, creds)?;
        let value = authenticator.authorization(&RequestContext::new(method, uri))?;
        self.digest_authenticator = Some(authenticator);
        req.header(AUTHORIZATION, value)
            .send()
            .await
            .map_err(|source| TokioError::Http {
                url: uri.to_string(),
                source,
            })
    }

    /// Whether `url` has the same origin (scheme, host, port — RFC 6454 §4)
    /// as the configured playlist URL; only such requests carry credentials.
    fn is_playlist_origin(&self, url: &str) -> bool {
        match (&self.playlist_origin_url, reqwest::Url::parse(url)) {
            (Some(playlist), Ok(url)) => playlist.origin() == url.origin(),
            _ => false,
        }
    }

    /// [`Self::fetch_bytes_inner`], abandoned the moment
    /// [`TokioClientConfig::cancel`] fires.
    async fn fetch_bytes(
        &mut self,
        url: &str,
        byte_range: Option<(u64, u64)>,
        timeout: Duration,
    ) -> Result<Vec<u8>, TokioError> {
        let cancel = self.config.cancel.clone();
        tokio::select! {
            r = self.fetch_bytes_inner(url, byte_range, timeout) => r,
            () = cancel.cancelled() => Err(TokioError::Cancelled),
        }
    }

    async fn fetch_bytes_inner(
        &mut self,
        url: &str,
        byte_range: Option<(u64, u64)>,
        timeout: Duration,
    ) -> Result<Vec<u8>, TokioError> {
        let req = self.apply_auth_preemptive(
            build_request(&self.http, url, byte_range, timeout)?,
            "GET",
            url,
        );
        let resp = req.send().await.map_err(|source| TokioError::Http {
            url: url.to_string(),
            source,
        })?;

        let resp = if resp.status() == StatusCode::UNAUTHORIZED {
            let retry_req = build_request(&self.http, url, byte_range, timeout)?;
            self.retry_after_unauthorized("GET", url, retry_req, resp)
                .await?
        } else {
            resp
        };

        let status = resp.status();
        if !status.is_success() {
            return Err(TokioError::Status {
                url: url.to_string(),
                status,
            });
        }
        let limit = self.config.max_body_bytes;
        let too_large = || TokioError::BodyTooLarge {
            url: url.to_string(),
            limit,
        };
        if resp
            .content_length()
            .is_some_and(|len| len > u64::try_from(limit).unwrap_or(u64::MAX))
        {
            return Err(too_large());
        }
        let mut resp = resp;
        let mut body = Vec::new();
        while let Some(chunk) = resp.chunk().await.map_err(|source| TokioError::Http {
            url: url.to_string(),
            source,
        })? {
            if body.len().saturating_add(chunk.len()) > limit {
                return Err(too_large());
            }
            body.extend_from_slice(&chunk);
        }
        Ok(body)
    }

    /// Retry a playlist fetch indefinitely (capped exponential backoff) —
    /// see the module docs' "Error recovery" section for why a playlist
    /// reload, unlike a resource fetch, has no bounded-retry fallback.
    ///
    /// A blocking or delta reload (`url` carries `_HLS_msn`/`_HLS_part`/
    /// `_HLS_skip`) answered with a `4xx` that repeating cannot change — an
    /// origin that restarted below the requested Media Sequence Number
    /// answers `400` (RFC 8216bis §6.2.5.2) — is retried **once** at once as
    /// a plain GET of the playlist URL instead of being repeated forever
    /// (audit r09-C3, issue #1031); the plain URL then keeps the usual
    /// backoff.
    async fn fetch_playlist_resilient(
        &mut self,
        url: &str,
        timeout: Duration,
    ) -> Result<Vec<u8>, TokioError> {
        let mut delays = schedule(&self.config, None);
        let mut current = url.to_string();
        loop {
            match self.fetch_bytes(&current, None, timeout).await {
                Ok(bytes) => return Ok(bytes),
                Err(TokioError::Cancelled) => return Err(TokioError::Cancelled),
                Err(source) if current != self.playlist_url && !is_retryable(&source) => {
                    current.clone_from(&self.playlist_url);
                }
                Err(_source) => {
                    // An unbounded schedule never ends; the fallback only
                    // keeps this total.
                    let d = delays.next().unwrap_or(self.config.max_retry_backoff);
                    if self.sleep_or_cancelled(d).await {
                        return Err(TokioError::Cancelled);
                    }
                }
            }
        }
    }

    /// Retry a resource fetch up to [`TokioClientConfig::max_resource_retries`]
    /// times (capped exponential backoff), then give up.
    async fn fetch_resource_bounded(
        &mut self,
        url: &str,
        byte_range: Option<(u64, u64)>,
    ) -> Result<Vec<u8>, TokioError> {
        let attempts = usize::try_from(self.config.max_resource_retries.max(1)).unwrap_or(1);
        // `attempts` total attempts means `attempts - 1` sleeps between them.
        let mut delays = schedule(&self.config, Some(attempts - 1));
        let mut last_err = None;
        for _ in 0..attempts {
            match self
                .fetch_bytes(url, byte_range, self.config.request_timeout)
                .await
            {
                Ok(bytes) => return Ok(bytes),
                Err(source @ TokioError::Cancelled) => return Err(source),
                Err(source) if !is_retryable(&source) => return Err(source),
                Err(source) => {
                    last_err = Some(source);
                    // No sleep after the last attempt.
                    let Some(d) = delays.next() else { break };
                    if self.sleep_or_cancelled(d).await {
                        return Err(TokioError::Cancelled);
                    }
                }
            }
        }
        Err(last_err.expect("loop runs at least once (attempts >= 1)"))
    }
}

/// The retry delays for `config`: `backon` exponential (minimum
/// `retry_backoff`, doubling, capped at `max_retry_backoff`), optionally
/// jittered; `max_sleeps = None` never ends (playlist reloads).
fn schedule(
    config: &TokioClientConfig,
    max_sleeps: Option<usize>,
) -> impl Iterator<Item = Duration> + use<> {
    let mut builder = ExponentialBuilder::default()
        .with_min_delay(config.retry_backoff)
        .with_max_delay(config.max_retry_backoff);
    if config.jitter {
        builder = builder.with_jitter();
    }
    builder = match max_sleeps {
        Some(n) => builder.with_max_times(n),
        None => builder.without_max_times(),
    };
    // `backon` adds jitter AFTER its own max-delay clamp, so an unclamped
    // delay can reach ~2x the cap: clamp again, making
    // `max_retry_backoff` a hard maximum for every delay.
    let cap = config.max_retry_backoff;
    builder.build().map(move |d| d.min(cap))
}

/// Whether a failed resource fetch is worth retrying: everything except a
/// `4xx` that repeating the same request cannot change (`408 Request Timeout`
/// and `429 Too Many Requests` are transient, RFC 9110 §15.5.9 / RFC 6585 §4).
fn is_retryable(err: &TokioError) -> bool {
    match err {
        TokioError::BodyTooLarge { .. } => false,
        TokioError::Status { status, .. } => {
            !status.is_client_error()
                || *status == StatusCode::REQUEST_TIMEOUT
                || *status == StatusCode::TOO_MANY_REQUESTS
        }
        _ => true,
    }
}

/// Builds a plain (unauthenticated) GET request for `url`, with an optional
/// `Range` header (RFC 8216bis §4.4.4.9 partial-part byte ranges) — factored
/// out so [`TokioClient::fetch_bytes`] can build a fresh, independent request
/// both for the first attempt and for the post-`401` Digest retry (a
/// `reqwest::RequestBuilder` is consumed by `.send()`, so the retry can't
/// reuse the first one).
fn build_request(
    client: &Client,
    url: &str,
    byte_range: Option<(u64, u64)>,
    timeout: Duration,
) -> Result<reqwest::RequestBuilder, TokioError> {
    let mut req = client.get(url).timeout(timeout);
    if let Some((offset, length)) = byte_range {
        // `saturating_sub` here only guards the `length == 0` underflow
        // case (`0 - 1`) — it must never silently saturate the *addition*
        // below, since a saturated `end` would produce a `Range:` header
        // for the wrong bytes rather than failing this one request. See
        // `TokioError::ByteRangeOverflow`.
        let overflow = || TokioError::ByteRangeOverflow {
            url: url.to_string(),
            offset,
            length,
        };
        let end = offset
            .checked_add(length.saturating_sub(1))
            .ok_or_else(overflow)?;
        let range = headers::Range::bytes(offset..=end).map_err(|_| overflow())?;
        let mut map = reqwest::header::HeaderMap::new();
        map.typed_insert(range);
        req = req.headers(map);
    }
    Ok(req)
}

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

    // Biting test for the u64-overflow defect (issue's `tokio_client.rs:497`
    // finding): `build_request`'s old `offset + length.saturating_sub(1)`
    // guarded only the `-1` underflow, not the addition itself. A byte range
    // whose `offset` is near `u64::MAX` — reachable via a malicious/corrupt
    // playlist's `BYTERANGE`, or a long-lived omitted-offset cursor walked
    // up by `crate::client::HlsClient` — must be rejected, not panic (debug)
    // or silently wrap to a bogus `Range:` header naming the wrong bytes
    // (release).
    //
    // MUTATION VERIFIED: reverting to the original
    // `let end = offset + length.saturating_sub(1);` (dropping
    // `checked_add`/`Err`) makes this test fail: the debug build panics with
    // "attempt to add with overflow" inside `build_request` before
    // `expect_err` ever runs (confirmed by running it), rather than
    // returning `TokioError::ByteRangeOverflow`. Recompiled and re-ran to
    // observe that exact panic, then restored the checked_add.
    #[test]
    fn build_request_rejects_a_range_whose_end_overflows_u64() {
        let client = Client::new();
        let err = build_request(
            &client,
            "http://example.com/seg.m4s",
            Some((u64::MAX - 1, 5)),
            Duration::from_secs(5),
        )
        .expect_err("offset near u64::MAX + length must overflow and be rejected");
        assert!(
            matches!(
                err,
                TokioError::ByteRangeOverflow {
                    offset,
                    length: 5,
                    ..
                } if offset == u64::MAX - 1
            ),
            "wrong error variant/fields: {err:?}"
        );
    }

    // The flip side: an ordinary, non-overflowing byte range must still
    // produce the expected `Range:` header — the overflow guard must not
    // have broken the common case.
    #[test]
    fn build_request_sets_range_header_for_a_normal_byte_range() {
        let client = Client::new();
        let req = build_request(
            &client,
            "http://example.com/seg.m4s",
            Some((10, 20)),
            Duration::from_secs(5),
        )
        .expect("a normal byte range must succeed");
        let built = req.build().expect("request builds");
        let header = built
            .headers()
            .get(reqwest::header::RANGE)
            .expect("Range header must be present");
        assert_eq!(header, "bytes=10-29");
    }

    #[test]
    fn retry_schedule_without_jitter_is_exponential_and_capped() {
        let cfg = TokioClientConfig::default().with_jitter(false);
        let got: Vec<_> = schedule(&cfg, None).take(6).collect();
        let want = [200, 400, 800, 1600, 2000, 2000].map(Duration::from_millis);
        // backon multiplies in `f32` (its factor is `f32`), so a doubled delay
        // can be a few nanoseconds off the ideal; the cap is exact.
        for (i, (g, w)) in got.iter().zip(want).enumerate() {
            let diff = g.abs_diff(w);
            assert!(
                diff < Duration::from_micros(1),
                "delay {i} = {g:?}, want {w:?} (min 200 ms, doubling, capped at 2 s)"
            );
        }
        assert_eq!(got[4], Duration::from_secs(2), "capped exactly");
        assert_eq!(got[5], Duration::from_secs(2), "stays at the cap");
    }

    #[test]
    fn retry_schedule_jitter_stays_within_one_extra_delay_and_never_exceeds_the_cap() {
        let cfg = TokioClientConfig::default();
        let cap = cfg.max_retry_backoff;
        // Many samples: jitter is random, a single draw proves little.
        for _ in 0..500 {
            for (i, got) in schedule(&cfg, None).take(8).enumerate() {
                let base = Duration::from_millis(200u64 << i.min(4)).min(cap);
                assert!(got >= base, "delay {i} = {got:?} below its base {base:?}");
                assert!(
                    got <= cap,
                    "delay {i} = {got:?} exceeds max_retry_backoff {cap:?}"
                );
                assert!(
                    got < base * 2 || got == cap,
                    "delay {i} = {got:?} outside [{base:?}, {:?})",
                    base * 2
                );
            }
        }
        assert_eq!(schedule(&cfg, Some(2)).count(), 2, "3 attempts = 2 sleeps");
    }

    #[test]
    fn client_errors_other_than_408_and_429_are_not_retried() {
        let status = |status| TokioError::Status {
            url: "http://h/x".into(),
            status,
        };
        assert!(!is_retryable(&status(StatusCode::NOT_FOUND)));
        assert!(!is_retryable(&status(StatusCode::FORBIDDEN)));
        assert!(is_retryable(&status(StatusCode::REQUEST_TIMEOUT)));
        assert!(is_retryable(&status(StatusCode::TOO_MANY_REQUESTS)));
        assert!(is_retryable(&status(StatusCode::SERVICE_UNAVAILABLE)));
    }

    #[test]
    fn credentials_origin_check_compares_scheme_host_and_port() {
        let client = TokioClient::new("http://origin.example:8080/live/index.m3u8").unwrap();
        assert!(client.is_playlist_origin("http://origin.example:8080/seg0.ts"));
        assert!(client.is_playlist_origin("HTTP://Origin.Example:8080/a/b.ts"));
        assert!(!client.is_playlist_origin("http://other.example:8080/seg0.ts"));
        assert!(!client.is_playlist_origin("https://origin.example:8080/seg0.ts"));
        assert!(!client.is_playlist_origin("http://origin.example/seg0.ts"));
        assert!(!client.is_playlist_origin("not a url"));
        // A default port written out is the same origin.
        let client = TokioClient::new("https://origin.example/live/index.m3u8").unwrap();
        assert!(client.is_playlist_origin("https://origin.example:443/seg0.ts"));
    }

    // Security-blocker regression (pre-release audit): `TokioClientConfig`
    // derives `Debug` and embeds `Option<Credentials>` directly (`auth`) — it
    // must inherit `Credentials`'s redacting `Debug`, never the raw secret.
    #[test]
    fn tokio_client_config_debug_does_not_leak_embedded_credentials_secret() {
        let config = TokioClientConfig {
            auth: Some(Credentials::new("admin", "a-very-secret-password")),
            ..TokioClientConfig::default()
        };
        let debug = format!("{config:?}");
        assert!(
            !debug.contains("a-very-secret-password"),
            "leaked via TokioClientConfig Debug: {debug}"
        );
        assert!(debug.contains("***"), "expected redaction marker: {debug}");
    }
}