decdn_node/runtime/

eviction.rs

//! Cache-eviction driver loop (#1173, ADR 040 §Whole-blob reclaim; range
//! eviction is upstream-gated).
//!
//! A single async task, owned by the node wiring layer (per
//! `appendix-poc-production-seams.md`), that enforces the cache ceiling the
//! write path deliberately does not. It periodically measures on-disk footprint
//! and, once over the high-water mark, releases the blobs the injected
//! `EvictionPolicy` ranks for eviction (via
//! [`CacheEngine::release_for_eviction`], the soft-evict path that is *not* the
//! durable operator-evict) down to the target mark, bounded by a per-sweep
//! budget.
//!
//! ## The ceiling is disk-aware
//!
//! The ceiling is not the static `cache.cache_size_mb`. That is an upper bound;
//! each tick the driver probes free disk on the `cache_dir` volume and clamps
//! the ceiling down to keep `cache.disk_headroom_mb` free, defended against any
//! process (#1930). See `effective_cap`. So a large `cache_size_mb` lets a node
//! size its cache to whatever disk the machine offers, and the high-water and
//! target marks track that dynamic ceiling.
//!
//! ## The driver's actuator is indirect — hence pending-reclaim accounting
//!
//! `release_for_eviction` does **not** free disk. It drops the blob's
//! GC-protecting tag; the bytes are reclaimed later by the independent
//! iroh-blobs GC sweep (`cache.gc_interval_sec`, default 300s). The driver ticks
//! far more often than that (`eviction_tick_secs`, default 1s), so a naive loop
//! that re-measured raw disk each tick would see its own releases have no effect
//! and keep releasing — roughly `gc_interval_sec / eviction_tick_secs` sweeps'
//! worth (~300 × 16 ≈ 4800 blobs at defaults) for a pressure event needing a
//! handful. That would drain the cache on a single high-water crossing.
//!
//! So the driver tracks `pending_reclaim`: bytes it has released but not yet
//! seen reclaimed. Decisions are made against the *effective* footprint
//! (`raw − pending_reclaim`) — what disk will look like once GC lands — while
//! the `decdn_cache_bytes` gauge still reports honest raw usage. When a
//! measurement comes in below the previous one, GC has run, and the observed
//! drop is credited against `pending_reclaim`. The accounting is self-correcting:
//! it needs no coupling to the GC schedule and recovers if a release is never
//! reclaimed.
//!
//! With GC disabled (`gc_interval_sec == 0`) nothing is ever reclaimed, so the
//! driver releases roughly one target's worth once and then idles rather than
//! stripping every tag in the cache. The runtime warns at startup in that case —
//! the driver cannot reclaim space and the ceiling is unenforceable.
//!
//! ## Hysteresis
//!
//! A latch: the driver starts evicting when the effective footprint crosses
//! `eviction_high_water_pct` and does not stop until it reaches
//! `eviction_target_pct`, then idles until the next high-water crossing. The
//! ≥5-point gap between the two (enforced at config resolution) is what prevents
//! thrash on writes hovering near the trigger.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use decdn_cache::{CacheEngine, CacheMetrics};
use tokio::sync::oneshot;

/// One mebibyte, the unit `cache.cache_size_mb` is expressed in.
const BYTES_PER_MB: u64 = 1_048_576;

/// Saturating `u64 -> i64` for gauge writes (`iroh_metrics::Gauge` is `i64`).
fn as_gauge(v: u64) -> i64 {
    i64::try_from(v).unwrap_or(i64::MAX)
}

