decdn-cache 0.1.0

//! Origin pull-through retry policy with exponential backoff and jitter (#285).
//!
//! Wraps a single pull-through attempt in a bounded retry loop. The adapter
//! (or the engine's per-attempt closure) classifies each failure as
//! [`OriginPullError::Transient`] or [`OriginPullError::Permanent`]; this
//! module only decides *how many* transient retries to run and *how long*
//! to sleep between them.
//!
//! ## Body-phase retry (#519)
//!
//! The loop also covers body-phase failures via two paths, both gated on
//! the same `max_retries` budget:
//!
//! - **Buffer-then-commit (small blobs):** when the adapter advertises a
//!   `size_hint` at or below [`RetryPolicy::buffered_max_bytes`] (default
//!   4 MiB), the engine's per-attempt closure drains the stream into a
//!   bounded `BytesMut` before handing it to iroh-blobs. Drain errors are
//!   classified via the internal `classify_io_error` helper (which
//!   recognises typed `BlobTooLargeMarker` and `OriginError::*` as
//!   Permanent, treats `io::ErrorKind::{ConnectionReset, TimedOut,
//!   UnexpectedEof, …}` and the catch-all `Other` as Transient) and
//!   re-fed to the loop, so small blobs get mid-stream retries at a
//!   bounded memory cost.
//! - **Abort + restart (large blobs):** the engine's per-attempt closure
//!   commits to `iroh-blobs::add_stream` directly. On mid-stream
//!   `io::Error`, the partial `TempTag` is dropped (iroh-blobs GC
//!   reclaims the bytes at `cache.gc_interval_sec` cadence) and the
//!   error is classified, restarting the loop from the headers phase.
//!   Memory stays bounded; disk amplification is
//!   `(1 + max_retries) * max_blob_bytes` worst case until GC.
//!
//! ## Coalescing interaction and the retry-amplification bound (#1615)
//!
//! [`crate::CacheEngine::get`] coalesces concurrent misses for the same hash
//! through a single owner task (#305). The retry loop runs *inside* that owner;
//! waiters block on the shared `Notify` until the owner either succeeds or
//! exhausts. After an owner exhausts, one waiter re-elects itself as the next
//! owner and runs its own full retry budget — so, taken alone, a burst of `N`
//! concurrent waiters on a down origin could drive up to `N` *sequential* full
//! retry budgets.
//!
//! That amplification is bounded, but by the per-origin circuit-breaker
//! ([`crate::circuit_breaker`], #963) rather than by anything in this module.
//! Each budget-exhaustion against a down origin records one `Unavailable`
//! outcome; after `failure_threshold` consecutive exhaustions the breaker
//! trips OPEN and every later waiter short-circuits *before* the retry loop
//! runs. The worst case is therefore `failure_threshold` full budgets per
//! origin per cooldown window — a constant, independent of `N`. The
//! integration test `coalesced_owner_exhaustion_bounded_by_circuit_breaker`
//! pins this, and also pins the O(`N`) fallback that returns when an operator
//! disables the breaker.
//!
//! A tighter per-hash bound — the owner's failure *shared* to all waiters so
//! not even `failure_threshold` budgets run — is deliberately NOT implemented.
//! Coalesced waiters do not all have the same origin reach: a `populate_local`
//! owner (local origins only, #1116) and a `get` / `populate` waiter (full
//! chain including the paid `Peer` origin) can coalesce on one hash, and
//! re-election is load-bearing there — a local-only owner's `NotFound` must
//! not short-circuit a full-chain waiter that could still succeed via `Peer`.
//! Sharing the owner's failure blindly would turn that into a wrong answer;
//! correct sharing would have to be origin-reach-aware. The circuit-breaker
//! already sheds the outage load without that complexity, so the shared budget
//! is not worth its risk on this hot, cancellation-sensitive path.

use std::future::Future;
use std::io;
use std::sync::Arc;
use std::time::Duration;

