decdn-client 0.1.0

//! The deCDN client: fetch content-addressed blobs from deCDN nodes, pay for
//! them per megabyte from a USDC payment pool, and verify every byte against
//! its BLAKE3 hash as it arrives.
//!
//! The `decdn` CLI's `get` is built on this crate, and its bundle route is a
//! thin caller of [`bundle::BundlePuller`]. So is the node's
//! own cache-miss pull: a node that misses in its cache is a paying client of
//! the node upstream of it.
//!
//! # Choose an entry point
//!
//! | You want to | Use |
//! |---|---|
//! | Save one blob, or a set of blobs, to files as fast as possible, and resume a partial download | [`Downloader`] |
//! | Read one blob in order, and pay only for what your reader reaches | [`Streamer`] |
//! | Pull a whole bundle manifest into a directory, with discovery, funding, resume and dedup handled | [`bundle::BundlePuller`] |
//!
//! The [`Downloader`] stripes a blob
//! across every holder at once, and writes each verified range at its offset in
//! a `.partial` file beside the destination. A rerun fetches only the ranges
//! that file does not hold yet. The [`Streamer`] fetches at most one read-ahead
//! window ([`PullConfig::read_ahead_bytes`]) ahead of its reader, so a reader
//! that stops early stops the spend within that window. The
//! [`bundle::BundlePuller`] runs every step below for you over one pool: it
//! takes the chain ([`session::ChainConfig`]), a buyer-pool store, and a
//! [`bundle::Funding`], and records each lane's payment itself.
//!
//! # Before you fetch
//!
//! A fetch needs a funded payment pool, an iroh endpoint, and one paid lane per
//! node that holds the blob. The `download` and `stream` examples in this
//! crate's `examples/` directory run the whole sequence:
//!
//! 1. **Load the buyer key.** Vouchers and pool transactions are signed with a
//!    [`PrivateKeySigner`].
//! 2. **Reuse or open a pool.** Read the pool the buyer store already records
//!    ([`decdn_incentive::BuyerPoolStore::get_by_owner`]). If there is none,
//!    approve the deposit ([`buyer_pool::ensure_allowance`]), open a pool
//!    ([`buyer_pool::open_pool`]), and record the returned state at once: the
//!    deposit is escrowed from the moment `open_pool` returns.
//! 3. **Bring up an endpoint** with [`endpoint::client_endpoint`].
//! 4. **Find the holders.** Read the registered nodes
//!    ([`discovery::bootstrap_nodes`]), pick a sample
//!    ([`discovery::select_candidates`]), and probe each one
//!    ([`probe::probe_once`]). Verify every answer
//!    ([`probe::verify_probe_response`]) before it counts, and drop an answer
//!    whose `has_blob` and coverage disagree. A probe answer's size
//!    ([`discovery::Probed::total_bytes`]) is an unsigned hint: pass it as the
//!    fetch's first size claim, which the fetch grows or shrinks as verified
//!    bytes land.
//! 5. **Build one lane per holder.** Pin a [`PoolContext`] to the holder's
//!    provider address with [`buyer_pool::self_owned_lane_ctx`], resuming at
//!    what the lane has already paid, and bind it to your endpoint
//!    ([`PoolContext::bind_endpoint`]). A node that misses the blob in its
//!    cache serves an unbound lane nothing. Create its ledger with
//!    [`PoolContext::new_ledger`], and wrap both in a [`PeerSource`] inside a
//!    [`StreamCandidate::new`]. Cap the lane's rate at the rate the node signed
//!    in its probe answer ([`effective_rate_ceiling`]).
//! 6. **Fetch** with [`Downloader::fetch_to_paths`] or [`Streamer::open`], over
//!    a [`StaticSources`] of the lanes. Every lane streams at once unless
//!    [`Downloader::max_lanes`] caps it.
//! 7. **Record what each lane paid**, whatever the outcome: build a
//!    [`VoucherProgress`] from the ledger's settlement, and apply the
//!    [`buyer_pool::ProgressWrite`] it calls for to the buyer store.
//!
//! # What the crate guarantees
//!
//! - **Every byte is verified.** Delivery is BLAKE3/bao verified streaming.
//!   The decoder checks each chunk group against the hash as it lands, so a
//!   corrupt range aborts the pull at that group. Nothing unverified reaches a
//!   [`VerifiedReader`] or a finished file.
//! - **You pay only for verified bytes.** You pay per chunk
//!   ([`decdn_protocol::CHUNK_BYTES`]) once the decoder has verified it. A
//!   lying node is paid for the bytes before its first bad group and for
//!   nothing after, never for the size it claimed.
//! - **Quotes are checked against your ceiling.** A stream response whose
//!   signed rate exceeds the `max_rate_per_mb` given to [`PeerSource::new`] is
//!   refused before any payment ([`RateAboveCeiling`]).
//! - **Recovery is free.** Every lane draws on one shared pool. A source that
//!   faults cools and returns, and its range goes to another holder meanwhile.
//!   A fetch ends on done, a fatal fault, or the stop policy
//!   ([`StopPolicy`]). No delivered byte is fetched or paid for twice.
//!
//! # What stays yours
//!
//! - **Persistence.** The faces never write the buyer store. Record every
//!   lane's payment after each fetch (step 7). A lane you do not record resumes
//!   from a stale watermark next time, and its provider rejects the vouchers.
//! - **Funding policy.** A source whose next voucher the deposit cannot cover,
//!   or that refuses the pool as `Unfunded`, waits for the deposit to rise.
//!   Once no other source serves and a fresh discovery finds nothing new, the
//!   fetch runs one funding recovery step through your [`Funder`]
//!   ([`Funder::recover`]): it may top the pool up, open a new pool
//!   ([`PoolReplaced`]), or report that it has no way to add funds. A further
//!   step needs a newly verified byte since the last one. When no step is
//!   allowed or possible, the fetch fails with a [`NoAffordableSource`],
//!   "funding needed".
//! - **Delegated credentials.** A delegate that pays under someone else's
//!   pool hands the face a [`CredentialSlot`] and watches its
//!   [`FundingEvent`]s. Swap a new key and capability into the slot when the
//!   capability runs low: the fetch retires the old lanes at a voucher
//!   boundary and goes on under the new key. Without a swap, a delegated
//!   fetch ends with a typed [`FundingNeeded`].
//! - **Which holders to use.** Discovery gives candidates; ordering and
//!   admission ([`discovery::admit_sources`]) are the caller's choice.
//!
//! # Mistakes to avoid
//!
//! - **One candidate per provider.** Each [`StreamCandidate`] pays one
//!   `(signer, provider)` lane. Two candidates for one provider sign competing
//!   vouchers on the same lane.
//! - **Always set a rate ceiling.** A `max_rate_per_mb` of `0` means no
//!   ceiling, so a node can quote low on the probe and high on the stream.
//!   Pass [`effective_rate_ceiling`] of the probed rate and any absolute cap you
//!   hold.
//! - **Never resume a lane below its on-chain watermark.** A voucher at or below
//!   the watermark redeems nothing, so the provider streams bytes it can never
//!   cash. A self-owned lane the buyer store has not seen resumes from the
//!   chain's `getWatermark`. A delegated lane has no watermark on chain until a
//!   redemption records it, so one the store has not seen starts at zero and
//!   heals from the watermark bundle the node returns (ADR 003 § Capability
//!   delegation).
//! - **Keep the stream drive running.** [`Streamer::open`] returns a reader and
//!   a [`StreamDrive`]. Run your read inside [`StreamDrive::alongside`]: the
//!   drive pays and drains open legs, and it must not wait behind a blocked
//!   consumer.
//! - **Progress is in content bytes.** A [`ProgressCallback`] on the faces
//!   reports verified content bytes against the blob's size.
//!
//! # How delivery works
//!
//! Each lane sends a [`StreamRequest`], verifies the signed [`StreamResponse`],
//! and then reads `ChunkData` and pays as it goes. It releases one hash-chain
//! preimage per delivered `CHUNK_BYTES` chunk, and signs a voucher to open a
//! chain, to roll one, and to settle a residual shorter than a chunk.
//!
//! The `ChunkData` payload is bao's interleaved verified-stream encoding, not
//! raw bytes. Every consumer feeds it to a `bao-tree` verifying decoder as it
//! arrives. The decoder checks every chunk group against the requested root. A
//! range fetched from any offset therefore verifies on its own, with no
//! dependency on earlier bytes (ADR 038).
//!
//! # Stability
//!
//! The documented items are the SDK, and the public-API snapshot in
//! `tests/public_api.rs` guards them. Their enums and the error types a fetch
//! returns are `#[non_exhaustive]`, so a new variant or field is not a
//! breaking change. Items hidden from these docs are the pull engine's seams
//! that the node and the `decdn` CLI drive directly. They carry no stability
//! promise.

// Docs live in `bundle.rs` as `//!`, for the same reason as `sink` below.
pub mod bundle;
// Docs live in `session.rs` as `//!`, for the same reason as `sink` below.
/// Buyer-side `PaymentPool` open kernel (#940), shared by the node service
/// and the CLI.
pub mod buyer_pool;
/// Zero-config tunables for the consumption faces over the paid pull engine
/// ([`config::PullConfig`], #1848): every knob defaults, so a caller overrides
/// only what it must, and construction needs no network or chain access.
pub mod config;
/// A caller-owned QUIC connection kept warm across many hash fetches
/// ([`connection::WarmConnection`]): one dial amortized over every hash, one
/// bi-stream per hash (no wire change), closed once on the handle's own `Drop`;
/// and the command-wide map of one such connection per node
/// ([`connection::Connections`]).
pub mod connection;
/// Range-keyed discovery coverage-map primitive plus the two
/// objective-specific planners over it (#1506): [`coverage_plan::plan_covered_runs`]
/// (node — concentrate + sticky) and `coverage_plan::spread_segments` (client —
/// spread for parallelism). Pure, no I/O.
pub(crate) mod coverage_plan;
/// A delegated fetch's voucher credential, which the application swaps when
/// the capability runs low, and the funding signal that says when.
pub mod credential;
/// Client-side node discovery (#936): read + select the active node set from
/// `CapacityBond.getRegisteredNodes`, then rank probed blob-holders.
pub mod discovery;
/// The `Downloader` consumption face (#1848): fetch a set of content-addressed
/// blobs (a bundle, or a single blob) to files in a directory, out-of-order and
/// at full throughput, so every byte is bao-verified and a resumed fetch
/// re-pulls only the missing ranges.
pub mod downloader;
/// The #1608 gap-driven fetch driver: [`driver::drive`] fills only the
/// [`missing_ranges`](decdn_bao_range::RangedStore::missing_ranges) of a request,
/// paying the minimum, by folding the resume / reseed loop behind the
/// [`pacer::Pacer`] axis.
pub mod driver;
/// One-shot client `Endpoint` construction: relay + discovery resolution for
/// the `cdn/client/v1` and `cdn/probe/v1` dial paths (#935/#936).
pub mod endpoint;
pub mod session;
// Docs live in `uri.rs` as `//!`, for the same reason as `sink` below.
pub mod uri;
// Hidden engine seams with `//!` docs in their own files: how a paid fetch
// finds holders, builds and funds its lanes, and records what they paid. The
// `decdn` CLI drives them; they carry no stability promise.
/// What a failed lane, lane build, or discovery means for the acquire loop
/// (ADR 039 § Failure handling: reassign-only tail): a pure classifier over
/// `anyhow::Error`.
#[doc(hidden)]
pub mod fault;
mod first_open;
#[doc(hidden)]
pub mod funding;
/// Command-wide health of each provider (ADR 039 § Failure handling:
/// reassign-only tail): cooling backoff on a delivery fault, parking on an
/// unaffordable price.
#[doc(hidden)]
pub mod health;
#[doc(hidden)]
pub mod lane_sources;
#[doc(hidden)]
pub mod lanes;
mod ledger;
/// Run-scoped registry of live per-lane voucher ledgers:
/// [`ledgers::LaneLedgers`] maps each `(pool_id, signer, provider)` lane to
/// the one [`ledgers::LaneHandle`] every concurrent fetch on that lane shares.
#[doc(hidden)]
pub mod ledgers;
#[doc(hidden)]
pub mod ordered_writes;
/// The pure pacing axis (#1608): [`pacer::Pacer`] / [`pacer::BudgetPacer`] decide
/// draw / wait / done / refuse for the gap-driven driver, with no I/O.
#[doc(hidden)]
pub mod pacer;
/// Persisted per-peer knowledge base: registry-fed identity plus interaction-fed
/// latency and price, keyed by iroh [`iroh::PublicKey`], one JSON file per peer.
#[doc(hidden)]
pub mod peer_store;
/// Reusable `cdn/probe/v1` client.
pub mod probe;
/// Sub-frame byte-progress observation (#1797): a `ProgressReader` that tallies bytes
/// off the QUIC stream beneath the message decode, and the `ThroughputFloor` that judges
/// those bytes against a minimum rate over a trailing window.
mod progress;
/// Wallet-filled HTTP provider builder for opening, topping up, closing and reclaiming
/// payment pools.
pub mod provider;
/// Client-side [`decdn_bao_range::RangedStore`] backend (#1621): a
/// `.partial` + sidecar store built on `bao-tree`/`decdn-bao-range` only.
#[doc(hidden)]
pub mod ranged_store;
/// The `APP_ERR_RATE_LIMITED` (`0x10`) transport shed, typed for the pull
/// orchestrator (ADR 013 §Application Error Codes).
pub(crate) mod rate_limited;
/// The funding recovery step at the exhausted candidate set (ADR 003 § Funding
/// recovery): the per-fetch [`recovery::RecoveryGate`] and its progress rule.
pub mod recovery;
mod scheduler;
/// Pure segmentation and tail-steal helpers for the multi-source scheduler
/// (ADR 039 § Dynamic segmentation and tail-stealing): no I/O, no async.
mod segment;
/// When a command gives up (ADR 039 § Failure handling: reassign-only tail):
/// the progress clock and the stop policy that watches it.
pub mod stop;
#[doc(hidden)]
pub mod stream_cap;
/// The `Streamer` consumption face (#1848): stream one blob's verified,
/// contiguous front to a consumer as it arrives, paced by consumption and bounded
/// to one read-ahead window ahead of the read cursor. A fetch-like single-blob
/// face over the same engine, with no chunk dedup.
pub mod streamer;
#[doc(hidden)]
pub mod targets;
// Docs live in `sink.rs` as `//!`. Deliberately NOT documented here as well:
// rustdoc resolves intra-doc links on a `mod` item in THIS file's scope, so the
// module's own links (`content_paid_frontier`, …) would go unresolved
// and fail the doc gate (`CARGO_BUILD_WARNINGS=deny`).
pub mod sink;
/// The sourcing and funding seams of a fetch (#1608): [`source::BlobSource`]
/// (raw-bao byte source for a range) and [`source::Funder`] (the injected
/// funding recovery seam), plus scripted test doubles.
pub mod source;
/// The sources of one blob (ADR 039 § Source set and selection):
/// [`source_set::SourceSet`] tracks
/// each holder's health, lane, and retry timing behind the injected
/// [`source_set::SourceProvider`] seam.
pub mod source_set;

// The SDK surface: the two faces, the setup sequence the crate guide walks
// through, and every type their signatures name.
pub use config::PullConfig;
pub use connection::Connections;
pub use credential::{
    CREDENTIAL_SWAP_WAIT, CapabilityCause, Credential, CredentialMismatch, CredentialSlot,
    FundingEvent, FundingNeeded,
};
pub use downloader::{DownloadTarget, Downloader};
pub use driver::PoolExhausted;
pub use ledger::{Cumulative, PoolLedger};
pub use recovery::RecoveryGate;
pub use sink::{BlobCache, NoCache, SinkFuture};
pub use source::{Funder, PeerSource, PoolReplaced, Recovery, SourceFuture};
pub use source_set::{Holder, NoAffordableSource, NoNodeWillServe, NoSourceHasBlob, StaticSources};
pub use stop::{ClockHold, GaveUp, ProgressClock, StopPolicy};
pub use streamer::{LiveReader, StreamCandidate, StreamDrive, Streamer, VerifiedReader};

// The engine seams the node and the `decdn` CLI drive directly. They are
// public so those crates can reach them, hidden from the docs and the
// public-API snapshot, and carry no stability promise.
#[doc(hidden)]
pub use connection::WarmConnection;
#[doc(hidden)]
pub use coverage_plan::{CoveredRun, SourceCoverage, plan_covered_runs};
#[doc(hidden)]
pub use credential::{QuoteMax, RunDemand};
#[doc(hidden)]
pub use decdn_bao_range::RangedStore;
#[doc(hidden)]
pub use driver::{LegNoProgress, PacingWait, SharedPool, WaitReason, drive, first_leg};
#[doc(hidden)]
pub use fault::{FatalScope, Fault, HealExhausted, LaneBuildFault, classify};
#[doc(hidden)]
pub use first_open::first_open;
#[doc(hidden)]
pub use health::{Health, PeerHealth};
#[doc(hidden)]
pub use ledger::{ChainCommit, EpochAction, Metered, Rebase, Released};
#[doc(hidden)]
pub use ledgers::{LaneHandle, LaneLedgers};
#[doc(hidden)]
pub use pacer::{
    BudgetPacer, DownstreamFrontier, PULL_WINDOW_FLOOR, PaceDecision, PaceState, Pacer, RampPacer,
    VERIFY_LAG_BYTES, WindowPacer,
};
#[doc(hidden)]
pub use peer_store::{PeerRecord, PeerStore, StoreConfig};
#[doc(hidden)]
pub use ranged_store::ClientRangedStore;
#[doc(hidden)]
pub use rate_limited::UpstreamRateLimited;
#[doc(hidden)]
pub use recovery::{RECOVERY_SETTLE, Stepped};
#[doc(hidden)]
pub use scheduler::{
    AcquireEnv, AcquireTarget, ConsumptionPacing, GROWTH_RETRY, GrowFor, LANE_WATCHDOG, LaneLease,
    LaneWiden, acquire,
};
#[doc(hidden)]
pub use source::{
    BaoRangeReader, BlobSource, IngestEnd, IngestFuture, IngestStore, PrimedSource, SourceStream,
};
#[doc(hidden)]
pub use source_set::{LaneRange, SourceProvider, SourceSet};

pub(crate) use ledger::StreamProof;

#[cfg(any(test, feature = "test-util"))]
pub use source::{FakeFunder, ScriptedReader, ScriptedSource};

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

use alloy::dyn_abi::Eip712Domain;
use alloy::primitives::{Address, B256, Signature, U256};
use alloy::signers::SignerSync;
use alloy::signers::local::PrivateKeySigner;
#[cfg(any(test, feature = "test-util"))]
use bao_tree::BaoTree;
#[cfg(any(test, feature = "test-util"))]
use bao_tree::io::BaoContentItem;
#[cfg(any(test, feature = "test-util"))]
use bao_tree::io::fsm::{ResponseDecoder, ResponseDecoderNext};
use bytes::Bytes;
#[cfg(any(test, feature = "test-util"))]
use decdn_bao_range::align_range;
use decdn_bao_range::align_range_clamped;
#[cfg(any(test, feature = "test-util"))]
use decdn_bao_range::{AlignedRange, IROH_BLOCK_SIZE};
use decdn_incentive::{
    BuyerPoolState, EPHEMERAL_BINDING_NONCE, SignedCapability, SignedVoucher, StreamSlashData,
    Voucher, binding_signing_hash, signed_to_wire_voucher,
};
use decdn_protocol::client::{
    ClientBinding, ClientMessage, StreamError, StreamRequest, StreamRequestExt, StreamResponse,
    StreamResponseExt, VoucherRejectReason, WatermarkBundle, WireCapability,
};
use decdn_protocol::{
    CHUNK_BYTES, ChunkPreimage, decode_message, encode_message, read_frame, write_frame,
};
use iroh::endpoint::{RecvStream, SendStream};
use iroh::{Endpoint, EndpointAddr};
#[cfg(any(test, feature = "test-util"))]
use sink::{PullReader, StashedFault};

/// Per-lane context the requester needs to sign pool vouchers.
///
/// A voucher is scoped to a `(pool_id, signer, provider)` lane. `pool_id` is the
/// on-chain deposit, the signer is `client_signer.address()` (the capability key
/// the pool owner delegated spend to — the owner's own key in the single-user
/// case), and `provider` is the delivering node this stream pays.
///
/// The `prior_*` fields capture the lane's cumulative state from earlier streams
/// so a **reused** lane resumes correctly: vouchers are cumulative across the
/// lane's lifetime, so the node's `last_bytes_delivered` / `last_amount` are
/// non-zero after the first stream. Starting a fresh stream from zero would be
/// rejected (`AmountRegression` / `BytesRegression`). For a brand-new lane pass
/// `U256::ZERO` for both.
///
/// Build one with [`Self::new`] or [`Self::for_pool`], then pin it with
/// [`Self::with_provider`] and bind it with [`Self::bind_endpoint`].
#[derive(Clone)]
#[cfg_attr(not(feature = "test-util"), non_exhaustive)]
pub struct PoolContext {
    /// On-chain `poolId` the vouchers draw from.
    pub pool_id: B256,
    /// The delivering node this stream pays — the voucher's `provider` field. A
    /// capability voucher is scoped to one provider and is invalid if redeemed
    /// against another.
    pub provider: Address,
    /// On-chain pool deposit (informational here; the node enforces it).
    pub deposit: U256,
    /// Client key that signs vouchers — its address is the lane's capability
    /// `signer`.
    pub client_signer: Arc<PrivateKeySigner>,
    /// `PaymentPool` EIP-712 domain.
    pub voucher_domain: Eip712Domain,
    /// Cumulative bytes paid for on this lane before this stream.
    pub prior_bytes_delivered: U256,
    /// Cumulative amount paid on this lane before this stream.
    pub prior_amount: U256,
    /// Optional ADR 005 client identity binding (address + `BindNodeId`
    /// signature over the requester's own iroh `NodeId`, see
    /// [`sign_client_binding`]). Attached to every `cdn/client/v1` request's
    /// `ext` so the serving node can recover the buyer address, confirm it owns
    /// the pool (`pull_authorized`), and reactively populate from its configured
    /// origin (#1115). `None` ⇒ no binding is sent (an unconfigured
    /// `capacity_bond` on the client, or an on-chain/registered requester).
    pub client_binding: Option<ClientBinding>,
    /// Optional pool owner capability delegating spend to `client_signer`
    /// (ADR 003 §Capability delegation, `issue_self_capability` /
    /// `buyer_pool::open_pool`). Attached to every `cdn/client/v1` request's
    /// `ext` alongside `client_binding` so the serving node can register the
    /// signer on that signer's first on-chain redemption. Node-agnostic — the
    /// same `SignedCapability` is valid at every node this context streams
    /// from, since it grants spend against the pool rather than a specific
    /// provider. `None` ⇒ no capability is sent (the signer is already
    /// registered on-chain, or this stream reuses a lane a prior stream this
    /// session already delivered the capability for).
    pub capability: Option<SignedCapability>,
}

impl PoolContext {
    /// A context that signs vouchers on `pool_id` with `client_signer` under
    /// `voucher_domain`, against a deposit of `deposit`. It pays no provider
    /// yet, starts its lane at zero, and carries no binding or capability:
    /// [`Self::with_provider`], [`Self::bind_endpoint`] and
    /// [`Self::with_capability`] add each. A delegated signer, which holds no
    /// pool row of its own, starts here.
    #[must_use]
    pub const fn new(
        pool_id: B256,
        deposit: U256,
        client_signer: Arc<PrivateKeySigner>,
        voucher_domain: Eip712Domain,
    ) -> Self {
        Self {
            pool_id,
            // A pool fans out to many providers; the fetch target is pinned
            // per-pull via [`Self::with_provider`], not stored in the pool.
            provider: Address::ZERO,
            deposit,
            client_signer,
            voucher_domain,
            prior_bytes_delivered: U256::ZERO,
            prior_amount: U256::ZERO,
            client_binding: None,
            capability: None,
        }
    }

