decdn-node 0.1.0

//! Cache-fill tiers for the serve-miss path.

use super::{
    Arc, CacheError, ClientHandler, Duration, FillOutcome, Hash, LaneDeliveryState, LaneKey, Mutex,
};
use tracing::Instrument as _;

impl ClientHandler {
    /// The lane that authorizes this request to make the node spend on a cache
    /// miss, or `None` when it may not. Every fill tier (own-origin, local
    /// populate, window and buffered pull-through, #831) and the pre-spend floor
    /// reservation are gated on it. Authority needs a verified client
    /// binding whose recovered address is the lane's `signer`.
    /// Pool *existence* is public (on-chain `PoolOpened`), so it cannot
    /// authorize spend — only proof of *voucher authority* can, since only the
    /// that signer can produce a voucher this lane will accept. Any capability
    /// holder can sign vouchers the pool accepts, but each on its own lane. An unbound
    /// request, or a binding that does not match that signer, must not make this
    /// node front upstream USDC or its own origin's egress.
    ///
    /// Spend authority is a *signer* question. In the shared-payment-pool model
    /// the lane is keyed by that signer directly, so authority reduces to "does
    /// a lane exist for `(pool_id, bound_signer, this operator)`?" — the seller
    /// resolves the lane from exactly that triple (brief §E1). `lane_key` is
    /// that triple (`None` for an unbound request) and `known_lane` is the lane
    /// `serve_stream` resolved for it, once, after the request's own capability
    /// intake. Authority is derived from that single resolution and never from a
    /// second registry read: a lane a sibling stream registers while this request
    /// is in flight cannot flip the answer between fill tiers (#2315). The
    /// returned pair is what the serve legs take, so a tier that is authorized
    /// always holds the lane it serves on.
    ///
    /// The spend-side origin-blacklist gate (ADR 011) is a separate *funder*
    /// question. ADR 011 sanctions the *money*: a takedown names the address that
    /// funded the delivery (`getPool.owner`), not whichever throwaway key happened
    /// to sign the vouchers. So a blacklisted delegate signer over a clean funder
    /// is not refused, and a clean delegate signer never launders a blacklisted
    /// funder. That gate runs at the serve gate's open-time funder check in
    /// `dispatch.rs`, which precedes every fill tier — a blacklisted funder is
    /// refused there before this authority is consulted, so no fill fronts USDC
    /// or origin egress on a blacklisted funder's behalf.
    pub(super) fn pull_authorized(
        lane_key: Option<LaneKey>,
        known_lane: Option<&Arc<Mutex<LaneDeliveryState>>>,
    ) -> Option<(LaneKey, &Arc<Mutex<LaneDeliveryState>>)> {
        lane_key.zip(known_lane)
    }

    /// Attempt to fill a cache miss by pulling from an upstream node (#831). The
    /// cache engine's `NodeOrigin` (last in the origin chain) does the discovery
    /// → probe → ranked paid pull → populate; here we trigger it via
    /// `cache.populate` (which fills the store WITHOUT returning the blob or
    /// bumping `bytes_returned` — this is an internal fill, not client egress),
    /// bounded by `timeout` so a slow upstream can't pin the delivery path. The
    /// subsequent normal delivery streams the populated bytes from the store and
    /// accounts the served-bytes metrics there. Returns whether the blob is now
    /// present locally, and — when it is not — whether the cause was a clean miss
    /// or a hard fault (#1129).
    pub(super) async fn try_pull_through(&self, hash: Hash, timeout: Duration) -> FillOutcome {
        let span = fill_span("node", hash);
        let outcome = self
            .node_pull_through(hash, timeout)
            .instrument(span.clone())
            .await;
        span.record("outcome", outcome.as_str());
        outcome
    }

    /// The body of [`Self::try_pull_through`], run inside its span.
    async fn node_pull_through(&self, hash: Hash, timeout: Duration) -> FillOutcome {
        match tokio::time::timeout(timeout, self.cache.populate(hash)).await {
            Ok(Ok(())) => FillOutcome::Filled,
            // A clean miss — no origin/provider had it — is the normal
            // unfillable case (`NotFound`/`NoOrigin`); log at debug and move on.
            Ok(Err(e @ (CacheError::NotFound { .. } | CacheError::NoOrigin { .. }))) => {
                tracing::debug!(%hash, error = %e.display_chain(), "node-to-node pull-through found no source");
                FillOutcome::CleanMiss
            }
            // A backend fault — the node is degraded, not empty. Report it as a fault so
            // the caller can refuse `InternalError` if no further tier fills (#1129),
            // rather than reporting a broken origin as a miss.
            //
            // This arm names neither lifetime: the node-origin also surfaces its own
            // buyer-side faults here (a broken signer, an unusable deadline config, an
            // unreadable pool store) as `OriginPullError::Permanent`, and those recur
            // for every hash until an operator acts. Telling an operator to wait for a
            // permanent defect to pass is worse than saying nothing.
            Ok(Err(
                e @ (CacheError::OriginError { .. }
                | CacheError::Store(_)
                | CacheError::Internal(_)),
            )) => {
                self.metrics.node_pull_through_error();
                tracing::warn!(%hash, error = %e.display_chain(), "node-to-node pull-through hit a backend fault");
                FillOutcome::HardFault
            }
            // Everything else (`BlobTooLarge`, `HashMismatch`, `VerifyFailed`,
            // `EvictionLimitExceeded`) is DETERMINISTIC: it will recur on every
            // request for this hash, so it is not evidence the node is degraded.
            // Reporting it as `InternalError` (wire `Declined`) would drop this
            // node for the hash over one bad upstream copy.
            // Meter it (the operator still needs to see it) but let it fall
            // through as a plain miss.
            Ok(Err(e)) => {
                self.metrics.node_pull_through_error();
                tracing::warn!(%hash, error = %e.display_chain(), "node-to-node pull-through hit a permanent cache-engine error");
                FillOutcome::CleanMiss
            }
            Err(_) => self.on_pull_through_timeout(hash, timeout).await,
        }
    }