/// Resolved driver parameters. Every field is range-checked at config
/// resolution (`resolve_cache_into`): `high_water_pct ∈ [60, 95]`,
/// `target_pct ∈ [40, 90]` **and** `target_pct <= high_water_pct - 5` (the
/// structural hysteresis gap), `per_sweep_budget ∈ [1, 256]`,
/// `tick ∈ [1s, 60s]`. The type itself carries no invariant — it is a wiring
/// DTO with a single production construction site.
#[derive(Debug, Clone, Copy)]
pub struct EvictionParams {
    /// Configured cache size in MiB; the upper bound on the ceiling (the disk
    /// clamp can only lower it, never raise it).
    pub cache_size_mb: u64,
    /// Free disk in MiB the driver keeps unused on the `cache_dir` volume by
    /// *any* process (#1930). Each tick the effective ceiling is capped so the
    /// cache cannot grow into the last `disk_headroom_mb` of free space; below
    /// it, the driver evicts to claw disk back. `0` opts out of the disk clamp
    /// (only `cache_size_mb` binds).
    pub disk_headroom_mb: u64,
    /// Crossing this percentage of the effective ceiling starts a sweep.
    pub high_water_pct: u64,
    /// A sweep runs until usage falls to this percentage. The gap to
    /// `high_water_pct` is the hysteresis that stops it flapping.
    pub target_pct: u64,
    /// Maximum blobs released in one sweep, so a sweep cannot monopolize the
    /// store.
    pub per_sweep_budget: u64,
    /// How often the driver re-reads usage.
    pub tick: Duration,
}

/// Per-tick ceiling inputs, precomputed once in [`run`] from [`EvictionParams`].
/// The effective ceiling itself is recomputed every tick from these plus the
/// live footprint and free disk (#1930), so a shared volume's ceiling tracks
/// real free space rather than a static config number.
#[derive(Debug, Clone, Copy)]
struct Ceiling {
    /// `cache_size_mb × MiB` — the absolute upper bound on the ceiling.
    config_limit_bytes: u64,
    /// `disk_headroom_mb × MiB` — free space kept on the volume.
    headroom_bytes: u64,
    /// Percent of the effective ceiling above which a sweep starts.
    high_water_pct: u64,
    /// Percent of the effective ceiling a sweep evicts down to.
    target_pct: u64,
}

/// Cross-tick driver state. Kept out of `run`'s body so `tick` can be unit-tested.
#[derive(Debug, Default)]
struct DriverState {
    /// Hysteresis latch: `true` between a high-water crossing and reaching target.
    evicting: bool,
    /// Bytes released but not yet observed reclaimed by the iroh-blobs GC.
    pending_reclaim: u64,
    /// Previous raw measurement, used to detect a GC reclaim (a decrease).
    last_raw: Option<u64>,
    /// Whether the disk-headroom clamp — not `cache_size_mb` — is the binding
    /// ceiling. Tracked so the driver logs the transition once, not every tick.
    disk_clamped: bool,
    /// Whether the free-disk probe is currently failing. Tracked so a persistent
    /// statvfs error logs once on the way in and once on recovery, not per tick.
    disk_probe_failing: bool,
}

/// `base * pct / 100` in u128 space, saturated back to u64.
fn pct_of(base: u64, pct: u64) -> u64 {
    u64::try_from(u128::from(base).saturating_mul(u128::from(pct)) / 100).unwrap_or(u64::MAX)
}

/// The effective cache ceiling for one tick: the smaller of the configured
/// `cache_size_mb` budget and what free disk allows, recomputed every tick so a
/// shared volume's ceiling tracks real free space (#1930).
///
/// Free-space tracking keeps `disk_headroom_mb` of the volume unused by *any*
/// process. The cache may grow into whatever is free beyond that headroom, on
/// top of the bytes it already holds:
/// `disk_ceiling = footprint + max(0, free - headroom)`. At or below headroom
/// the ceiling collapses to the current footprint, so the driver evicts to claw
/// disk back toward the margin. A `free` of `u64::MAX` — the statvfs-failed
/// sentinel the driver substitutes on a probe error — saturates the disk term
/// so the configured budget binds, and a transient probe failure never shrinks
/// the cache to its footprint.
const fn effective_cap(
    config_limit_bytes: u64,
    footprint: u64,
    free_bytes: u64,
    headroom_bytes: u64,
) -> u64 {
    let disk_ceiling = footprint.saturating_add(free_bytes.saturating_sub(headroom_bytes));
    if config_limit_bytes < disk_ceiling {
        config_limit_bytes
    } else {
        disk_ceiling
    }
}

