decdn-cache 0.0.1

//! `FillSession` — the cache-resident fill-coordination primitive the
//! serve-miss legs share (ADR 038).
//!
//! # Why this exists
//!
//! The serve leg must emit ONE coherent whole-range bao verified stream
//! while the pull leg fills the cache incrementally beside it. `bao_tree`'s async
//! [`encode_ranges_validated`](bao_tree::io::fsm::encode_ranges_validated) is the
//! right engine — it walks the range pre-order, reading leaf DATA and, for every
//! internal node, the `(left, right)` hash pair from an [`Outboard`]. The data the
//! serve reads from the cache (via ranged export, gated on the present-range watch).
//! But the OUTBOARD it cannot: iroh-blobs keeps a partial blob's outboard in actor
//! memory and exposes no reader, and a pre-order walk needs proof hashes for
//! not-yet-filled right-spine subtrees (the root's `right_hash` is emitted before
//! any right-half data lands) that are NOT derivable from present data.
//!
//! So the cache keeps its own copy, captured as each range admits (the pull leg
//! already receives those exact proof nodes in the verified upstream bao it ingests,
//! and the cache's `admit_bao_stream` captures them via [`FillSession::capture`] into
//! a whole-tree buffer). The serve leg reads them through a [`SessionOutboardReader`],
//! whose [`Outboard::load`] AWAITS a node that has not been captured yet — racing the
//! registry-wide fill liveness so a fill that dies (or a genuinely missing node after
//! a clean pull) fails the serve rather than hanging.
//!
//! # The outboard is per-HASH, shared by every fill of that hash
//!
//! A blob is BLAKE3-addressed, so its tree geometry — and thus every captured
//! `(left, right)` node — is a property of the HASH, not of any single pull. When two
//! serve-misses want overlapping bytes of one hash, one opens a pull for its
//! remainder and the other for its own; both admit into the SAME store and each admit
//! captures proof nodes into the ONE per-hash [`HashOutboard`]. Because
//! `attach ∪ remainder = R` by construction, the two pulls together capture every
//! proof node a coherent encode of `R` needs, into that single buffer. A serve leg
//! reading `R` therefore does not care which pull filled what — it reads the shared
//! per-hash outboard and the source-agnostic present-range watch.
//!
//! The one thing that must generalize past a single pull is termination: "await this
//! node, or fail" has to consult ALL live fills whose `covered` ranges intersect the
//! node's byte span, not one pull's terminal signal. [`FillSession::range_still_live`]
//! (backed by [`FillRegistry::range_still_live`]) answers exactly that — it settles
//! a parked reader once no live session's `covered` still intersects the range,
//! which is the N-fill generalization of the single-pull failure race. A settled
//! read is served from the local store when the store holds what it needs, and
//! fails otherwise; it never hangs.
//!
//! # Ownership seam (cache owns the structure, the node drives it)
//!
//! The pull leg (paid peer or own-origin) is node-side and alone knows when/why it
//! terminated. So the session exposes [`FillSession::mark_ended`]; the node's pull
//! leg calls it on every exit. The cache owns the DATA STRUCTURE; the node DRIVES
//! it. [`FillError`] is a flattened terminal message (`anyhow::Error` is not `Clone`).

use std::collections::HashMap;
use std::io;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex as StdMutex, OnceLock, PoisonError, Weak};

use bao_tree::io::fsm::Outboard;
use bao_tree::{BaoTree, BlockSize, ChunkNum, ChunkRanges, TreeNode, blake3};
use decdn_bao_range::align_range_clamped;
use tokio::sync::Notify;
use tokio_util::sync::CancellationToken;

use crate::Hash;
use crate::engine::CacheEngine;
use crate::error::{CacheResult, ErrorChain};

/// iroh-blobs' block size — 16 KiB chunk groups (`2^4` 1 KiB chunks). Identical by
/// construction to `decdn_bao_range::IROH_BLOCK_SIZE` and
/// `iroh_blobs::store::IROH_BLOCK_SIZE`; the tree geometry (and thus every node
/// offset) must match the store's, since the captured hashes are the store's.
const IROH_BLOCK_SIZE: BlockSize = BlockSize::from_chunk_log(4);

/// Bytes per outboard node (a `(left, right)` blake3 hash pair).
const HASH_PAIR_BYTES: usize = 64;

/// Bytes per bao chunk (the `ChunkNum` unit). `ChunkNum(n)` starts at `n * 1024`.
const CHUNK_BYTES: u64 = 1024;

/// A flattened terminal error for a fill: the node's pull leg records why its pull
/// ended, and the serve-side [`SessionOutboardReader`] / data reader surface it.
/// A `String` message rather than `anyhow::Error` because the outcome must be
/// `Clone` (every observer reads it) and `anyhow::Error` is not.
#[derive(Debug, Clone)]
pub struct FillError(String);

impl FillError {
    /// Wrap a flattened terminal message.
    #[must_use]
    pub fn new(msg: impl Into<String>) -> Self {
        Self(msg.into())
    }
}

impl std::fmt::Display for FillError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::error::Error for FillError {}

/// The pre-order outboard buffer + a per-node "captured" flag. Guarded together;
/// the lock is never held across an await.
///
/// Both vectors are allocated lazily on the first capture (see
/// [`HashOutboard::write_pair`]), not at construction: empty ⇒ nothing captured
/// yet, which [`HashOutboard::try_load`] reads as "absent" exactly as an all-false
/// `captured` did. This keeps the MB-scale zeroing for a large blob off the global
/// registry lock that [`HashOutboard::new`] runs under.
#[derive(Debug)]
struct OutboardState {
    /// `tree.outboard_size()` bytes once sized: node `n`'s pair lives at
    /// `tree.pre_order_offset(n) * 64`. Empty until the first capture sizes it.
    bytes: Vec<u8>,
    /// `captured[i]` is set once node index `i` (its `pre_order_offset`) is saved.
    /// Empty until the first capture sizes it, alongside `bytes`.
    captured: Vec<bool>,
}

/// The captured whole-tree outboard for ONE hash, shared by every [`FillSession`]
/// that fills that hash. A blob's tree geometry is fixed by its BLAKE3 root, so a
/// node captured by any pull — of any range, from any source — is valid for every
/// serve leg reading that hash. The registry keys exactly one of these per hash and
/// every session for the hash references it, which is what makes partial-overlap
/// proof-sourcing free: pull A (admitting `attach`) and pull B (admitting
/// `remainder`) capture into the same buffer, and a serve leg for `R = attach ∪
/// remainder` reads every node either supplied.
#[derive(Debug)]
pub struct HashOutboard {
    tree: BaoTree,
    root: blake3::Hash,
    state: StdMutex<OutboardState>,
    /// Notified after every [`Self::capture`] so a parked [`Outboard::load`]
    /// re-checks. Runtime-agnostic, so it crosses the pull/serve runtimes safely.
    captured: Notify,
    /// Notified when ANY session for this hash ends or is cancelled, so a parked
    /// reader re-checks [`FillRegistry::range_still_live`] and fails a range no live
    /// pull will fill. Shared as an `Arc` so a data reader on the node side can hold
    /// an owned handle without naming this cache-private type.
    liveness: Arc<Notify>,
}

impl HashOutboard {
    /// Build the empty outboard for a `total_bytes`-byte blob rooted at `root`.
    ///
    /// O(1): the `tree.outboard_size()`-byte buffer is NOT allocated here — the
    /// first [`Self::write_pair`] sizes it under the per-hash `state` lock. This
    /// keeps a session's outboard construction under the global [`FillRegistry`]
    /// lock free of MB-scale zeroing, so one large-blob claim does not stall
    /// every other hash's claim/wakeup for the allocation duration, and a session
    /// that only adopts the canonical outboard allocates nothing to discard.
    #[must_use]
    fn new(root: blake3::Hash, total_bytes: u64) -> Arc<Self> {
        Arc::new(Self {
            tree: BaoTree::new(total_bytes, IROH_BLOCK_SIZE),
            root,
            state: StdMutex::new(OutboardState {
                bytes: Vec::new(),
                captured: Vec::new(),
            }),
            captured: Notify::new(),
            liveness: Arc::new(Notify::new()),
        })
    }