    /// Attempt to fill a cache miss from the node's OWN configured origins only
    /// (#1116), via [`CacheEngine::populate_local`](super::CacheEngine::populate_local) — which skips the paid `Peer`
    /// node→node origin, so this fronts no upstream USDC. Bounded by `timeout`
    /// like [`Self::try_pull_through`]. A clean miss (`NotFound`/`NoOrigin` — the
    /// operator's origin lacks it, or only a `Peer` origin is configured) is the
    /// normal unfillable case; the caller then falls through to the node→node
    /// paths. A [`FillOutcome::HardFault`] means the operator's OWN origin faulted
    /// (an S3 5xx, an open breaker, an fs I/O error) — the caller may still try a
    /// further tier, but must not let a later clean miss launder the fault into a
    /// signed `NotFound` (#1129). A timeout is a clean miss: nothing is warmed in
    /// the background (#1610 removed the detached warm), so the blob is re-fetched
    /// on the next real client request (see [`Self::on_local_populate_timeout`]).
    pub(super) async fn try_local_populate(&self, hash: Hash, timeout: Duration) -> FillOutcome {
        let span = fill_span("local", hash);
        let outcome = self
            .local_populate(hash, timeout)
            .instrument(span.clone())
            .await;
        span.record("outcome", outcome.as_str());
        outcome
    }

    /// The body of [`Self::try_local_populate`], run inside its span.
    async fn local_populate(&self, hash: Hash, timeout: Duration) -> FillOutcome {
        match tokio::time::timeout(timeout, self.cache.populate_local(hash)).await {
            Ok(Ok(())) => FillOutcome::Filled,
            Ok(Err(e @ (CacheError::NotFound { .. } | CacheError::NoOrigin { .. }))) => {
                tracing::debug!(%hash, error = %e.display_chain(), "reactive local-origin pull-through found no source");
                FillOutcome::CleanMiss
            }
            // Transient — the operator's own origin is down. See
            // [`Self::try_pull_through`] for why the split is exactly these two
            // variants and not a catch-all.
            Ok(Err(
                e @ (CacheError::OriginError { .. }
                | CacheError::Store(_)
                | CacheError::Internal(_)),
            )) => {
                self.metrics.node_pull_through_error();
                tracing::warn!(%hash, error = %e.display_chain(), "reactive local-origin pull-through hit a transient backend fault");
                FillOutcome::HardFault
            }
            // Deterministic (`BlobTooLarge` / `HashMismatch` / `VerifyFailed`):
            // recurs every request, so it is not evidence this node is degraded.
            Ok(Err(e)) => {
                self.metrics.node_pull_through_error();
                tracing::warn!(%hash, error = %e.display_chain(), "reactive local-origin pull-through hit a permanent cache-engine error");
                FillOutcome::CleanMiss
            }
            Err(_) => self.on_local_populate_timeout(hash, timeout).await,
        }
    }

    /// Handle a reactive local-origin populate deadline expiry (#1116). Serves the
    /// blob if a concurrent fill landed it in the store at the instant the
    /// deadline fired (so a node with node→node OFF doesn't report `CacheMiss` for
    /// a blob that is now present), otherwise meters the timeout and reports the
    /// miss.
    ///
    /// A deadline expiry is a [`FillOutcome::CleanMiss`], not a fault — see
    /// [`Self::on_pull_through_timeout`] for why.
    pub(super) async fn on_local_populate_timeout(
        &self,
        hash: Hash,
        timeout: Duration,
    ) -> FillOutcome {
        match self.cache.has(hash).await {
            Ok(true) => return FillOutcome::Filled,
            Ok(false) => {}
            Err(e) => {
                // A store-lookup fault at the deadline is a store error, not a
                // timeout: meter it as an error and return — do NOT also count a
                // timeout or emit a misleading "timed out" line for it. It is a
                // genuine local store fault, so the node is degraded, not empty.
                self.metrics.node_pull_through_error();
                tracing::warn!(%hash, error = %e.display_chain(), "reactive local-origin pull-through store lookup failed after deadline");
                return FillOutcome::HardFault;
            }
        }
        self.metrics.node_pull_through_timeout();
        tracing::debug!(%hash, ?timeout, "reactive local-origin pull-through timed out");
        FillOutcome::CleanMiss
    }