    /// [`Self::new`] for a pool the buyer store records: its `pool_id` and
    /// committed deposit.
    #[must_use]
    pub const fn for_pool(
        state: &BuyerPoolState,
        client_signer: Arc<PrivateKeySigner>,
        voucher_domain: Eip712Domain,
    ) -> Self {
        Self::new(state.pool_id, state.deposit, client_signer, voucher_domain)
    }

    /// Pin the delivering node this context pays (the voucher's `provider`) and
    /// seed the lane's prior cumulative totals. A pool serves many providers, so
    /// the target is chosen per-fetch rather than at pool open.
    #[must_use]
    pub const fn with_provider(
        mut self,
        provider: Address,
        prior_bytes_delivered: U256,
        prior_amount: U256,
    ) -> Self {
        self.provider = provider;
        self.prior_bytes_delivered = prior_bytes_delivered;
        self.prior_amount = prior_amount;
        self
    }

    /// This context's `(pool_id, signer, provider)` lane — the scope of one
    /// hash chain, and the key every seed is derived under.
    #[must_use]
    pub fn lane(&self) -> decdn_incentive::LaneKey {
        decdn_incentive::LaneKey {
            pool_id: self.pool_id,
            signer: self.client_signer.address(),
            provider: self.provider,
        }
    }

    /// The ledger for this lane, seeded from its persisted cumulative and
    /// nothing else.
    ///
    /// A chain is never resumed across a restart — the resume folds the
    /// frontier the node proved into a signed amount and opens a fresh chain
    /// (ADR 003 §Resumption folds) — so there is no chain secret to reproduce
    /// here, no counter to read back, and nothing that could have been
    /// persisted wrong. What carries across an invocation is the money owed.
    #[must_use]
    pub fn new_ledger(&self) -> PoolLedger {
        PoolLedger::new(Cumulative {
            bytes: self.prior_bytes_delivered,
            amount: self.prior_amount,
        })
    }

    /// Attach an ADR 005 client identity binding so this context's
    /// `cdn/client/v1` requests prove pool ownership to the serving node,
    /// enabling reactive cache-miss origin pull-through (#1115).
    /// [`Self::bind_endpoint`] signs and attaches one in a single call. A
    /// hand-built `ClientBinding` whose
    /// `ethereum_address` and signature don't correspond (or that doesn't own the
    /// pool) is rejected by the serving node, so this only ever hurts the caller
    /// itself.
    #[must_use]
    pub fn with_client_binding(mut self, binding: ClientBinding) -> Self {
        self.client_binding = Some(binding);
        self
    }

    /// Bind `endpoint` to this context's signer, so a node that misses the blob
    /// in its cache may pull it from its origin for this buyer. Without a
    /// binding, such a node refuses the open as `NotFound`. `bind_domain` is
    /// `decdn_incentive::bind_node_id_domain(chain_id, capacity_bond)`.
    ///
    /// # Errors
    ///
    /// A signing error from `client_signer`.
    pub fn bind_endpoint(
        self,
        endpoint: &Endpoint,
        bind_domain: &Eip712Domain,
    ) -> anyhow::Result<Self> {
        let own_node_id = B256::from(*endpoint.id().as_bytes());
        let binding = sign_client_binding(&self.client_signer, own_node_id, bind_domain)?;
        Ok(self.with_client_binding(binding))
    }

    /// Attach a pool owner capability so this context's `cdn/client/v1`
    /// requests carry it to the serving node at session start, letting the
    /// node register `client_signer` on that signer's first on-chain
    /// redemption (ADR 003 §Capability delegation). Pass the
    /// `SignedCapability` produced by `buyer_pool::open_pool` /
    /// `issue_self_capability`; a capability whose `signer` doesn't match
    /// `client_signer` or that fails owner-signature recovery only ever hurts
    /// the caller itself, exactly like [`Self::with_client_binding`].
    #[must_use]
    pub const fn with_capability(mut self, capability: SignedCapability) -> Self {
        self.capability = Some(capability);
        self
    }
}

/// Sign an ADR 005 ephemeral client identity binding: an EIP-712
/// `BindNodeId(nodeId, nonce = 0)` attestation over the requester's OWN iroh
/// `NodeId`, signed with the buyer key. The serving node recovers the signer via
/// `ecrecover` (`verify_binding`) and checks it signs for a known lane on the named pool before
/// honoring a cache-miss origin pull (`pull_authorized`, ADR 003 §Off-Chain
/// Ephemeral Binding). Used by the CLI client fetch (#1115) and by node-to-node
/// pulls, where `node_origin` binds its upstream requests so an upstream can
/// chain a further reactive origin pull (#1117).
///
/// `own_node_id` MUST be the requester's own endpoint `NodeId` — what the peer
/// authenticates as `conn.remote_id()` — NOT the target node's. `bind_domain` is
/// `decdn_incentive::bind_node_id_domain(chain_id, capacity_bond)`.
///
/// # Errors
///
/// Propagates a signing error from the buyer signer.
#[doc(hidden)]
pub fn sign_client_binding(
    signer: &PrivateKeySigner,
    own_node_id: B256,
    bind_domain: &Eip712Domain,
) -> anyhow::Result<ClientBinding> {
    let hash = binding_signing_hash(own_node_id, EPHEMERAL_BINDING_NONCE, bind_domain);
    let binding_signature = signer
        .sign_hash_sync(&hash)
        .map_err(|e| anyhow::anyhow!("sign client binding: {e}").context(LocalPullFault))?
        .as_bytes()
        .to_vec();
    Ok(ClientBinding {
        ethereum_address: signer.address().into(),
        binding_signature,
    })
}

/// Build the trailing [`StreamRequestExt`] carrying the context's client
/// identity binding and/or owner capability, or `None` when the context
/// carries neither — in which case `encode_stream_request` appends no ext
/// bytes. Applied by `open_stream`, the single request site, so the capability
/// rides the same session-start request as the binding.
fn client_binding_ext(ctx: &PoolContext) -> Option<StreamRequestExt> {
    if ctx.client_binding.is_none() && ctx.capability.is_none() {
        return None;
    }
    Some(StreamRequestExt {
        binding: ctx.client_binding.clone(),
        capability: ctx.capability.as_ref().map(|signed| WireCapability {
            spending_cap: signed.capability.spending_cap,
            expiry: signed.capability.expiry,
            owner_signature: signed.signature.as_bytes().to_vec(),
        }),
    })
}

impl std::fmt::Debug for PoolContext {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PoolContext")
            .field("pool_id", &self.pool_id)
            .field("provider", &self.provider)
            .field("deposit", &self.deposit)
            .finish_non_exhaustive()
    }
}

/// The lane's committed voucher watermark, returned by [`UpstreamPull::finish`] /
/// [`UpstreamPull::abort`] / [`UpstreamPull::progress`] and threaded through the
/// `test-util` `stream_fetch_tracked` wrapper as an out-param, so the caller can
/// persist what it paid (#852).
///
/// A pull runs through a [`PoolLedger`] seeded from the lane's prior cumulative
/// state; this watermark is that ledger read back
/// ([`VoucherProgress::from_ledger`]) — so it
/// always holds the **absolute** cumulative totals of the last presumed-accepted
/// voucher (not per-stream deltas), exactly the `(bytes_delivered, amount)` pair
/// the buyer-pool lane record expects. Because the
/// copy-back runs on every return path (including the `Err`/timeout arms), the
/// latest totals survive a mid-stream failure or a paid-but-corrupt delivery.
///
/// **Implicit acceptance.** The watermark tracks vouchers the upstream is
/// presumed to have accepted — continued delivery is acceptance (ADR 005), so a
/// sent voucher advances it and only an explicit `VoucherRejected` rewinds it.
///
/// Read the persistable totals via [`VoucherProgress::advanced`], which yields
/// `None` when nothing was paid on this stream (so there is nothing to persist).
#[derive(Clone, Copy, Debug, Default)]
pub struct VoucherProgress {
    /// Cumulative lane bytes paid for as of the last committed voucher.
    bytes_delivered: U256,
    /// Cumulative lane amount paid as of the last committed voucher.
    amount: U256,
    /// Whether the watermark moved past the lane's seed on this stream.
    advanced: bool,
    /// The node's watermark the lane's ledger rebased to, behind it on
    /// `amount` ([`PoolLedger::rebase`]), and no persist has recorded yet. The caller
    /// overwrites the lane record with it once, then advances to the totals
    /// above.
    rebase_anchor: Option<Cumulative>,
}

impl VoucherProgress {
    /// Build the watermark from a ledger [`Cumulative`] plus the lane's seed
    /// amount. `advanced` is set when the cumulative `amount` rose past the seed,
    /// preserving the "something was paid iff the watermark advanced" contract
    /// even when the ledger was shared across concurrent streams.
    ///
    /// Public because a caller that owns its ledger persists from it directly
    /// rather than through the `&mut VoucherProgress` out-param — including from a
    /// `Drop`, where nothing can be awaited and [`PoolLedger::committed`] is the
    /// only readable source (#1145 review).
    #[must_use]
    pub fn from_cumulative(cum: Cumulative, prior_amount: U256) -> Self {
        Self {
            bytes_delivered: cum.bytes,
            amount: cum.amount,
            advanced: cum.amount > prior_amount,
            rebase_anchor: None,
        }
    }

    /// Attach the watermark the ledger rebased DOWN to and has not persisted
    /// yet ([`PoolLedger::take_unsaved_rebase`]).
    #[must_use]
    pub const fn with_rebase_anchor(mut self, anchor: Option<Cumulative>) -> Self {
        self.rebase_anchor = anchor;
        self
    }

    /// The watermark to overwrite the lane record with before advancing, when
    /// the ledger rebased down to the node's and no persist has recorded it
    /// yet. A monotone advance refuses a lower watermark, so without the
    /// overwrite the next run would sign from the anchor the node refused.
    #[must_use]
    pub const fn rebase_anchor(&self) -> Option<Cumulative> {
        self.rebase_anchor
    }

    /// The same watermark, read from a live ledger.
    ///
    /// Prefer this wherever a ledger is in hand: it reads
    /// [`PoolLedger::settlement`] rather than a copied-back cumulative, so a
    /// voucher left armed by an ambiguous send is still reported as owed.
    ///
    /// It also TAKES the ledger's unsaved rebase anchor, so call it only where
    /// the result is persisted: the anchor is handed out once.
    #[must_use]
    pub fn from_ledger(ledger: &PoolLedger, prior_amount: U256) -> Self {
        // Take the anchor first: every settlement read after it is at or above it.
        let anchor = ledger.take_unsaved_rebase();
        Self::from_cumulative(ledger.settlement(), prior_amount).with_rebase_anchor(anchor)
    }

    /// The cumulative `(bytes_delivered, amount)` to persist via the lane record,
    /// or `None` if nothing new was paid on this stream (nothing to record).
    #[must_use]
    pub fn advanced(&self) -> Option<(U256, U256)> {
        self.advanced.then_some((self.bytes_delivered, self.amount))
    }

    /// The cumulative `(bytes_delivered, amount)` this watermark carries,
    /// whether or not it advanced past the seed.
    #[must_use]
    pub const fn totals(&self) -> (U256, U256) {
        (self.bytes_delivered, self.amount)
    }
}

/// The upstream delivered bytes that failed bao verification against the
/// requested content root — a paid-but-corrupt delivery (the content-addressing
/// invariant, ADR 014/038). Under ADR 038 the verifier is the per-chunk-group
/// `bao-tree` decoder fed as bytes arrive, not a whole-blob re-hash, so this
/// fires the moment any group's proof mismatches. Returned (via `anyhow`) by
/// this crate's decoding consumers — [`ClientRangedStore::ingest_stream`] and the
/// `test-util` in-memory decoder — so callers can `downcast_ref` to classify corruption
/// (e.g. a reputation `Corruption` outcome) without matching on the error
/// message string. The `Display` text is kept stable for logs and the existing
/// requester tests. The node's cache admit reports the same fault as
/// `CacheError::VerifyFailed` / `CacheError::HashMismatch`, which `decdn-node`'s
/// `is_bao_corruption` folds together with this type.
#[derive(Debug)]
pub struct HashMismatch;

impl std::fmt::Display for HashMismatch {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("received bytes do not match requested hash")
    }
}

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

/// Reject a `total_bytes == 0` claim for a non-empty root as the typed
/// [`HashMismatch`] (#1054).
///
/// A 0-byte blob aligns to an empty range: the bao decoder has no chunk group to
/// anchor and would accept the empty stream for ANY root, and a store sized from a
/// `total_bytes == 0` claim has no gap to pull and is complete as created — so
/// nothing downstream ever verifies the empty stream against the root. The only
/// hash an empty blob can carry is the empty root; a source that claims `0` for any
/// other hash is a paid-but-wrong delivery. This is the one explicit verify site
/// the trivial-empty-range bypass has, shared by [`driver::drive`] (before the
/// empty store is finalized) and the in-memory `fetch_in_memory_once` path so the
/// two cannot drift. A `total_bytes > 0` claim is untouched: the streaming decoder
/// verifies it group by group.
pub(crate) fn reject_empty_claim_for_nonempty_root(
    total_bytes: u64,
    hash: [u8; 32],
) -> anyhow::Result<()> {
    if total_bytes == 0 && hash != *blake3::hash(&[]).as_bytes() {
        return Err(anyhow::Error::new(HashMismatch));
    }
    Ok(())
}

/// Typed sentinel for a pull aborted because the bytes that ACTUALLY arrived
/// crossed the buyer's `max_blob_size_bytes` ceiling (#1895). The peer's signed
/// `total_bytes` claim never drives a refusal — it is peer-controlled and
/// unverified (`StreamResponse::validate()` does not bound it) — so the ceiling
/// binds on cumulative RECEIVED, BLAKE3-verified bytes instead. Returned (not a
/// bare string) so the pull orchestrator can `downcast_ref` and classify it as a
/// buyer-side policy rejection — distinct from a hash mismatch or an unreachable
/// peer — rather than mis-attributing it to the provider's reputation. `Display`
/// carries `BlobTooLarge` so logs and the requester tests can match on it.
///
/// **Units.** `reached` and `ceiling` are ALWAYS the same unit within one error —
/// the comparison at each enforcement site is apples-to-apples — but that unit
/// differs by site, so neither field is a raw `max_blob_size_bytes` config value
/// across all call sites. The receive loop ([`UpstreamPull::next_chunk`]) meters bao
/// WIRE bytes (content plus interleaved proof, ADR 038) and compares against the
/// wire size of a ceiling-sized blob. The gap-driven driver (`fill_gap`) meters
/// CONTENT bytes (the store's delivered frontier) and the resume-offset guard in
/// [`open_progressive_pull`] meters a CONTENT offset, both against the configured
/// `max_blob_size_bytes` directly. [`acquire`] meters CONTENT bytes against
/// [`AcquireEnv::max_blob_bytes`]: it clamps a size claim to the cap and
/// reports a blob whose bytes run past it once every byte below the cap is
/// present and no leg proved a size within it. Every one is a faithful "the
/// byte position crossed the ceiling" report.
#[derive(Debug)]
#[cfg_attr(not(feature = "test-util"), non_exhaustive)]
pub struct BlobTooLarge {
    /// The byte position that crossed `ceiling` — wire bytes taken off the stream
    /// (receive loop), the store's content frontier (gap-driven driver), or a
    /// content resume offset already past it. Same unit as `ceiling` (see the
    /// type's **Units** note).
    pub reached: u64,
    /// The ceiling `reached` crossed, in the same unit as `reached`.
    pub ceiling: u64,
}

/// The requested range starts at or past the end of the blob the node signed
/// for, so no byte of it can be served (#1120). A node that signs a size of
/// zero for a bounded open of `[0, len)` raises it too.
///
/// Typed rather than a bare string because it is the ONE signal that proves a
/// caller's partial download does not belong to this blob — a stale `.partial`
/// left under the same `--output` by a fetch of a different or larger blob, or
/// one that completed but was killed before its rename. A resumable client keys
/// its discard-and-refetch on this and nothing else: classifying by exclusion
/// ("any error that is not a voucher rejection") would sweep in ordinary stalls
/// and resets and destroy a perfectly good prefix the user has already paid for.
///
/// Note this is a statement about the *offset*, not about the peer: an honest
/// node answers this way to a range cut from a wrong size. A node's reputation
/// must not score it. The acquire loop classifies it as a
/// [`Fault::Source`]: the source cools and its range
/// moves to other sources.
#[derive(Debug)]
#[doc(hidden)]
pub struct ResumeOffsetPastEnd {
    /// Whole-blob size the node signed for.
    pub total_bytes: u64,
    /// The offset we asked to resume from.
    pub byte_offset: u64,
}

impl std::fmt::Display for ResumeOffsetPastEnd {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "cannot resume at byte {}: the blob is only {} bytes",
            self.byte_offset, self.total_bytes
        )
    }
}

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

impl std::fmt::Display for BlobTooLarge {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "reached byte {}, crossing the {}-byte size ceiling (BlobTooLarge)",
            self.reached, self.ceiling
        )
    }
}

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

/// Typed sentinel for a server that signed an open-stage `StreamResponse`
/// (`ok == true`) quoting a per-MB `rate_per_mb` above the buyer's effective
/// ceiling (#1375). The buyer aborts **before paying**, so the profitable
/// "quote low on the probe, quote high on the stream, then deliver" bait-and-switch
/// never gets paid.
///
/// The ceiling is the lower of two bounds the caller supplies: the probe rate the
/// candidate was selected on (a stream quote may not exceed what the probe
/// advertised) and an optional absolute buyer config (`max_rate_per_mb`). Because
/// the probe-relative bound is always applied by the node pull path, a *completed*
/// delivery is by construction priced at or below the probe rate — so it can never
/// be rate-manipulation (`SlashJudge` requires `stream.ratePerMb > probe.ratePerMb`),
/// and the challengeable case is exactly this abort.
///
/// Carries the operator's own signed `response` — already verified against
/// `expected_signer` by `verify_response` before construction, so it still
/// recovers to the delivering node and a caller could replay it (paired with the
/// same node's `ProbeResponse`) to `SlashJudge` with no re-signing.
///
/// **Slashability caveat.** The evidence is only rate-*manipulation* evidence when
/// the **probe-relative** bound was the one exceeded: `SlashJudge` requires
/// `stream.ratePerMb > probe.ratePerMb`. The ceiling here is
/// `min(probe_rate, config_absolute)` ([`effective_rate_ceiling`]), so when the
/// buyer's absolute `config` bound is the binding one the quote may sit at or below
/// the probe rate and be perfectly honest. A challenger must therefore still
/// confirm `quote > probeRate` before submitting; this type does not assert it.
///
/// **Consumption.** The signed response is retained *on the error* so a caller can
/// act on it, but the node's pull-failure handler does not itself submit a
/// challenge — auto-slashing is deferred (the same footing as the daemon never
/// auto-challenging an [`UpstreamRefused`], whose evidence is exercised only by the
/// e2e harness today). `evidence()` is the only accessor; the field is private and
/// the sole constructor derives it from the verified response, so the
/// verified-signature invariant cannot be bypassed by a hand-built value.
pub struct RateAboveCeiling {
    quoted_rate_per_mb: u64,
    ceiling_rate_per_mb: u64,
    response: StreamResponse,
}

impl RateAboveCeiling {
    /// Build the abort error for an open-stage quote above `ceiling`. Crate-private
    /// and the SINGLE construction path (`open_progressive_pull` routes through
    /// it), so `quoted_rate_per_mb` is always
    /// *derived* from `response.body.rate_per_mb` and cannot desync from the
    /// retained evidence — the same "derive, don't trust the caller to set it
    /// consistently" discipline as [`UpstreamRefused::open`]. Callers MUST have run
    /// `verify_response` on `response` first (the field docs' signature invariant).
    fn over_ceiling(response: StreamResponse, ceiling_rate_per_mb: u64) -> anyhow::Error {
        anyhow::Error::new(Self {
            quoted_rate_per_mb: response.body.rate_per_mb,
            ceiling_rate_per_mb,
            response,
        })
    }

    /// The per-MB rate the server quoted in the signed `StreamResponse`.
    #[must_use]
    pub const fn quoted_rate_per_mb(&self) -> u64 {
        self.quoted_rate_per_mb
    }

    /// The effective ceiling the quote exceeded (min of probe rate and config).
    #[must_use]
    pub const fn ceiling_rate_per_mb(&self) -> u64 {
        self.ceiling_rate_per_mb
    }

    /// The operator's own signed `StreamResponse`, verified against
    /// `expected_signer` before this value was built. See the type docs for the
    /// slashability caveat (only `quote > probeRate` is manipulation).
    #[must_use]
    pub const fn evidence(&self) -> &StreamResponse {
        &self.response
    }
}

impl std::fmt::Debug for RateAboveCeiling {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Elide the signed body + 65-byte `slash_sig` to a length, as `UpstreamRefused`
        // does, so a logged abort stays readable.
        f.debug_struct("RateAboveCeiling")
            .field("quoted_rate_per_mb", &self.quoted_rate_per_mb)
            .field("ceiling_rate_per_mb", &self.ceiling_rate_per_mb)
            .field("evidence_slash_sig_len", &self.response.slash_sig.len())
            .finish()
    }
}

impl std::fmt::Display for RateAboveCeiling {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "server quoted {} per MB, exceeding the buyer ceiling {} per MB (RateAboveCeiling)",
            self.quoted_rate_per_mb, self.ceiling_rate_per_mb
        )
    }
}

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

/// Combine the two buyer rate ceilings into one effective bound, treating `0` as
/// "unbounded" on each input (the probe-relative bound and the absolute config
/// bound are both optional). Returns `0` only when BOTH are unbounded.
#[must_use]
pub const fn effective_rate_ceiling(probe_relative: u64, config_absolute: u64) -> u64 {
    match (probe_relative, config_absolute) {
        (0, c) => c,
        (p, 0) => p,
        (p, c) if p < c => p,
        (_, c) => c,
    }
}

/// Typed sentinel for the buyer's own per-candidate pull deadline firing (#857).
/// Returned (not a bare string) so the pull orchestrator can `downcast_ref` and
/// recognize that the timeout is OUR local deadline — a possibly mis-sized
/// configuration value — not evidence the provider is unreachable, and so must
/// not tar the provider's local reputation score. `Display` keeps the
/// stable `timed out` text for logs (and for the `!contains("timed out")`
/// negative assertion in `node_to_node_pull_through`'s deadline test).
///
/// Since #1134 it is raised by OUR OWN wall clocks only, and the message is
/// deliberately stage-NEUTRAL because there are THREE of them: the shared
/// `open_stream` open bound (`PullDeadlines::open`), the optional overall
/// `hard_cap` the `test-util` wrappers apply, and the stall clock elapsing before
/// the FIRST byte (`cumulative == 0`) in the streaming loop — where it is
/// measuring the server's time-to-first-byte, not mid-stream inactivity. `decdn-node`
/// also wraps `open_progressive_pull` in its per-candidate budget as belt-and-braces,
/// which raises the same sentinel.
///
/// The one thing it is NOT is a mid-stream STALL once bytes have flowed — that is
/// [`PullStalled`], a fact about the PEER that scores its reputation, whereas this is a
/// fact about our own possibly mis-sized budget (or an honest slow server's TTFB) and
/// must not. Keeping the two apart is the whole point of the split, and the `cumulative
/// == 0` carve-out is where the two loops draw the line (see [`PullStalled`]).
///
/// Callers that want to name the stage add a `.context(…)` layer; `downcast_ref`
/// still recovers the sentinel through it — pinned by
/// `buyer_side_sentinels_survive_anyhow_downcast` in `decdn-node`'s `node_origin.rs`,
/// not in this crate.
#[derive(Debug)]
#[cfg_attr(not(feature = "test-util"), non_exhaustive)]
pub struct PullTimeout {
    /// How long the pull ran before the budget elapsed.
    pub after: Duration,
}