    /// Write one internal node's `(left, right)` hash pair into `state`, returning
    /// whether the pair was actually persisted. `false` for a leaf (no outboard
    /// slot), an offset past `usize`, or a slot outside the buffers — nothing was
    /// stored, so the caller must NOT wake readers on its behalf. `true` only once
    /// both the hash bytes and the captured flag are set, which is exactly the state
    /// [`Self::try_load`] reads back. Idempotent: re-saving a node (a re-admitted
    /// range) overwrites with the same bytes. Does NOT notify; the caller wakes
    /// parked readers once the batch is in, so an N-node admit fires `captured` a
    /// single time rather than N.
    fn write_pair(
        &self,
        state: &mut OutboardState,
        node: TreeNode,
        pair: (blake3::Hash, blake3::Hash),
    ) -> bool {
        let Some(offset) = self.tree.pre_order_offset(node) else {
            return false; // leaf: no hash pair in the outboard
        };
        let Ok(idx) = usize::try_from(offset) else {
            return false; // offset beyond usize: no addressable slot
        };
        // Lazily size the outboard on the first capture, under this per-hash
        // `state` lock rather than the global registry lock `new` runs under. A
        // real interior node (a `Some` offset) implies `outboard_size() >= 64`, so
        // an empty buffer here always means "not yet sized", never a zero-node
        // tree. Sized exactly once: every later capture sees a non-empty buffer.
        // Bail if the size is not `usize`-representable rather than allocating
        // `usize::MAX` — the same "not addressable, not capturable" degrade as the
        // `idx` guard above, never a process-aborting allocation.
        if state.bytes.is_empty() {
            let Ok(size) = usize::try_from(self.tree.outboard_size()) else {
                return false;
            };
            state.bytes = vec![0u8; size];
            state.captured = vec![false; size / HASH_PAIR_BYTES];
        }
        let byte_off = idx.saturating_mul(HASH_PAIR_BYTES);
        let (l, r) = pair;
        let bytes_written = if let Some(slot) =
            state.bytes.get_mut(byte_off..byte_off + HASH_PAIR_BYTES)
            && let Some((left, right)) = slot.split_at_mut_checked(32)
        {
            left.copy_from_slice(l.as_bytes());
            right.copy_from_slice(r.as_bytes());
            true
        } else {
            false
        };
        let flag_set = if let Some(flag) = state.captured.get_mut(idx) {
            *flag = true;
            true
        } else {
            false
        };
        bytes_written && flag_set
    }

    /// Capture one internal node's `(left, right)` hash pair, then wake parked
    /// readers. A `node` with no outboard slot (a leaf) is ignored and does not notify.
    fn capture(&self, node: TreeNode, pair: (blake3::Hash, blake3::Hash)) {
        let wrote = {
            let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
            self.write_pair(&mut state, node, pair)
        };
        if wrote {
            self.captured.notify_waiters();
        }
    }

    /// Capture a batch of node pairs under one lock acquisition, then fire
    /// `captured` exactly once. A serve leg parked on any of these nodes reads them
    /// all back the same as a per-node [`Self::capture`] loop would, but every parked
    /// reader wakes once per batch instead of once per node. The admit path hands
    /// each leaf's run of proof nodes here just before the leaf. Leaves are ignored;
    /// the notify fires only if at least one internal node was written.
    fn capture_many(
        &self,
        pairs: impl IntoIterator<Item = (TreeNode, (blake3::Hash, blake3::Hash))>,
    ) {
        let wrote = {
            let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
            pairs.into_iter().fold(false, |acc, (node, pair)| {
                self.write_pair(&mut state, node, pair) || acc
            })
        };
        if wrote {
            self.captured.notify_waiters();
        }
    }

    /// Read node `idx`'s captured pair, or `None` if not captured yet. Never holds
    /// the lock across an await.
    fn try_load(&self, idx: usize) -> Option<(blake3::Hash, blake3::Hash)> {
        let state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
        if state.captured.get(idx).copied() != Some(true) {
            return None;
        }
        let byte_off = idx.saturating_mul(HASH_PAIR_BYTES);
        let slot = state.bytes.get(byte_off..byte_off + HASH_PAIR_BYTES)?;
        let (l, r) = slot.split_at_checked(32)?;
        let left: [u8; 32] = l.try_into().ok()?;
        let right: [u8; 32] = r.try_into().ok()?;
        Some((blake3::Hash::from(left), blake3::Hash::from(right)))
    }
}

/// A forward-only content frontier shared between a fill's serve legs and its pull.
///
/// Holders read it with [`Self::get`]. Only the owning [`FillSession`] raises it,
/// and every raise that moves the value wakes a pull parked in
/// [`DownstreamWatch::past`]. No holder can lower the frontier or raise it without
/// that wakeup, so a parked pull never misses an advance.
#[derive(Debug, Clone)]
pub struct Frontier {
    /// The frontier value, a high-water mark.
    value: Arc<AtomicU64>,
    /// The session's downstream wakeup, notified when [`Self::raise`] moves `value`.
    advanced: Arc<Notify>,
}

impl Frontier {
    /// A frontier at `start` that notifies `advanced` on each advance.
    fn new(start: u64, advanced: Arc<Notify>) -> Self {
        Self {
            value: Arc::new(AtomicU64::new(start)),
            advanced,
        }
    }

    /// The current frontier value.
    #[must_use]
    pub fn get(&self) -> u64 {
        self.value.load(Ordering::Acquire)
    }

    /// Raise the frontier to `to` and wake the parked pull if the value moved. A
    /// lower `to` changes nothing.
    fn raise(&self, to: u64) {
        if self.value.fetch_max(to, Ordering::AcqRel) < to {
            self.advanced.notify_waiters();
        }
    }
}

/// A fill's serve demand: one standing demand per serve leg that has run out of
/// encoded bytes while its encode waits on content this fill produces.
///
/// A leg stands its demand through its [`DemandSlot`] and withdraws it once its
/// encode moves again or the leg ends. [`Self::get`] reads the NEAREST standing
/// demand, so a leg parked far down the blob never hides a leg parked at the pull's
/// frontier. Every change to the nearest demand wakes a pull parked in
/// [`DownstreamWatch::past`].
#[derive(Debug, Clone)]
pub struct Demand {
    /// The content end each starved serve leg awaits, keyed by its slot id.
    legs: Arc<StdMutex<HashMap<u64, u64>>>,
    /// The session's downstream wakeup, shared with its paid frontier.
    advanced: Arc<Notify>,
}

impl Demand {
    /// An empty demand that notifies `advanced` when its nearest value changes.
    fn new(advanced: Arc<Notify>) -> Self {
        Self {
            legs: Arc::new(StdMutex::new(HashMap::new())),
            advanced,
        }
    }

    /// The nearest content end a starved serve leg awaits, or `0` when none waits.
    #[must_use]
    pub fn get(&self) -> u64 {
        nearest(&self.legs.lock().unwrap_or_else(PoisonError::into_inner))
    }

    /// Record that leg `leg` awaits content up to `end`.
    fn stand(&self, leg: u64, end: u64) {
        self.update(|legs| {
            legs.insert(leg, end);
        });
    }

    /// Drop leg `leg`'s standing demand, if any.
    fn withdraw(&self, leg: u64) {
        self.update(|legs| {
            legs.remove(&leg);
        });
    }

    /// Apply `change` and wake the parked pull when the nearest demand moved.
    fn update(&self, change: impl FnOnce(&mut HashMap<u64, u64>)) {
        let moved = {
            let mut legs = self.legs.lock().unwrap_or_else(PoisonError::into_inner);
            let before = nearest(&legs);
            change(&mut legs);
            nearest(&legs) != before
        };
        if moved {
            self.advanced.notify_waiters();
        }
    }
}

/// The smallest standing demand in `legs`, or `0` when it is empty.
fn nearest(legs: &HashMap<u64, u64>) -> u64 {
    legs.values().copied().min().unwrap_or(0)
}

/// Source of [`DemandSlot`] ids, unique across every fill in the process.
static NEXT_DEMAND_SLOT: AtomicU64 = AtomicU64::new(1);

/// One serve leg's handle on the serve demand of its blob's live fills. Minted by
/// [`FillSession::demand_slot`].
///
/// [`Self::stand`] routes the leg's demand to the live fill whose covered range
/// holds the awaited byte, and withdraws it from every other fill of the hash:
/// only that fill's pull can produce the byte. Dropping the slot withdraws the
/// demand, so a finished or abandoned leg leaves nothing standing.
#[derive(Debug)]
pub struct DemandSlot {
    /// The serve leg's own session, the route to the hash's registry.
    session: Arc<FillSession>,
    /// This slot's key in every [`Demand`] it stands in.
    id: u64,
}