use bytes::{Bytes, BytesMut};
use decdn_config_types::RetryPolicy;
use futures_util::StreamExt;
use iroh_blobs::Hash;

use crate::error::{OriginError, OriginPullError};
use crate::metrics::CacheMetrics;
use crate::origin::{BlobTooLargeMarker, OriginByteStream};

/// Compute the backoff delay before the `attempt`-th retry (0-indexed:
/// `attempt = 0` is the first retry, immediately after the initial
/// failure). The schedule is `min(initial * 2^attempt, max)` with
/// equal-jitter applied.
///
/// Lives here (not on [`RetryPolicy`], which moved to the
/// `decdn-config-types` leaf crate) because it needs `rand` for the
/// jitter sample — keeping `rand` out of the leaf is the point of #578.
///
/// Saturating math throughout: `checked_shl` handles huge `attempt`
/// values without panicking on `1u64 << 64`, and the eventual
/// `min(_, max_backoff_ms)` collapses everything to the cap.
fn delay_for(policy: RetryPolicy, attempt: u32) -> Duration {
    let multiplier = 1u64.checked_shl(attempt).unwrap_or(u64::MAX);
    let exp = policy.initial_backoff_ms.saturating_mul(multiplier);
    let base = exp.min(policy.max_backoff_ms);

    // Equal jitter: factor lies in [1 - r/2, 1 + r/2).
    let r = policy.jitter_ratio.clamp(0.0, 1.0);
    let factor = 1.0_f64 - (r / 2.0) + rand::random::<f64>() * r;
    #[allow(clippy::cast_precision_loss)]
    let base_f = base as f64;
    #[allow(clippy::cast_precision_loss)]
    let upper = u64::MAX as f64;
    let scaled = (base_f * factor).clamp(0.0, upper);
    #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
    let ms = scaled as u64;
    Duration::from_millis(ms)
}

/// Whether a terminal pull-through failure left the origin looking
/// *unreachable* (a transient that exhausted the retry budget) or merely
/// unable to serve *this* object (an immediate permanent — 4xx, decode,
/// cap). Used by the per-origin circuit-breaker (#963) to decide whether
/// an attempt counts toward the trip threshold; the plain `Result` the
/// engine folds this into collapses both into `OriginPullError::Permanent`,
/// erasing the distinction the breaker needs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TerminalFailure {
    /// The terminal error was a transient that exhausted `max_retries`
    /// (or a `max_retries = 0` single transient). The origin looks
    /// unavailable.
    TransientExhausted,
    /// The terminal error was an immediate `Permanent` — the origin
    /// responded but can't serve this object. Not an outage.
    Permanent,
}

/// The retry loop, with a terminal-failure classification for the
/// circuit-breaker (#963). On `Ok`, the second tuple element is `None`.
/// On `Err`, it carries whether the failure was a transient-exhaustion
/// (origin-unavailable) or an immediate permanent (per-object).
pub(crate) async fn run_with_retry_classified<F, Fut, T>(
    policy: RetryPolicy,
    metrics: Option<&Arc<CacheMetrics>>,
    hash: Hash,
    mut attempt_fn: F,
) -> (Result<T, OriginPullError>, Option<TerminalFailure>)
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, OriginPullError>>,
{
    let mut attempt: u32 = 0;
    loop {
        match attempt_fn().await {
            Ok(value) => return (Ok(value), None),
            Err(OriginPullError::Permanent(e)) => {
                return (
                    Err(OriginPullError::Permanent(e)),
                    Some(TerminalFailure::Permanent),
                );
            }
            Err(OriginPullError::Transient(e)) => {
                if attempt >= policy.max_retries {
                    // Only count exhaustion when at least one retry actually
                    // fired. With `max_retries = 0` (operator opted out)
                    // a single transient failure is just a failure, not a
                    // burned-out retry budget — bumping the counter would
                    // ruin alerts that page on actual exhaustion.
                    if attempt > 0
                        && let Some(m) = metrics
                    {
                        m.origin_retry_exhausted.inc();
                    }
                    tracing::error!(
                        %hash,
                        attempts = attempt.saturating_add(1),
                        max_retries = policy.max_retries,
                        error = %format_args!("{e:#}"),
                        "origin fetch exhausted retry budget; surfacing as permanent",
                    );
                    return (
                        Err(OriginPullError::Permanent(e)),
                        Some(TerminalFailure::TransientExhausted),
                    );
                }
                let sleep = delay_for(policy, attempt);
                let sleep_ms = u64::try_from(sleep.as_millis()).unwrap_or(u64::MAX);
                tracing::warn!(
                    %hash,
                    attempt = attempt + 1,
                    of = policy.max_retries,
                    sleep_ms,
                    error = %format_args!("{e:#}"),
                    "origin fetch transient failure; retrying after backoff",
                );
                tokio::time::sleep(sleep).await;
                attempt = attempt.saturating_add(1);
            }
        }
    }
}