impl std::fmt::Display for PullTimeout {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "upstream pull timed out after {:?}", self.after)
    }
}

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

/// Typed sentinel for the upstream rejecting a voucher we presented mid-stream
/// (#857) — e.g. a cumulative amount regression (#852), deposit exhaustion, or
/// a wrong pool. This is OUR payment-side fault, not the provider's, so the pull
/// orchestrator `downcast_ref`s it to skip the candidate WITHOUT recording a
/// reputation observation (mirroring the buyer pool-open-failure arm). Named
/// with the `Upstream` prefix to disambiguate from the protocol-level
/// `StreamError::VoucherRejected` reason enum, whose `reason` it carries verbatim
/// (the `Copy` `VoucherRejectReason`, not a lossy stringification) so a future
/// caller can branch on retry-vs-top-up-vs-abandon without re-parsing a message.
/// `Display` keeps the stable `voucher rejected` text for logs.
///
/// `bundle` mirrors the wire [`WatermarkBundle`] verbatim (issue #1481): `Some`
/// only for a watermark-gated reason ([`VoucherRejectReason::is_watermark_gated`]),
/// and only when the node verified the rejected voucher recovered to the pool
/// capability's pinned `signer` before attaching it. A caller that sees `Some`
/// can self-heal when the bundle advances its ledger — reseed to the bundle's
/// watermark ([`crate::ledger::Cumulative::from`]) and resume from
/// `bytes_delivered` — or, on `Underpaid`, rebase down to it, rather than
/// treating the rejection as terminal. `SpendingCapExhausted`
/// with `bundle: None` means there is no signer-verified watermark to resume
/// from (or, more commonly, that a wallet-less delegate simply has no local
/// means to add deposit) — the caller must surface that to the app rather
/// than loop. `CapabilityExpired` and `PoolExhausted` are never
/// watermark-gated (no bundle is ever attached) and are always terminal —
/// the fix is a fresh capability or an owner top-up, not a resync.
#[derive(Debug)]
#[cfg_attr(not(feature = "test-util"), non_exhaustive)]
pub struct UpstreamVoucherRejected {
    /// Why the seller refused the voucher.
    pub reason: VoucherRejectReason,
    /// The seller's signer-verified watermark, when it sent one, so the buyer
    /// can resync its cumulative state and retry. `None` means there is
    /// nothing to resync from and the caller must surface the failure.
    pub bundle: Option<WatermarkBundle>,
    /// The ledger generation the rejected voucher was signed under
    /// (`StreamProof::Voucher`), when the stream knows it. `None` for a
    /// rejected reveal, or a rejection read before this stream claimed
    /// anything; [`PoolLedger::rebase`] treats it as current.
    pub proof_generation: Option<u64>,
}

impl std::fmt::Display for UpstreamVoucherRejected {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // The `{:?}` rendering of the reason is load-bearing: the loopback tests
        // assert `.contains("CapExceeded")` / `.contains("Expired")`-style substrings
        // on this string. A custom `Display` for `VoucherRejectReason` would have to
        // reproduce the variant names verbatim, so keep the Debug rendering here.
        write!(f, "voucher rejected: {:?}", self.reason)
    }
}

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

/// Typed sentinel for an upstream *refusing* delivery up front — a
/// `StreamResponse` with `body.ok == false` (#1144). Carries the wire
/// [`StreamError`] verbatim (mirroring [`UpstreamVoucherRejected`]'s typed
/// `reason`, not a lossy stringification) so the pull orchestrator can
/// `downcast_ref` and give each refusal the verdict it deserves, rather than
/// folding every one into "unreachable".
///
/// That distinction is the whole point. A refusal is proof the peer is *reachable
/// and answering*, so no class is evidence that it is degraded (ADR 005 §Open-time
/// refusal classes):
///
/// - `NotFound` — "not now". It is NODE-scoped, not blob-scoped
///   (`ServeRejectReason::wire_error` collapses a miss, a load shed, and several
///   pool and signer states onto it), so it earns only a short suppression.
/// - `Declined` — "not this hash during this fetch", DURABLE for this (peer, hash):
///   asking again soon gets the same answer, so the consumer drops the peer for the
///   hash rather than burning a candidate slot on it.
/// - `Unfunded` — "not at the requester's current funding". It says nothing about
///   the peer; it counts toward the requester's funding recovery.
/// - `VoucherRejected` — a payment fault on this lane. Listed because it genuinely
///   arrives here: it is the code designated for MID-stream, and the mid-stream receive
///   arms wrap any `StreamError` into this sentinel (#1145 review).
///
/// `node_origin::classify_refusal` is the consumer and matches this exhaustively, so a new
/// wire code breaks that build rather than silently inheriting a verdict.
///
/// Only the wire class is recoverable here, never the finer server-side
/// `ServeRejectReason` — that collapse is intentional and must not be reversed.
/// `Display` keeps the stable `delivery refused: {code}` text that logs, the CLI's
/// cache-miss annotation, and the loopback tests match on.
/// The two refusal shapes are a private inner enum, not `pub` variants: enum
/// variant fields cannot be made private, so exposing them would let any caller
/// build an `Open { .. }` directly and bypass the crate-private `open`
/// constructor's error-from-response derivation — reintroducing exactly the
/// doc-comment-only invariant this type exists to remove (#1377). The newtype
/// keeps the only paths to a value the two constructors, so the "present response
/// ⇒ open-stage, verified" invariant is a property of the type rather than a
/// convention.
#[derive(Clone)]
pub struct UpstreamRefused {
    kind: Kind,
}

#[derive(Clone)]
enum Kind {
    /// Open-stage refusal: the upstream signed a [`StreamResponse`] with
    /// `body.ok == false`, already verified against `expected_signer` by
    /// [`verify_response`] before the value is built, so `error` is derived
    /// *from* the response's trailing [`StreamResponseExt`] and `response`
    /// always recovers to `expected_signer`.
    Open {
        error: StreamError,
        // Boxed (issue #1481 review): `StreamError::VoucherRejected` grew an
        // optional `WatermarkBundle`, which pushed the unboxed variant past
        // clippy's `large_enum_variant` threshold relative to `MidStream`.
        response: Box<StreamResponse>,
    },
    /// Mid-stream refusal: a bare [`ClientMessage::StreamError`] frame that
    /// arrived after the open stage. It carries no signature (#1042), so there
    /// is never a signed `StreamResponse` to retain — `None` by construction.
    MidStream { error: StreamError },
}

impl UpstreamRefused {
    /// Build the open-stage refusal for a `body.ok == false` response, deriving
    /// the wire `error` *from* the response's trailing [`StreamResponseExt`] so
    /// the two can never disagree (#1377 invariant 2). The one construction site
    /// [`open_progressive_pull`]'s open stage uses, so every caller classifies a
    /// refusal the same way.
    ///
    /// Returns `anyhow::Error` rather than `Self` because the `None`-error arm is
    /// a protocol violation, not a refusal: callers MUST have run
    /// [`verify_response`] first, whose `StreamResponseExt::validate` rejects
    /// `ok == false` with no error code (`MissingStreamError`). Reaching the
    /// `None` arm means that invariant was bypassed, so it is surfaced as the
    /// violation it is rather than defaulted to an invented code — which would
    /// launder a malformed refusal into a plausible-looking one.
    ///
    /// Takes the whole `response` rather than just its `error` so the operator's
    /// own signed refusal survives on the value ([`Self::evidence`], #1042) —
    /// `body` is exactly the field set the `SlashJudge` EIP-712 `StreamResponse`
    /// typehash covers and `slash_sig` is the operator's secp256k1 signature over
    /// it.
    ///
    /// A refusal (`ok == false`) is **not** on-chain slash evidence: rate
    /// manipulation requires `ok == true` and blacklist violation requires a
    /// served claim (ADR 014), so a node may sign refusals freely. The retention
    /// is kept on its own merits — it is an attributable, non-repudiable record
    /// of *why* a paid pull was declined, which the caller can log, surface, or
    /// present in a dispute without re-signing anything.
    fn open(response: StreamResponse, ext: &StreamResponseExt) -> anyhow::Error {
        // `ok == true` is not a refusal at all — building an `Open` from it would
        // mint an evidence-carrying refusal with `ok == true`, violating invariant
        // 3. `validate` already rejects `(ok: true, error: Some)` as
        // `StreamErrorWithOk`, so reaching here with `ok == true` means that
        // invariant was bypassed; surface it as the protocol violation it is rather
        // than fold it into a typed refusal. Makes invariant 3 a construction
        // property, not just a call-site precondition (#1377 review).
        if response.body.ok {
            return anyhow::anyhow!(
                "open-stage refusal constructed from an ok == true response \
                 (StreamResponse::validate invariant bypassed)"
            );
        }
        match ext.error.clone() {
            Some(error) => anyhow::Error::new(Self {
                kind: Kind::Open {
                    error,
                    response: Box::new(response),
                },
            }),
            None => anyhow::anyhow!(
                "delivery refused but the validated response carried no error code \
                 (StreamResponseExt::validate invariant bypassed)"
            ),
        }
    }

    /// Build the refusal for a **mid-stream** [`ClientMessage::StreamError`]
    /// frame — the refusal that arrives *after* the open stage, in reply to a
    /// delivery chunk or a voucher.
    ///
    /// Such a frame carries no signature (#1042), so [`Self::evidence`] is `None`
    /// here **by construction** and this is the only place that decision is made.
    /// The four mid-stream receive sites route through it so none can drift into
    /// synthesising a `StreamResponse` — which would hand an observer an unsigned
    /// artifact the evidence contract promises always recovers to the delivering
    /// node (#1378). The open-stage counterpart is the crate-private `open`.
    ///
    /// Public because callers outside the crate (and their tests) legitimately
    /// build mid-stream refusals; it is safe to expose precisely because it
    /// *cannot* carry a signed response — the encapsulation that matters guards
    /// the attributable `open` ([`Self::evidence`]), which stays crate-private
    /// so a caller cannot mint a refusal attributed to an operator.
    #[must_use]
    pub const fn mid_stream(error: StreamError) -> Self {
        Self {
            kind: Kind::MidStream { error },
        }
    }

    /// The wire code the upstream signed. Always a `StreamError` as it appeared
    /// on the wire — never a server-side `ServeRejectReason`, whose seven-way
    /// collapse onto `NotFound` is deliberate and one-way
    /// (`handlers::client::wire_error`). Total over both shapes.
    #[must_use]
    pub const fn error(&self) -> &StreamError {
        match &self.kind {
            Kind::Open { error, .. } | Kind::MidStream { error } => error,
        }
    }

    /// The upstream's own signed [`StreamResponse`], present iff the refusal
    /// arrived at the **open stage**. `None` for a mid-stream frame.
    ///
    /// An attributable record, not loose diagnostics — already verified against
    /// `expected_signer` by `verify_response` before this value is built, so a
    /// present value always recovers to `expected_signer`: the operator address
    /// the caller bound this pull to, which is what `SlashJudge._checkRegistered`
    /// resolves `nodeId` against.
    ///
    /// Note it is not by itself *slash* evidence — no offense admits a refusal
    /// (`ok == false`) as its stream leg (ADR 014). What it gives the caller is a
    /// non-repudiable statement of why the pull was declined, signed by the
    /// operator and usable verbatim, with no re-signing.
    #[must_use]
    pub fn evidence(&self) -> Option<&StreamResponse> {
        match &self.kind {
            Kind::Open { response, .. } => Some(response.as_ref()),
            Kind::MidStream { .. } => None,
        }
    }
}

impl std::fmt::Debug for UpstreamRefused {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Elide the 65-byte `slash_sig` (and the full signed body) to a length so
        // `tracing::error!(?err)` stays readable — the derived `Debug` would dump
        // the entire attestation on every logged refusal.
        let mut dbg = f.debug_struct("UpstreamRefused");
        dbg.field("error", self.error());
        match self.evidence() {
            Some(resp) => dbg.field("evidence_slash_sig_len", &resp.slash_sig.len()),
            None => dbg.field("evidence", &Option::<()>::None),
        };
        dbg.finish()
    }
}

impl std::fmt::Display for UpstreamRefused {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // The `{:?}` rendering is load-bearing, exactly as in
        // `UpstreamVoucherRejected`: the loopback tests assert
        // `.contains("NotFound")` on this string, and `StreamError` has no
        // `Display`. What renders here is always a WIRE code — never a node-side
        // `ServeRejectReason` such as `UnknownLane`, which collapses to
        // `NotFound` before it leaves the server (see `wire_error`). Matching on a
        // reject-reason name would therefore never fire.
        write!(f, "delivery refused: {:?}", self.error())
    }
}

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

/// Typed sentinel for an upstream that went SILENT mid-stream — bytes were flowing, and
/// then no byte of progress for the stall budget (#1134). Distinct from [`PullTimeout`],
/// which is a bound of OURS elapsing, and the distinction is load-bearing:
///
/// A whole-transfer deadline cannot tell "the provider is dead" from "this blob is big" or
/// "this link is slow", so `PullTimeout` must NOT tar the provider — it fires on perfectly
/// healthy transfers. A stall deadline resets on every byte received, so it fires ONLY when
/// a provider stops delivering while we wait. That IS evidence the peer is unreachable, and
/// `classify_pull_failure` scores it as such in the local per-peer EWMA (ADR 008).
///
/// # What the bound rests on
///
/// Two things, and it is worth being precise about both, because scoring a peer on a bound
/// that does not hold is how an honest node's local reputation gets unfairly defamed.
///
/// **The non-empty-`ChunkData` invariant (#1088).** With empty frames banned, "a frame
/// arrived" and "bytes made progress" are the same statement, so a peer cannot hold the
/// floor open with padding. The receive loop ([`UpstreamPull::next_chunk`]) has no
/// independent progress check behind that floor: it is safe because no frame a peer can
/// send makes zero progress, not because it verifies that it did. The floor is
/// structural rather than advisory — `ChunkData`'s field is private, and its constructor
/// and decode gate both reject an empty payload — so it cannot be relaxed by forgetting
/// to call a validator (#1145 review).
///
/// **At least one byte having already arrived.** The reset is what makes a stall the peer's
/// fault, so before the FIRST byte there has been no reset and the argument does not apply:
/// the clock is measuring the server's time-to-first-byte, which scales with blob size
/// (the serve path materialises the whole bao wire via `export_bao_range` before it can emit
/// chunk #1). Both loops therefore raise [`PullTimeout`] — exonerating — when the budget
/// elapses at `cumulative == 0`, and this sentinel only once bytes have flowed (#1145
/// review). Without that split, a 1 GiB blob off a cold disk would score an honest server as
/// unreachable for the crime of being big.
#[derive(Debug)]
#[cfg_attr(not(feature = "test-util"), non_exhaustive)]
pub struct PullStalled {
    /// The inactivity budget that elapsed after bytes had been flowing.
    pub after: Duration,
}

impl std::fmt::Display for PullStalled {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "upstream stalled: no progress for {:?}", self.after)
    }
}

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

/// Marker for a pull failure that is OURS, not the upstream's — a broken local
/// signer, an encode fault, a bad range computation (#1145 review).
///
/// Everything else that reaches `classify_pull_failure`'s catch-all is attributable
/// to the peer: a failed dial, a dropped connection, a bad signature, an unexpected
/// frame. Scoring the peer `Unreachable` there is correct, and is in fact the
/// PRIMARY way a dead node is detected — so the catch-all must stay as it is.
///
/// The exception was this class. A node whose own signer is broken cannot pay
/// anybody, and would walk the candidate list tarring every honest
/// provider it met with an `Unreachable` — a local EWMA hit; ADR 008 scoring is
/// local-only — on the strength of its own fault. Attach this marker at a
/// local-fault site and the classifier exonerates the peer and warns about us instead.
///
/// # It also decides what a CLIENT is told (#1560)
///
/// Reputation is not the only consequence. A second consumer — the node crate's
/// `record_pool_open_failure` — reads this marker to choose between answering a
/// downstream client `StreamError::NotFound` ("we could not obtain this blob") and
/// `Declined` ("this node will not serve this hash during this fetch"). That path
/// involves no peer and no reputation at all. One marked failure still answers
/// `NotFound`: a wallet that cannot fund a deposit, because the funding problem is
/// this node's, not the requester's.
///
/// So attaching this marker at a NEW site changes serve-path refusals, not just scoring.
/// Attach it when the failure means *this node* cannot serve anyone — a broken signer, an
/// unreadable or unwritable store, a poisoned lock, an unfunded wallet. Do NOT attach it to
/// a condition that is specific to one peer, one lane, or one blob, however much it is
/// "our side" of the exchange: a wedged lane to a single provider is ours and is still a
/// clean miss, because the node can serve every other request perfectly well.
///
/// It is a marker, so it composes: `.context(LocalPullFault)` on any error.
#[derive(Debug)]
#[doc(hidden)]
pub struct LocalPullFault;

impl std::fmt::Display for LocalPullFault {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("local buyer-side fault (not attributable to the upstream)")
    }
}

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

/// The bounds on a pull, each matched to the stage it governs (#1134, #1797).
///
/// A single whole-transfer deadline is mostly useless as a health signal: it has to be
/// sized against `blob size × link speed`, so it kills legitimate large or slow-but-healthy
/// transfers while a value small enough to catch a dead peer quickly cannot serve a big blob
/// at all. The operator ends up tuning a number that has nothing to do with node health.
///
/// Split by stage instead:
///
/// - **`window` + `floor_bps`** bound the STREAMING stage by THROUGHPUT. The requester
///   counts bytes read off the QUIC stream (sub-frame, so the measure is frame-size
///   independent) and trips when the bytes across the trailing `window` fall below
///   `floor_bps · window`. This catches both a wedge (throughput drops to zero) and a
///   slow drip (throughput below the floor), and is indifferent to transfer size and link
///   speed. `floor_bps == 0` leaves pure idle detection: at least one byte per window.
///   This is the primary mechanism. The abort is requester-local policy and does not score
///   the peer.
/// - **`hard_cap`** is an optional overall wall-clock escape hatch, `None` by
///   default. It exists for a caller that must bound total runtime regardless; it
///   is not how a stalled peer is detected.
///
/// - **`open`** bounds the OPEN stage (dial → request → verified `StreamResponse`)
///   by WALL CLOCK. That stage is bounded work whose duration does not depend on
///   the blob, so a slow one really is a stall and a wall clock is the right tool.
///
/// Every stage must carry a bound of its own. It is not enough for a caller to
/// wrap the whole pull in a timeout and call the open "bounded": the handshake
/// happens *inside* [`open_progressive_pull`], and the production callers run
/// with no overall cap, so leaving the bound to the caller would leave the
/// `StreamResponse` read with no bound at all — a peer that accepts a connection
/// and then says nothing would hang the pull forever. `open` exists so that
/// cannot be expressed.
/// # The relational invariant
///
/// `hard_cap`, when set, must STRICTLY EXCEED `open + window`. Both clocks below run inside
/// the cap's, and in the worst case consecutively — the open can legitimately consume its
/// whole budget before the streaming stage begins, and the floor needs one full `window` to
/// fire — so a cap that does not outlast both means the cap always fires first and the
/// throughput floor can never trip under it. The pull then looks fully configured while its
/// peer-health signal is dead.
///
/// [`Self::capped`] is fallible and the fields are private BECAUSE of that (#1145 review).
/// The invariant belongs on this type: with `pub` fields, callers build it with a struct
/// literal and the only check is a hardcoded `timeout > 2 × window` in the CLI's
/// `ClientFetchArgs::validate`, correct only because those call sites set `open` from the same
/// knob as `window`. Adding an
/// `--open-timeout-ms` flag would have made it silently wrong, in the direction that reopens
/// the hole. The invariant belongs to the type that has the values.
#[derive(Debug, Clone, Copy)]
pub struct PullDeadlines {
    /// Wall-clock bound on the open stage: dial, request, and the signed
    /// `StreamResponse`. Bounded work — a slow one is a stall.
    open: Duration,
    /// Trailing window over which the streaming-stage throughput floor is measured.
    window: Duration,
    /// Minimum bytes-per-second the streaming stage must sustain over `window`. `0`
    /// disables the throughput test and leaves pure idle detection (one byte per window).
    floor_bps: u64,
    /// Optional overall wall-clock cap on the whole exchange. `None` = uncapped;
    /// `open` and `window` between them are what keep an uncapped pull from hanging.
    hard_cap: Option<Duration>,
}

/// A [`PullDeadlines`] whose bounds cannot do their job. Carries the values so a CLI
/// can render the arithmetic back to the user rather than just saying "invalid".
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum DeadlineError {
    /// A zero budget elapses on its first poll, so the stage it bounds can never run.
    ZeroBudget,
    /// The cap does not outlast `open + window`, so it always fires first and the
    /// throughput floor — the signal that a healthy stream is still making progress —
    /// can never fire.
    CapCannotOutlastItsStages {
        /// The open-stage budget.
        open: Duration,
        /// The throughput-floor window that follows it.
        window: Duration,
        /// The whole-pull cap, which must exceed `open + window`.
        hard_cap: Duration,
    },
}

impl std::fmt::Display for DeadlineError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ZeroBudget => write!(f, "a deadline of zero elapses before any work can run"),
            Self::CapCannotOutlastItsStages {
                open,
                window,
                hard_cap,
            } => write!(
                f,
                "the overall cap ({hard_cap:?}) must exceed the open bound ({open:?}) plus the \
                 throughput-floor window ({window:?}): both run inside it and in the worst case \
                 consecutively, so below that the cap always elapses first and a stalled \
                 provider can never be detected"
            ),
        }
    }
}

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

impl PullDeadlines {
    /// The recommended shape: a wall clock on the open, inactivity on the stream,
    /// and no overall cap — so a pull of any size completes as long as the upstream
    /// keeps feeding it bytes.
    ///
    /// # Errors
    ///
    /// [`DeadlineError::ZeroBudget`] for a zero `open` or `window`. `floor_bps` may be zero
    /// (idle-detection mode: one byte per window).
    ///
    /// A zero `open` or `window` must fail here, not pass. One might argue it is a bound that
    /// fires too EAGERLY — loud, immediately obvious — rather than one that silently never
    /// fires. That is exactly backwards: a zero window makes the throughput floor demand
    /// progress over no time at all, so the streaming stage can never satisfy it.
    ///
    /// The only thing standing between config and that state was a `> 0` check in a resolver
    /// in another crate — the same advisory-invariant shape `ChunkData` had before #1088, and
    /// the reason this type owns its bounds at all.
    pub const fn new(
        open: Duration,
        window: Duration,
        floor_bps: u64,
    ) -> Result<Self, DeadlineError> {
        if open.is_zero() || window.is_zero() {
            return Err(DeadlineError::ZeroBudget);
        }
        Ok(Self {
            open,
            window,
            floor_bps,
            hard_cap: None,
        })
    }

    /// As [`Self::new`], plus an overall wall-clock cap on the whole exchange.
    ///
    /// # Errors
    ///
    /// [`DeadlineError::CapCannotOutlastItsStages`] if `hard_cap` does not strictly exceed
    /// `open + window`, and [`DeadlineError::ZeroBudget`] for a zero `open` or `window`. See
    /// the type's own docs for why this is the one constructor that must be fallible.
    pub fn capped(
        open: Duration,
        window: Duration,
        floor_bps: u64,
        hard_cap: Duration,
    ) -> Result<Self, DeadlineError> {
        if open.is_zero() || window.is_zero() {
            return Err(DeadlineError::ZeroBudget);
        }
        if hard_cap <= open.saturating_add(window) {
            return Err(DeadlineError::CapCannotOutlastItsStages {
                open,
                window,
                hard_cap,
            });
        }
        Ok(Self {
            open,
            window,
            floor_bps,
            hard_cap: Some(hard_cap),
        })
    }