impl DemandSlot {
    /// Stand this leg's demand for content up to `end`, clamped to the blob size.
    /// It replaces the leg's previous demand. An `end` of `0` withdraws it.
    pub fn stand(&self, end: u64) {
        let end = end.min(self.session.total_bytes());
        let Some(byte) = end.checked_sub(1) else {
            self.withdraw();
            return;
        };
        match self.session.registry_and_hash() {
            Some((registry, hash)) => registry.route_demand(hash, self.id, byte, end),
            None => self.session.route_own_demand(self.id, byte, end),
        }
    }

    /// Withdraw this leg's demand from every fill of the hash.
    pub fn withdraw(&self) {
        match self.session.registry_and_hash() {
            Some((registry, hash)) => registry.withdraw_demand(hash, self.id),
            None => self.session.serve_demand.withdraw(self.id),
        }
    }
}

impl Drop for DemandSlot {
    fn drop(&mut self) {
        self.withdraw();
    }
}

/// A pull leg's view of its session's two downstream frontiers: the paid frontier
/// ([`FillSession::served_frontier`]) and the serve demand
/// ([`FillSession::serve_demand`]). Read-only and cheap to clone, so a pull on its
/// own runtime holds an owned copy. Minted by [`FillSession::downstream_watch`].
#[derive(Debug, Clone)]
pub struct DownstreamWatch {
    /// The session's paid content frontier.
    served_paid: Frontier,
    /// The session's serve demand. Shares its wakeup with `served_paid`.
    serve_demand: Demand,
}

impl DownstreamWatch {
    /// The current paid content frontier.
    #[must_use]
    pub fn served_paid(&self) -> u64 {
        self.served_paid.get()
    }

    /// The nearest content end a starved serve leg awaits, or `0` when none waits.
    #[must_use]
    pub fn serve_demand(&self) -> u64 {
        self.serve_demand.get()
    }

    /// Resolve once `served_paid` lies past `observed_paid` or the serve demand
    /// differs from `observed_demand`. Returns at once when either already holds.
    /// The demand counts any change: a nearer leg parking lowers it, and the pull
    /// must re-decide for that leg as much as for a farther one.
    ///
    /// Registers the wakeup BEFORE it re-reads the frontiers. `notify_waiters` wakes
    /// only waiters registered when it fires and stores no permit, so an advance
    /// that lands between the caller reading the frontiers and this park would
    /// otherwise be lost, and the pull would wait forever (#1673).
    pub async fn past(&self, observed_paid: u64, observed_demand: u64) {
        let notified = self.served_paid.advanced.notified();
        tokio::pin!(notified);
        loop {
            notified.as_mut().enable();
            if self.served_paid() > observed_paid || self.serve_demand() != observed_demand {
                return;
            }
            notified.as_mut().await;
            notified.set(self.served_paid.advanced.notified());
        }
    }
}

/// One in-flight fill of a `total_bytes`-byte blob rooted at `root`.
///
/// Holds a reference to the per-hash [`HashOutboard`] (captured by the cache admit
/// path via [`Self::capture`], read by [`SessionOutboardReader`]), the pull's
/// terminal outcome ([`Self::mark_ended`] / [`Self::outcome`]), the ranges this fill
/// covers ([`Self::set_covered`]), and the shared downstream frontiers the pull
/// leg paces against ([`Self::served_frontier`] / [`Self::serve_demand`], awaited
/// through [`Self::downstream_watch`]). Shared
/// behind an `Arc`; every field is interior-mutable, so both legs hold
/// `Arc<FillSession>` and coordinate on it.
#[derive(Debug)]
pub struct FillSession {
    /// The per-hash captured outboard. A freshly built session starts with its own,
    /// but registration ([`FillRegistry::register_fill`] / [`FillRegistry::claim`])
    /// swaps in the hash's canonical one via [`Self::adopt_outboard`], so all
    /// sessions for a hash — and every reader they mint — share the ONE buffer. The
    /// swap runs under the registry lock before the session is handed to a pull leg
    /// or a serve leg, so no capture or read observes a stale outboard.
    outboard: StdMutex<Arc<HashOutboard>>,
    /// The pull leg's terminal outcome — `None` while running, `Some(Ok)` on a
    /// clean end, `Some(Err)` on a failure.
    ended: StdMutex<Option<Result<(), FillError>>>,
    /// The client's PAID content frontier: the serve leg raises it after each
    /// voucher batch commits ([`Self::advance_served`]), the pull leg's pacer reads
    /// it to bound `pulled − served_paid ≤ window`, plus one floor to serve
    /// `serve_demand`. A cloneable handle, so a pull leg on its own runtime can hold
    /// an owned copy.
    served_paid: Frontier,
    /// The ABSOLUTE content offset the owning request starts at — the value
    /// `served_paid` is seeded with. A pull leg's `RampPacer` ramps on
    /// `served_paid − served_start` plus the owning stream's carried lane credit,
    /// so a resumed request ramps from that credit, not from its blob offset.
    served_start: u64,
    /// Where the starved serve legs reading this fill's content wait. A serve leg's
    /// frame consumer stands a demand through its [`DemandSlot`] when it has no
    /// encoded bytes left and its encode is parked on a leaf or a proof node no pull
    /// has produced ([`SessionOutboardReader::parked_on`]), and withdraws it once
    /// the encode moves again. A park with encoded bytes still buffered is
    /// look-ahead and stands nothing, so a client that stops paying cannot drag a
    /// pull past its window. When the nearest demand lies within one chunk group
    /// past a pull's frontier, that pull's pacer draws one window floor even with
    /// its window full. Without it, a pull window that closes before the serve leg's
    /// credit window leaves both legs waiting on each other. Shares one wakeup with
    /// `served_paid`, so a parked pull wakes on either.
    serve_demand: Demand,
    /// The chunk ranges THIS fill will produce — exactly the bytes this pull
    /// fetches (its `missing_ranges ∩ R`). [`FillRegistry::range_still_live`]
    /// intersects a reader's node range against the union of all live sessions'
    /// `covered` to decide whether to keep awaiting or fail. Set at registration via
    /// [`Self::set_covered`]; defaults to [`ChunkRanges::all`] so a session built but
    /// not yet range-scoped coalesces conservatively.
    covered: StdMutex<ChunkRanges>,
    /// Live observer count: the pull owner plus every serve leg attached to this
    /// fill. An [`ObserverLease`] increments on attach and decrements on drop; when
    /// it reaches 0 while the pull is still running ([`Self::outcome`] is `None`),
    /// the last-out drop fires [`Self::cancel`] to stop ingest (#1610).
    observers: AtomicUsize,
    /// Cancels the pull when the last observer leaves before the pull ends. The
    /// node's pull leg selects on it to abort ingest; the tagged partial persists
    /// for a later resume.
    cancel: CancellationToken,
    /// A back-link to the owning registry + this session's hash, set once at
    /// registration ([`Self::bind_registry`]). It lets a reader consult ALL
    /// live sessions for the hash via [`FillRegistry::range_still_live`]. A session
    /// built standalone (in tests) leaves this unset and falls back to its own
    /// single-fill liveness — correct, because a standalone session is the only fill.
    registry: OnceLock<(Weak<FillRegistry>, Hash)>,
    /// The node's pull-thread [`JoinHandle`](std::thread::JoinHandle), parked here by
    /// the owner right after it spawns the pull, so whichever observer leaves LAST
    /// joins it. That frees an owner whose OWN client finishes first from parking in
    /// the join while other observers still stream: a non-last-out release returns
    /// immediately, and the true last-out [`ObserverLease`] drop takes the handle back
    /// out to join OFF the map lock. `None` when no pull was spawned (an attach, or a
    /// spawn that failed).
    pull_thread: StdMutex<Option<std::thread::JoinHandle<()>>>,
}

impl FillSession {
    /// Construct the fill session for a `total_bytes`-byte blob rooted at `root`,
    /// with its served frontier at 0. An owning branch for a request that starts
    /// past byte 0 uses [`Self::starting_at`] instead.
    #[must_use]
    pub fn new(root: blake3::Hash, total_bytes: u64) -> Arc<Self> {
        Self::starting_at(root, total_bytes, 0)
    }