/// Fold a fresh raw measurement into the driver's pending-reclaim bookkeeping
/// and return the **effective** footprint to decide against.
///
/// A decrease since the previous tick means the iroh-blobs GC sweep landed, so
/// the observed drop is credited against what we are still waiting on. An
/// increase is unrelated new writes and leaves `pending_reclaim` alone. The
/// effective footprint is `raw - pending_reclaim`: what disk will look like once
/// the releases already issued are reclaimed.
///
/// This is the whole fix for the driver's original non-convergence: without it,
/// releases (which only drop GC protection) never moved the measurement, so the
/// latch stayed engaged and the driver re-released every tick for the whole GC
/// interval — ~`gc_interval_sec / eviction_tick_secs` sweeps' worth.
const fn reconcile(state: &mut DriverState, raw: u64) -> u64 {
    if let Some(prev) = state.last_raw
        && raw < prev
    {
        state.pending_reclaim = state.pending_reclaim.saturating_sub(prev - raw);
    }
    state.last_raw = Some(raw);
    raw.saturating_sub(state.pending_reclaim)
}

/// One budget-bounded eviction pass. Releases up to `budget` candidates in the
/// order `policy` ranks them, stopping early once the projected effective
/// footprint reaches `target_bytes`. Returns the number of bytes actually
/// released (to be added to `pending_reclaim`).
///
/// Emits `evictions_starved` when there is nothing left to evict while still
/// over target — either because the candidate set is empty (everything pinned,
/// operator-evicted, or probe-held, or the open-time recency seed failed so
/// blobs from before the restart are not candidates; see
/// `decdn_cache_recency_seed_failures_total`) or because a whole pass released
/// nothing.
#[allow(clippy::too_many_arguments)]
async fn sweep(
    cache: &CacheEngine,
    metrics: &CacheMetrics,
    effective: u64,
    target_bytes: u64,
    budget: u64,
    cache_bytes: u64,
    sizes: &HashMap<decdn_cache::Hash, u64>,
    policy: &Arc<dyn decdn_cache::EvictionPolicy>,
    warming: &Arc<crate::warming_allowance::WarmingAllowance>,
) -> u64 {
    let candidates = cache.eviction_candidates();
    if candidates.is_empty() {
        // Over target but no eligible victim — the operational-alarm signal.
        metrics.evictions_starved.inc();
        return 0;
    }

    // The policy owns the whole sweep decision (ADR 040): it ranks candidates,
    // applies the target/budget stop conditions, and returns what to evict and
    // what to promote. The driver is a dumb executor.
    let segments = cache.segments_snapshot();
    let plan = policy.plan(&decdn_cache::EvictionContext {
        candidates: &candidates,
        sizes,
        segments: &segments,
        total_bytes: effective,
        target_bytes,
        budget,
        cache_bytes,
    });

    // Promotion is a pure in-memory segment move — no store I/O.
    for (hash, seg) in plan.promote {
        cache.set_segment(hash, seg);
    }

    let mut freed: u64 = 0;
    let mut removed: u64 = 0;
    for hash in plan.evict {
        match cache.release_for_eviction(hash).await {
            // `Ok(0)` means nothing was released — the hash was pinned (the
            // documented defence-in-depth refusal) or carried no protecting tag.
            // Crediting it would charge the budget and, worse, add its size to
            // `freed`, letting the sweep break early believing it had freed
            // bytes it never touched.
            Ok(0) => {}
            Ok(_) => {
                freed = freed.saturating_add(sizes.get(&hash).copied().unwrap_or(0));
                removed = removed.saturating_add(1);
                metrics.evictions.inc();
                // ADR 041: drop the warming tag for the evicted hash, so a later
                // reuse of this slot can never credit a stale source's allowance.
                warming.forget(hash);
            }
            Err(err) => {
                tracing::warn!(%hash, error = %err.display_chain(), "eviction driver: release_for_eviction failed; skipping hash");
            }
        }
    }

    if removed == 0 {
        // Had candidates but released none (all pinned/untagged/erroring). Same
        // operator signal as an empty candidate set — otherwise the driver looks
        // busy while making no progress.
        metrics.evictions_starved.inc();
    }
    metrics.evictions_bytes.inc_by(freed);
    freed
}