    /// The partial-serve gate (#1506): whether `[byte_offset, byte_offset +
    /// byte_len)` is already fully present in the LOCAL cache of a blob
    /// `serve_audit` reports unserveable (not `Complete`) — the counterpart
    /// to the will-serve coverage a partial holder now advertises (probe/DHT
    /// tasks 2-3). Returns the blob's total size when so, letting the caller
    /// skip straight to the size gate + delivery instead of running the
    /// miss-path fill legs; `None` otherwise.
    ///
    /// `None` covers two cases the caller treats identically — this is a
    /// best-effort optimization, never a decline path, so both fall through
    /// to the ordinary miss path unchanged: the requested span is genuinely
    /// not fully cached, or the hash is absent/evicted/refused.
    ///
    /// The size comes from the `observe()` bitfield
    /// ([`CacheEngine::present_ranges`](super::CacheEngine::present_ranges)),
    /// the SAME source [`CacheEngine::coverage`](super::CacheEngine::coverage)
    /// sizes a partial from when deciding what to advertise. Unlike
    /// `status()`/`inspect`, the bitfield knows a `Partial` blob's size as
    /// soon as any chunk carries it, not only once the FINAL chunk validates
    /// — so a front-prefix partial (holds the front, not the tail) sizes its
    /// range here too, matching what it already advertises as covered.
    /// Sizing serve and advertise from the same source matters: two different
    /// sources here would let the node advertise a block it then refuses to
    /// serve (#1506).
    ///
    /// Reuses [`CacheEngine::missing_ranges`](super::CacheEngine::missing_ranges)'s
    /// own size contract — the same `blob_size` the own-origin spine resolves
    /// via [`CacheEngine::origin_size`](super::CacheEngine::origin_size) for the
    /// ORIGIN-backed case; here the size comes from the LOCAL store's own
    /// bitfield, since a purely local partial has no origin to probe.
    pub(super) async fn partial_hit_size(
        &self,
        hash: Hash,
        byte_offset: u64,
        byte_len: u64,
    ) -> Option<u64> {
        let size = self.cache.present_ranges(hash).await.ok()?.size();
        if size == 0 {
            return None;
        }
        let missing = self
            .cache
            .missing_ranges(hash, byte_offset, byte_len, size)
            .await
            .ok()?;
        missing.is_empty().then_some(size)
    }

    /// Handle a foreground pull-through deadline expiry (#859). Serves the blob if
    /// it landed in the store in the race; otherwise meters the abandoned pull and
    /// reports the miss.
    ///
    /// A deadline expiry itself is a [`FillOutcome::CleanMiss`], NOT a
    /// [`FillOutcome::HardFault`] (#1129): we do not KNOW that anything is broken.
    /// The blob may well exist upstream and we simply ran out of patience. Reporting
    /// a slow upstream as `InternalError` (wire `Declined`) would drop a healthy
    /// node for the hash because someone ELSE was slow. Only a
    /// genuine store/origin *error* is a fault, including the `has`-lookup error
    /// below.
    pub(super) async fn on_pull_through_timeout(
        &self,
        hash: Hash,
        timeout: Duration,
    ) -> FillOutcome {
        // Race: the fill may have landed in the store at the instant the outer
        // deadline fired. If so, serve it — this was NOT an abandoned pull, so do
        // not count a timeout. A `has` *error* is a real store fault, not a clean
        // race-loss: surface it like the populate-engine-error arm above rather
        // than silently treating the store as empty, then fall through to the warm.
        let mut faulted = false;
        match self.cache.has(hash).await {
            Ok(true) => return FillOutcome::Filled,
            Ok(false) => {}
            Err(e) => {
                // A store-lookup fault at the deadline is a store error, NOT a
                // timeout. Meter it as an error only — counting it as a timeout too
                // would contaminate `node_pull_through_timeout`, whose whole job is
                // to separate "slow/wedged upstream" from "broken store", and send
                // an operator chasing the network while the disk is dying. Same rule
                // `on_local_populate_timeout` states.
                self.metrics.node_pull_through_error();
                tracing::warn!(%hash, error = %e.display_chain(), "node-to-node pull-through store lookup failed after deadline");
                faulted = true;
            }
        }
        if !faulted {
            // Genuinely abandoned at the deadline (metered here, after the race
            // check, so a blob that landed in time isn't over-counted as a timeout):
            // this distinguishes a slow/wedged upstream from "not on network".
            self.metrics.node_pull_through_timeout();
            tracing::debug!(%hash, ?timeout, "node-to-node pull-through timed out");
        }
        if faulted {
            FillOutcome::HardFault
        } else {
            FillOutcome::CleanMiss
        }
    }
}

/// The span around one cache-fill attempt for a serve miss. `tier` is `node`
/// (a paid pull from an upstream deCDN node) or `local` (the operator's own
/// origin). `outcome` is recorded once the attempt ends.
fn fill_span(tier: &'static str, hash: Hash) -> tracing::Span {
    tracing::info_span!(
        "pull_through",
        tier,
        %hash,
        outcome = tracing::field::Empty,
    )
}