    /// Construct the fill session for a `total_bytes`-byte blob rooted at `root`,
    /// with its served frontier at `served_start`: the ABSOLUTE content offset the
    /// request starts at, so a non-zero-offset request shows no phantom lead. The
    /// session starts with a private per-hash outboard, replaced by the registry's
    /// canonical one at registration.
    #[must_use]
    pub fn starting_at(root: blake3::Hash, total_bytes: u64, served_start: u64) -> Arc<Self> {
        let downstream_advanced = Arc::new(Notify::new());
        Arc::new(Self {
            outboard: StdMutex::new(HashOutboard::new(root, total_bytes)),
            ended: StdMutex::new(None),
            served_paid: Frontier::new(served_start, Arc::clone(&downstream_advanced)),
            served_start,
            serve_demand: Demand::new(downstream_advanced),
            covered: StdMutex::new(ChunkRanges::all()),
            observers: AtomicUsize::new(0),
            cancel: CancellationToken::new(),
            registry: OnceLock::new(),
            pull_thread: StdMutex::new(None),
        })
    }

    /// Park the owner's pull-thread handle on the session so whichever observer leaves
    /// LAST joins it. Called once, on an owning branch, right after the pull thread is
    /// spawned. The pure-attach path never calls this — it drives no pull of its own.
    ///
    /// Refuses to overwrite an already-parked handle: a session has exactly one owning
    /// pull, so a second park cannot happen — but if a future regression called this
    /// twice, overwriting would DROP the first handle and detach its thread, orphaning
    /// a pull that keeps paying/draining. So keep the first (the one the last-out
    /// observer will join) and `debug_assert` the double-park loudly in tests/dev.
    pub fn set_pull_handle(&self, handle: std::thread::JoinHandle<()>) {
        let mut slot = self
            .pull_thread
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        debug_assert!(
            slot.is_none(),
            "a FillSession parks exactly one pull-thread handle; a second park would \
             detach the first and orphan its pull"
        );
        if slot.is_none() {
            *slot = Some(handle);
        }
    }

    /// Take the parked pull-thread handle, if any. Called only by the last-out lease
    /// teardown, so exactly one caller ever receives it — the one that then joins the
    /// thread off-task.
    fn take_pull_handle(&self) -> Option<std::thread::JoinHandle<()>> {
        self.pull_thread
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .take()
    }

    /// The per-hash outboard this session currently references. Cloned out (a cheap
    /// `Arc` bump) rather than borrowed, so the swap lock is never held across a
    /// capture or a read.
    fn outboard(&self) -> Arc<HashOutboard> {
        self.outboard
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
    }

    /// Adopt the hash's canonical outboard (shared with sibling sessions). Called
    /// once, under the registry lock, before this session drives a pull or mints a
    /// reader — so every later capture / read lands in the one shared buffer.
    fn adopt_outboard(&self, shared: Arc<HashOutboard>) {
        *self.outboard.lock().unwrap_or_else(PoisonError::into_inner) = shared;
    }

    /// Bind this session to its registry + hash, so a reader it mints can ask the
    /// registry which live fills still cover a range. Called once at registration.
    fn bind_registry(&self, registry: Weak<FillRegistry>, hash: Hash) {
        let _ = self.registry.set((registry, hash));
    }

    /// Set the chunk ranges this fill will produce. Called once at registration,
    /// before the pull streams — the registry reads it to plan observer attach and
    /// to answer [`Self::range_still_live`].
    pub fn set_covered(&self, covered: ChunkRanges) {
        *self.covered.lock().unwrap_or_else(PoisonError::into_inner) = covered;
    }

    /// A clone of this fill's `covered` ranges.
    fn covered_ranges(&self) -> ChunkRanges {
        self.covered
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
    }

    /// The blob's total byte length. The attach path signs its `StreamResponse`
    /// from this without re-running the upstream header handshake — the session
    /// already knows the geometry from its `BaoTree`.
    #[must_use]
    pub fn total_bytes(&self) -> u64 {
        self.outboard().tree.size()
    }

    /// The ABSOLUTE content offset this session's owning request starts at.
    #[must_use]
    pub const fn served_start(&self) -> u64 {
        self.served_start
    }

    /// The current PAID content frontier (absolute): how far the observers of
    /// this fill have paid, and therefore how far past it the pull may run.
    #[must_use]
    pub fn served_paid(&self) -> u64 {
        self.served_paid.get()
    }

    /// The live observer count (pull owner + attached serve legs).
    #[must_use]
    pub fn observer_count(&self) -> usize {
        self.observers.load(Ordering::Acquire)
    }

    /// Whether the pull has been cancelled because the last observer left.
    #[must_use]
    pub fn is_cancelled(&self) -> bool {
        self.cancel.is_cancelled()
    }

    /// The pull leg's cancellation token — it selects on this to abort ingest when
    /// the last observer leaves before the pull completes.
    #[must_use]
    pub const fn cancel_token(&self) -> &CancellationToken {
        &self.cancel
    }

    /// Whether this session is dead: the pull has recorded a terminal outcome, or
    /// the fill was cancelled because its last observer left. A dead session must
    /// not block a fresh pull (its `covered` is excluded from the coverage union)
    /// and must not be an attach target.
    #[must_use]
    fn is_dead(&self) -> bool {
        self.outcome().is_some() || self.cancel.is_cancelled()
    }

    /// Mint a serve-side [`SessionOutboardReader`] over this hash's shared outboard.
    /// The reader snapshots the (already-adopted) outboard and holds this session so
    /// its termination race can reach the registry. A serve path attaches the store
    /// ([`SessionOutboardReader::with_store_fallback`]); without it, a load of a node
    /// no live fill captures fails.
    #[must_use]
    pub fn outboard_reader(self: &Arc<Self>) -> SessionOutboardReader {
        SessionOutboardReader {
            session: Arc::clone(self),
            outboard: self.outboard(),
            parked_on: Arc::new(AtomicU64::new(0)),
            store: None,
        }
    }

    /// The shared PAID content frontier the pull leg's pacer bounds against.
    #[must_use]
    pub const fn served_frontier(&self) -> &Frontier {
        &self.served_paid
    }

    /// Raise the PAID content frontier to `served` after a downstream voucher batch
    /// commits, waking the parked pull if it moved. A lower value changes nothing,
    /// so N observers advancing one shared frontier never regress each other.
    pub fn advance_served(&self, served: u64) {
        self.served_paid.raise(served);
    }

    /// Raise the PAID content frontier to `served` on behalf of an observer whose
    /// delivery starts at content `offset`, but only when this session's frontier
    /// already reaches `offset`.
    ///
    /// Under partial-overlap coalescing an observer consumes a SUFFIX of this fill's
    /// covered range, starting at `offset`. Its payment extends this fill's paid
    /// PREFIX only once that prefix reaches `offset`. Before that, the bytes in
    /// `[frontier, offset)` are unpaid, and raising the frontier past them would let
    /// the pull run its window over bytes no observer of this fill has paid for.
    pub fn extend_served_from(&self, offset: u64, served: u64) {
        if self.served_paid.get() >= offset {
            self.served_paid.raise(served);
        }
    }

    /// A read-only watch over both downstream frontiers, for a pull leg to pace
    /// against and park on.
    #[must_use]
    pub fn downstream_watch(&self) -> DownstreamWatch {
        DownstreamWatch {
            served_paid: self.served_paid.clone(),
            serve_demand: self.serve_demand.clone(),
        }
    }

    /// The serve demand of this fill. When its nearest value lies within one chunk
    /// group past a pull's frontier, that pull's pacer draws one window floor even
    /// with its window full; a demand further out is ignored.
    #[must_use]
    pub const fn serve_demand(&self) -> &Demand {
        &self.serve_demand
    }

    /// Mint the [`DemandSlot`] one serve leg reading this fill stands its demand
    /// through. Under coalescing the awaited byte may belong to a sibling fill of
    /// the hash, so the slot routes each demand through the registry.
    #[must_use]
    pub fn demand_slot(self: &Arc<Self>) -> DemandSlot {
        DemandSlot {
            session: Arc::clone(self),
            id: NEXT_DEMAND_SLOT.fetch_add(1, Ordering::Relaxed),
        }
    }

    /// The owning registry and this session's hash, when registered and the
    /// registry is still alive.
    fn registry_and_hash(&self) -> Option<(Arc<FillRegistry>, Hash)> {
        self.registry
            .get()
            .and_then(|(weak, hash)| weak.upgrade().map(|registry| (registry, *hash)))
    }