    /// The open-stage wall clock.
    #[must_use]
    pub const fn open(&self) -> Duration {
        self.open
    }

    /// The streaming-stage throughput-floor window.
    #[must_use]
    pub const fn window(&self) -> Duration {
        self.window
    }

    /// The streaming-stage throughput floor, in bytes per second. `0` = idle detection only.
    #[must_use]
    pub const fn floor_bps(&self) -> u64 {
        self.floor_bps
    }

    /// The overall cap, if any.
    #[must_use]
    pub const fn hard_cap(&self) -> Option<Duration> {
        self.hard_cap
    }

    /// The single-deadline shape: one budget serving as the open bound, the
    /// streaming window, AND the overall cap, with the throughput floor disabled
    /// (`floor_bps == 0`, idle detection only). **Test-only — do not reach for this in
    /// production.** That conflation is exactly what #1134 set out to remove, and the
    /// name reads far more like a legitimate policy choice than it is.
    ///
    /// Note what it quietly costs, beyond re-introducing the size-coupled deadline:
    /// because `hard_cap == window`, and the cap's clock starts at the top of the whole
    /// exchange while the streaming window starts only once the open has completed, **the
    /// hard cap always elapses first — so [`PullStalled`] can never fire under it.** A pull
    /// built this way silently cannot detect a stalled peer, and so cannot score one.
    /// Every loopback test using this helper is exercising a pull with the stall
    /// signal disabled; the stall path is covered by
    /// `node_origin_mid_stream_silence_does_not_score_stalled_upstream`, which builds its
    /// deadlines explicitly.
    ///
    /// That is a statement about the `test-util` `stream_fetch*` wrappers, where
    /// `with_hard_cap` wraps the whole exchange. [`open_progressive_pull`] and
    /// [`driver::drive`] never consult `hard_cap` at all, so a `whole_transfer` used there
    /// would leave `PullStalled` perfectly able to fire — which is not a reprieve, just a
    /// different reason not to reach for this (#1145 review).
    ///
    /// It is also the reason this constructor stays infallible while [`Self::capped`] is not:
    /// it deliberately builds the very state `capped` refuses.
    ///
    /// TEST-ONLY, and now unrepresentable in production by construction: gated behind the
    /// `test-util` feature (#1145 review), so a production caller cannot name it and reach for
    /// the disabled-floor / uncapped state — it wants [`Self::new`] (floor-bounded) or
    /// [`Self::capped`] (floor-bounded with a leak guard). Used by the loopback helper
    /// `stream_fetch` and directly by the `client_loopback` suite, whose blobs are small
    /// enough that none of this matters.
    #[cfg(any(test, feature = "test-util"))]
    #[must_use]
    pub const fn whole_transfer(timeout: Duration) -> Self {
        Self {
            open: timeout,
            window: timeout,
            floor_bps: 0,
            hard_cap: Some(timeout),
        }
    }
}

/// Fetch `hash` from `target` over `cdn/client/v1` into memory, paying as bytes
/// arrive and verifying each bao chunk group as it lands.
///
/// `expected_signer` is the delivering node's Ethereum address, used to verify
/// the response `slash_sig`. `byte_offset` resumes a partial fetch. Use
/// [`stream_fetch_tracked`] instead if you need to persist the voucher watermark
/// the lane reached (#852); this convenience wrapper discards it.
///
/// TEST-ONLY, like the whole `stream_fetch*` family: gated behind the `test-util`
/// feature alongside [`PullDeadlines::whole_transfer`], the single-deadline shape
/// it passes (#1145 review). The wrappers are thin in-memory drivers of the one
/// receive loop, [`UpstreamPull`], for suites that want the decoded bytes back;
/// production callers stream into a store through [`driver::drive`] /
/// [`PeerSource`].
///
/// # Errors
///
/// Fails on connect/transport errors, an invalid or zero-rate response, a
/// `slash_sig` that does not recover to `expected_signer`, a mismatched echoed
/// field, a mid-stream `VoucherRejected`, a hash mismatch, or a timeout.
#[cfg(any(test, feature = "test-util"))]
#[allow(clippy::too_many_arguments)]
pub async fn stream_fetch(
    endpoint: &Endpoint,
    target: EndpointAddr,
    ctx: &PoolContext,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    timeout: Duration,
) -> anyhow::Result<Bytes> {
    stream_fetch_tracked(
        endpoint,
        target,
        ctx,
        slash_domain,
        expected_signer,
        hash,
        // `stream_fetch` models a node-to-node pull; a suite that routes on a
        // namespace calls `stream_fetch_tracked` directly.
        decdn_protocol::client::NO_NAMESPACE,
        byte_offset,
        timestamp_us,
        // The single-deadline shape (#1134): this helper's callers are
        // loopback tests moving tiny blobs, for which one budget serving as both
        // the overall cap and the stall bound is harmless. Production paths take
        // `PullDeadlines` directly and split the two.
        PullDeadlines::whole_transfer(timeout),
        // No buyer-side blob-size or rate ceiling on this loopback helper; a suite
        // that exercises either passes it through `stream_fetch_tracked`.
        0,
        0,
        &mut VoucherProgress::default(),
    )
    .await
}

/// Like [`stream_fetch`], but drives the fetch over a caller-owned
/// [`WarmConnection`] instead of dialling a fresh connection (#1848). Two
/// calls with the same `warm` reuse one dialled connection — one bi-stream per
/// hash — which is exactly what the connection-reuse test asserts.
///
/// TEST-ONLY, like the rest of the `stream_fetch*` family. Runs a single attempt
/// against a one-shot [`PoolLedger`] seeded from `ctx.prior_*` (the loopback
/// blobs it serves never trigger the wallet-less resume loop).
///
/// # Errors
///
/// The same set as [`stream_fetch`].
#[cfg(any(test, feature = "test-util"))]
#[allow(clippy::too_many_arguments)]
pub async fn stream_fetch_on(
    warm: &WarmConnection,
    ctx: &PoolContext,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    timeout: Duration,
) -> anyhow::Result<Bytes> {
    let ledger = Arc::new(ctx.new_ledger());
    fetch_in_memory_once_on(
        warm,
        ctx,
        ledger,
        slash_domain,
        expected_signer,
        hash,
        // Models a node-to-node pull, matching `stream_fetch`.
        decdn_protocol::client::NO_NAMESPACE,
        byte_offset,
        timestamp_us,
        0,
        0,
        PullDeadlines::whole_transfer(timeout),
        None,
    )
    .await
}

/// Like `stream_fetch`, but reports the lane's acked voucher watermark via
/// the `progress` out-param so the caller can persist what it paid (#852).
///
/// `progress` is an out-param: on return it holds the cumulative
/// `(bytes_delivered, amount)` of the last *acked* voucher. Internally the pull
/// runs against a one-shot [`PoolLedger`] seeded from `ctx.prior_*`; the
/// ledger's snapshot is copied back into `progress` on every return path — `Ok`,
/// `Err`, or timeout — so the caller can record progress even for a mid-stream
/// failure or a paid-but-corrupt delivery. See [`VoucherProgress`].
///
/// # Errors
///
/// Same as `stream_fetch`.
#[cfg(any(test, feature = "test-util"))]
#[allow(clippy::too_many_arguments)]
pub async fn stream_fetch_tracked(
    endpoint: &Endpoint,
    target: EndpointAddr,
    ctx: &PoolContext,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    deadlines: PullDeadlines,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
    progress: &mut VoucherProgress,
) -> anyhow::Result<Bytes> {
    stream_fetch_tracked_with_progress(
        endpoint,
        target,
        ctx,
        slash_domain,
        expected_signer,
        hash,
        namespace_id,
        byte_offset,
        timestamp_us,
        deadlines,
        max_blob_size_bytes,
        max_rate_per_mb,
        progress,
        None,
    )
    .await
}

/// Delivery-progress callback: invoked with `(received, expected)` after each
/// verified leaf lands, so a caller (e.g. `decdn get`) can render a progress
/// bar. Both slots are **content** bytes: `received` is the content position the
/// verifying decoder has reached and `expected` is the blob's content
/// `total_bytes`, constant across the pull — so a bar keyed on the two fills to
/// exactly 100%. `drive` / `acquire` (the resumable [`crate::driver`]
/// path the CLI `get` uses) additionally emit the
/// already-present resume base (`base_present`) once before streaming begins; the
/// `test-util` `stream_fetch_tracked_with_progress` wrapper reports the same
/// unit from its in-memory decoder.
///
/// It must not panic (it runs inside the hot receive loop) and must be `Send +
/// Sync` so the pull future stays spawnable.
pub type ProgressCallback = dyn Fn(u64, u64) + Send + Sync;

/// Like [`stream_fetch_tracked`], but also reports per-chunk delivery progress
/// through `on_progress` (see [`ProgressCallback`]) — the byte-progress hook the
/// watermark-only [`VoucherProgress`] out-param does not provide. `None` behaves
/// exactly like [`stream_fetch_tracked`].
///
/// # Errors
///
/// Same as `stream_fetch`.
#[cfg(any(test, feature = "test-util"))]
#[allow(clippy::too_many_arguments)]
pub async fn stream_fetch_tracked_with_progress(
    endpoint: &Endpoint,
    target: EndpointAddr,
    ctx: &PoolContext,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    deadlines: PullDeadlines,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
    progress: &mut VoucherProgress,
    on_progress: Option<&ProgressCallback>,
) -> anyhow::Result<Bytes> {
    // One-shot ledger seeded from the lane's prior cumulative state. A single
    // (non-shared) pull owns its ledger; concurrent shared-lane pulls use
    // `stream_fetch_shared` with a caller-owned ledger instead.
    let ledger = Arc::new(ctx.new_ledger());
    let result = with_hard_cap(
        deadlines.hard_cap,
        fetch_inner(
            endpoint,
            target,
            ctx,
            slash_domain,
            expected_signer,
            hash,
            namespace_id,
            byte_offset,
            timestamp_us,
            max_blob_size_bytes,
            max_rate_per_mb,
            deadlines,
            &ledger,
            on_progress,
        ),
    )
    .await;
    // Copy the watermark back into `progress` on EVERY return path (Ok, Err,
    // timeout) BEFORE returning, so the latest totals survive a mid-stream failure
    // or a paid-but-corrupt delivery (#852).
    //
    // Read it from the LEDGER rather than a copied-back cumulative, so the figure
    // is `settlement` and not `committed` — matching the two sibling persist paths
    // (`UpstreamPull::progress`, the `node_origin` drop guard). A voucher left
    // armed by an ambiguous send is owed, and the error arm here is exactly where
    // that happens. On the `Ok` arm nothing is armed, so the two agree.
    *progress = VoucherProgress::from_ledger(&ledger, ctx.prior_amount);
    result
}

/// Apply the optional overall wall-clock cap of a [`PullDeadlines`] to `fut`.
///
/// `None` runs the pull uncapped — which is safe precisely because the `stall`
/// bound inside bounds every streaming read. A pull with no cap cannot hang; it
/// can only take as long as the upstream keeps feeding it bytes, which is the
/// point (#1134).
#[cfg(any(test, feature = "test-util"))]
async fn with_hard_cap<F>(hard_cap: Option<Duration>, fut: F) -> anyhow::Result<Bytes>
where
    F: std::future::Future<Output = anyhow::Result<Bytes>>,
{
    match hard_cap {
        Some(cap) => tokio::time::timeout(cap, fut)
            .await
            .map_err(|_| anyhow::Error::new(PullTimeout { after: cap }))?,
        None => fut.await,
    }
}

/// Like `stream_fetch`, but issues vouchers through a caller-owned shared
/// [`PoolLedger`] so multiple concurrent pulls on ONE pool lane coordinate.
///
/// The bug this fixes: each `stream_fetch`/`stream_fetch_tracked` call seeds its
/// own voucher state from `ctx.prior_*`, so N concurrent pulls on the same lane
/// each compute the next cumulative `amount` independently and collide — the node
/// accepts exactly one and rejects the rest as `AmountRegression`. Passing every
/// concurrent caller the SAME `&Arc<PoolLedger>` (shared across
/// `tokio::spawn`/`join!`) serializes their voucher issuance through the
/// ledger's mutex: each issue advances the cumulative `amount`/`bytes_delivered`
/// in turn, the lane advances monotonically, and all pulls succeed.
///
/// The caller owns the ledger's lifetime and persists what the lane paid from it
/// directly (this entrypoint does not surface a [`VoucherProgress`] — the shared ledger
/// IS the watermark). Persist via [`PoolLedger::settlement`], NOT `snapshot`:
/// `snapshot`/`committed` report only ACKED vouchers, so a voucher left in the ack wait
/// (the drop the node's `SettleOnDrop` guard handles) is under-reported and its deposit
/// stranded — `settlement` adds the in-flight voucher back (#1122/#1145). `snapshot` is
/// also `async`, so a `Drop` guard cannot call it at all.
///
/// # Errors
///
/// Same as `stream_fetch`.
#[cfg(any(test, feature = "test-util"))]
#[allow(clippy::too_many_arguments)]
pub async fn stream_fetch_shared(
    endpoint: &Endpoint,
    target: EndpointAddr,
    ctx: &PoolContext,
    ledger: &Arc<PoolLedger>,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    deadlines: PullDeadlines,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
) -> anyhow::Result<Bytes> {
    with_hard_cap(
        deadlines.hard_cap,
        fetch_inner(
            endpoint,
            target,
            ctx,
            slash_domain,
            expected_signer,
            hash,
            // Shared-lane pulls model the daemon as buyer on a node-to-node
            // fill; the requester already discovered the holder, so no namespace.
            decdn_protocol::client::NO_NAMESPACE,
            byte_offset,
            timestamp_us,
            max_blob_size_bytes,
            max_rate_per_mb,
            deadlines,
            ledger,
            // Shared concurrent pulls interleave many blobs on one lane; a
            // single unified byte-progress readout would be meaningless, so this
            // path never reports progress.
            None,
        ),
    )
    .await
}

/// Where an [`open_stream`] gets its QUIC connection.
enum ConnSource<'a> {
    /// Dial a fresh one-shot connection to `target`. The pull owns it and closes
    /// it on its terminal method (`finish`/`abort`/drop).
    Dial {
        endpoint: &'a Endpoint,
        target: EndpointAddr,
        /// The runtime the dial and the connection's driver run on; `None` runs
        /// them on the caller's (see [`connection::dial`]).
        runtime: Option<&'a tokio::runtime::Handle>,
    },
    /// Reuse a caller-owned [`WarmConnection`]'s connection. The pull borrows it
    /// and leaves it open for the next hash; the [`WarmConnection`] closes it once.
    Reuse(&'a iroh::endpoint::Connection),
}

/// The OPEN stage of a `cdn/client/v1` pull, the single request site behind
/// [`open_progressive_pull`]: get the connection from `source` (dial a fresh one,
/// or reuse a warm one), open the bi-stream, send the [`StreamRequest`], then read
/// and verify the signed [`StreamResponse`]. Returns the live connection, its
/// streams, and the verified response for the caller to stream from.
///
/// **Bounded as a whole by `open`** (#1134), and that bound lives HERE rather than
/// in the caller for a reason worth stating: the production pull paths run with no
/// overall wall-clock cap, so that a blob of any size can complete as long as bytes
/// keep arriving. Leaving the open to "whatever the caller wraps us in" therefore
/// means leaving it *unbounded* — and this is precisely the stage where a peer can
/// accept a connection, take our request, and then say nothing at all. Such a peer
/// would hang the pull until the QUIC idle timeout.
///
/// The stall bound cannot cover this stage: it measures inactivity BETWEEN bytes,
/// and here no byte has arrived yet. A wall clock is the right tool because the
/// open is bounded work whose duration does not scale with the blob.
///
/// A `slash_sig` that does not recover to `expected_signer`, a mismatched echoed
/// field, or a zero rate fails here (see [`verify_response`]) — before any byte is
/// paid for. A refusal (`body.ok == false`) is returned to the caller intact, since
/// only the caller knows how to classify it.
#[allow(clippy::too_many_arguments)]
async fn open_stream(
    source: ConnSource<'_>,
    ctx: &PoolContext,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    byte_len: u64,
    timestamp_us: u64,
    open: Duration,
) -> anyhow::Result<(
    iroh::endpoint::Connection,
    SendStream,
    RecvStream,
    StreamResponse,
    StreamResponseExt,
)> {
    tokio::time::timeout(open, async move {
        let conn = match source {
            ConnSource::Dial {
                endpoint,
                target,
                runtime,
            } => connection::dial(endpoint, target, runtime, "connect failed").await?,
            // Already dialled and warm — reuse the handle. A fresh `open_bi` below
            // gives this hash its own stream.
            ConnSource::Reuse(conn) => conn.clone(),
        };
        let (mut send, mut recv) = conn
            .open_bi()
            .await
            .map_err(|e| rate_limited::transport_error("open_bi failed", e))?;

        let req = StreamRequest {
            hash,
            // The serving node routes on this for its pull-through authorized-origin
            // gate and origin-directory fallback (ADR 005 §Namespace routing). A
            // client fetch passes the namespace it published under; a node-to-node
            // pull to a DHT-discovered holder passes `NO_NAMESPACE` (0) — the holder
            // already has the bytes (ADR 002 §Retrieval by namespace) — while a pull
            // to a directory-discovered cold origin passes the served namespace so
            // that origin's gate resolves and it can fill from its own backend.
            namespace_id,
            pool_id: ctx.pool_id.into(),
            byte_offset,
            // `0` = whole-tail fetch; a caller requesting a bounded middle gap
            // (the gap-driven driver, #1608, via `PeerSource`) passes the gap's
            // length so the server scopes both the serve and the payment to it.
            byte_len,
            timestamp_us,
        };
        // Two-phase encode (ADR 005): attach the client identity binding when the
        // context carries one, so the serving node can prove lane authority and
        // authorize a cache-miss origin pull (#1115). Absent ⇒ no ext bytes: the
        // base frame alone (unbound node-to-node / registered-client path).
        let ext = client_binding_ext(ctx);
        let payload = decdn_protocol::encode_stream_request(&req, ext.as_ref())
            .map_err(|e| anyhow::anyhow!("encode stream request: {e}").context(LocalPullFault))?;
        write_frame(&mut send, &payload)
            .await
            .map_err(|e| rate_limited::transport_error("write stream request", e))?;

        let (resp, resp_ext) = read_stream_response(&mut recv).await?;
        verify_response(
            &resp,
            &resp_ext,
            slash_domain,
            expected_signer,
            hash,
            ctx.pool_id,
            timestamp_us,
        )?;
        Ok((conn, send, recv, resp, resp_ext))
    })
    .await
    .map_err(|_| anyhow::Error::new(PullTimeout { after: open }))?
}

/// Wallet-less resume (issue #1481 §5): the maximum number of times a fetch will reopen a fresh
/// stream after a rejection carrying a watermark-gated bundle
/// ([`VoucherRejectReason::is_watermark_gated`]). Bounds a node that keeps rejecting (a buggy or
/// adversarial peer echoing a bundle that never lets the client catch up) to a handful of round
/// trips rather than looping forever; a healthy self-heal needs exactly one.
///
/// Public because the STREAMING fetch (#1120) drives its own reopen loop — it
/// owns the output file and must rewind it before each retry, which this crate
/// cannot do for it — and both loops must agree on the bound.
#[doc(hidden)]
pub const MAX_RESUME_ATTEMPTS: u32 = 3;

/// In-memory fetch with wallet-less resume (issue #1481 §5): if a mid-stream
/// voucher rejection carries a signer-verified [`WatermarkBundle`] for one of
/// the four regression/exhaustion reasons, reseed `ledger` from it and reopen
/// the pull — at the **same** `byte_offset` the caller originally requested,
/// not `bundle.bytes_delivered` — instead of surfacing the rejection as
/// terminal.
///
/// `bundle.bytes_delivered` is deliberately NOT used as the retry's wire
/// `byte_offset`, even though that is what makes a `WatermarkBundle` look
/// resumable at a glance. The two are different axes: `bytes_delivered` is
/// the LANE's cumulative payment counter (used to reconstruct each
/// voucher's EIP-712 `bytesDelivered`, ADR 005 — [`Voucher`] carries no such
/// field on the wire), not a position within THIS blob's byte range. A
/// lane can pay for many blobs; jumping the wire offset ahead to the
/// lane's cumulative would, for a caller that requested `byte_offset == 0`,
/// silently return a TRUNCATED tail instead of the full blob the caller is
/// relying on getting back. The bytes already decoded in the failed attempt
/// were dropped with it, so there is nothing to legitimately splice a jump
/// with. Retrying at the caller's original offset is always safe and is the
/// only thing this bundle field is actually needed for: fixing the ledger's
/// amount/bytes BASELINE so the resumed stream's vouchers verify against what
/// the node now expects, not re-deriving where in the blob to resume.
///
/// This is the retry loop for this case on the in-memory path —
/// [`UpstreamPull::pay_one`] never retries itself, because the stream it holds
/// is already dead by the time a mid-stream `VoucherRejected` reaches it (the
/// node finishes its send side before replying,
/// `handlers/client/wire.rs::write_reject`); a fresh stream can only be opened
/// by whoever owns the connection, which is here.
///
/// Every `stream_fetch*` wrapper goes through here, so a resumable rejection is
/// retried transparently and the caller only ever sees the FINAL outcome
/// (success, or the original terminal error once [`MAX_RESUME_ATTEMPTS`] is
/// exhausted or the reason/bundle isn't eligible).
///
/// The production callers do not use this wrapper: the gap-driven
/// [`driver::drive`] reaches the same reseed loop around [`open_progressive_pull`].
///
/// A bundle-less rejection, a non-gated reason, or a bundle that fails
/// [`WatermarkBundle::validate`] (malformed `last_signature` length) is
/// never treated as resumable and is returned to the caller unchanged on
/// the first attempt.
#[cfg(any(test, feature = "test-util"))]
#[allow(clippy::too_many_arguments)]
async fn fetch_inner(
    endpoint: &Endpoint,
    target: EndpointAddr,
    ctx: &PoolContext,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
    deadlines: PullDeadlines,
    ledger: &Arc<PoolLedger>,
    on_progress: Option<&ProgressCallback>,
) -> anyhow::Result<Bytes> {
    for attempt in 0..=MAX_RESUME_ATTEMPTS {
        let result = fetch_in_memory_once(
            endpoint,
            target.clone(),
            ctx,
            Arc::clone(ledger),
            slash_domain,
            expected_signer,
            hash,
            namespace_id,
            byte_offset,
            timestamp_us,
            max_blob_size_bytes,
            max_rate_per_mb,
            deadlines,
            on_progress,
        )
        .await;
        let Err(e) = result else {
            return result;
        };
        if attempt == MAX_RESUME_ATTEMPTS {
            return Err(e);
        }
        // A bundle that proves no desync (see `heal_watermark_desync`) is not worth
        // the resume budget: surface the real error instead.
        let healed = match rejection_watermark(&e, ctx) {
            Some(watermark) => heal_watermark_desync(&e, watermark, ledger).await,
            None => None,
        };
        let Some(healed) = healed else {
            return Err(e);
        };
        tracing::debug!(
            attempt,
            byte_offset,
            ?healed,
            "voucher rejection carried an authenticated watermark bundle; retrying at the same \
             byte_offset"
        );
    }
    // Unreachable: the loop above always returns on both the `Ok` and every `Err`
    // branch (either directly or after `attempt == MAX_RESUME_ATTEMPTS`
    // triggers on the final iteration). Kept as a typed bail rather than
    // `unreachable!()`/`panic!()` per the workspace anti-panic policy.
    Err(anyhow::anyhow!("resume loop exited without returning"))
}