/// Recompute the effective ceiling for one tick, publish the ceiling gauge, and
/// log the disk-clamp transition once (#1930).
///
/// `footprint` is the live raw on-disk footprint, paired with `free_bytes`
/// (both real on-disk state) inside [`effective_cap`]. Returns
/// `(cap, high_water_bytes, target_bytes)` for the latch and sweep. The clamp
/// becoming — or ceasing to be — the binding ceiling is the operator-visible
/// event, so it is logged on the transition, not every tick.
fn resolve_ceiling(
    ceiling: Ceiling,
    footprint: u64,
    free_bytes: u64,
    state: &mut DriverState,
    metrics: &CacheMetrics,
) -> (u64, u64, u64) {
    let cap = effective_cap(
        ceiling.config_limit_bytes,
        footprint,
        free_bytes,
        ceiling.headroom_bytes,
    );
    metrics.size_limit_bytes.set(as_gauge(cap));

    let clamped = cap < ceiling.config_limit_bytes;
    if clamped && !state.disk_clamped {
        tracing::warn!(
            effective_cap_bytes = cap,
            config_limit_bytes = ceiling.config_limit_bytes,
            free_bytes,
            headroom_bytes = ceiling.headroom_bytes,
            "eviction driver: free disk is the binding cache ceiling; cache.cache_size_mb \
             is clamped down to keep cache.disk_headroom_mb free on the volume"
        );
    } else if !clamped && state.disk_clamped {
        tracing::info!(
            config_limit_bytes = ceiling.config_limit_bytes,
            "eviction driver: free disk recovered; cache.cache_size_mb is the binding ceiling again"
        );
    }
    state.disk_clamped = clamped;

    (
        cap,
        pct_of(cap, ceiling.high_water_pct),
        pct_of(cap, ceiling.target_pct),
    )
}

/// One driver tick: measure, recompute the effective ceiling from the live
/// footprint and free disk (#1930), refresh cache-health gauges, reconcile
/// pending-reclaim against observed GC progress, apply the hysteresis latch, and
/// (when latched over target) run one budget-bounded [`sweep`].
///
/// `free_bytes` is the volume's available bytes probed for this tick, or
/// `u64::MAX` when the probe failed — the sentinel that disables the disk clamp
/// so `config_limit_bytes` binds (see [`effective_cap`]).
#[allow(clippy::too_many_arguments)]
async fn tick(
    cache: &CacheEngine,
    metrics: &CacheMetrics,
    ceiling: Ceiling,
    free_bytes: u64,
    budget: u64,
    state: &mut DriverState,
    policy: &Arc<dyn decdn_cache::EvictionPolicy>,
    warming: &Arc<crate::warming_allowance::WarmingAllowance>,
) {
    // Exactly ONE store walk per tick: this snapshot is both the footprint
    // source and the sweep's per-hash size lookup, so `sweep` takes it by
    // reference rather than re-walking. (An earlier shape called `total_bytes()`
    // here and `size_snapshot()` again in the sweep — two walks under sustained
    // pressure, and the two readings could disagree.)
    //
    // A snapshot failure skips the whole tick rather than evicting blind:
    // without sizes there is no way to know when target is reached, and
    // degrading to an empty map silently drained the full budget every tick.
    let sizes = match cache.size_snapshot().await {
        Ok(s) => s,
        Err(err) => {
            metrics.size_measure_failures.inc();
            tracing::warn!(
                error = %err.display_chain(),
                "eviction driver: cache size measurement failed; skipping tick \
                 (decdn_cache_bytes is now stale — alert on decdn_cache_size_measure_failures_total)"
            );
            return;
        }
    };

    let raw = sizes.values().fold(0u64, |acc, sz| acc.saturating_add(*sz));
    let effective = reconcile(state, raw);

    // Gauges report honest raw usage; `pending_reclaim` is driver bookkeeping.
    metrics.bytes.set(as_gauge(raw));
    metrics.pinned_count.set(as_gauge(
        u64::try_from(cache.pinned_snapshot().len()).unwrap_or(u64::MAX),
    ));

    // Effective ceiling for this tick: the configured budget, clamped down to
    // real free disk (#1930). Publishes the ceiling gauge and logs the clamp
    // transition.
    let (cap, high_water_bytes, target_bytes) =
        resolve_ceiling(ceiling, raw, free_bytes, state, metrics);

    // Hysteresis latch: only start evicting on a high-water crossing; once
    // latched, keep evicting until at/below target, then release.
    if state.evicting {
        if effective <= target_bytes {
            state.evicting = false;
            return;
        }
    } else if effective <= high_water_bytes {
        return;
    } else {
        state.evicting = true;
    }

    let freed = sweep(
        cache,
        metrics,
        effective,
        target_bytes,
        budget,
        cap,
        &sizes,
        policy,
        warming,
    )
    .await;
    state.pending_reclaim = state.pending_reclaim.saturating_add(freed);
}