    /// Stand slot `slot`'s demand for content up to `end` here when this live fill
    /// produces `byte`, and withdraw it otherwise.
    fn route_own_demand(&self, slot: u64, byte: u64, end: u64) {
        let chunk = bao_tree::ChunkNum::full_chunks(byte);
        let produces = !self.is_dead() && {
            let covered = self.covered.lock().unwrap_or_else(PoisonError::into_inner);
            !(&*covered & &ChunkRanges::from(chunk..chunk + 1)).is_empty()
        };
        if produces {
            self.serve_demand.stand(slot, end);
        } else {
            self.serve_demand.withdraw(slot);
        }
    }

    /// The per-hash liveness signal, notified whenever any session for this hash
    /// ends or is cancelled. A serve-side data reader awaits it (alongside the
    /// present-range watch) so a read of a range no live pull will fill settles
    /// rather than hanging: served from the store when it holds the bytes, failed
    /// otherwise. Handed out as an owned `Arc<Notify>` so the node's data reader
    /// need not name the cache-private [`HashOutboard`].
    #[must_use]
    pub fn liveness_signal(&self) -> Arc<Notify> {
        Arc::clone(&self.outboard().liveness)
    }

    /// Whether any LIVE fill of this hash still covers `range`, so a not-yet-present
    /// node / leaf in `range` may still be filled. A registry-bound session consults
    /// EVERY live session for the hash (the N-fill generalization); a standalone
    /// session (unregistered, in tests) is the only fill, so it answers from its own
    /// liveness. Returns `false` when nothing live covers `range` — the caller then
    /// fails the read instead of hanging.
    #[must_use]
    pub fn range_still_live(&self, range: &ChunkRanges) -> bool {
        match self.registry.get() {
            Some((weak, hash)) => match weak.upgrade() {
                Some(registry) => registry.range_still_live(*hash, range),
                // The registry (the whole cache) is gone; nothing can fill anything.
                None => false,
            },
            // Standalone: this session is the only fill, so its liveness decides.
            None => !self.is_dead() && !(&self.covered_ranges() & range).is_empty(),
        }
    }

    /// Capture one internal node's `(left, right)` hash pair into the per-hash
    /// outboard. Idempotent; a leaf (no outboard slot) is ignored. Every
    /// `admit_bao_stream` for the hash, from any pull or source, calls this — so the
    /// one shared buffer accumulates the whole tree's proof.
    pub fn capture(&self, node: TreeNode, pair: (blake3::Hash, blake3::Hash)) {
        self.outboard().capture(node, pair);
    }

    /// Capture a batch of node pairs into the per-hash outboard in one shot, waking
    /// parked serve legs once for the batch rather than once per node. The admit path
    /// hands each leaf's run of proof nodes here just before the leaf, so a serve leg
    /// reads a leaf's proof as soon as that leaf arrives, with at most one wake per
    /// leaf.
    pub fn capture_many(
        &self,
        pairs: impl IntoIterator<Item = (TreeNode, (blake3::Hash, blake3::Hash))>,
    ) {
        self.outboard().capture_many(pairs);
    }

    /// Record the pull leg's terminal outcome and wake parked readers. Idempotent
    /// on the wake; the outcome is set once by the pull leg on its single exit.
    /// Fires the per-hash liveness so a reader awaiting a range this fill covered
    /// re-checks [`Self::range_still_live`] at once.
    pub fn mark_ended(&self, result: Result<(), FillError>) {
        {
            let mut guard = self.ended.lock().unwrap_or_else(PoisonError::into_inner);
            *guard = Some(result);
        }
        self.outboard().liveness.notify_waiters();
    }

    /// The pull leg's terminal outcome, if it has ended: `Some(Ok)` clean,
    /// `Some(Err)` failed. `None` while the pull is still running.
    #[must_use]
    pub fn outcome(&self) -> Option<Result<(), FillError>> {
        self.ended
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
    }
}

/// Serve-side [`Outboard`] over a hash's shared [`HashOutboard`]. Its
/// [`Outboard::load`] awaits a not-yet-captured node, racing the registry-wide fill
/// liveness for the no-hang guarantee: once no live fill still covers the node's
/// byte range, it reads the node from the local store when the reader carries a
/// store fallback ([`Self::with_store_fallback`]), and fails when that cannot
/// supply it. Minted via [`FillSession::outboard_reader`].
#[derive(Debug)]
pub struct SessionOutboardReader {
    session: Arc<FillSession>,
    /// The shared per-hash outboard, snapshot at mint (after the session adopted the
    /// canonical one), so every `load` reads whatever any pull for the hash captured.
    outboard: Arc<HashOutboard>,
    /// The content end this reader waits on, written before each park: one byte
    /// into the node whose pair no pull has captured. The reader does NOT raise
    /// serve demand itself — an encoder reads ahead of what its consumer needs, so
    /// a park here alone does not mean the serve leg is stuck. The frame consumer
    /// stands this value via its [`DemandSlot`] only once it is
    /// starved of encoded bytes ([`Self::parked_on`]).
    parked_on: Arc<AtomicU64>,
    /// The local store, read for a node no live fill will capture
    /// ([`Self::with_store_fallback`]). `None` fails such a load.
    store: Option<StoreFallback>,
}

/// The local store a [`SessionOutboardReader`] reads a node's pair from when no
/// live fill will capture it.
#[derive(Debug)]
struct StoreFallback {
    engine: CacheEngine,
    /// The store's present chunks of the hash as last read. A node they already
    /// meet needs no fresh read.
    present: ChunkRanges,
}

impl StoreFallback {
    /// Capture into `outboard` the pairs `export_bao` emits for ONE chunk group the
    /// store holds under a node of chunk range `node_range`: every parent on the
    /// group's proof path, the node among them, plus the right-siblings over
    /// absent content. The store reads and verifies at most that one group's
    /// leaves, however much content the node spans. Captures nothing when the
    /// store holds nothing under the node. Returns whether the export yielded any
    /// pairs.
    async fn capture(
        &mut self,
        outboard: &HashOutboard,
        node_range: &ChunkRanges,
    ) -> CacheResult<bool> {
        let hash = Hash::from_bytes(*outboard.root.as_bytes());
        if (node_range & &self.present).is_empty() {
            self.present = self
                .engine
                .present_ranges(hash)
                .await?
                .chunk_ranges()
                .clone();
        }
        let held = node_range & &self.present;
        let Some(first) = held.boundaries().first().copied() else {
            return Ok(false);
        };
        let group_chunks = 1u64 << IROH_BLOCK_SIZE.chunk_log();
        let group_start = first.0 / group_chunks * group_chunks;
        let group = ChunkRanges::from(
            ChunkNum(group_start)..ChunkNum(group_start.saturating_add(group_chunks)),
        );
        let pairs = self.engine.outboard_pairs(hash, &(&held & &group)).await?;
        let exported = !pairs.is_empty();
        outboard.capture_many(pairs);
        Ok(exported)
    }
}

impl SessionOutboardReader {
    /// Read a node no live fill will capture from `engine`, the local store, before
    /// failing the load. The store can hold content under such a node: a pull that
    /// faulted landed part of its range first, or an earlier fill stored it. The
    /// store's outboard then holds the node's pair. Without the fallback such a
    /// load fails.
    #[must_use]
    pub fn with_store_fallback(mut self, engine: CacheEngine) -> Self {
        self.store = Some(StoreFallback {
            engine,
            present: ChunkRanges::empty(),
        });
        self
    }

    /// The cell this reader writes the content end it waits on into. A frame
    /// consumer shares it with its data reader and stands it as serve demand
    /// while it has no encoded bytes left to take.
    #[must_use]
    pub fn parked_on(&self) -> Arc<AtomicU64> {
        Arc::clone(&self.parked_on)
    }

    /// Log a node the store fallback supplied. After a failed pull, or once
    /// every covering pull ended, the store is the expected source: debug. After
    /// this session's own pull completed cleanly, the pull should have captured
    /// the node itself, so the fallback covers a capture defect: warn.
    fn log_store_hit(&self, node: TreeNode) {
        let hash = self.outboard.root;
        if matches!(self.session.outcome(), Some(Ok(()))) {
            tracing::warn!(
                %hash,
                ?node,
                "a clean pull never captured an outboard node; read it from the store"
            );
        } else {
            tracing::debug!(
                %hash,
                ?node,
                "read an outboard node no live fill will capture from the store"
            );
        }
    }