/// One attempt of the in-memory `stream_fetch*` path: open a progressive pull,
/// feed its wire bytes through the bao verifying decoder AS THEY ARRIVE, and
/// return the decoded plaintext span `[byte_offset, total_bytes)`.
///
/// This is the same receive loop the production callers drive
/// ([`UpstreamPull::next_chunk`] under a [`PullReader`]) with an in-memory `Vec`
/// standing in for the ranged store / cache admit; there is no second receive
/// implementation. A corrupt chunk group therefore aborts the pull at that group
/// (ADR 038 §Receive side), before the rest of the stream is pulled or paid for —
/// so a peer that signs an inflated `total_bytes` and streams framed garbage can
/// bill at most the bytes that arrived before the first unverifiable group, never
/// its claim.
///
/// # Errors
///
/// Everything [`open_progressive_pull`] raises, plus [`HashMismatch`] for a
/// paid-but-corrupt delivery (including a non-empty root claimed against an empty
/// blob, #1054), the typed pull faults [`PullReader`] stashes (stall, refusal,
/// voucher rejection), a truncated stream, or a short delivery at `finish`.
#[cfg(any(test, feature = "test-util"))]
#[allow(clippy::too_many_arguments)]
async fn fetch_in_memory_once(
    endpoint: &Endpoint,
    target: EndpointAddr,
    ctx: &PoolContext,
    ledger: Arc<PoolLedger>,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
    deadlines: PullDeadlines,
    on_progress: Option<&ProgressCallback>,
) -> anyhow::Result<Bytes> {
    let (header, pull) = open_progressive_pull(
        endpoint,
        target,
        ctx,
        ledger,
        slash_domain,
        expected_signer,
        hash,
        namespace_id,
        byte_offset,
        timestamp_us,
        max_blob_size_bytes,
        max_rate_per_mb,
        deadlines,
        // Whole tail: the in-memory wrapper has no store to compute gaps against.
        0,
        // One long-lived test runtime: the dial runs on it.
        None,
    )
    .await?;
    let total_bytes = header.total_bytes;
    // A 0-byte blob (#1054) aligns to an empty range the decoder cannot anchor, so
    // the empty stream must be proven against the empty root explicitly rather than
    // accepted for an arbitrary one (see [`reject_empty_claim_for_nonempty_root`]).
    if total_bytes == 0 {
        reject_empty_claim_for_nonempty_root(total_bytes, hash)?;
        pull.finish().await?;
        return Ok(Bytes::new());
    }
    let aligned = align_range(byte_offset, 0, total_bytes)
        .map_err(|e| anyhow::anyhow!("range alignment: {e}").context(LocalPullFault))?;
    let reader = PullReader::new(pull);
    let (plaintext, reader) =
        decode_to_vec(hash, total_bytes, &aligned, reader, on_progress).await?;
    // Wire completeness (the full promised bao wire size arrived before
    // `StreamEnd`) and a clean close; the watermark it returns is read back from
    // the shared ledger by the caller, so it is not needed here.
    reader.into_inner().finish().await?;
    // The decoded buffer spans the aligned superset `[fetch_start, total_bytes)`;
    // trim back to the caller's requested span. Guard the bound explicitly:
    // `Bytes::slice` panics out of range, and a short decode must surface as a
    // clean error.
    let lead = usize::try_from(byte_offset.saturating_sub(aligned.fetch_start()))?;
    let bytes = Bytes::from(plaintext);
    if lead > bytes.len() {
        anyhow::bail!("decoded range shorter than requested span");
    }
    Ok(bytes.slice(lead..))
}

/// Like [`fetch_in_memory_once`], but opens the pull on a caller-owned
/// [`WarmConnection`] so the in-memory `stream_fetch_on` wrapper can prove
/// connection reuse across hashes (#1848).
///
/// # Errors
///
/// The same set as [`fetch_in_memory_once`].
#[cfg(any(test, feature = "test-util"))]
#[allow(clippy::too_many_arguments)]
async fn fetch_in_memory_once_on(
    warm: &WarmConnection,
    ctx: &PoolContext,
    ledger: Arc<PoolLedger>,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
    deadlines: PullDeadlines,
    on_progress: Option<&ProgressCallback>,
) -> anyhow::Result<Bytes> {
    let (header, pull) = open_progressive_pull_on(
        warm.connection(),
        connection::ConnOwner::Borrowed,
        ctx,
        ledger,
        slash_domain,
        expected_signer,
        hash,
        namespace_id,
        byte_offset,
        timestamp_us,
        max_blob_size_bytes,
        max_rate_per_mb,
        deadlines,
        // Whole tail: the in-memory wrapper has no store to compute gaps against.
        0,
    )
    .await?;
    let total_bytes = header.total_bytes;
    if total_bytes == 0 {
        reject_empty_claim_for_nonempty_root(total_bytes, hash)?;
        pull.finish().await?;
        return Ok(Bytes::new());
    }
    let aligned = align_range(byte_offset, 0, total_bytes)
        .map_err(|e| anyhow::anyhow!("range alignment: {e}").context(LocalPullFault))?;
    let reader = PullReader::new(pull);
    let (plaintext, reader) =
        decode_to_vec(hash, total_bytes, &aligned, reader, on_progress).await?;
    reader.into_inner().finish().await?;
    let lead = usize::try_from(byte_offset.saturating_sub(aligned.fetch_start()))?;
    let bytes = Bytes::from(plaintext);
    if lead > bytes.len() {
        anyhow::bail!("decoded range shorter than requested span");
    }
    Ok(bytes.slice(lead..))
}

/// Feed a bao interleaved wire stream through the verifying decoder, appending
/// each verified leaf to an in-memory buffer, and return the buffer plus the
/// reader for the caller to finish. Every chunk group is checked against the
/// content-hash root `root` as it is decoded, so a corrupt group — at any offset,
/// including a resumed tail — is rejected at that group without needing earlier
/// bytes or the rest of the stream (ADR 038 §Receive side).
///
/// The decoded buffer spans the chunk-group-aligned **superset**
/// `[aligned.fetch_start(), aligned.fetch_end())` the server serves (a bao proof
/// anchors whole 16 KiB groups); the caller trims the lead. The buffer is grown
/// leaf by leaf rather than pre-sized: `aligned` derives from the peer's
/// unverified `total_bytes`, so a capacity hint would let an inflated claim drive
/// an over-allocation before a single byte verifies.
///
/// `on_progress`, when set, is called with `(content_position, total_bytes)` after
/// each verified leaf lands — the same content-byte unit [`driver::drive`] reports.
///
/// Same loop shape as [`ClientRangedStore::ingest_stream`], minus the durable
/// checkpointing; a typed fault the reader stashed ([`StashedFault`]) takes
/// precedence over the decoder's own complaint, and a decode failure is
/// classified by [`sink::classify_decode_error`] ([`HashMismatch`] for a
/// verification failure, a truncation error for a short stream).
///
/// # Errors
///
/// The reader's stashed typed fault, [`HashMismatch`], or a truncation / IO fault.
#[cfg(any(test, feature = "test-util"))]
async fn decode_to_vec<R>(
    root: [u8; 32],
    total_bytes: u64,
    aligned: &AlignedRange,
    reader: R,
    on_progress: Option<&ProgressCallback>,
) -> anyhow::Result<(Vec<u8>, R)>
where
    R: iroh_io::AsyncStreamReader + StashedFault + Send,
{
    let tree = BaoTree::new(total_bytes, IROH_BLOCK_SIZE);
    let root = blake3::Hash::from_bytes(root);
    let mut decoder = ResponseDecoder::new(root, aligned.chunk_ranges().clone(), tree, reader);
    let mut out = Vec::new();
    loop {
        match decoder.next().await {
            ResponseDecoderNext::More((rest, Ok(BaoContentItem::Leaf(leaf)))) => {
                out.extend_from_slice(&leaf.data);
                if let Some(cb) = on_progress {
                    let position = aligned
                        .fetch_start()
                        .saturating_add(u64::try_from(out.len()).unwrap_or(u64::MAX));
                    cb(position, total_bytes);
                }
                decoder = rest;
            }
            ResponseDecoderNext::More((rest, Ok(BaoContentItem::Parent(_)))) => decoder = rest,
            ResponseDecoderNext::More((rest, Err(decode_err))) => {
                let mut r = rest.finish();
                if let Some(fault) = r.take_fault() {
                    return Err(fault);
                }
                return Err(sink::classify_decode_error(decode_err));
            }
            ResponseDecoderNext::Done(mut r) => {
                if let Some(fault) = r.take_fault() {
                    return Err(fault);
                }
                return Ok((out, r));
            }
        }
    }
}

/// The security-critical core shared by every "did WE sign this?" check: does `signature` recover
/// to `expected` over `voucher` under `domain`?
///
/// A node's claim about a lane's watermark arrives over an unauthenticated application-level
/// message — a mid-stream `StreamError::VoucherRejected` on the fetch path (#1042) — so the claimed
/// `amount`/`bytes_delivered` are attacker-controllable on the wire. Without this check a malicious
/// or buggy upstream could hand back an inflated watermark and have the client act on it: `reseed`
/// its ledger and then sign a voucher for the echoed `amount + delta` on the retried pull. That is
/// a voucher this client genuinely holds the key to sign and the node can redeem on-chain up to the
/// deposit, draining the pool while delivering ~nothing.
///
/// The client's OWN last-accepted voucher signature closes that hole: a node can only echo back a
/// signature the client itself produced, so a tuple that verifies is by construction one this client
/// already committed to. That the node's stored signature always matches its stored tuple is the
/// write-side invariant of the seller-side lane state, which writes `last_amount`/`last_bytes` and
/// the accepted signature together and only after verifying the signature against the lane's pinned
/// capability `signer`. A signature that does not recover to `expected` is treated as a hostile or
/// corrupt echo, never as evidence.
///
/// Callers MUST verify over the exact tuple they are about to adopt as their new committed baseline
/// — never a tuple derived from it — so there is no window in which the proof covers one state and
/// the action commits another. Taking the whole [`Voucher`] rather than its fields loose is what
/// lets a caller verify and then act on *the same value*.
#[must_use]
pub(crate) fn voucher_signed_by(
    voucher: &Voucher,
    signature: &[u8],
    expected: Address,
    domain: &Eip712Domain,
) -> bool {
    let Ok(signature) = Signature::try_from(signature) else {
        return false;
    };
    SignedVoucher {
        voucher: voucher.clone(),
        signature,
    }
    .recover_signer(domain)
    .is_ok_and(|recovered| recovered == expected)
}

/// Extract a resumable, AUTHENTICATED [`WatermarkBundle`] from a pull error, or `None` if the
/// error is not an `UpstreamVoucherRejected`, its reason is not
/// [`VoucherRejectReason::is_watermark_gated`], it carries no bundle, the bundle fails
/// [`WatermarkBundle::validate`] (malformed `last_signature` length), or either of the two
/// security-critical checks fails: `last_signature` must recover to this client's OWN
/// voucher-signing address over the bundle's anchor and chain section, and the bundle's `tip`
/// must PROVE its `verified_index`, by hashing forward to the root that signature covers.
///
/// Both halves of the bundle are folded into money a resuming signer signs, so both have to be
/// evidence. The anchor half is the client's own signature ([`voucher_signed_by`]; see its doc
/// for why acting on an unverified watermark is a pool-draining hole rather than a robustness
/// nicety). The chain half carries no signature at all, which is what makes the tip check the
/// only thing standing between an inflated `verified_index` and a client that signs it.
pub(crate) fn resumable_watermark<'a>(
    err: &'a anyhow::Error,
    ctx: &PoolContext,
) -> Option<&'a WatermarkBundle> {
    let rejected = err.downcast_ref::<UpstreamVoucherRejected>()?;
    if !rejected.reason.is_watermark_gated() {
        return None;
    }
    let bundle = rejected.bundle.as_ref()?;
    if bundle.validate().is_err() {
        return None;
    }
    let claimed = Voucher {
        pool_id: ctx.pool_id,
        signer: ctx.client_signer.address(),
        provider: ctx.provider,
        amount: U256::from(bundle.amount),
        bytes_delivered: U256::from(bundle.bytes_delivered),
        chain_root: B256::from(bundle.chain_root),
        chunk_price: U256::from(bundle.chunk_price),
    };
    if !voucher_signed_by(
        &claimed,
        &bundle.last_signature,
        ctx.client_signer.address(),
        &ctx.voucher_domain,
    ) {
        return None;
    }
    frontier_is_proved(bundle).then_some(bundle)
}

/// How [`heal_watermark_desync`] resolved a rejection's authenticated watermark.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Healed {
    /// The bundle was AHEAD of our committed watermark: the node holds a voucher
    /// we lost. [`PoolLedger::reseed`] moved us up to it.
    Reseeded,
    /// The bundle was BEHIND our committed watermark on `amount`, and the
    /// ledger moved to it: on `Underpaid`, we hold vouchers the node never
    /// accepted ([`PoolLedger::rebase`]); on `BytesRegression`, the bundle is
    /// ahead of us on bytes, because the lane resumed below a voucher the node
    /// holds ([`PoolLedger::rebase_ahead_on_bytes`]).
    Rebased,
    /// The ledger already covers this rejection's watermark: an `Underpaid` or
    /// `BytesRegression` for a voucher signed before the latest rebase, or an
    /// `UnderFold`, `AmountRegression` or `BytesRegression` whose bundle the
    /// ledger has reached. The watermark
    /// did not move, and the pull retries from the ledger. An `UnderFold` that
    /// no heal has taken yet also retires the live chain the node refused
    /// ([`PoolLedger::retire_unadopted_chain`]), so the retry opens a fresh one.
    Stale,
}

/// The authenticated watermark a voucher rejection carries, as the cumulative a
/// ledger re-anchors to, or `None` when [`resumable_watermark`] refuses it.
#[must_use]
pub(crate) fn rejection_watermark(err: &anyhow::Error, ctx: &PoolContext) -> Option<Cumulative> {
    resumable_watermark(err, ctx).map(Cumulative::from)
}

/// Heal the watermark desync a voucher rejection reports, from `watermark` — the
/// authenticated bundle [`rejection_watermark`] extracted — and say how, or
/// `None` when it proves no desync and the caller must surface the real error.
///
/// A bundle ahead of our committed watermark reseeds, whatever the reason. A
/// bundle that does not advance our amount rebases on
/// [`VoucherRejectReason::Underpaid`], and on a
/// [`VoucherRejectReason::BytesRegression`] whose bundle is ahead of us on
/// bytes (see [`heal_bytes_regression`]). A rebase applies only to a voucher
/// signed under the ledger's current generation; see [`PoolLedger::rebase`].
/// On [`VoucherRejectReason::UnderFold`] or
/// [`VoucherRejectReason::AmountRegression`] it retries without moving the
/// ledger (below). On every other reason it proves no desync: the node attaches
/// a bundle to every watermark-gated rejection once a voucher is accepted, so an
/// exhausted lane echoes our own watermark straight back.
///
/// An `UnderFold` or `AmountRegression` whose folded bundle does not advance our
/// amount splits two ways. Both reasons say a proof trailed the node's
/// watermark. When our ledger already covers the bundle on both axes, the
/// stream retries ([`Healed::Stale`]): either a sibling stream took the same
/// rejection and healed first, or the ledger metered reveals under a root the
/// node refused, which an `UnderFold` retires. When it covers the amount but
/// not the bytes, the ledger paid the node's whole claim but signed fewer bytes
/// than the node holds. No bundle can heal that; it is terminal and logged at
/// `warn!`.
pub(crate) async fn heal_watermark_desync(
    err: &anyhow::Error,
    watermark: Cumulative,
    ledger: &PoolLedger,
) -> Option<Healed> {
    if ledger.reseed(watermark) {
        tracing::info!(
            amount = %watermark.amount,
            bytes = %watermark.bytes,
            "voucher rejection carried a watermark ahead of ours; reseeded"
        );
        return Some(Healed::Reseeded);
    }
    let rejected = err.downcast_ref::<UpstreamVoucherRejected>()?;
    if matches!(
        rejected.reason,
        VoucherRejectReason::UnderFold | VoucherRejectReason::AmountRegression
    ) {
        return already_covered(rejected.reason, watermark, ledger).await;
    }
    match rejected.reason {
        VoucherRejectReason::Underpaid => {
            let outcome = ledger.rebase(watermark, rejected.proof_generation).await;
            rebase_outcome(
                rejected.reason,
                outcome,
                watermark,
                rejected.proof_generation,
            )
        }
        VoucherRejectReason::BytesRegression => {
            heal_bytes_regression(watermark, rejected.proof_generation, ledger).await
        }
        _ => None,
    }
}

/// Resolve a `BytesRegression` whose bundle does not advance our amount.
///
/// A bundle ahead of us on bytes is our own voucher that the node holds and
/// our ledger does not. The lane resumed from a lower anchor, typically the
/// chain watermark after the payer's store lost the lane, while the node holds
/// an unredeemed voucher above it. The ledger priced its spans from that lower
/// anchor, so its amount passed the node's anchor while its bytes still
/// trailed it. The rebase moves the ledger to the node's anchor
/// ([`PoolLedger::rebase_ahead_on_bytes`]). The payer never pays more than it
/// signed for, because the anchor is its own voucher.
///
/// A bundle the ledger already covers on both axes retries from the ledger
/// ([`Healed::Stale`]). The rejected voucher trailed the ledger: a sibling
/// stream healed first, or a sibling voucher committed past the node's anchor,
/// or the voucher was signed before the latest rebase. Every voucher the
/// ledger signs next is at or above the node's anchor on both axes. A
/// `BytesRegression` with no bundle is the single-signer fault that no heal
/// takes; it never reaches this function.
async fn heal_bytes_regression(
    watermark: Cumulative,
    proof_generation: Option<u64>,
    ledger: &PoolLedger,
) -> Option<Healed> {
    let outcome = ledger
        .rebase_ahead_on_bytes(watermark, proof_generation)
        .await;
    if outcome != Rebase::Refused {
        return rebase_outcome(
            VoucherRejectReason::BytesRegression,
            outcome,
            watermark,
            proof_generation,
        );
    }
    let committed = ledger.committed();
    if committed.amount >= watermark.amount && committed.bytes >= watermark.bytes {
        tracing::debug!(
            bundle_amount = %watermark.amount,
            bundle_bytes = %watermark.bytes,
            committed_amount = %committed.amount,
            committed_bytes = %committed.bytes,
            ?proof_generation,
            "bytes-regression rejection carried a watermark our ledger already covers; \
             retrying from the ledger"
        );
    } else {
        // The bundle is ahead of us on amount, which `reseed` refused a moment
        // ago: a rewind landed in between. The retry's next rejection carries
        // the same bundle, and the reseed takes it then.
        tracing::warn!(
            bundle_amount = %watermark.amount,
            bundle_bytes = %watermark.bytes,
            committed_amount = %committed.amount,
            committed_bytes = %committed.bytes,
            ?proof_generation,
            "bytes-regression rejection carried a watermark ahead of our ledger on amount \
             that neither a reseed nor a rebase took; retrying"
        );
    }
    Some(Healed::Stale)
}

/// Resolve a trailing-proof rejection (`UnderFold` or `AmountRegression`) whose
/// folded bundle does not advance our amount. When our ledger covers its bytes
/// too, the stream retries. Otherwise the ledger covers the amount but not the
/// bytes, which no bundle can heal.
///
/// A covering ledger is in one of two states. Either a sibling stream took the
/// same rejection and healed the lane first, or the ledger meters a root the
/// node refused to adopt. The second state follows an `UnderFold` on a fresh
/// root: the stream released reveals under it before the rejection landed, and
/// only the last one rewound. A retry would re-anchor under that root and draw
/// the same rejection, so an `UnderFold` retires it
/// ([`PoolLedger::retire_unadopted_chain`]), unless a heal has already taken
/// this bundle.
async fn already_covered(
    reason: VoucherRejectReason,
    watermark: Cumulative,
    ledger: &PoolLedger,
) -> Option<Healed> {
    let committed = ledger.committed();
    if committed.bytes < watermark.bytes {
        tracing::warn!(
            bundle_amount = %watermark.amount,
            bundle_bytes = %watermark.bytes,
            committed_amount = %committed.amount,
            committed_bytes = %committed.bytes,
            "trailing-proof rejection carried a watermark our amount covers but our bytes \
             do not; the ledger paid the amount but not the bytes"
        );
        return None;
    }
    let retired = if reason == VoucherRejectReason::UnderFold {
        ledger.retire_unadopted_chain(watermark).await
    } else {
        None
    };
    log_covered(watermark, committed, retired);
    Some(Healed::Stale)
}

/// Log how [`already_covered`] resolved a rejection whose watermark our
/// `committed` ledger covers, and the root it `retired`, if any.
fn log_covered(watermark: Cumulative, committed: Cumulative, retired: Option<B256>) {
    if let Some(root) = retired {
        tracing::info!(
            retired_root = %root,
            bundle_amount = %watermark.amount,
            bundle_bytes = %watermark.bytes,
            committed_amount = %committed.amount,
            committed_bytes = %committed.bytes,
            "under-fold rejection: the node never adopted our live chain, and our ledger \
             already covers its watermark; folded the ledger and retired the chain, so the \
             retry opens a fresh one"
        );
    } else if committed == watermark {
        tracing::debug!(
            amount = %watermark.amount,
            bytes = %watermark.bytes,
            "trailing-proof rejection already healed by a sibling stream; retrying"
        );
    } else {
        tracing::debug!(
            bundle_amount = %watermark.amount,
            bundle_bytes = %watermark.bytes,
            committed_amount = %committed.amount,
            committed_bytes = %committed.bytes,
            "trailing-proof rejection carried a watermark our ledger is already past; \
             retrying from the ledger"
        );
    }
}

/// Log what a [`PoolLedger::rebase`] did and say how the heal resolved.
fn rebase_outcome(
    reason: VoucherRejectReason,
    outcome: Rebase,
    watermark: Cumulative,
    proof_generation: Option<u64>,
) -> Option<Healed> {
    match outcome {
        Rebase::Rebased { from } => {
            tracing::warn!(
                ?reason,
                from_amount = %from.amount,
                from_bytes = %from.bytes,
                to_amount = %watermark.amount,
                to_bytes = %watermark.bytes,
                "upstream refused our vouchers: the watermark it accepted differs from our \
                 lane's; rebased to it"
            );
            Some(Healed::Rebased)
        }
        Rebase::Stale => {
            tracing::debug!(
                ?reason,
                ?proof_generation,
                "rejection of a voucher signed before the latest rebase; retrying from the \
                 healed anchor"
            );
            Some(Healed::Stale)
        }
        Rebase::Refused => None,
    }
}

/// Whether the bundle's `tip` proves the `verified_index` it reports:
/// `keccak^verified_index(tip) == chain_root`.
///
/// A bundle reports two things and the resuming signer folds BOTH into the amount it re-signs.
/// The anchor half is covered by the client's own signature. The chain half is covered by
/// nothing — `verified_index` is a number the node writes, and the signature it echoes is over a
/// voucher that contains no index. A node that reported 255 where the client released 3 would
/// otherwise collect `252 × chunk_price` for chunks it never delivered, from a client that signed
/// for them itself.
///
/// The tip closes that, and it is the only thing that can: a preimage is unforgeable, so the
/// deepest one the node can present is the deepest one it was actually GIVEN, and hashing it
/// forward to a root the client's own signature commits to turns `verified_index` from a claim
/// into a proof. The walk is `verified_index` keccaks against local state — no signer, no round
/// trip.
///
/// `verified_index == 0` folds nothing, so it needs no proof and carries no tip (the node sends
/// all-zero). That is the never-metered and sealed case, and the only one where an all-zero tip
/// is legitimate.
#[must_use]
fn frontier_is_proved(bundle: &WatermarkBundle) -> bool {
    bundle.verified_index == 0
        || decdn_incentive::chain::verify_forward(
            B256::from(bundle.tip),
            bundle.verified_index,
            B256::from(bundle.chain_root),
        )
}