/// True when the origin's advertised `size_hint` is small enough to
/// buffer in memory for body-phase retry classification. Unknown
/// (`None`) hint falls through to streaming — buffering an unknown-size
/// body would let a lying origin overrun the operator's memory budget.
pub(crate) const fn should_buffer(size_hint: Option<u64>, buffered_max_bytes: u64) -> bool {
    if buffered_max_bytes == 0 {
        return false;
    }
    match size_hint {
        Some(n) => n <= buffered_max_bytes,
        None => false,
    }
}

/// Drain `stream` into a contiguous `Bytes`, capping the buffer at
/// `cap` bytes. Mirrors [`crate::origin::OriginFetch::collect_to_bytes`] but adds:
///
/// - A running cap: chunks past `cap` produce an `io::Error` whose
///   inner is a [`BlobTooLargeMarker`], so the caller can downcast and
///   surface `CacheError::BlobTooLarge` instead of a generic origin
///   error.
/// - Per-chunk `pull_through_bytes` metric bumps that mirror the streaming
///   path's egress accounting (every byte the origin sent is
///   billed, even on a drain that ultimately fails).
pub(crate) async fn drain_to_bytes(
    mut stream: OriginByteStream,
    cap: u64,
    metrics: Option<&Arc<CacheMetrics>>,
) -> io::Result<Bytes> {
    // Bound the upfront allocation regardless of `cap` so a hostile
    // origin advertising `Content-Length: 10 TiB` (truncated to `cap`
    // by the caller, but still potentially large) can't pre-commit a
    // huge virtual region before the first byte arrives. Matches the
    // 1 MiB initial-capacity hint used in `OriginFetch::collect_to_bytes`.
    const INITIAL_CAP: usize = 1 << 20; // 1 MiB
    let initial = usize::try_from(cap).unwrap_or(INITIAL_CAP).min(INITIAL_CAP);
    let mut buf = BytesMut::with_capacity(initial);
    let mut total: u64 = 0;
    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        if let Some(m) = metrics {
            m.pull_through_bytes
                .inc_by(u64::try_from(chunk.len()).unwrap_or(u64::MAX));
        }
        total = total.saturating_add(chunk.len() as u64);
        if total > cap {
            return Err(io::Error::other(BlobTooLargeMarker { max_bytes: cap }));
        }
        buf.extend_from_slice(&chunk);
    }
    Ok(buf.freeze())
}