    /// The terminal error for a node no live fill will supply: a precise message
    /// when this session's own pull ended (the standalone / single-fill case), a
    /// generic one when the covering fills were siblings.
    fn dead_range_error(&self, node: TreeNode) -> io::Error {
        match self.session.outcome() {
            Some(Err(msg)) => io::Error::other(format!(
                "upstream pull failed before supplying outboard node {node:?}: {msg}"
            )),
            Some(Ok(())) => io::Error::other(format!(
                "upstream pull completed but outboard node {node:?} was never captured"
            )),
            None => io::Error::other(format!(
                "no live fill covers outboard node {node:?}; every covering pull ended"
            )),
        }
    }
}

impl Outboard for SessionOutboardReader {
    fn root(&self) -> blake3::Hash {
        self.outboard.root
    }

    fn tree(&self) -> BaoTree {
        self.outboard.tree
    }

    async fn load(&mut self, node: TreeNode) -> io::Result<Option<(blake3::Hash, blake3::Hash)>> {
        let Some(offset) = self.outboard.tree.pre_order_offset(node) else {
            return Ok(None); // leaf: bao_tree sources this from the data reader
        };
        let idx = usize::try_from(offset).unwrap_or(usize::MAX);
        let node_range = ChunkRanges::from(node.chunk_range());
        loop {
            // Register the wakers BEFORE inspecting shared state, so a capture /
            // fill-end recorded concurrently cannot slip between the check and the
            // await.
            let captured = self.outboard.captured.notified();
            let liveness = self.outboard.liveness.notified();
            tokio::pin!(captured);
            tokio::pin!(liveness);
            captured.as_mut().enable();
            liveness.as_mut().enable();

            if let Some(pair) = self.outboard.try_load(idx) {
                return Ok(Some(pair));
            }

            // Not captured yet. If no live fill still covers this node's range, no
            // pull will capture the node — decide now. A clean end may lag one
            // capture behind the outcome, so re-check once. Then read the node
            // from the store, which may hold content under it that a fill landed
            // before it ended, and fail only when the store cannot supply it.
            if !self.session.range_still_live(&node_range) {
                if let Some(pair) = self.outboard.try_load(idx) {
                    return Ok(Some(pair));
                }
                let exported = match self.store.as_mut() {
                    Some(fallback) => fallback.capture(&self.outboard, &node_range).await,
                    None => Ok(false),
                };
                match exported {
                    Ok(exported) => {
                        if let Some(pair) = self.outboard.try_load(idx) {
                            if exported {
                                self.log_store_hit(node);
                            }
                            return Ok(Some(pair));
                        }
                        if exported {
                            return Err(io::Error::other(format!(
                                "the store holds content under outboard node {node:?} but \
                                 its export did not yield the node"
                            )));
                        }
                    }
                    Err(err) => {
                        return Err(io::Error::other(format!(
                            "{}; the store read for it failed: {}",
                            self.dead_range_error(node),
                            ErrorChain::new(&err)
                        )));
                    }
                }
                return Err(self.dead_range_error(node));
            }

            // A pull captures this pair only once it fetches into the node's range.
            // Record the first byte of that range as what this encode waits on; the
            // frame consumer turns it into serve demand if it starves on this park.
            let node_start = node.chunk_range().start.to_bytes();
            self.parked_on
                .store(node_start.saturating_add(1), Ordering::Relaxed);

            // Await the next capture or a liveness change, then re-check.
            tokio::select! {
                biased;
                () = liveness.as_mut() => {}
                () = captured.as_mut() => {}
            }
        }
    }
}

/// RAII observer handle on a [`FillSession`]. One is minted for the pull owner and
/// one per attached serve leg ([`FillRegistry::claim`]). Releasing it decrements the
/// session's observer count; the last-out release, if the pull is still running,
/// cancels the pull so a fill no client is waiting on stops ingesting (#1610) — and
/// hands back the session's parked pull-thread handle so THAT caller joins it
/// off-task (the owner-join hand-off: whoever leaves LAST joins the pull, freeing an
/// owner whose own client finished first).
///
/// The decrement + cancel decision runs UNDER the registry map lock — the same
/// lock `claim`/`register_fill` hold when they attach an observer. That mutual
/// exclusion closes the resurrection race: a concurrent claim cannot `fetch_add` a
/// session 0→1 in the window between this drop's last-out `fetch_sub` (1→0) and its
/// `cancel()`, so no live observer can end up bound to a cancelled fill. The lease
/// therefore holds a [`Weak`] back to the registry (plus the session's `hash`) to
/// reach that lock at drop. If the upgrade fails the registry is gone, so the fill
/// is moot and the drop just skips. It only TAKES the pull handle under the lock;
/// the JOIN always runs off it.
#[derive(Debug)]
pub struct ObserverLease {
    session: Arc<FillSession>,
    registry: Weak<FillRegistry>,
    hash: Hash,
    /// Cleared once teardown has run. [`Self::release`] consumes the lease and runs
    /// teardown explicitly; the value then still `Drop`s, so this flag makes the
    /// second pass a no-op — the decrement fires exactly once.
    armed: bool,
}

impl ObserverLease {
    /// Release the lease at a serve leg's teardown, handing back the parked
    /// pull-thread handle when THIS release is the last one out. The caller joins that
    /// handle OFF the registry map lock and off its accept task (on a blocking
    /// thread). A non-last-out release returns `None`, so an owner whose own client
    /// finished first returns at once, leaving the pull filling for the remaining
    /// observers — the last of which cancels it (#1610 preserved). Named so teardown
    /// reads as intent rather than an incidental drop.
    #[must_use]
    pub fn release(mut self) -> Option<std::thread::JoinHandle<()>> {
        self.teardown()
    }

    /// The last-out decrement + cancel, shared by [`Self::release`] and [`Drop`].
    /// Runs under the registry map lock — the SAME lock `claim`/`register_fill` hold
    /// to attach — so the decrement, the map removal, and the cancel decision cannot
    /// interleave with an attach (the #1610 resurrection-race fix). It only TAKES the
    /// pull handle here; it never JOINS under the lock. Idempotent via `armed`.
    fn teardown(&mut self) -> Option<std::thread::JoinHandle<()>> {
        if !self.armed {
            return None;
        }
        self.armed = false;
        let Some(registry) = self.registry.upgrade() else {
            // Registry dropped — the fill (and its map entry) are gone; nothing can
            // attach, so the decrement + cancel are moot. Skip.
            return None;
        };
        // Take the SAME lock `claim`/`register_fill` hold to attach, so the
        // decrement, the map removal, and the cancel decision cannot interleave with
        // an attach.
        let mut map = registry.map.lock().unwrap_or_else(PoisonError::into_inner);
        // `fetch_sub` returns the PREVIOUS value; `1` means this drop took the
        // count to 0.
        let prev = self.session.observers.fetch_sub(1, Ordering::AcqRel);
        if prev != 1 {
            return None;
        }
        // Last observer left: free the session from the map so dead sessions do not
        // accumulate forever. Remove by pointer identity so a sibling session for the
        // same hash survives; drop the `hash` key if its Vec empties (the per-hash
        // outboard then drops once every reader releases it).
        if let Some(entry) = map.get_mut(&self.hash) {
            entry.sessions.retain(|s| !Arc::ptr_eq(s, &self.session));
            if entry.sessions.is_empty() {
                map.remove(&self.hash);
            }
        }
        // Cancel only if the pull has not already ended — a completed pull needs no
        // cancel, and a failed one already terminated. The session is removed from the
        // map regardless. Firing the per-hash liveness wakes any reader parked on a
        // range this fill covered so it re-checks at once.
        if self.session.outcome().is_none() {
            self.session.cancel.cancel();
        }
        self.session.outboard().liveness.notify_waiters();
        // Hand the parked pull-thread handle to this last-out caller to join off-task.
        // Taking (not joining) it here keeps the map lock free of a blocking join.
        self.session.take_pull_handle()
    }
}

impl Drop for ObserverLease {
    fn drop(&mut self) {
        let _ = self.teardown();
    }
}