/// Whether a failed open of a bounded range could mean only that the range's
/// START is past the blob's end: the size it was cut from was wrong. An end
/// past the blob clamps and serves rather than failing, so this fires only on
/// a start with no bytes to serve.
///
/// A holder refuses such a range before it signs (`RangeNotSatisfiable`, which
/// reaches the wire as [`StreamError::Declined`]). That refusal names only the
/// class, so it removes the holder for this fetch and does not qualify here. A
/// relay signs its own size instead, and the range then fails to align against
/// it here ([`decdn_bao_range::RangeVerifyError::RangeOutOfBounds`] or
/// [`ResumeOffsetPastEnd`]). A caller that cut the range from an unsigned size
/// asks for the whole blob on these, and treats every other failure as the
/// open's own outcome (#2063).
#[must_use]
#[doc(hidden)]
pub fn is_range_past_end(err: &anyhow::Error) -> bool {
    err.downcast_ref::<ResumeOffsetPastEnd>().is_some()
        || err.chain().any(|cause| {
            matches!(
                cause.downcast_ref::<decdn_bao_range::RangeVerifyError>(),
                Some(decdn_bao_range::RangeVerifyError::RangeOutOfBounds { .. })
            )
        })
}

/// The wire-byte bound for a fetch of `[byte_offset, byte_offset + byte_len)`
/// (`byte_len == 0` meaning "to end") of a `total_bytes` blob: the bao-encoded
/// size of the chunk-group-aligned range (content plus interleaved proof, ADR
/// 038 §Payment metering), exactly the byte count the server emits. The server
/// widens the request to enclosing 16 KiB groups and clamps an end past the
/// blob to the blob's end (a claimed size is a hint); `total_bytes` here is the response
/// header's signed size, so [`align_range_clamped`] /
/// [`AlignedRange::wire_len`](decdn_bao_range::AlignedRange::wire_len)
/// reproduce the same clamp the server applied, keeping the two sides in
/// lock-step. The one site every pull derives its wire bound (and its
/// received-byte ceiling) through. A range with no byte below `total_bytes`
/// faults with [`RangeVerifyError::RangeOutOfBounds`] (see
/// [`is_range_past_end`]). Every input but the offset comes from the peer's
/// signed size or the caller's own cap, so the fault is never marked
/// [`LocalPullFault`].
///
/// [`RangeVerifyError::RangeOutOfBounds`]: decdn_bao_range::RangeVerifyError::RangeOutOfBounds
fn aligned_wire_len(byte_offset: u64, byte_len: u64, total_bytes: u64) -> anyhow::Result<u64> {
    let aligned = align_range_clamped(byte_offset, byte_len, total_bytes)
        .map_err(|e| anyhow::Error::new(e).context("range alignment"))?;
    Ok(aligned.wire_len())
}

/// The wire bound of an open of `[byte_offset, byte_offset + byte_len)`
/// (`byte_len == 0` meaning "to end") against the size `total_bytes` the peer
/// signed ([`aligned_wire_len`]). A range that asks for bytes and starts at or
/// past that size is refused first, with [`ResumeOffsetPastEnd`]: an offset
/// past a resumed blob's end, or a size of zero signed for a bounded open. The
/// open of the empty blob, `(0, 0)` against a size of zero, passes.
fn served_wire_len(byte_offset: u64, byte_len: u64, total_bytes: u64) -> anyhow::Result<u64> {
    if total_bytes <= byte_offset && (byte_offset > 0 || byte_len > 0) {
        return Err(anyhow::Error::new(ResumeOffsetPastEnd {
            total_bytes,
            byte_offset,
        }));
    }
    aligned_wire_len(byte_offset, byte_len, total_bytes)
}

/// How often the throughput floor samples the byte counter: a fraction of the window, so a
/// stall is detected within roughly one extra sample period beyond the window, floored at
/// 100 ms so a tiny window cannot spin the sampler.
pub(crate) fn stall_sample_period(window: Duration) -> Duration {
    (window / 8).max(Duration::from_millis(100))
}

/// How long the receive loop waits for a terminal message after a voucher write
/// fails (below). The node writes `StreamEnd` (or a `StreamError`) *before* the
/// teardown that stops our send, so the terminal signal is already in flight and
/// arrives at once; the bound only stops a peer that stops our send and then goes
/// silent from pinning this recovery read. It is an error-path bound, not a
/// steady-state one.
const TERMINAL_AFTER_WRITE_TIMEOUT: Duration = Duration::from_secs(5);

/// How long one phase of a leg may run before the leg logs it at debug
/// ([`log_proof_stall`]). A node stops a stream whose proof it has waited on
/// for 10 s, so this names the slow phase before that happens.
const PROOF_STALL_DEBUG_AFTER: Duration = Duration::from_secs(5);

/// Log at debug that `phase` of one leg took `elapsed`, once it reaches
/// [`PROOF_STALL_DEBUG_AFTER`]. Each phase names one place a leg can stall:
/// `unpolled` (the caller did not read the stream, so a proof the node waits
/// for was not sent), `pay` (waiting for the lane's ledger, signing and
/// writing the proof), `first_frame` (a fresh leg waited from its open for
/// its first frame, any time before the caller's first read included), or
/// `read` (the leg waited for bytes). A `read` stall
/// with `unproved > 0` can be the node waiting for a proof this leg does not
/// yet owe; with `unproved == 0` the leg owes nothing and the node is slow to
/// send. `unpolled` and `pay` are logged when the phase ends; `first_frame`
/// and `read` are logged while the leg still waits.
fn log_proof_stall(
    pull: (&dyn std::fmt::Display, [u8; 32], u64),
    cumulative: u64,
    unproved: u64,
    phase: &'static str,
    elapsed: Duration,
) {
    if elapsed < PROOF_STALL_DEBUG_AFTER {
        return;
    }
    let (peer, hash, byte_offset) = pull;
    tracing::debug!(
        peer = %peer,
        hash = %blake3::Hash::from_bytes(hash).to_hex(),
        byte_offset,
        cumulative,
        unproved,
        phase,
        elapsed_ms = elapsed.as_millis(),
        "a leg phase ran past {PROOF_STALL_DEBUG_AFTER:?}"
    );
}

/// Header fields from the upstream `StreamResponse`, surfaced by
/// [`open_progressive_pull`] before the first chunk so the fused serve path
/// (#856) knows `total_bytes` up front — it must sign its OWN downstream
/// `StreamResponse` (which commits to a `total_bytes`) before forwarding a byte.
#[derive(Debug, Clone, Copy)]
#[doc(hidden)]
pub struct UpstreamPullHeader {
    /// Whole-blob size the upstream promised (echoed into our downstream
    /// `StreamResponse`).
    pub total_bytes: u64,
    /// Upstream rate; informational for the caller (the buyer pays it inside
    /// [`UpstreamPull::next_chunk`]).
    pub rate_per_mb: u64,
    /// Upstream metering quantum in bytes: the buyer releases one hash-chain
    /// preimage per chunk as they arrive. Always `CHUNK_BYTES` on a paid
    /// upstream — it is a protocol constant, not a negotiated value — and `0`
    /// on an unpaid source, which is the only reason it is carried at all.
    pub interval_bytes: u64,
}

/// A live, progressive `cdn/client/v1` pull (#856) — the ONE receive loop for
/// paid delivery. Opened by [`open_progressive_pull`] (which has already done the
/// handshake and verified the response), driven chunk-by-chunk via
/// [`Self::next_chunk`], and closed by [`Self::finish`] (a wire-completeness
/// check — integrity is verified per bao chunk group by the consumer's decoder
/// as the bytes land, not by a whole-blob re-hash) or [`Self::abort`].
///
/// It yields each chunk to the caller — a [`sink::PullReader`] feeding a
/// verifying decoder into the ranged store, the cache admit, or (`test-util`)
/// memory — instead of buffering the whole blob, and pays the upstream per chunk
/// of wire bytes that decoder has verified ([`Self::mark_verified`]). This is what
/// lets the serving node cap its speculative exposure to a bounded window rather
/// than fronting the entire upstream cost before any downstream voucher arrives,
/// and what keeps a lying peer's bill at the bytes before its first unverifiable
/// chunk group.
///
/// The acked voucher watermark is OWNED here (not threaded as a `&mut`
/// out-param) and read back via [`Self::progress`] /
/// returned by `finish`/`abort` — the caller (`node_origin`) persists it. On any
/// exit, the caller MUST call `progress`/`finish`/`abort` to recover the
/// watermark for `record_progress` (#852); a [`Drop`] guard closes the
/// connection if none ran, but cannot return the watermark, so the obligation
/// stands.
///
/// **Deadlines.** The *open* (handshake) phase is bounded inside
/// [`open_progressive_pull`] by `PullDeadlines::open`, via the shared `open_stream`
/// helper — NOT by the caller (#1134). `decdn-node` does additionally wrap the open
/// in its per-candidate budget, but that is belt-and-braces: a bound left to the
/// caller is a bound a caller can forget.
///
/// The streaming `next_chunk`/`finish` reads are bounded here, by a THROUGHPUT FLOOR
/// (#1797): bytes are counted off the QUIC stream sub-frame, and a read is abandoned when
/// the bytes across the trailing `window` fall below `floor_bps · window`. The detector
/// persists across reads, so its window spans the whole stream rather than one call, and a
/// self-inflicted window-pacing pause is excluded from the window.
///
/// A wall clock would be the wrong bound to reach for here: this type exists to
/// stream blobs of any size, so any fixed deadline would either kill a healthy
/// large transfer or be too loose to catch a dead one. A byte-throughput floor is
/// indifferent to size and link speed, and frame-size-independent.
#[doc(hidden)]
pub struct UpstreamPull {
    conn: iroh::endpoint::Connection,
    /// What this pull owns of its connection: a one-shot dial it closes on its
    /// terminal method, a caller-owned [`WarmConnection`] it borrows, or a
    /// [`Connections`] map's connection it holds and unpins on a transport fault.
    owner: connection::ConnOwner,
    send: SendStream,
    recv: RecvStream,
    ctx: PoolContext,
    /// The lane's voucher ledger, SHARED with every other concurrent pull on this
    /// lane (#1145 review). Not a per-pull one-shot: that made two concurrent pulls
    /// each compute the same next cumulative `amount` independently and collide
    /// (`AmountRegression`).
    ledger: Arc<PoolLedger>,
    hash: [u8; 32],
    /// The requested content range's start, kept for the stall diagnostics.
    byte_offset: u64,
    /// The requested content range's length (`0` = to the end), kept for the
    /// stall diagnostics.
    byte_len: u64,
    rate_per_mb: u64,
    /// This stream's chain anchor — which epoch it has told the upstream about.
    meter: StreamMeter,
    /// Throughput-floor window for every streaming read (#1797). Names the `after` on a
    /// [`PullStalled`] / [`PullTimeout`]; the floor itself owns the window and floor rate.
    window: Duration,
    /// Bytes read off the QUIC stream so far, tallied sub-frame by the `ProgressReader`
    /// each `next_chunk`/`finish` read wraps `recv` in. The floor reads this counter.
    progress_counter: Arc<std::sync::atomic::AtomicU64>,
    /// The byte-progress stall detector, persisted across reads so its window spans the
    /// whole stream, not one `next_chunk` call.
    floor: progress::ThroughputFloor,
    /// The floor's sampling tick.
    sampler: tokio::time::Interval,
    /// Promised **wire** bytes for this stream: the bao-encoded size of the
    /// chunk-group-aligned range (content plus interleaved proof, ADR 038), not
    /// the content-byte remainder. Bounds the receive loop and the closing voucher.
    expected_wire_bytes: u64,
    /// The buyer's received-byte ceiling (#1895) as a WIRE bound: the wire size of a
    /// ceiling-sized blob's content from this stream's offset, so `next_chunk` can
    /// enforce `max_blob_size_bytes` on the bytes that ACTUALLY arrive without
    /// decoding. `0` = unlimited.
    max_received_wire: u64,
    /// Wire bytes received so far on this stream.
    cumulative: u64,
    /// Wire bytes the caller's decoder has verified against the root, never
    /// above `cumulative`. Payment follows this count, not `cumulative`.
    verified: u64,
    /// Verified wire bytes not yet covered by a proof.
    unproved: u64,
    /// `StreamEnd` seen — `next_chunk` returns `None` and `finish` skips the
    /// drain.
    ended: bool,
    /// When `next_chunk` last returned a chunk, so the next call can tell how
    /// long the caller left the stream unread ([`log_proof_stall`]).
    returned_at: Option<tokio::time::Instant>,
    /// When the open returned, so a fresh leg's wait for its first frame is
    /// timed from there ([`log_proof_stall`]).
    opened_at: tokio::time::Instant,
}

impl std::fmt::Debug for UpstreamPull {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("UpstreamPull")
            .field("hash", &blake3::Hash::from_bytes(self.hash))
            .field("expected_wire_bytes", &self.expected_wire_bytes)
            .field("cumulative", &self.cumulative)
            .field("ended", &self.ended)
            .finish_non_exhaustive()
    }
}

/// Open a progressive `cdn/client/v1` pull (#856): connect, send the
/// [`StreamRequest`], read and verify the signed [`StreamResponse`] (so
/// `total_bytes` is known up front), and return its header plus a live
/// [`UpstreamPull`] to drive. Zero-rate rejection, `slash_sig` recovery, echoed
/// field checks, the buyer's rate ceiling, and the `total_bytes >= byte_offset`
/// floor are all enforced BEFORE the first chunk. The `max_blob_size_bytes`
/// ceiling is NOT one of them (#1895): the peer's `total_bytes` claim is
/// unverified, so the ceiling is enforced on the bytes that ACTUALLY arrive,
/// inside [`UpstreamPull::next_chunk`], as [`BlobTooLarge`]. `0` = unlimited.
///
/// # Errors
///
/// Connect / transport faults, a refused or zero-rate response, a bad `slash_sig`,
/// a mismatched echoed field, a [`RateAboveCeiling`] quote, a
/// [`ResumeOffsetPastEnd`] offset, or a resume offset already at/past the size
/// ceiling ([`BlobTooLarge`]). The size ceiling otherwise surfaces later, as a
/// [`BlobTooLarge`] abort once received bytes cross it.
///
/// The open stage is bounded by `deadlines.open`, inside the shared `open_stream`
/// helper — NOT left to the caller (#1134). `node_origin` additionally wraps this
/// call in its per-candidate budget, which is belt-and-braces rather than the sole
/// bound. `deadlines.window` and `deadlines.floor_bps` are the throughput floor the returned
/// [`UpstreamPull`] carries into every streaming read; `deadlines.hard_cap` is not consulted
/// here (the caller owns the streaming lifetime on this path).
///
/// `ledger` is the LANE's voucher ledger, not this pull's: pass the same
/// `Arc<PoolLedger>` to every concurrent pull on one lane, or they will each
/// compute the same next cumulative `amount` independently and collide
/// (`AmountRegression`, #1145 review). The caller reads what to
/// persist from it — including after a drop — via [`PoolLedger::settlement`].
///
/// `dial_runtime` is the runtime the dial and the connection's QUIC driver run
/// on; `None` runs them on the caller's. A caller that drops the runtime it pulls
/// on passes a long-lived one — see
/// [`PeerSource::with_dial_runtime`](crate::source::PeerSource::with_dial_runtime).
///
/// Runs inside an `open_progressive_pull` span that covers the dial and the
/// signed-response handshake — the time to first byte. Its `hash`, `pool_id`,
/// `byte_offset`, `peer` and `local_node_id` fields match the serving node's
/// `serve_stream` span, with `peer` and `local_node_id` swapped.
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(
    name = "open_progressive_pull",
    skip_all,
    fields(
        error = tracing::field::Empty,
        otel.status_code = tracing::field::Empty,
        otel.kind = "client",
        peer = %target.id,
        local_node_id = %endpoint.id(),
        hash = %decdn_protocol::ContentHash::from_bytes(hash),
        pool_id = %ctx.pool_id,
        byte_offset = byte_offset,
        byte_len = byte_len,
    )
)]
#[doc(hidden)]
pub async fn open_progressive_pull(
    endpoint: &Endpoint,
    target: EndpointAddr,
    ctx: &PoolContext,
    ledger: Arc<PoolLedger>,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
    deadlines: PullDeadlines,
    byte_len: u64,
    dial_runtime: Option<&tokio::runtime::Handle>,
) -> anyhow::Result<(UpstreamPullHeader, UpstreamPull)> {
    open_progressive_pull_impl(
        ConnSource::Dial {
            endpoint,
            target,
            runtime: dial_runtime,
        },
        // One-shot dial: the pull owns this connection and closes it on teardown.
        connection::ConnOwner::Owned,
        ctx,
        ledger,
        slash_domain,
        expected_signer,
        hash,
        namespace_id,
        byte_offset,
        timestamp_us,
        max_blob_size_bytes,
        max_rate_per_mb,
        deadlines,
        byte_len,
    )
    .await
}

/// Like [`open_progressive_pull`], but opens the pull on the warm connection
/// `conn` instead of dialling a fresh one (#1848). `owner` names where `conn`
/// comes from: a caller-owned [`WarmConnection`] or a [`Connections`] map's
/// connection. The warm connection is reused across
/// many hashes (one dial, a fresh bi-stream per hash: one stream = one hash, no
/// wire change) and stays open when this pull ends, so the pull never closes
/// it. A map's connection is held by the pull while it runs, and a transport
/// fault on the open or on a later read unpins it from the map
/// ([`Connections::invalidate`]).
///
/// Every other argument behaves exactly as on [`open_progressive_pull`]; see its
/// docs. There is no `endpoint`/`target` pair — the warm connection already names
/// its peer.
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(
    name = "open_progressive_pull",
    skip_all,
    fields(
        error = tracing::field::Empty,
        otel.status_code = tracing::field::Empty,
        otel.kind = "client",
        peer = %conn.remote_id(),
        hash = %decdn_protocol::ContentHash::from_bytes(hash),
        pool_id = %ctx.pool_id,
        byte_offset = byte_offset,
        byte_len = byte_len,
    )
)]
pub(crate) async fn open_progressive_pull_on(
    conn: &iroh::endpoint::Connection,
    owner: connection::ConnOwner,
    ctx: &PoolContext,
    ledger: Arc<PoolLedger>,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
    deadlines: PullDeadlines,
    byte_len: u64,
) -> anyhow::Result<(UpstreamPullHeader, UpstreamPull)> {
    let opened = open_progressive_pull_impl(
        ConnSource::Reuse(conn),
        // A warm connection: leave it open for the next hash.
        owner.clone(),
        ctx,
        ledger,
        slash_domain,
        expected_signer,
        hash,
        namespace_id,
        byte_offset,
        timestamp_us,
        max_blob_size_bytes,
        max_rate_per_mb,
        deadlines,
        byte_len,
    )
    .await;
    if let Err(err) = &opened {
        owner.unpin_on_fault(err).await;
    }
    opened
}

/// The shared body behind [`open_progressive_pull`] (dial) and
/// [`open_progressive_pull_on`] (reuse). `owner` records what the pull owns of
/// its connection, so the returned [`UpstreamPull`] knows whether its terminal
/// method closes the connection (owned, one-shot) or leaves it open for the
/// next hash (warm), and whether a read fault unpins it from a map.
#[allow(clippy::too_many_arguments)]
async fn open_progressive_pull_impl(
    source: ConnSource<'_>,
    owner: connection::ConnOwner,
    ctx: &PoolContext,
    ledger: Arc<PoolLedger>,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    namespace_id: [u8; 32],
    byte_offset: u64,
    timestamp_us: u64,
    max_blob_size_bytes: u64,
    max_rate_per_mb: u64,
    deadlines: PullDeadlines,
    // Upper bound on the requested range: `[byte_offset, byte_offset + byte_len)`.
    // `0` means "to end" (the whole tail, which every non-gap caller requests).
    // A gap-driven caller (`source::PeerSource`, #1608) passes the exact gap
    // length so the server scopes both the serve and the payment to it, rather
    // than streaming the whole remainder.
    byte_len: u64,
) -> anyhow::Result<(UpstreamPullHeader, UpstreamPull)> {
    let result: anyhow::Result<(UpstreamPullHeader, UpstreamPull)> = async {
        let window = deadlines.window;
        let floor_bps = deadlines.floor_bps;
        let (conn, send, recv, resp, resp_ext) = open_stream(
            source,
            ctx,
            slash_domain,
            expected_signer,
            hash,
            // Node-to-node pull. `NO_NAMESPACE` (0) for a DHT-discovered *holder* — it
            // already holds the bytes, so the downstream node needs no namespace (ADR
            // 002 §Retrieval by namespace). But a directory-discovered *cold origin* is
            // reached with the served request's namespace: it must fill from its own
            // backend, and its pull-through authorized-origin gate resolves on that
            // namespace (ADR 005 §Namespace routing). The caller passes whichever
            // applies.
            namespace_id,
            byte_offset,
            byte_len,
            timestamp_us,
            deadlines.open,
        )
        .await?;
        if !resp.body.ok {
            return Err(UpstreamRefused::open(resp, &resp_ext));
        }
        // The peer's signed `total_bytes` never drives a refusal (#1895): it is
        // peer-controlled and unverified (`StreamResponse::validate()` does not bound
        // it), so the `max_blob_size_bytes` ceiling is enforced on the bytes that
        // ACTUALLY arrive, inside `UpstreamPull::next_chunk`, not on the claim. Refusing
        // on an inflated claim would let a holder centralise a small blob's traffic
        // across every finite-ceiling relay; a lie cannot produce bytes that verify
        // against the true root, and an honest giant is streamed and paid for only up
        // to one ceiling.
        // Buyer-side rate ceiling (#1375): refuse an over-ceiling quote before the
        // first paid interval, carrying the signed quote out as rate-manipulation
        // evidence. `0` = unbounded.
        if max_rate_per_mb > 0 && resp.body.rate_per_mb > max_rate_per_mb {
            return Err(RateAboveCeiling::over_ceiling(resp, max_rate_per_mb));
        }
        let rate_per_mb = resp.body.rate_per_mb;
        // Wire-byte bound (bao-encoded size of the aligned range), not content bytes:
        // the window path forwards this stream verbatim and pays the upstream in wire
        // bytes (ADR 038 §Payment metering).
        let total_bytes = resp.body.total_bytes;
        let expected_wire_bytes = served_wire_len(byte_offset, byte_len, total_bytes)?;
        // Received-byte ceiling (#1895), expressed as a WIRE bound so `next_chunk` can
        // enforce it without decoding: the wire size of a ceiling-sized blob's content
        // from this offset. Enforced on the bytes that ACTUALLY arrive, never on the
        // peer's unverified `total_bytes` claim. A resume offset already at/past the
        // ceiling means the blob is genuinely oversized — abort before the first chunk.
        // `0` = unlimited.
        let max_received_wire = if max_blob_size_bytes == 0 {
            0
        } else if byte_offset >= max_blob_size_bytes {
            return Err(anyhow::Error::new(BlobTooLarge {
                reached: byte_offset,
                ceiling: max_blob_size_bytes,
            }));
        } else {
            aligned_wire_len(byte_offset, 0, max_blob_size_bytes)?
        };
        let header = UpstreamPullHeader {
            total_bytes,
            rate_per_mb,
            interval_bytes: CHUNK_BYTES,
        };
        let progress_counter = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let floor = progress::ThroughputFloor::new(
            progress::FloorConfig { window, floor_bps },
            Arc::clone(&progress_counter),
            tokio::time::Instant::now(),
        );
        let mut sampler = tokio::time::interval(stall_sample_period(window));
        sampler.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        let pull = UpstreamPull {
            window,
            progress_counter,
            floor,
            sampler,
            conn,
            owner,
            send,
            recv,
            ctx: ctx.clone(),
            ledger,
            hash,
            byte_offset,
            byte_len,
            rate_per_mb,
            meter: StreamMeter::default(),
            expected_wire_bytes,
            max_received_wire,
            cumulative: 0,
            verified: 0,
            unproved: 0,
            ended: false,
            returned_at: None,
            opened_at: tokio::time::Instant::now(),
        };
        Ok((header, pull))
    }
    .await;
    record_open_result(&tracing::Span::current(), &result);
    result
}