/// Classify a body-phase `io::Error` into the retry loop's
/// [`OriginPullError`] taxonomy. Used by both the drain path
/// ([`drain_to_bytes`]) and the streaming path (engine's
/// side-channel reader) so the classification is consistent
/// across the two body-phase entry points.
///
/// The decision tree:
///
/// 1. If the `io::Error` wraps a typed inner via
///    `io::Error::other(typed)`, recover the typed variant first:
///    [`BlobTooLargeMarker`] → `Permanent(BlobTooLarge)`,
///    [`OriginError`] (e.g. `DecompressionFailed`) → `Permanent`.
///    These are deterministic protocol or cap violations — retry
///    won't help and would just waste budget.
///
/// 2. Otherwise dispatch on `io::ErrorKind`. The Transient set:
///    `Interrupted`, `TimedOut`, `WouldBlock`, `ResourceBusy`,
///    `BrokenPipe`, `ConnectionReset`, `ConnectionAborted`,
///    `ConnectionRefused`, `NotConnected`, `UnexpectedEof`, and the
///    catch-all `Other`. `UnexpectedEof` is the load-bearing kind
///    for mid-body resets behind a flaky LB — reqwest surfaces
///    truncated Content-Length responses through it. `Other`
///    defaults to Transient because the HTTP and S3 origin adapters
///    wrap reqwest / SDK body errors via `io::Error::other(...)` —
///    mostly network/body failures (resets, timeouts, truncation), so
///    Transient is the safer default for them.
///    Anything not in the Transient set is classified Permanent
///    (fail-fast over retry-storm for unrecognised modes).
pub(crate) fn classify_io_error(e: io::Error) -> OriginPullError {
    // Peek the inner *without* consuming so we can preserve the
    // original `e` (and its `raw_os_error` / Display) when no typed
    // marker matches. `io::Error::from_raw_os_error` / `last_os_error`
    // produce errors whose `into_inner()` returns `None` — consuming
    // up front would silently strip the OS error code.
    let has_typed_marker = e
        .get_ref()
        .is_some_and(|r| r.is::<BlobTooLargeMarker>() || r.is::<OriginError>());
    if !has_typed_marker {
        let kind = e.kind();
        return classify_by_kind(kind, e);
    }
    // We confirmed a typed marker via `is::<>`; consume and chain
    // `Box::downcast` (returns the original box back through `Err`
    // on a mismatch). Both downcasts can only fail if a third typed
    // variant slipped in between the `get_ref` peek and the consume
    // — defensive fall-back rewraps and routes through `classify_by_kind`.
    let kind = e.kind();
    let Some(inner) = e.into_inner() else {
        // Structurally unreachable: `get_ref` returned `Some` so the
        // Repr is `Custom`, and `Custom`'s `into_inner` always returns
        // `Some`. If std's invariant ever changes, fall through.
        return classify_by_kind(kind, io::Error::from(kind));
    };
    let inner = match inner.downcast::<BlobTooLargeMarker>() {
        Ok(marker) => {
            return OriginPullError::Permanent(anyhow::anyhow!(
                "origin body exceeded max_blob_bytes={}",
                marker.max_bytes
            ));
        }
        Err(b) => b,
    };
    match inner.downcast::<OriginError>() {
        Ok(typed) => OriginPullError::Permanent(anyhow::Error::from(*typed)),
        Err(b) => classify_by_kind(kind, io::Error::new(kind, b)),
    }
}

/// Classify a body-phase failure by `io::ErrorKind` alone. Takes the
/// original `io::Error` so `raw_os_error()` and any Display payload
/// flow through unchanged into the surfaced `anyhow::Error`. Used
/// when the typed-marker downcast didn't fire.
fn classify_by_kind(kind: io::ErrorKind, e: io::Error) -> OriginPullError {
    let err = anyhow::Error::new(e);
    match kind {
        io::ErrorKind::Interrupted
        | io::ErrorKind::TimedOut
        | io::ErrorKind::WouldBlock
        | io::ErrorKind::ResourceBusy
        | io::ErrorKind::BrokenPipe
        | io::ErrorKind::ConnectionReset
        | io::ErrorKind::ConnectionAborted
        | io::ErrorKind::ConnectionRefused
        | io::ErrorKind::NotConnected
        | io::ErrorKind::UnexpectedEof
        | io::ErrorKind::Other => OriginPullError::Transient(err),
        _ => OriginPullError::Permanent(err),
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] // tests
mod tests;