/// The atomic outcome of [`FillRegistry::claim`]: a serve-miss either coalesces
/// wholly onto live pulls (`Attach`), owns a fresh pull for its whole range
/// (`Owner`), or — on a partial overlap — owns a fresh pull for its non-overlapping
/// REMAINDER while attaching to a live sibling for the overlap (`Mixed`). Every
/// variant carries the [`FillSession`]s and [`ObserverLease`]s the caller holds
/// for the life of its serve leg. `claim` decides the split AND registers any new
/// owner session under ONE map-lock acquisition, so two concurrent fresh misses for
/// the same blob cannot both open a pull.
#[derive(Debug)]
pub enum FillClaim {
    /// A live pull already covers the whole request; run a serve leg over `session`
    /// and open NO new pull. `lease` is an observer lease on the existing fill.
    Attach {
        /// The live session to serve from (largest overlap with the request).
        session: Arc<FillSession>,
        /// The observer lease held for the life of the attaching serve leg.
        lease: ObserverLease,
    },
    /// No live pull covers the request; the caller owns the freshly-registered
    /// `session` (scoped to the WHOLE request) and must spawn the pull. `lease` is
    /// the owner lease (observer count starts at 1).
    Owner {
        /// The freshly-registered session the caller must drive a pull for.
        session: Arc<FillSession>,
        /// The owner lease (count 1) held for the life of the owning pull.
        lease: ObserverLease,
    },
    /// A partial overlap: a live sibling covers a prefix/suffix of the request. The
    /// caller OWNS a fresh pull for the contiguous `remainder` (`[remainder_offset,
    /// remainder_offset + remainder_len)`) and ATTACHES to the sibling for the
    /// overlap, serving the whole request from the shared per-hash cache + outboard.
    /// Two pulls, each fetching its bytes once.
    Mixed {
        /// The freshly-registered session covering the remainder; drive its pull.
        owner: Arc<FillSession>,
        /// The owner lease (count 1) for the remainder pull.
        owner_lease: ObserverLease,
        /// The live sibling covering the overlap; its pull is shared, not re-opened.
        attach: Arc<FillSession>,
        /// The observer lease on the sibling, held for the life of the serve leg.
        attach_lease: ObserverLease,
        /// The remainder's byte offset — the pull leg fetches `[offset, offset+len)`.
        remainder_offset: u64,
        /// The remainder's byte length.
        remainder_len: u64,
    },
}

/// Total chunk count spanned by a (disjoint, sorted) chunk-range set. Boundaries
/// come in `[start, end)` pairs; a trailing unpaired boundary (an open-ended set)
/// contributes nothing, which is fine for the finite `covered ∩ R` overlaps this
/// measures.
fn chunk_span(ranges: &ChunkRanges) -> u64 {
    let (pairs, _open_end) = ranges.boundaries().as_chunks::<2>();
    pairs
        .iter()
        .map(|[start, end]| end.0.saturating_sub(start.0))
        .sum()
}

/// The single contiguous byte span `[offset, offset+len)` of a chunk-range set, or
/// `None` if it is empty or split into two or more disjoint pieces. A `Mixed` claim
/// only opens a remainder pull when the remainder is one contiguous range (a
/// prefix/suffix overlap), because the pull leg fetches one `[offset, len)` gap; a
/// remainder split by an interior overlap falls back to a conservative whole-request
/// pull. `total` clamps the final (possibly ragged) group to the blob end.
fn contiguous_byte_span(ranges: &ChunkRanges, total: u64) -> Option<(u64, u64)> {
    match ranges.boundaries() {
        [start, end] => {
            let offset = start.0.saturating_mul(CHUNK_BYTES);
            let end_bytes = end.0.saturating_mul(CHUNK_BYTES).min(total);
            Some((offset, end_bytes.saturating_sub(offset)))
        }
        _ => None,
    }
}

/// One hash's live in-flight fills plus the ONE per-hash outboard they all capture
/// into and serve from.
#[derive(Debug)]
struct HashEntry {
    /// The canonical per-hash outboard, adopted by every session for this hash.
    outboard: Arc<HashOutboard>,
    /// The live fills for this hash (each a distinct pull). Removed by pointer
    /// identity when a session's last observer leaves.
    sessions: Vec<Arc<FillSession>>,
}

/// Range-aware coalescing registry: the live in-flight fills, keyed by hash, plus a
/// per-hash captured outboard. A serve-miss consults [`Self::claim`] to split its
/// range into an attach (covered by a running pull) and a remainder (fetch it), and
/// [`Self::range_still_live`] answers a parked reader's "keep awaiting or fail?".
/// Purely synchronous range math under a std `Mutex`; the lock is never held across
/// an await.
#[derive(Debug, Default)]
pub struct FillRegistry {
    map: StdMutex<HashMap<Hash, HashEntry>>,
    /// Cache metrics for the coalescing decisions this registry makes; `None`
    /// in unit fixtures.
    metrics: Option<Arc<crate::metrics::CacheMetrics>>,
}

impl FillRegistry {
    /// An empty registry with no metrics handle (unit fixtures).
    #[must_use]
    pub fn new() -> Self {
        Self::with_metrics(None)
    }

    /// An empty registry that reports its coalescing decisions to `metrics`.
    #[must_use]
    pub fn with_metrics(metrics: Option<Arc<crate::metrics::CacheMetrics>>) -> Self {
        Self {
            map: StdMutex::new(HashMap::new()),
            metrics,
        }
    }

    /// Increment `session`'s observer count and mint its RAII lease. The caller
    /// MUST already hold the map lock — the increment shares the lock the lease's
    /// drop takes for its decrement, so attach and teardown are mutually exclusive.
    fn mint_lease(self: &Arc<Self>, session: &Arc<FillSession>, hash: Hash) -> ObserverLease {
        session.observers.fetch_add(1, Ordering::AcqRel);
        ObserverLease {
            session: Arc::clone(session),
            registry: Arc::downgrade(self),
            hash,
            armed: true,
        }
    }

    /// Insert a freshly-built `session` for `hash` under the held map lock: adopt the
    /// hash's canonical outboard (or seed it from this session if it is the first),
    /// bind the session's registry back-link, and publish it. The session's
    /// `covered` must already be set by the caller.
    fn insert_session(
        self: &Arc<Self>,
        map: &mut HashMap<Hash, HashEntry>,
        hash: Hash,
        session: &Arc<FillSession>,
    ) {
        let entry = map.entry(hash).or_insert_with(|| HashEntry {
            outboard: session.outboard(),
            sessions: Vec::new(),
        });
        session.adopt_outboard(Arc::clone(&entry.outboard));
        session.bind_registry(Arc::downgrade(self), hash);
        entry.sessions.push(Arc::clone(session));
    }

    /// Publish a new pull's `session` (already scoped via [`FillSession::set_covered`]
    /// to the bytes it will fetch) and return the owner lease (observer count starts
    /// at 1). The entry stays in the map until the owning pull is joined node-side.
    /// A test-facing helper; production registers via [`Self::claim`].
    pub fn register_fill(
        self: &Arc<Self>,
        hash: Hash,
        session: &Arc<FillSession>,
    ) -> ObserverLease {
        let mut map = self.map.lock().unwrap_or_else(PoisonError::into_inner);
        self.insert_session(&mut map, hash, session);
        self.mint_lease(session, hash)
    }

    /// The blob length of a LIVE in-flight fill for `hash`, if one runs — an
    /// ADVISORY peek the serve-miss path uses to skip its upstream header handshake
    /// when a pull it can coalesce onto already exists. Returns the `total_bytes` of
    /// the first non-dead session under `hash` (every session for one hash shares
    /// the blob's geometry, so any live one's length is THE length); `None` when no
    /// live fill exists (empty, or only dead sessions remain). Advisory only: a
    /// concurrent last-observer drop can retire the peeked session before the
    /// caller's atomic [`Self::claim`], which then owns a fresh pull and opens the
    /// leg late — so this NEVER replaces `claim`, it only removes the handshake in
    /// the common coalescing case.
    #[must_use]
    pub fn in_flight_total(&self, hash: Hash) -> Option<u64> {
        let map = self.map.lock().unwrap_or_else(PoisonError::into_inner);
        map.get(&hash)?
            .sessions
            .iter()
            .find(|session| !session.is_dead())
            .map(|session| session.total_bytes())
    }

    /// Route slot `slot`'s demand for content up to `end` to the live fill of `hash`
    /// that produces `byte`, and withdraw it from every other fill of the hash
    /// ([`DemandSlot::stand`]).
    fn route_demand(&self, hash: Hash, slot: u64, byte: u64, end: u64) {
        let map = self.map.lock().unwrap_or_else(PoisonError::into_inner);
        if let Some(entry) = map.get(&hash) {
            for session in &entry.sessions {
                session.route_own_demand(slot, byte, end);
            }
        }
    }