impl UpstreamPull {
    /// Wire bytes this stream will deliver: the bao-encoded size of the
    /// chunk-group-aligned range (content plus interleaved proof, ADR 038). The
    /// window serve loop uses it as the pull budget `total` — it must be wire
    /// bytes, since `pulled`/`served_paid` count forwarded wire bytes.
    #[must_use]
    pub const fn expected_wire_bytes(&self) -> u64 {
        self.expected_wire_bytes
    }

    /// The watermark to PERSIST on any exit — including an error from `next_chunk`, and
    /// including a DROP (#852, #1122).
    ///
    /// Reads [`PoolLedger::settlement`], not a copied-back field. A field can only be
    /// updated on a path that RUNS, and this pull's does not always run: the serve loop
    /// drives it as a future on the `accept` task, which is dropped on shutdown or a
    /// downstream reset, and `settlement` additionally covers a voucher left in flight when
    /// that happened. See the ledger's docs.
    #[must_use]
    pub fn progress(&self) -> VoucherProgress {
        VoucherProgress::from_ledger(&self.ledger, self.ctx.prior_amount)
    }

    /// Record that the caller's decoder has verified the first `wire` bytes of
    /// this stream against the root. The count only moves forward and never
    /// passes the bytes received. The newly verified bytes become payable: the
    /// next [`Self::next_chunk`], [`Self::finish`] or [`Self::stop`] pays for
    /// them, and a byte that never verifies is never paid for.
    pub(crate) fn mark_verified(&mut self, wire: u64) {
        let wire = wire.min(self.cumulative);
        if wire > self.verified {
            self.unproved = self.unproved.saturating_add(wire - self.verified);
            self.verified = wire;
        }
    }

    /// Pay for the verified bytes no proof covers yet: one reveal per whole
    /// chunk, plus one closing signature for the residual once every promised
    /// byte has verified. The sends commit optimistically; only a rejection
    /// comes back, on a later read.
    ///
    /// A write failure may only mean the node stopped our send after it
    /// finished, or is rejecting us, so the terminal signal it left wins over
    /// the opaque write failure ([`Self::terminal_after_write_failure`]).
    async fn pay_verified(&mut self) -> anyhow::Result<()> {
        let complete = self.verified >= self.expected_wire_bytes;
        if self.unproved < CHUNK_BYTES && !(complete && self.unproved > 0) {
            return Ok(());
        }
        let unproved = self.unproved;
        // Gate the floor across our own payment: while we owe the covering proof
        // the upstream legitimately pauses (ADR 005 §Payment pacing), so that
        // pause is self-inflicted, not a sender stall (#1797).
        let pay_started = tokio::time::Instant::now();
        self.floor.pause(pay_started);
        let paid = self.pay_one(unproved, complete).await;
        let paid_at = tokio::time::Instant::now();
        self.floor.resume(paid_at);
        log_proof_stall(
            (&self.conn.remote_id(), self.hash, self.byte_offset),
            self.cumulative,
            unproved,
            "pay",
            paid_at.saturating_duration_since(pay_started),
        );
        match paid {
            Ok(remaining) => self.unproved = remaining,
            // Boxed so this cold error-path future does not enlarge the steady
            // receive loop's future (`clippy::large_futures`); the allocation
            // only happens on the failure path. A confirmed voucher covers every
            // byte this stream verified, so nothing stays unproved.
            Err(write_err) => {
                if Box::pin(self.terminal_after_write_failure(write_err)).await? {
                    self.unproved = 0;
                }
            }
        }
        Ok(())
    }

    /// Send one voucher for `delta_bytes` newly verified since the last voucher,
    /// through the LANE's ledger — shared with every other concurrent pull on it, so
    /// their vouchers are serialized into strict cumulative-amount order rather than
    /// colliding. Sends optimistically (#1484): the ack is
    /// read back by [`Self::next_chunk`] / [`Self::finish`], not awaited here.
    async fn pay_one(&mut self, unproved: u64, complete: bool) -> anyhow::Result<u64> {
        let ledger = Arc::clone(&self.ledger);
        self.meter
            .pay(
                &mut self.send,
                &self.ctx,
                &ledger,
                self.rate_per_mb,
                unproved,
                complete,
            )
            .await
    }

    /// Read one message under the persistent throughput floor (#1797). Wraps `recv` in a
    /// `ProgressReader` feeding the shared counter, and races the read against
    /// the floor's sampling tick: a read slow enough to lose the race is judged, and a
    /// sub-floor window aborts with [`PullTimeout`] before the first byte (our own
    /// blob-size-dependent budget) or [`PullStalled`] after it — neither scores the peer.
    ///
    /// A transport fault on a [`Connections`] map's connection (either floor verdict,
    /// or a connection that has closed under the read) unpins that connection from
    /// the map, so the node's next leg dials again.
    async fn read_under_floor(&mut self) -> anyhow::Result<ClientMessage> {
        let read = self.read_under_floor_once().await;
        if let Err(err) = &read {
            self.owner.unpin_on_fault(err).await;
        }
        read
    }

    /// [`Self::read_under_floor`] before a transport fault unpins the connection.
    async fn read_under_floor_once(&mut self) -> anyhow::Result<ClientMessage> {
        let cumulative = self.cumulative;
        let window = self.window;
        let recv = &mut self.recv;
        let sampler = &mut self.sampler;
        let floor = &mut self.floor;
        let (peer, hash, byte_offset) = (self.conn.remote_id(), self.hash, self.byte_offset);
        let unproved = self.unproved;
        // A fresh leg's wait runs from its open, any time before the caller's
        // first read included, so a first frame that never comes is named
        // once, as `first_frame` (#2230).
        let (phase, read_started) = if cumulative == 0 {
            ("first_frame", self.opened_at)
        } else {
            ("read", tokio::time::Instant::now())
        };
        // The stall line runs on its own timer, not on the sampler, whose
        // period follows the configurable window and can be later than the
        // node's proof timeout.
        let read_stall_at = read_started + PROOF_STALL_DEBUG_AFTER;
        let mut read_stall_logged = false;
        let mut reader = progress::ProgressReader::new(recv, Arc::clone(&self.progress_counter));
        // Pin ONE read future and poll it across ticks. `read_client_message` is not
        // cancellation-safe — `read_frame` fills a frame with `read_exact`, so dropping the
        // future mid-frame loses the bytes already consumed and desynchronises the stream.
        // Recreating it per tick would corrupt every frame that spans a tick; the pinned
        // future is dropped only when the floor actually aborts (#1797). `biased` cannot
        // starve the floor: a ready read arm means a whole frame arrived, which is byte
        // progress the counter already holds, so a skipped tick only ever coincides with
        // throughput the floor would clear.
        let read = read_client_message(&mut reader);
        tokio::pin!(read);
        loop {
            tokio::select! {
                biased;
                r = &mut read => return r,
                _ = sampler.tick() => {
                    let now = tokio::time::Instant::now();
                    let last_sample = floor.last_sampled_at();
                    if let progress::FloorVerdict::Stalled = floor.evaluate(now) {
                        // What tells a quiet peer from a reader that was not
                        // polled (#2211): the gap since the previous sample.
                        tracing::debug!(
                            peer = %self.conn.remote_id(),
                            hash = %blake3::Hash::from_bytes(self.hash).to_hex(),
                            byte_offset = self.byte_offset,
                            byte_len = self.byte_len,
                            cumulative,
                            expected_wire_bytes = self.expected_wire_bytes,
                            window_bytes = floor.window_bytes(),
                            unpolled_ms = last_sample
                                .map(|t| now.saturating_duration_since(t).as_millis()),
                            window_ms = window.as_millis(),
                            "upstream pull fell under its throughput floor"
                        );
                        return Err(if cumulative == 0 {
                            anyhow::Error::new(PullTimeout { after: window })
                        } else {
                            anyhow::Error::new(PullStalled { after: window })
                        });
                    }
                }
                // A long wait here is the leg waiting on the node, while the
                // node may wait on a proof or still be sending its first frame.
                () = tokio::time::sleep_until(read_stall_at), if !read_stall_logged => {
                    read_stall_logged = true;
                    let waited = read_started.elapsed();
                    let pull = (&peer as &dyn std::fmt::Display, hash, byte_offset);
                    log_proof_stall(pull, cumulative, unproved, phase, waited);
                }
            }
        }
    }

    /// Pay for the bytes the caller's decoder has verified since the last proof
    /// ([`Self::mark_verified`]), then read the next `ChunkData` and return it
    /// for that decoder (and, on the node's serve leg, to forward downstream).
    /// The payment goes first because the node may be waiting on it: a read
    /// before it could wait on bytes the node holds back until it is paid.
    /// Returns `Ok(None)` on `StreamEnd`.
    ///
    /// # Errors
    ///
    /// More bytes than promised, received bytes crossing the buyer's size ceiling
    /// ([`BlobTooLarge`], raised before the crossing chunk is paid), a mid-stream `StreamError`
    /// (typed [`UpstreamRefused`]), an unexpected message, a [`UpstreamVoucherRejected`] /
    /// transport error while paying, or —
    /// on the throughput floor — [`PullStalled`] once bytes have flowed, or [`PullTimeout`] if
    /// the floor trips before the first byte (`cumulative == 0`). A malformed `ChunkData` is
    /// not raised here: an empty payload is rejected at decode by `ChunkData`'s
    /// `serde(try_from)` (#1088) and an oversized frame by the framing layer's
    /// `MAX_MESSAGE_SIZE` before it allocates, so both surface out of `read_client_message`.
    pub async fn next_chunk(&mut self) -> anyhow::Result<Option<Bytes>> {
        if self.ended {
            return Ok(None);
        }
        if let Some(returned_at) = self.returned_at.take() {
            log_proof_stall(
                (&self.conn.remote_id(), self.hash, self.byte_offset),
                self.cumulative,
                self.unproved,
                "unpolled",
                returned_at.elapsed(),
            );
        }
        self.pay_verified().await?;
        if self.ended {
            return Ok(None);
        }
        // Read exactly one message. In the pool model there is no positive ack to
        // consume (acceptance is implicit — continued delivery IS acceptance, ADR
        // 005), so every message either carries a chunk, ends the stream, or is a
        // mid-stream error/unexpected frame; none is skipped.
        {
            // Byte-progress bound (#1797): the read is judged by the persistent throughput
            // floor, which counts bytes off the QUIC stream sub-frame. A `ChunkData` carries
            // at least one byte (#1088 bans empty frames), so every frame is a unit of
            // progress. The floor's window spans the whole stream, not this call, and the
            // caller's downstream-forward time is not charged against it — the counter only
            // moves on bytes the reader pulls.
            let msg = self.read_under_floor().await?;
            match msg {
                ClientMessage::ChunkData(chunk) => {
                    // The payload is bounded on both sides by construction (#1088): the
                    // ceiling caps per-frame allocation, and the non-empty floor ties every
                    // frame to payload, so a peer cannot advance this loop with a run of
                    // empty frames that move neither `cumulative` nor the voucher
                    // accounting. This path has no belt-and-braces payload check behind
                    // that floor, and does not need one now the floor is structural.
                    let chunk_len = chunk.bytes().len() as u64;
                    self.cumulative = self.cumulative.saturating_add(chunk_len);
                    if self.cumulative > self.expected_wire_bytes {
                        anyhow::bail!(
                            "server sent {} bytes, more than the {} promised",
                            self.cumulative,
                            self.expected_wire_bytes
                        );
                    }
                    // Received-byte ceiling (#1895): enforce the size cap on the bytes
                    // that ACTUALLY arrive, never on the peer's unverified `total_bytes`
                    // claim. `max_received_wire` is the wire size of a ceiling-sized
                    // blob's content, so this is the wire form of "received content >
                    // ceiling" — distinct from the claim-derived `expected_wire_bytes`
                    // overrun guard above. Abort BEFORE paying for the chunk that
                    // crosses it, so the spend stays bounded to roughly one ceiling.
                    // `0` = unlimited.
                    if self.max_received_wire > 0 && self.cumulative > self.max_received_wire {
                        return Err(anyhow::Error::new(BlobTooLarge {
                            reached: self.cumulative,
                            ceiling: self.max_received_wire,
                        }));
                    }
                    // No per-chunk hashing here: this stream's bytes are bao wire
                    // bytes the caller feeds to a verifying decoder — the cache's
                    // (`admit_bao_stream`), the ranged store's, or the in-memory
                    // test wrapper's — which checks every chunk group against the
                    // root as it lands (ADR 038) and reports what it verified
                    // through `mark_verified`. Only verified bytes are paid for, so
                    // a peer that sends a bad group bills nothing for it.
                    self.returned_at = Some(tokio::time::Instant::now());
                    Ok(Some(Bytes::from(chunk.into_bytes())))
                }
                // A mid-stream `StreamError` is either a `VoucherRejected` (our
                // payment fault, with the self-heal bundle) or a refusal; both
                // surface as errors.
                ClientMessage::StreamError(e) => {
                    Err(voucher_rejection(&self.ledger, &self.meter, e))
                }
                ClientMessage::StreamEnd => {
                    self.ended = true;
                    Ok(None)
                }
                other => {
                    anyhow::bail!("unexpected message mid-delivery: {}", variant_name(&other))
                }
            }
        }
    }

    /// Decide what a voucher write failure really means.
    ///
    /// A node stops our send half only once it needs nothing more from us: a
    /// completed delivery leaves a `StreamEnd` on the wire, a mid-stream rejection a
    /// `StreamError`. The write half and the read half run in lock-step in the
    /// receive loop, so a raw voucher-write failure — the end-of-stream
    /// `STOP_SENDING(0)` a node emits when it finishes and drops `recv` — would
    /// otherwise mask that terminal signal and abort a complete, paid fetch (or
    /// swallow a typed rejection the fault classification keys on). Read the terminal
    /// signal, briefly, and prefer it.
    ///
    /// A `StreamEnd` marks the stream ended, and the next read returns `None`. When the failed write was
    /// a voucher, the `StreamEnd` also confirms it. An honest node ends a stream
    /// only once every interval, the closing one included, is credited. So
    /// payment-based completion ([`driver::drive`]) sees this leg paid through its
    /// end and does not re-pull and re-bill the tail. Confirming costs nothing
    /// extra, because [`PoolLedger::settlement`] already reports the armed voucher.
    ///
    /// A `StreamError` surfaces as the typed rejection. The node delivers ahead of
    /// payment within its credit window, so `ChunkData` it sent before it rejected
    /// can sit in front of the `StreamError`: the read passes over those frames,
    /// unread, to reach the terminal message. The rejected leg pays for none of
    /// them, and a caller that resumes pulls them again. A `StreamEnd` behind
    /// skipped `ChunkData` confirms nothing, because this leg did not take in those
    /// bytes. A write we caused ourselves (a [`LocalPullFault`]) is never masked,
    /// nor is a stream that yields no terminal signal before the bound.
    ///
    /// Returns whether a voucher was confirmed. A `StreamEnd` after a failed reveal
    /// or re-anchor write, or after a sibling's voucher replaced this stream's
    /// armed one, ends the stream with nothing confirmed.
    async fn terminal_after_write_failure(
        &mut self,
        write_err: anyhow::Error,
    ) -> anyhow::Result<bool> {
        if write_err.is::<LocalPullFault>() {
            return Err(write_err);
        }
        // Any outcome other than a terminal message means nothing terminal was
        // waiting, so the write failure stands as the honest outcome (its typed
        // cause still downcasts through the added context); what the recovery saw
        // rides along so the log can tell a silent peer from a chatty one.
        // The frames it passes over stay under the same ceilings as the ones it
        // takes in (`next_chunk`): an honest node sends no byte past the
        // request's promised wire length, and our own received-byte ceiling
        // stops the pull as it would on any read.
        let mut skipped: u64 = 0;
        let terminal = tokio::time::timeout(TERMINAL_AFTER_WRITE_TIMEOUT, async {
            loop {
                match self.read_under_floor().await {
                    Ok(ClientMessage::ChunkData(chunk)) => {
                        skipped = skipped.saturating_add(chunk.bytes().len() as u64);
                        let seen = self.cumulative.saturating_add(skipped);
                        if seen > self.expected_wire_bytes {
                            anyhow::bail!(
                                "server sent {seen} bytes, more than the {} promised",
                                self.expected_wire_bytes
                            );
                        }
                        if self.max_received_wire > 0 && seen > self.max_received_wire {
                            return Err(anyhow::Error::new(BlobTooLarge {
                                reached: seen,
                                ceiling: self.max_received_wire,
                            }));
                        }
                    }
                    other => return other,
                }
            }
        })
        .await;
        match terminal {
            Ok(Ok(ClientMessage::StreamEnd)) if skipped > 0 => Err(write_err.context(format!(
                "the peer ended the stream after {skipped} bytes of ChunkData this leg did not \
                 take in once the voucher write failed"
            ))),
            Ok(Ok(ClientMessage::StreamEnd)) => {
                self.ended = true;
                let attempted = write_err
                    .downcast_ref::<UnconfirmedVoucher>()
                    .map(|voucher| voucher.amount);
                let confirmed = match attempted {
                    Some(amount) => self.ledger.confirm_armed_stamped(amount).await,
                    None => None,
                };
                if let Some((confirmed, generation)) = confirmed {
                    tracing::debug!(
                        amount = %confirmed.amount,
                        error = %format_args!("{write_err:#}"),
                        "upstream ended the stream after a voucher write failed; confirmed \
                         the voucher"
                    );
                    self.meter.last_proof = Some(StreamProof::Voucher {
                        amount: confirmed.amount,
                        generation,
                    });
                    Ok(true)
                } else {
                    tracing::warn!(
                        attempted = ?attempted,
                        error = %format_args!("{write_err:#}"),
                        "upstream ended the stream after a proof write failed; nothing \
                         confirmed, so the leg's tail stays unpaid"
                    );
                    Ok(false)
                }
            }
            Ok(Ok(ClientMessage::StreamError(e))) => {
                Err(voucher_rejection(&self.ledger, &self.meter, e))
            }
            Ok(Ok(other)) => Err(write_err.context(format!(
                "no terminal signal after the voucher write failed; peer sent {}",
                variant_name(&other)
            ))),
            // The ceiling stays typed: it is terminal, not a peer to fail over from.
            Ok(Err(read_err)) if read_err.is::<BlobTooLarge>() => Err(read_err),
            Ok(Err(read_err)) => Err(write_err.context(format!(
                "no terminal signal after the voucher write failed; recovery read failed: \
                 {read_err:#}"
            ))),
            Err(_) => Err(write_err.context(format!(
                "no terminal signal within {TERMINAL_AFTER_WRITE_TIMEOUT:?} of the voucher \
                 write failing"
            ))),
        }
    }

    /// Finalize a completed pull: pay for the verified bytes no proof covers yet
    /// (the closing voucher, which the node waits for before its `StreamEnd`),
    /// drain to `StreamEnd` if needed, enforce wire-byte completeness (the full
    /// promised bao wire size was received), close the connection cleanly, and
    /// return the final acked watermark to persist. Per ADR 038 this does not
    /// re-hash the whole blob — bao verification is the caller's decoder's job,
    /// group by group as the bytes land.
    ///
    /// # Errors
    ///
    /// A short delivery (fewer wire bytes than promised before `StreamEnd`), a stream/protocol
    /// error while paying or draining, or — on the same inactivity bound as `next_chunk` —
    /// [`PullStalled`] if the upstream goes silent before `StreamEnd`.
    pub async fn finish(mut self) -> anyhow::Result<VoucherProgress> {
        if !self.ended {
            self.pay_verified().await?;
        }
        while !self.ended {
            // Same byte-progress bound as `next_chunk` (#1797): an upstream that never sends
            // its `StreamEnd` must not hold the drain open forever.
            let msg = self.read_under_floor().await?;
            match msg {
                ClientMessage::StreamEnd => self.ended = true,
                ClientMessage::ChunkData(_) => {
                    anyhow::bail!("server sent ChunkData after the promised total")
                }
                // The closing voucher was committed optimistically when it was sent
                // (implicit acceptance, ADR 005), so the drain only needs to watch
                // for a late `VoucherRejected` / other `StreamError`, which still
                // surfaces as an error.
                ClientMessage::StreamError(e) => {
                    return Err(voucher_rejection(&self.ledger, &self.meter, e));
                }
                other => {
                    anyhow::bail!("unexpected message at stream end: {}", variant_name(&other))
                }
            }
        }
        // Completeness for every fetch (full and resumed): bao verification is
        // the caller's decoder's job, group by group as the bytes land, so
        // `finish` does not re-hash the whole blob. A truncated stream (fewer
        // wire bytes than promised) can't be decoded, so require the full
        // promised wire size as the completeness signal.
        if self.cumulative < self.expected_wire_bytes {
            self.close_transport(0, b"short-delivery");
            anyhow::bail!(
                "server sent {} of {} promised wire bytes before StreamEnd",
                self.cumulative,
                self.expected_wire_bytes
            );
        }
        self.close_transport(0, b"done");
        Ok(self.progress())
    }

    /// Abandon the pull (e.g. the downstream client dropped, so we stop pulling
    /// and paying). Closes the connection and returns the acked watermark so the
    /// caller can still persist what it paid (#852).
    #[must_use]
    pub fn abort(mut self) -> VoucherProgress {
        self.close_transport(0, b"client-abandoned");
        self.progress()
    }

    /// Stop the pull before its promised end, after paying for every wire byte
    /// the caller's decoder verified. A steal that takes the tail of this pull's
    /// range stops the pull at the split on its open stream, so the stolen tail
    /// goes to one source and this one opens no new stream for the part it
    /// keeps. Frames past the split that the stream already carried are not
    /// verified, so they are not paid for, and the stolen tail is paid once.
    ///
    /// A rejection the node already sent waits behind the `ChunkData` it sent
    /// first, so the stop first reads what has arrived, without waiting: a
    /// `VoucherRejected` there rewinds the rejected proof and surfaces as the
    /// typed rejection, as it does at [`Self::finish`]. A rejection still in
    /// flight reaches the lane's next leg instead. The closing voucher then
    /// settles the verified residual, the send half is finished, and the
    /// transport is torn down (this stream, or the whole connection when the
    /// pull owns it), so the node sees its sending stopped. A node that stops
    /// sending before it reads that voucher does not redeem it, and the lane's
    /// next cumulative voucher covers the same bytes.
    ///
    /// A closing voucher that fails to send stays armed in the ledger:
    /// [`PoolLedger::settlement`] counts it, and the lane's next voucher builds
    /// on it. The caller's paid frontier, which reads
    /// [`PoolLedger::committed`], stays behind the verified bytes, so its next
    /// leg re-delivers and pays for them once more.
    ///
    /// # Errors
    ///
    /// The typed rejection of a `StreamError` that had already arrived, or a
    /// closing voucher that fails on our side ([`LocalPullFault`]). Any other
    /// send failure is logged and leaves the residual as above.
    pub async fn stop(mut self) -> anyhow::Result<VoucherProgress> {
        if let Some(rejection) = self.arrived_rejection() {
            self.close_transport(0, b"stopped-at-split");
            return Err(rejection);
        }
        let paid = self.pay_one(self.unproved, true).await;
        let _ = self.send.finish();
        self.close_transport(0, b"stopped-at-split");
        match paid {
            Ok(remaining) => self.unproved = remaining,
            Err(err) if err.is::<LocalPullFault>() => return Err(err),
            Err(err) => tracing::warn!(
                peer = %self.conn.remote_id(),
                cumulative = self.cumulative,
                unproved = self.unproved,
                error = %format_args!("{err:#}"),
                "the closing voucher of a pull stopped at a split did not send"
            ),
        }
        Ok(self.progress())
    }