/// Probe free disk for one tick, or return the `u64::MAX` clamp-disabled
/// sentinel on a `statvfs` failure (#1930). One cheap syscall, dwarfed by the
/// store walk `tick` already does. On failure the driver falls back to the
/// config-only ceiling rather than evicting blind. The failure and recovery
/// transitions are each logged once, not every tick.
fn probe_free_bytes(cache_dir: &std::path::Path, state: &mut DriverState) -> u64 {
    match decdn_common::disk::statvfs_target(cache_dir) {
        Ok(space) => {
            if state.disk_probe_failing {
                tracing::info!(
                    cache_dir = %cache_dir.display(),
                    "eviction driver: free-disk probe recovered; disk-headroom clamp re-enabled"
                );
                state.disk_probe_failing = false;
            }
            space.avail
        }
        Err(err) => {
            if !state.disk_probe_failing {
                tracing::warn!(
                    error = %err,
                    cache_dir = %cache_dir.display(),
                    "eviction driver: free-disk probe failed; disk-headroom clamp disabled \
                     until it recovers (cache.cache_size_mb still enforced)"
                );
                state.disk_probe_failing = true;
            }
            u64::MAX
        }
    }
}

/// Run the eviction driver until `shutdown` fires. Intended to be
/// `tasks.spawn(run(...))`-ed into the runtime `JoinSet`; returns when the
/// oneshot is dropped or sent.
pub async fn run(
    cache: CacheEngine,
    metrics: Arc<CacheMetrics>,
    params: EvictionParams,
    cache_dir: std::path::PathBuf,
    policy: Arc<dyn decdn_cache::EvictionPolicy>,
    warming: Arc<crate::warming_allowance::WarmingAllowance>,
    mut shutdown: oneshot::Receiver<()>,
) {
    let ceiling = Ceiling {
        config_limit_bytes: params.cache_size_mb.saturating_mul(BYTES_PER_MB),
        headroom_bytes: params.disk_headroom_mb.saturating_mul(BYTES_PER_MB),
        high_water_pct: params.high_water_pct,
        target_pct: params.target_pct,
    };

    let mut ticker = tokio::time::interval(params.tick);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    ticker.tick().await; // burn the immediate first tick

    let mut state = DriverState::default();

    loop {
        tokio::select! {
            biased;
            _ = &mut shutdown => {
                tracing::debug!("eviction driver shutdown signal received");
                return;
            }
            _ = ticker.tick() => {}
        }

        // `disk_headroom_mb = 0` opts out of the disk clamp (only `cache_size_mb`
        // binds), so skip the syscall entirely and force the clamp-disabled
        // sentinel — no probe, no probe-failure logging.
        let free_bytes = if ceiling.headroom_bytes == 0 {
            u64::MAX
        } else {
            probe_free_bytes(&cache_dir, &mut state)
        };

        tick(
            &cache,
            &metrics,
            ceiling,
            free_bytes,
            params.per_sweep_budget,
            &mut state,
            &policy,
            &warming,
        )
        .await;
    }
}

#[cfg(test)]
mod tests;