    /// Withdraw slot `slot`'s demand from every fill of `hash`.
    fn withdraw_demand(&self, hash: Hash, slot: u64) {
        let map = self.map.lock().unwrap_or_else(PoisonError::into_inner);
        if let Some(entry) = map.get(&hash) {
            for session in &entry.sessions {
                session.serve_demand.withdraw(slot);
            }
        }
    }

    /// Whether any LIVE fill of `hash` still covers `range`. A parked reader calls
    /// this (via [`FillSession::range_still_live`]) to decide "keep awaiting a
    /// capture / present-range advance, or fail because no pull will ever fill this".
    /// `true` iff some live session's `covered` intersects `range`.
    #[must_use]
    pub fn range_still_live(&self, hash: Hash, range: &ChunkRanges) -> bool {
        let map = self.map.lock().unwrap_or_else(PoisonError::into_inner);
        match map.get(&hash) {
            Some(entry) => entry
                .sessions
                .iter()
                .any(|s| !s.is_dead() && !(&s.covered_ranges() & range).is_empty()),
            None => false,
        }
    }

    /// Atomically decide attach/own/mixed AND register any new owner session, all
    /// under ONE map-lock acquisition. This closes the coalescing TOCTOU race a
    /// separate plan-then-register pair would leave open: between two lock
    /// acquisitions another thread can register, so two concurrent fresh misses for
    /// the same blob would both see an empty registry and both open a pull. `claim`
    /// holds the lock across the whole decision, so exactly one of two racing
    /// identical claims registers an owner and the other attaches.
    ///
    /// `R = align_range_clamped(offset, len, total)`: an end past `total` clamps to
    /// it, so a request whose claimed size grows still coalesces
    /// with a live fill instead of always falling to Owner. On an align error (a
    /// start at or past `total`) or empty `R`, the caller takes the Owner path (its
    /// own fetch surfaces any out-of-bounds error).
    /// Otherwise, over LIVE sessions **whose paid frontier reaches `R`'s aligned
    /// fetch start** (a session behind that frontier is excluded — attaching to it
    /// would park this request on a pull that advances only as the OTHER client
    /// pays, #2062; the exclusion trades duplicate egress for the overlap, counted
    /// as `fill_not_coalesced`), `covered_union = ⋃ covered`,
    /// `attach = R ∩ covered_union`, `remainder = R − covered_union`:
    /// - `attach` empty → OWNER of the whole `R`.
    /// - `remainder` empty (a live pull covers all of `R`) → ATTACH to the
    ///   largest-overlap session; `make_session` is NOT called.
    /// - both non-empty, and the largest-overlap sibling covers ALL of `attach`, and
    ///   `R − sibling.covered` is ONE contiguous span → MIXED: own a pull for that
    ///   remainder, attach the sibling for the overlap.
    /// - otherwise (a multi-sibling union, or an interior overlap splitting the
    ///   remainder in two) → OWNER of the whole `R`, the conservative fallback.
    ///   It never wedges; like the frontier exclusion above it can double-pull
    ///   an overlap a live sibling also covers.
    ///
    /// `make_session` runs under the lock only on an owning branch (never
    /// built-and-dropped on a pure attach); it does no await (a std lock) and only
    /// allocates, so holding the lock across it is safe.
    pub fn claim(
        self: &Arc<Self>,
        hash: Hash,
        offset: u64,
        len: u64,
        total: u64,
        make_session: impl FnOnce() -> Arc<FillSession>,
    ) -> FillClaim {
        let mut map = self.map.lock().unwrap_or_else(PoisonError::into_inner);

        // R = the chunk ranges the request spans. An align error (a start at or
        // past the blob end) or an empty R has no coalescable range, so fall
        // straight to the Owner path. An end past the blob clamps instead.
        let (r, fetch_start) = match align_range_clamped(offset, len, total) {
            Ok(aligned) => (aligned.chunk_ranges().clone(), aligned.fetch_start()),
            Err(_) => (ChunkRanges::empty(), 0),
        };

        if !r.is_empty() {
            let mut covered_union = ChunkRanges::empty();
            let mut best: Option<(Arc<FillSession>, ChunkRanges, u64)> = None;
            let mut not_coalesced = false;
            if let Some(entry) = map.get(&hash) {
                for session in &entry.sessions {
                    // Dead sessions (ended or cancelled) may never deliver, so their
                    // `covered` neither blocks a fresh pull nor is an attach target.
                    // Reading `is_dead` under the map lock pins it against a
                    // concurrent last-out lease drop (which fires `cancel()` under
                    // this same lock).
                    if session.is_dead() {
                        continue;
                    }
                    // A live fill advances only as ITS payer pays: the pull runs one
                    // credit window past `served_paid` and parks. A request that
                    // starts AHEAD of that frontier would park on it too, and an
                    // observer's payments extend a fill's paid prefix only once the
                    // prefix reaches the observer's start (`extend_served_from`).
                    // With the owner not paying — a header-only open (the CLI's
                    // multi-source size probe), or a stalled client — the
                    // observer starves until its own stall
                    // budget expires. So such a request is not coalesced: it OWNS
                    // its own span, and the two fills coexist under the hash. A
                    // request at or behind the frontier attaches; its payments
                    // extend the shared prefix at once.
                    //
                    // The cost is duplicate egress on the overlap, and only on the
                    // part of it not yet in the store: the owner's pull fills just
                    // the gaps the store is missing when it starts, so an overlap
                    // the live fill already landed is never fetched again. The
                    // part still to land is at most the live fill's credit window
                    // plus its in-flight draws, and only attaching could save it,
                    // which is what starves. So the rule stays, and a claim that
                    // skips an overlapping session is counted once.
                    if fetch_start > session.served_paid() {
                        if !(&r & &session.covered_ranges()).is_empty() {
                            not_coalesced = true;
                            tracing::debug!(
                                %hash,
                                fetch_start,
                                served_paid = session.served_paid(),
                                "fill not coalesced: request starts ahead of the live fill's paid frontier",
                            );
                        }
                        continue;
                    }
                    let covered = session.covered_ranges();
                    let overlap = &r & &covered;
                    if !overlap.is_empty() {
                        let span = chunk_span(&overlap);
                        if best
                            .as_ref()
                            .is_none_or(|(_, _, best_span)| span > *best_span)
                        {
                            best = Some((Arc::clone(session), covered.clone(), span));
                        }
                    }
                    covered_union |= covered;
                }
            }

            if not_coalesced && let Some(m) = &self.metrics {
                m.fill_not_coalesced.inc();
            }

            let attach = &r & &covered_union;
            let remainder = &r - &covered_union;

            if !attach.is_empty()
                && let Some((sibling, sib_covered, _)) = best
            {
                if remainder.is_empty() {
                    // Whole R covered by live pulls → pure ATTACH to the sibling.
                    let lease = self.mint_lease(&sibling, hash);
                    return FillClaim::Attach {
                        session: sibling,
                        lease,
                    };
                }
                // Partial overlap. Coalesce only when ONE sibling covers all of the
                // overlap AND the bytes it does not cover form ONE contiguous span
                // the remainder pull can fetch as a single gap. Otherwise fall
                // through to a conservative whole-R owner (no double-pull, no wedge).
                let attach_covered_by_sibling = (&attach - &sib_covered).is_empty();
                let remainder_vs_sibling = &r - &sib_covered;
                if attach_covered_by_sibling
                    && let Some((rem_offset, rem_len)) =
                        contiguous_byte_span(&remainder_vs_sibling, total)
                {
                    let owner = make_session();
                    owner.set_covered(remainder_vs_sibling);
                    self.insert_session(&mut map, hash, &owner);
                    let owner_lease = self.mint_lease(&owner, hash);
                    let attach_lease = self.mint_lease(&sibling, hash);
                    return FillClaim::Mixed {
                        owner,
                        owner_lease,
                        attach: sibling,
                        attach_lease,
                        remainder_offset: rem_offset,
                        remainder_len: rem_len,
                    };
                }
            }
        }

        // OWNER: no overlap, an empty/unalignable R, or the conservative fallback.
        // Scope the session to the WHOLE R (so a later identical claim attaches),
        // register + mint the owner lease under the SAME lock.
        let owner = make_session();
        owner.set_covered(r);
        self.insert_session(&mut map, hash, &owner);
        let lease = self.mint_lease(&owner, hash);
        FillClaim::Owner {
            session: owner,
            lease,
        }
    }
}

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

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