    /// The typed rejection of a `StreamError` that has already arrived, read
    /// without waiting past the `ChunkData` in front of it. Every frame it
    /// takes in counts as received, as in [`Self::next_chunk`]; no decoder
    /// verifies it, so no voucher pays for it. A frame past the request's
    /// promised wire length or past our received-byte ceiling ends the read:
    /// an honest node sends neither. `None` once nothing more has arrived, on
    /// any other message, or on a read fault: the caller closes the stream
    /// either way.
    fn arrived_rejection(&mut self) -> Option<anyhow::Error> {
        loop {
            match futures_util::FutureExt::now_or_never(self.read_under_floor_once())? {
                Ok(ClientMessage::ChunkData(chunk)) => {
                    let seen = self.cumulative.saturating_add(chunk.bytes().len() as u64);
                    if seen > self.expected_wire_bytes
                        || (self.max_received_wire > 0 && seen > self.max_received_wire)
                    {
                        return None;
                    }
                    self.cumulative = seen;
                }
                Ok(ClientMessage::StreamError(e)) => {
                    return Some(voucher_rejection(&self.ledger, &self.meter, e));
                }
                _ => return None,
            }
        }
    }

    /// Tear down this pull's transport on any exit.
    ///
    /// An OWNED connection (a one-shot dial) is closed, which ends the QUIC
    /// connection so the upstream's serve task stops and the paid stream does not
    /// linger half-open. A WARM connection (a [`WarmConnection`] reused across
    /// hashes, caller-owned or from a [`Connections`] map) is left open for the
    /// next hash, and only THIS stream is torn down: the send half is finished (a
    /// clean FIN) and the recv half is stopped. The warm connection's own `Drop`
    /// closes the connection once, later.
    ///
    /// `Connection::close`, `SendStream::finish`, and `RecvStream::stop` are all
    /// first-wins / idempotent, so an explicit terminal method (`finish`/`abort`)
    /// keeps its richer reason and the `Drop` safety net becomes a no-op.
    fn close_transport(&mut self, code: u32, reason: &[u8]) {
        if self.owner.owns() {
            self.conn.close(code.into(), reason);
        } else {
            let _ = self.send.finish();
            let _ = self.recv.stop(code.into());
        }
    }
}

impl Drop for UpstreamPull {
    /// Safety net for the "call a terminal method on every exit" contract: if a
    /// caller returns or panics without `finish`/`abort`, still tear this pull's
    /// transport down through `close_transport` so the QUIC stream and the
    /// upstream's server-side serve task don't linger and keep that paid stream
    /// half-open. `Connection::close` / `SendStream::finish` / `RecvStream::stop`
    /// are all first-wins and idempotent, so an explicit teardown in
    /// `finish`/`abort` keeps its richer reason and this is a no-op when one of
    /// them ran; it only takes effect on a dropped-without-finalize path.
    ///
    /// It tears down the transport and nothing else, and that is sufficient: the
    /// acked watermark lives in the lane's [`PoolLedger`], which OUTLIVES the
    /// pull (it is shared with the other pulls on the lane), not in a field of
    /// this struct. A dropped pull's caller reads it with [`PoolLedger::settlement`]
    /// and persists it — `node_origin` does exactly that from its own `Drop` guard
    /// (#1145 review).
    fn drop(&mut self) {
        self.close_transport(0, b"upstream-pull-dropped");
    }
}

/// Issue one cumulative voucher for `delta_bytes` newly delivered since the last
/// voucher and SEND it. Acceptance is implicit (ADR 005): continued delivery IS
/// acceptance, so there is no ack to wait for — the send itself commits.
///
/// Voucher issuance runs through the lane's [`PoolLedger`], which serializes the
/// compute → sign → send critical section across every concurrent stream drawing
/// on the lane — so vouchers reach the node in strict cumulative order — and
/// releases its issuance lock the instant the send returns. Parallel range
/// streams to one provider do not serialize their payments behind each other's
/// round trips, because there is no round trip: only a rejection comes back, and
/// it arrives as its own mid-stream `StreamError`.
///
/// Each voucher's own *delta* (`ceil(delta_bytes * rate / 1 MiB)`) covers its own
/// bytes at the advertised rate (the node checks deltas, not the rounded
/// cumulative). A successful send advances the committed watermark optimistically;
/// an ambiguous send failure leaves the voucher armed so [`PoolLedger::settlement`]
/// still reports it (settle high — the upstream persists a voucher before it would
/// reject it, ADR 003). The client never pays ahead of what it received (vouchers
/// are cumulative over delivered bytes).
///
/// Returns the sent voucher as the [`StreamProof`] the stream records, stamped
/// with the ledger generation it was signed under, and the chain root the
/// voucher anchors this stream to (`None` for a sealed voucher, which clears the
/// node's anchor). The root is the one the voucher signed, read under the
/// issuance lock: a read of the ledger after the send can already name a
/// sibling's newer chain.
async fn send_voucher(
    send: &mut SendStream,
    ctx: &PoolContext,
    ledger: &PoolLedger,
    rate_per_mb: u64,
    delta_bytes: u64,
    epoch: EpochAction,
) -> anyhow::Result<(StreamProof, Option<B256>)> {
    if ctx.provider.is_zero() {
        return Err(anyhow::anyhow!(
            "voucher provider is not pinned (Address::ZERO) — call PoolContext::with_provider \
             before signing"
        )
        .context(LocalPullFault));
    }
    let mut attempted = None;
    let mut signed_root = B256::ZERO;
    let issued = ledger
        .issue_stamped(delta_bytes, rate_per_mb, epoch, |next, chain| {
            attempted = Some(next.amount);
            signed_root = chain.chain_root;
            sign_and_write_voucher(send, ctx, next, chain)
        })
        .await;
    issued
        .map(|(next, generation)| {
            (
                StreamProof::Voucher {
                    amount: next.amount,
                    generation,
                },
                (!signed_root.is_zero()).then_some(signed_root),
            )
        })
        .map_err(|err| match attempted {
            Some(amount) => err.context(UnconfirmedVoucher { amount }),
            None => err,
        })
}

/// Names the voucher a failed [`send_voucher`] left armed, so a terminal
/// `StreamEnd` read afterwards confirms that voucher and no other
/// ([`PoolLedger::confirm_armed_stamped`]). Carried as error context: it composes with,
/// and still downcasts through, the send error it wraps.
#[derive(Debug)]
struct UnconfirmedVoucher {
    /// The cumulative amount the armed voucher signs.
    amount: U256,
}

impl std::fmt::Display for UnconfirmedVoucher {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "voucher for cumulative amount {} did not confirm",
            self.amount
        )
    }
}

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

/// Sign one voucher over `next`/`chain` with the pool context's key and write it
/// to `send` — the body both [`send_voucher`] and [`PoolLedger::reanchor`] hand to
/// the ledger, so the two differ only in which cumulative the ledger hands back.
async fn sign_and_write_voucher(
    send: &mut SendStream,
    ctx: &PoolContext,
    next: Cumulative,
    chain: ChainCommit,
) -> anyhow::Result<()> {
    let signed = Voucher {
        pool_id: ctx.pool_id,
        signer: ctx.client_signer.address(),
        provider: ctx.provider,
        amount: next.amount,
        bytes_delivered: next.bytes,
        chain_root: chain.chain_root,
        chunk_price: chain.chunk_price,
    }
    .sign(ctx.client_signer.as_ref(), &ctx.voucher_domain)
    .map_err(|e| anyhow::anyhow!("voucher signing failed: {e}").context(LocalPullFault))?;
    let wire_voucher = signed_to_wire_voucher(&signed)
        .map_err(|e| anyhow::anyhow!("voucher exceeds wire width: {e}").context(LocalPullFault))?;
    write_message(send, &ClientMessage::Voucher(wire_voucher)).await
}

/// One stream's view of the lane's hash chain (ADR 003 §Concurrent Streams).
///
/// The chain's scope is the **lane**, not the stream: every stream from one
/// signer to one node shares a single root and a single index, because a
/// released preimage advances the lane and cannot be attributed to whichever
/// stream happened to carry it. What IS per-stream is the *anchor* — which
/// epoch this stream has told the node about.
///
/// That distinction is the whole reason this type exists. A bare preimage names
/// no chain, so the node places it against the anchor of the stream it arrived
/// on. If a stream released a reveal before carrying that epoch's root voucher,
/// the node could not name the chain the reveal belongs to and would reject it
/// `UnanchoredPreimage`. So each stream re-anchors itself whenever the lane
/// rolls, rather than trusting a sibling to have done it — QUIC orders within a
/// stream, so a fast stream that has already adopted a new root can never
/// invalidate a slower sibling still finishing the old one.
#[derive(Debug, Default)]
struct StreamMeter {
    /// The chain this stream has already carried the root voucher for, named by
    /// its root. `None` until it has anchored to anything.
    anchored_root: Option<B256>,
    /// The last proof THIS stream claimed money with, so a `VoucherRejected`
    /// read here rewinds that proof and not whatever a sibling stream has done
    /// on the lane since (see [`PoolLedger::resolve_reject`]). `None` before the
    /// stream has claimed anything, and left alone by a re-anchor — which claims
    /// nothing and so has nothing to rewind.
    last_proof: Option<StreamProof>,
}

impl StreamMeter {
    /// Ensure this stream has told the node which chain its reveals belong to,
    /// opening a chain if the lane meters none, and sending that chain's root
    /// voucher first if this stream has not carried it.
    ///
    /// Called from one place — a stream about to release a reveal, which is the
    /// only thing that needs a chain to exist. Residual settlement commits to
    /// whatever the lane already meters against and goes straight to
    /// [`send_voucher`].
    async fn anchor(
        &mut self,
        send: &mut SendStream,
        ctx: &PoolContext,
        ledger: &PoolLedger,
        rate_per_mb: u64,
    ) -> anyhow::Result<()> {
        if ledger.chain_root().is_none() {
            let (proof, root) =
                send_voucher(send, ctx, ledger, rate_per_mb, 0, EpochAction::Open).await?;
            self.last_proof = Some(proof);
            // The root the voucher signed: the send is what opened the chain,
            // and a sibling may have rolled it again since.
            self.anchored_root = root;
            return Ok(());
        }
        if ledger.chain_root().is_some() && self.anchored_root != ledger.chain_root() {
            // A sibling rolled the lane, or this is our first reveal on a chain a
            // sibling opened. Either way this stream must name the chain before it
            // can reveal against it (ADR 003 §Concurrent Streams, Rule 1).
            //
            // This must NOT go through `send_voucher`: every voucher the ledger
            // issues folds the chain's accrual, and a voucher that folds must also
            // roll — so re-anchoring that way would retire the chain it meant to
            // join, strand every sibling's in-flight reveals under the retired
            // root, and buy a signature plus a fresh keccak ladder each time.
            // `reanchor` re-states the signed anchor under the live root instead,
            // folding nothing.
            //
            // Take the root `reanchor` reports, not one read before the send: the
            // lane's live chain is only knowable while the issuance lock is held,
            // and storing a stale root would leave this stream re-anchoring on
            // every chunk.
            self.anchored_root = ledger
                .reanchor(|anchor, chain| sign_and_write_voucher(send, ctx, anchor, chain))
                .await?;
        }
        Ok(())
    }

    /// Meter one delivered chunk: release the next preimage on this stream,
    /// rolling the chain first if the epoch is spent.
    ///
    /// The reveal costs one hash and one 33-byte message — no signature, no
    /// acknowledgement, and nothing durable to write — which is the whole point
    /// of the chain. A rollover costs one signature, and happens once per 255
    /// chunks rather than once per chunk.
    async fn release_chunk(
        &mut self,
        send: &mut SendStream,
        ctx: &PoolContext,
        ledger: &PoolLedger,
        rate_per_mb: u64,
    ) -> anyhow::Result<()> {
        // One rollover, never more: the first meter can find the epoch spent,
        // and the roll that follows opens a chain with index 1 available by
        // construction. A second exhaustion would mean the ledger contradicted
        // itself. A `Moved` tick re-anchors and meters again; each one means a
        // sibling rolled the lane, so the lane progresses every time it repeats.
        let mut rolled = false;
        loop {
            self.anchor(&mut *send, ctx, ledger, rate_per_mb).await?;
            // Reborrow per iteration: the closure takes the stream by unique
            // reference for the duration of the send, and the loop needs it back.
            let wire = &mut *send;
            match ledger
                .meter(self.anchored_root, |released: Released| async move {
                    write_message(
                        wire,
                        &ClientMessage::ChunkPreimage(ChunkPreimage {
                            preimage: released.preimage.into(),
                            index: released.index,
                        }),
                    )
                    .await
                })
                .await?
            {
                Metered::Released(released) => {
                    self.last_proof = Some(StreamProof::Reveal {
                        chain_root: released.chain_root,
                        index: released.index,
                    });
                    return Ok(());
                }
                Metered::Moved => {}
                Metered::Exhausted if !rolled => {
                    // Roll. The new voucher's `amount` already folds this
                    // chain's frontier — every reveal advanced the committed
                    // cumulative as it went — so the fold is the frontier
                    // actually reached, never a flat 255 (ADR 003 §Rollover).
                    let (proof, root) =
                        send_voucher(&mut *send, ctx, ledger, rate_per_mb, 0, EpochAction::Roll)
                            .await?;
                    self.last_proof = Some(proof);
                    self.anchored_root = root;
                    rolled = true;
                }
                Metered::Exhausted => {
                    // A ledger self-contradiction, not a peer fault: marked as OURS so
                    // it is scored as a local fault and never rescued by the
                    // voucher-write recovery into a clean completion.
                    return Err(
                        anyhow::anyhow!("hash chain still exhausted after a rollover")
                            .context(LocalPullFault),
                    );
                }
            }
        }
    }

    /// Settle a residual smaller than one chunk with a signed voucher.
    ///
    /// A preimage always advances the claim by a *whole* chunk, so a partial
    /// trailing chunk cannot be priced by one — it settles through a signature,
    /// exact to one token base unit.
    ///
    /// It asks for [`EpochAction::Keep`], and usually does not get it. Any
    /// transfer that released at least one reveal has accrual outstanding, and a
    /// voucher that folds must also roll ([`PoolLedger::issue`]) — so the typical
    /// close retires the chain and opens a fresh one, and a sibling stream's
    /// in-flight reveals under the old root land as superseded and re-anchor.
    /// That is the correct trade and not an oversight: the fold is what makes the
    /// residual exact, and a folded frontier must never coexist with the root
    /// that proved it. `Keep` is what a transfer smaller than one whole chunk
    /// gets — nothing accrued, nothing to fold — and there it commits the sealed
    /// section, since such a lane never opened a chain at all.
    async fn settle_residual(
        &mut self,
        send: &mut SendStream,
        ctx: &PoolContext,
        ledger: &PoolLedger,
        rate_per_mb: u64,
        residual_bytes: u64,
    ) -> anyhow::Result<()> {
        let (proof, root) = send_voucher(
            send,
            ctx,
            ledger,
            rate_per_mb,
            residual_bytes,
            EpochAction::Keep,
        )
        .await?;
        self.last_proof = Some(proof);
        self.anchored_root = root;
        Ok(())
    }

    /// Emit the proofs `delivered` new bytes have earned: one reveal per whole
    /// chunk, plus a closing signature for any residual once the transfer is
    /// complete.
    ///
    /// Returns the bytes still unproved — always below one chunk.
    async fn pay(
        &mut self,
        send: &mut SendStream,
        ctx: &PoolContext,
        ledger: &PoolLedger,
        rate_per_mb: u64,
        mut unproved: u64,
        complete: bool,
    ) -> anyhow::Result<u64> {
        while unproved >= CHUNK_BYTES {
            self.release_chunk(send, ctx, ledger, rate_per_mb).await?;
            unproved -= CHUNK_BYTES;
        }
        if complete && unproved > 0 {
            self.settle_residual(send, ctx, ledger, rate_per_mb, unproved)
                .await?;
            unproved = 0;
        }
        Ok(unproved)
    }
}

/// Turn a mid-stream `StreamError` read off the receive loop into the typed error
/// the loop surfaces. There is no positive ack in the pool model — continued
/// delivery is acceptance (ADR 005) — so this handles only the rejection slot.
///
/// A `VoucherRejected` is OUR payment-side fault: rewind the proof `meter` last sent
/// (known-not-taken, so it must not be settled optimistically) and carry its typed
/// reason — plus the wallet-less-resume `bundle` (#1481) — so the orchestrator can
/// exonerate the provider (#857) and the reseed loops (`driver::fill_gap`, the
/// `test-util` `fetch_inner`) can self-heal from an authenticated watermark
/// instead of treating the rejection as terminal. Any
/// OTHER `StreamError` is the upstream refusing mid-stream; carry the typed wire
/// class as [`UpstreamRefused`] so a refusing peer is scored on its real class
/// rather than the `Unreachable` catch-all (#1145 review).
fn voucher_rejection(
    ledger: &PoolLedger,
    meter: &StreamMeter,
    error: StreamError,
) -> anyhow::Error {
    match error {
        StreamError::VoucherRejected { reason, bundle } => {
            // Rewind the proof THIS stream sent. The lane is shared, so the proof
            // it last issued may belong to a sibling; naming ours is what keeps
            // the rejection from un-committing a voucher the node accepted.
            if let Some(proof) = meter.last_proof {
                ledger.resolve_reject(proof);
            }
            let proof_generation = match meter.last_proof {
                Some(StreamProof::Voucher { generation, .. }) => Some(generation),
                Some(StreamProof::Reveal { .. }) | None => None,
            };
            anyhow::Error::new(UpstreamVoucherRejected {
                reason,
                bundle,
                proof_generation,
            })
        }
        other => anyhow::Error::new(UpstreamRefused::mid_stream(other)),
    }
}

/// Validate + verify a `StreamResponse` on receive (ADR 005, ADR 014 §1, #252).
fn verify_response(
    resp: &StreamResponse,
    ext: &StreamResponseExt,
    slash_domain: &Eip712Domain,
    expected_signer: Address,
    hash: [u8; 32],
    pool_id: B256,
    timestamp_us: u64,
) -> anyhow::Result<()> {
    // #252 + slash_sig length/upper-bound checks on the frozen base...
    resp.validate()
        .map_err(|e| anyhow::anyhow!("invalid stream response: {e}"))?;
    // ...and the ok/error agreement, which spans the signed base and the unsigned
    // extension, so neither validator can see it alone (ADR 013 §Tier 1).
    ext.validate(resp.body.ok)
        .map_err(|e| anyhow::anyhow!("invalid stream response: {e}"))?;
    // Echoed-field correlation (ADR 005).
    if resp.body.hash != hash {
        anyhow::bail!("response hash does not match request");
    }
    if resp.body.pool_id != pool_id.as_slice() {
        anyhow::bail!("response pool_id does not match request");
    }
    if resp.body.timestamp_us != timestamp_us {
        anyhow::bail!("response timestamp_us not echoed");
    }
    // slash_sig must recover to the delivering node's Ethereum address.
    let sig = Signature::try_from(resp.slash_sig.as_slice())
        .map_err(|e| anyhow::anyhow!("slash_sig parse: {e}"))?;
    StreamSlashData::from_response_body(&resp.body)
        .verify_signer(&sig, expected_signer, slash_domain)
        .map_err(|e| anyhow::anyhow!("slash_sig verification failed: {e}"))?;
    Ok(())
}

async fn write_message(send: &mut SendStream, msg: &ClientMessage) -> anyhow::Result<()> {
    // The encode is ours; the write below is the peer's connection.
    let payload = encode_message(msg)
        .map_err(|e| anyhow::anyhow!("encode failed: {e}").context(LocalPullFault))?;
    // A mid-stream voucher write, past the open handshake: `APP_ERR_RATE_LIMITED`
    // (`0x10`) is only ever sent before any application stream exists (ADR 013
    // §Application Error Codes), so it cannot reach this site. Keep the plain text
    // rather than route through `transport_error` — typing a shed here would trust
    // that invariant instead of scoping the recovery to the open-stage sites.
    write_frame(send, &payload)
        .await
        .map_err(|e| anyhow::anyhow!("write failed: {e}"))
}

/// Read the open-stage `StreamResponse` together with its trailing
/// [`StreamResponseExt`] (ADR 013 §Tier 1, two-phase).
///
/// Separate from [`read_client_message`] because only this message carries an
/// extension: every mid-stream variant is a single postcard value, and widening
/// the shared reader would push an always-`default()` ext onto all of them.
async fn read_stream_response(
    recv: &mut RecvStream,
) -> anyhow::Result<(StreamResponse, StreamResponseExt)> {
    let frame = read_frame(recv)
        .await
        .map_err(|e| rate_limited::transport_error("frame read failed", e))?;
    let (msg, tail) = decode_message::<ClientMessage>(&frame)
        .map_err(|e| anyhow::anyhow!("decode failed: {e}"))?;
    let ClientMessage::StreamResponse(response) = msg else {
        anyhow::bail!("expected StreamResponse, got {}", variant_name(&msg));
    };
    let ext = decdn_protocol::parse_stream_response_ext(tail)
        .map_err(|e| anyhow::anyhow!("decode stream response ext: {e}"))?;
    Ok((response, ext))
}

async fn read_client_message<R: tokio::io::AsyncRead + Unpin>(
    recv: &mut R,
) -> anyhow::Result<ClientMessage> {
    // A mid-stream read, past the open handshake. `0x10` reaches only the
    // open-stage sites (`connect` / `open_bi` / the initial request write and
    // [`read_stream_response`]), never here — the limiters shed before any stream
    // exists (ADR 013 §Application Error Codes). Plain text, not `transport_error`.
    let frame = read_frame(recv)
        .await
        .map_err(|e| anyhow::anyhow!("frame read failed: {e}"))?;
    let (msg, _rest) = decode_message::<ClientMessage>(&frame)
        .map_err(|e| anyhow::anyhow!("decode failed: {e}"))?;
    Ok(msg)
}

const fn variant_name(msg: &ClientMessage) -> &'static str {
    match msg {
        ClientMessage::StreamRequest(_) => "StreamRequest",
        ClientMessage::StreamResponse(_) => "StreamResponse",
        ClientMessage::ChunkData(_) => "ChunkData",
        ClientMessage::Voucher(_) => "Voucher",
        ClientMessage::ChunkPreimage(_) => "ChunkPreimage",
        ClientMessage::StreamEnd => "StreamEnd",
        ClientMessage::StreamError(_) => "StreamError",
    }
}

/// Record a failed [`open_progressive_pull`] on its span once: the error text
/// and an error status, so a failed dial or handshake is not exported as a
/// slow success.
fn record_open_result<T>(span: &tracing::Span, result: &anyhow::Result<T>) {
    if let Err(e) = result {
        span.record("error", tracing::field::display(format_args!("{e:#}")));
        span.record("otel.status_code", "ERROR");
    }
}

#[cfg(test)]
mod tests;