dig_download/

source.rs

//! [`RangeTransport`] — fetch one byte range (or an availability answer) from one provider — plus
//! per-source health tracking and the real dig-nat-backed implementation.
//!
//! The orchestrator fans byte ranges across providers by calling [`RangeTransport::fetch_range`]
//! concurrently, one future per (provider, range). The trait abstracts the peer transport so the
//! scheduler is tested over an in-memory mock (see [`crate::testkit`]); the real
//! [`NatRangeTransport`] rides dig-nat (`dig.getAvailability` + `dig.fetchRange` over an mTLS mux
//! stream). A provider that fails or serves a bad range is penalized via [`SourceTracker`] so the
//! scheduler stops leaning on it.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use dig_dht::ProviderRecord;
use dig_nat::{
    AvailabilityItem, AvailabilityResponse, ChunkLensAssembler, ChunkLensError, RangeFrame,
    RangeRequest,
};
use dig_peer::DigPeer;
use tokio::io::{AsyncRead, AsyncReadExt};

use crate::error::DownloadError;

/// The verification metadata a range's frames carry (L7 §9): the whole-resource shape a downloader
/// uses to establish or check the [`ResourceCommitment`](crate::verify::ResourceCommitment).
///
/// # Read this as "the stream's declared identity", not "frame 1's fields"
///
/// [`total_length`](Self::total_length), [`chunk_lens`](Self::chunk_lens),
/// [`chunk_count`](Self::chunk_count) and [`root`](Self::root) describe the WHOLE resource, so every
/// frame of a conforming stream repeats the same values. [`assemble_range_stream`] therefore captures
/// them from the first frame and RE-CHECKS every later frame against them — a holder that revises its
/// declared shape mid-stream is rejected rather than silently believed on whichever frame arrived
/// first. [`chunk_index`](Self::chunk_index) is the one field that legitimately varies per frame; see
/// its own note.
///
/// `#[non_exhaustive]`: the L7 range preamble gains optional fields as the wire grows (`chunk_count`
/// and the paged prologue are recent additions), and each one arrives here. Build one with
/// [`from_frame`](Self::from_frame) or from [`Default`] plus the `declaring_*` setters.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct RangeMeta {
    /// The full resource ciphertext length.
    pub total_length: Option<u64>,
    /// Per-chunk ciphertext lengths of the whole resource, in order. For a resource whose layout is
    /// too large to state on one frame this is one PAGE of the array rather than all of it — which is
    /// why [`chunk_count`](Self::chunk_count) exists to say how many entries the whole array has.
    pub chunk_lens: Option<Vec<u64>>,
    /// How many entries the whole resource's `chunk_lens` array has.
    ///
    /// Present so a reader can tell a COMPLETE single-frame layout from one page of a paged prologue:
    /// `chunk_lens.len() < chunk_count` means the array continues on later frames. Absent means the
    /// holder declared no count, and `chunk_lens` is taken to be the whole array (the pre-paging
    /// shape, which stays readable — §5.1).
    pub chunk_count: Option<u64>,
    /// Index into `chunk_lens` of the first chunk in THIS frame.
    ///
    /// Unlike the other fields this is per-frame, not per-resource: a chunk-aligned continuation frame
    /// states where it begins. Frames arrive in ascending byte offset, so the declared index is
    /// non-decreasing across a conforming stream.
    pub chunk_index: Option<u64>,
    /// The chain-anchored generation root (64-hex).
    pub root: Option<String>,
    /// The whole-resource merkle inclusion proof (base64), or `None` for a capsule.
    pub inclusion_proof: Option<String>,
}

impl RangeMeta {
    /// The verification metadata one [`RangeFrame`] declares.
    pub fn from_frame(frame: &RangeFrame) -> Self {
        RangeMeta {
            total_length: frame.total_length,
            chunk_lens: frame.chunk_lens.clone(),
            chunk_count: frame.chunk_count,
            chunk_index: frame.chunk_index,
            root: frame.root.clone(),
            inclusion_proof: frame.inclusion_proof.clone(),
        }
    }

    /// Declare the whole-resource shape (`total_length` + `chunk_lens` + the array's `chunk_count`).
    /// The `#[non_exhaustive]` construction path for a test fixture or a non-dig-nat transport.
    pub fn declaring_layout(
        mut self,
        total_length: u64,
        chunk_lens: Vec<u64>,
        chunk_count: u64,
    ) -> Self {
        self.total_length = Some(total_length);
        self.chunk_count = Some(chunk_count);
        self.chunk_lens = Some(chunk_lens);
        self
    }

    /// Declare the chain anchor (`root` + the whole-resource `inclusion_proof`).
    pub fn declaring_anchor(mut self, root: String, inclusion_proof: Option<String>) -> Self {
        self.root = Some(root);
        self.inclusion_proof = inclusion_proof;
        self
    }

    /// Declare which chunk of the resource this frame's bytes begin at.
    pub fn declaring_chunk_index(mut self, chunk_index: u64) -> Self {
        self.chunk_index = Some(chunk_index);
        self
    }
}

/// A fetched, reassembled byte range: the assembled ciphertext for the requested `[offset, offset+len)`
/// plus the first-frame verification metadata. The orchestrator verifies this against the resource
/// commitment, then writes `bytes` at `request_offset` in the sink.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FetchedRange {
    /// The absolute resource offset the range was requested at (== [`RangeRequest::offset`]).
    pub request_offset: u64,
    /// The reassembled range ciphertext.
    pub bytes: Vec<u8>,
    /// The first-frame verification metadata for this range.
    pub meta: RangeMeta,
}

/// Fetch content ranges + availability from providers. The one network capability the orchestrator
/// needs, abstracted for testability (mock in [`crate::testkit`]; real [`NatRangeTransport`]).
#[async_trait]
pub trait RangeTransport: Send + Sync {
    /// Ask `provider` which of `items` it holds (`dig.getAvailability`) — the pre-check before fanning
    /// ranges. The answer's `total_length` / `chunk_count` also seed range planning.
    async fn query_availability(
        &self,
        provider: &ProviderRecord,
        items: Vec<AvailabilityItem>,
    ) -> Result<AvailabilityResponse, DownloadError>;

    /// Fetch the byte range described by `req` from `provider` (`dig.fetchRange`), streaming +
    /// reassembling the frames into a [`FetchedRange`]. A transport failure (connect/stream error) is
    /// a recoverable [`DownloadError::Transport`] — the orchestrator retries the range elsewhere.
    async fn fetch_range(
        &self,
        provider: &ProviderRecord,
        req: &RangeRequest,
    ) -> Result<FetchedRange, DownloadError>;
}

/// Health of one provider as a range source — failure count + a backoff window during which the
/// scheduler avoids it.
#[derive(Debug, Clone, Default)]
pub struct SourceHealth {
    /// Consecutive failures (reset on success).
    pub failures: u32,
    /// Total ranges this source has successfully served (for rebalancing / diagnostics).
    pub served: u64,
    /// Do not schedule this source again until this instant (set on failure, capped-exponential).
    pub backoff_until: Option<Instant>,
}

/// Tracks per-provider [`SourceHealth`] so the scheduler prefers healthy sources and backs off failed
/// ones (bounded exponential backoff), without ever permanently banning a source that might recover.
#[derive(Debug, Default)]
pub struct SourceTracker {
    health: HashMap<String, SourceHealth>,
    base_backoff: Duration,
    max_backoff: Duration,
}

impl SourceTracker {
    /// A tracker with the given base + max backoff (backoff doubles per consecutive failure, capped).
    pub fn new(base_backoff: Duration, max_backoff: Duration) -> Self {
        SourceTracker {
            health: HashMap::new(),
            base_backoff,
            max_backoff,
        }
    }

    /// Whether `peer_id` is schedulable at `now` (not inside a backoff window).
    pub fn is_available(&self, peer_id: &str, now: Instant) -> bool {
        match self.health.get(peer_id) {
            Some(h) => match h.backoff_until {
                Some(until) => now >= until,
                None => true,
            },
            None => true,
        }
    }

    /// Record a successful range served by `peer_id` (clears failures + backoff).
    pub fn record_success(&mut self, peer_id: &str) {
        let h = self.health.entry(peer_id.to_string()).or_default();
        h.failures = 0;
        h.served += 1;
        h.backoff_until = None;
    }

    /// Record a failure by `peer_id` at `now` and set its (capped-exponential) backoff window.
    pub fn record_failure(&mut self, peer_id: &str, now: Instant) {
        let base = self.base_backoff;
        let max = self.max_backoff;
        let h = self.health.entry(peer_id.to_string()).or_default();
        h.failures = h.failures.saturating_add(1);
        let shift = h.failures.saturating_sub(1).min(16);
        let backoff = base.checked_mul(1u32 << shift).unwrap_or(max).min(max);
        h.backoff_until = Some(now + backoff);
    }

    /// The number of successfully-served ranges recorded for `peer_id`.
    pub fn served(&self, peer_id: &str) -> u64 {
        self.health.get(peer_id).map(|h| h.served).unwrap_or(0)
    }

    /// The consecutive-failure count recorded for `peer_id`.
    pub fn failures(&self, peer_id: &str) -> u32 {
        self.health.get(peer_id).map(|h| h.failures).unwrap_or(0)
    }
}

/// The whole-resource identity a range stream declared on its FIRST frame, enforced against every
/// later frame of the same stream.
///
/// # Why a later frame has to be checked at all
///
/// The L7 frame contract splits its optional fields in two (dig-nat `mux.rs`, `SPEC.md` §5.1.1):
/// `total_length` / `root` / `chunk_count` / `chunk_index` are **identity — every frame**, while
/// `chunk_lens` / `inclusion_proof` are **prologue — once per stream** and MUST NOT be repeated.
/// Reading only frame 1 and discarding the rest, as this reader used to, means a holder can declare an
/// honest shape on the frame the commitment binds to and a different one on every frame after it, and
/// be believed on the first. Nothing downstream recovers that: the commitment was already adopted.
///
/// # The rule is over the class, not over one behaviour
///
/// A revision is a revision whichever direction it goes and whichever field carries it, so this rejects
/// **any** disagreement with frame 1 — a changed value, and equally a value APPEARING on a later frame
/// that frame 1 left unstated, since the reader binds to frame 1 and a late arrival is information it
/// deliberately withheld from the frame that mattered. A later frame that simply OMITS an identity field
/// is allowed: it asserts nothing, so there is nothing to disbelieve, and refusing it would reject a
/// holder whose only fault is terseness while gaining no property.
#[derive(Debug, Default)]
struct StreamIdentity {
    root: Option<String>,
    total_length: Option<u64>,
    chunk_count: Option<u64>,
    /// The highest `chunk_index` declared so far. Frames arrive in ascending byte offset and this names
    /// the frame's own first chunk, so it may ADVANCE but never rewind — the one identity-adjacent
    /// field that legitimately differs per frame.
    highest_chunk_index: Option<u64>,
    /// One past the last `chunk_lens` entry any accepted page has filled — the boundary that separates a
    /// conforming NEXT page from a restatement of ground already covered.
    prologue_frontier: u64,
}

/// What a conforming LATER frame turned out to be.
#[derive(Debug, PartialEq, Eq)]
enum LaterFrame {
    /// A continuation carrying bytes and no prologue — the ordinary case.
    Continuation,
    /// A NEW `chunk_lens` page: conforming, and the next installment of a paged prologue.
    NewProloguePage {
        /// The entry the page begins at.
        offset: u64,
        /// How many entries it carries.
        entries: usize,
    },
}

impl StreamIdentity {
    /// Capture the identity the stream's first frame declares.
    fn from_first_frame(frame: &RangeFrame) -> Self {
        let page_entries = frame.chunk_lens.as_ref().map_or(0, Vec::len) as u64;
        StreamIdentity {
            root: frame.root.clone(),
            total_length: frame.total_length,
            chunk_count: frame.chunk_count,
            highest_chunk_index: frame.chunk_index,
            prologue_frontier: frame
                .chunk_lens_offset
                .unwrap_or(0)
                .saturating_add(page_entries),
        }
    }

    /// Check one LATER frame against the captured identity, advancing the chunk-index and prologue
    /// frontiers, and classify what the frame is.
    ///
    /// `Err` names the field and both values, because "the holder revised its declared shape" is only
    /// actionable if the reader can say which declaration moved.
    fn check_later_frame(&mut self, frame: &RangeFrame) -> Result<LaterFrame, String> {
        check_unrevised("root", self.root.as_deref(), frame.root.as_deref())?;
        check_unrevised("total_length", self.total_length, frame.total_length)?;
        check_unrevised("chunk_count", self.chunk_count, frame.chunk_count)?;

        // `inclusion_proof` is once-per-stream with NO paged form — there is only ever one proof — so any
        // later frame carrying it is restating, whether or not it agrees. Refusing the restatement rather
        // than comparing it is what makes "frame 1 said A, frame 5 said B" inexpressible instead of
        // merely unpersuasive.
        if frame.inclusion_proof.is_some() {
            return Err("a later frame restates inclusion_proof, which is once-per-stream".into());
        }

        if let Some(index) = frame.chunk_index {
            if let Some(highest) = self.highest_chunk_index {
                if index < highest {
                    return Err(format!(
                        "chunk_index {index} rewinds below {highest}; frames arrive in ascending offset"
                    ));
                }
            }
            self.highest_chunk_index = Some(index);
        }

        // `chunk_lens` is the one prologue field WITH a paged form, so "MUST NOT be repeated" cannot be
        // read as "only the first frame may carry it" — the wire contract explicitly has successive frames
        // each carry a page, stamped with the offset it starts at. What is forbidden is restating ground
        // already covered; what is prescribed is advancing past it. The field that tells those apart is
        // `chunk_lens_offset`, so the rule is stated over it.
        //
        // Reading the ban as first-frame-only would hand a CONFORMING paging holder a protocol-violation
        // verdict, and the paging work would then have to delete this branch outright. Classifying the
        // frame instead means that work only changes what the CALLER does with a new page.
        let Some(page) = &frame.chunk_lens else {
            return Ok(LaterFrame::Continuation);
        };
        // Absent means "begins at 0" per the wire contract, which the first frame's page already covered,
        // so an unstamped later page lands below the frontier and is caught here as the restatement it is.
        let offset = frame.chunk_lens_offset.unwrap_or(0);
        if offset < self.prologue_frontier {
            return Err(format!(
                "a later frame's chunk_lens page at offset {offset} re-covers entries below {}, which an \
                 earlier page already filled; a paged prologue must ADVANCE, never restate",
                self.prologue_frontier
            ));
        }
        self.prologue_frontier = offset.saturating_add(page.len() as u64);
        Ok(LaterFrame::NewProloguePage {
            offset,
            entries: page.len(),
        })
    }

    /// How many `chunk_lens` entries the stream has delivered so far.
    fn entries_delivered(&self) -> u64 {
        self.prologue_frontier
    }
}

/// Reject a later frame's `declared` value for an identity `field` unless it agrees with what the
/// first frame `committed` — including the case where the first frame committed nothing at all.
fn check_unrevised<T: PartialEq + std::fmt::Debug>(
    field: &str,
    committed: Option<T>,
    declared: Option<T>,
) -> Result<(), String> {
    match (committed, declared) {
        (_, None) => Ok(()), // asserts nothing
        (Some(committed), Some(declared)) if committed == declared => Ok(()),
        (Some(committed), Some(declared)) => Err(format!(
            "{field} changed mid-stream: first frame declared {committed:?}, a later frame {declared:?}"
        )),
        (None, Some(declared)) => Err(format!(
            "{field} {declared:?} appears only on a later frame; the first frame left it unstated"
        )),
    }
}

/// Reassemble a `dig.fetchRange` frame stream into `(bytes, meta)`: read [`RangeFrame`]s in ascending
/// offset order, placing each frame's bytes at its (range-relative) offset and capturing the
/// stream's declared verification metadata. Stops on the frame marked `complete` or clean
/// end-of-stream.
///
/// Bounded by `max_len` (the expected range length) so a misbehaving peer cannot stream unbounded
/// bytes into memory: a frame that overshoots the window is CLIPPED to it (servers answer at chunk
/// granularity, so a 1-byte metadata probe is legitimately served a whole chunk), assembly stops as
/// soon as the window is full, and only a frame starting at or beyond `max_len` is an error.
///
/// The returned [`RangeMeta`] is the FIRST frame's declaration, and every later frame is checked
/// against it — a holder that revises `root` / `total_length` / `chunk_count`
/// mid-stream, or restates the once-per-stream prologue, fails the whole fetch instead of being
/// believed on frame 1. Nothing beneath this reader performs that check.
///
/// # Paged prologue
///
/// A resource whose `chunk_lens` array is too large to state on one frame is served as a **paged
/// prologue**: the first frame declares the whole array's `chunk_count` but carries only its first
/// page, and successive frames each carry another page stamped with the entry `chunk_lens_offset` it
/// begins at (dig-nat `SPEC.md` §5.1.1). This reader reassembles those pages into one array via
/// [`ChunkLensAssembler`], and sets [`RangeMeta::chunk_lens`] to the FULL array only once every page
/// has landed. The reassembly is **fail-closed**: `chunk_lens` is a decrypt input (per-chunk
/// AES-GCM-SIV needs the whole array, whose entries must sum to `total_length`), so a stream whose
/// prologue ends short — or whose page is misaligned, duplicated, or overshoots — yields NO layout at
/// all rather than a partial one, and the holder is skipped (a RECOVERABLE error) rather than believed.
///
/// This is the pure, network-free core of [`NatRangeTransport::fetch_range`] and is
/// unit-tested by feeding encoded frames through an in-memory reader.
pub async fn assemble_range_stream<R: AsyncRead + Unpin>(
    reader: &mut R,
    max_len: u64,
) -> Result<(Vec<u8>, RangeMeta), DownloadError> {
    let mut buf: Vec<u8> = Vec::new();
    let mut meta = RangeMeta::default();
    let mut identity: Option<StreamIdentity> = None;
    // Reassembles the `chunk_lens` array of a paged prologue across frames. Built lazily on the first
    // frame that declares more entries than it carries, and `None` for the ordinary single-frame layout.
    let mut assembler: Option<ChunkLensAssembler> = None;
    // One past the furthest byte any frame has contributed — the progress the termination guard at the
    // bottom of the loop requires each non-final frame to advance.
    let mut byte_frontier: u64 = 0;
    loop {
        let frame = RangeFrame::decode(reader)
            .await
            .map_err(|e| DownloadError::Transport {
                provider: String::new(),
                reason: format!("range frame decode: {e}"),
            })?;
        let Some(frame) = frame else {
            break; // clean end-of-stream
        };
        // Whether THIS frame advanced the paged prologue. A prologue page may carry zero data bytes, so
        // the termination guard must count an accepted page as progress or a legitimately data-less page
        // would look like a stalled stream.
        let mut accepted_page = false;
        match identity.as_mut() {
            None => {
                meta = RangeMeta::from_frame(&frame);
                identity = Some(StreamIdentity::from_first_frame(&frame));
                // A layout too large for one frame is paged: the first frame declares the whole array's
                // `chunk_count` while carrying only its first page. Begin reassembly so the pages that
                // arrive on later frames land in one array. `ChunkLensAssembler::new` refuses a
                // `chunk_count` above `MAX_RESOURCE_CHUNK_COUNT` before it allocates.
                if let Some(chunk_count) = frame.chunk_count {
                    let delivered = frame.chunk_lens.as_ref().map_or(0, Vec::len) as u64;
                    if chunk_count > delivered {
                        let mut asm = ChunkLensAssembler::new(chunk_count as usize)
                            .map_err(chunk_lens_error)?;
                        if let Some(page) = &frame.chunk_lens {
                            asm.accept_page(frame.chunk_lens_offset.unwrap_or(0), page)
                                .map_err(chunk_lens_error)?;
                            accepted_page = true;
                        }
                        assembler = Some(asm);
                    }
                }
            }
            Some(identity) => {
                let verdict = identity.check_later_frame(&frame).map_err(|reason| {
                    DownloadError::Transport {
                        provider: String::new(),
                        reason,
                    }
                })?;
                if let LaterFrame::NewProloguePage { offset, .. } = verdict {
                    // The identity guard confirmed this page ADVANCES the prologue; the assembler applies
                    // the finer placement rules (alignment, exact page length, no duplicate slot) that
                    // dig-nat owns. A hostile page is a RECOVERABLE rejection that skips THIS holder.
                    let page = frame.chunk_lens.as_ref().expect(
                        "a NewProloguePage verdict is only produced for a frame with chunk_lens",
                    );
                    match assembler.as_mut() {
                        Some(asm) => {
                            asm.accept_page(offset, page).map_err(chunk_lens_error)?;
                            accepted_page = true;
                        }
                        // The first frame declared no multi-page layout, yet a later frame pages one: the
                        // frames disagree about the resource's shape. Refuse fail-closed rather than adopt
                        // a page for an array this reader never sized.
                        None => {
                            return Err(DownloadError::PagedPrologueUnsupported {
                                provider: String::new(),
                                chunk_count: identity.chunk_count.unwrap_or_default(),
                                delivered: identity.entries_delivered(),
                            });
                        }
                    }
                }
            }
        }
        // The paged prologue is stream metadata that must fully drain even when the window is already
        // full or the request wanted metadata only — so every early loop exit waits on it.
        let prologue_drained = assembler
            .as_ref()
            .map_or(true, ChunkLensAssembler::is_complete);
        // A zero-length request asks for metadata ONLY (there is no window to place bytes in). The paged
        // prologue IS that metadata, so drain it before stopping; a page that carries no data bytes is
        // handled here rather than falling through to byte placement.
        if max_len == 0 {
            if prologue_drained {
                break;
            }
            // A non-final frame that advanced nothing (no accepted page) cannot progress the prologue —
            // stop it looping forever. Mirrors the byte-window termination guard below.
            if !accepted_page {
                return Err(DownloadError::Transport {
                    provider: String::new(),
                    reason: "a metadata stream ended its prologue short without progressing".into(),
                });
            }
            continue;
        }
        // A frame that starts at or past the end of the requested window carries bytes that can never
        // belong to this range — a real protocol violation, not a granularity mismatch.
        if frame.offset >= max_len {
            return Err(DownloadError::Transport {
                provider: String::new(),
                reason: format!(
                    "range frame at offset {} starts beyond expected length {max_len}",
                    frame.offset
                ),
            });
        }
        // CLIP an over-long frame instead of rejecting it: a server legitimately answers at CHUNK
        // granularity, so a 1-byte metadata probe is served a whole chunk (#836). Taking only the
        // requested window keeps memory bounded by `max_len` AND keeps such a holder usable.
        let start = frame.offset as usize;
        let take = frame.bytes.len().min((max_len - frame.offset) as usize);
        let end = start + take;
        if buf.len() < end {
            // FALLIBLE growth. `max_len` is derived from a peer-DECLARED chunk length, so even bounded
            // by the commitment's ceiling it can exceed what this host can hold — and an infallible
            // `resize` aborts the process through the uncatchable `handle_alloc_error` (#1608). A frame
            // sparse in the window (`offset` near `max_len`, a few bytes of payload) makes that
            // reachable from ONE small frame, so exhaustion must be an ordinary recoverable error the
            // scheduler routes around, not a death.
            buf.try_reserve(end - buf.len())
                .map_err(|e| DownloadError::Transport {
                    provider: String::new(),
                    reason: format!("cannot allocate a {end}-byte range assembly buffer: {e}"),
                })?;
            buf.resize(end, 0); // within the reservation above — no further allocation
        }
        buf[start..end].copy_from_slice(&frame.bytes[..take]);
        // The window filling is NOT sufficient to stop while a paged prologue is still draining: the
        // layout is stream metadata the reader must hold in full, and the probe that requested one byte
        // is exactly the read that must keep going until the last page lands.
        if frame.complete || (prologue_drained && buf.len() as u64 >= max_len) {
            break;
        }

        // TERMINATION. Every loop exit above depends on the window filling or the holder saying it is
        // done, so a frame that does neither and advances nothing lets the holder stream forever: a
        // `{ offset: 0, bytes: [], complete: false }` frame with all identity omitted satisfies every
        // check — omission is conforming by design — while `take` is 0 and `buf` never grows. A holder
        // re-sending the SAME low offset does the same thing with non-empty bytes. Either one hangs the
        // job on a few dozen bytes per frame, holding the staging claim that this crate's own comment
        // calls the denial primitive the claim exists to prevent.
        //
        // The guard is over the CLASS — a frame that does not extend the assembled prefix — rather than
        // over the empty-payload instance of it, so the re-send variant is caught by the same rule.
        // Bytes arrive in ascending offset, so a conforming continuation always extends past the frontier.
        // A frame that placed no new bytes but ADVANCED the paged prologue is progress too: prologue-only
        // pages legitimately carry zero data, so they are exempt from the byte-frontier rule.
        if end as u64 <= byte_frontier && !accepted_page {
            return Err(DownloadError::Transport {
                provider: String::new(),
                reason: format!(
                    "a frame at offset {} extends the range to {end}, past nothing (already at \
                     {byte_frontier}), and does not complete it; the stream cannot progress",
                    frame.offset
                ),
            });
        }
        byte_frontier = byte_frontier.max(end as u64);
    }

    // The layout is adopted ONLY as a COMPLETE array. An assembler that never saw its last page yields
    // nothing (fail-closed): a partial `chunk_lens` sums short of `total_length` and would decrypt every
    // chunk to garbage, so the holder is skipped (a recoverable refusal) rather than believed.
    if let Some(asm) = assembler {
        match asm.into_chunk_lens() {
            Ok(full) => meta.chunk_lens = Some(full),
            Err(_) => {
                return Err(DownloadError::PagedPrologueUnsupported {
                    provider: String::new(),
                    chunk_count: meta.chunk_count.unwrap_or_default(),
                    delivered: identity
                        .as_ref()
                        .map_or(0, StreamIdentity::entries_delivered),
                });
            }
        }
    }
    Ok((buf, meta))
}

/// Map a [`ChunkLensError`] from the paged-prologue assembler to a RECOVERABLE [`DownloadError`], so a
/// hostile or short prologue skips the offending holder rather than failing the whole download. The
/// `provider` is left empty for the transport layer to stamp (see [`DownloadError::attributed_to`]).
fn chunk_lens_error(e: ChunkLensError) -> DownloadError {
    DownloadError::Transport {
        provider: String::new(),
        reason: format!("chunk_lens prologue rejected: {e}"),
    }
}

/// The maximum number of trailer bytes drained from a range stream after the complete/last frame,
/// before the mux stream is closed. A well-behaved peer sends nothing (or a tiny framing tail) after
/// the last frame, so this bound is generous; it exists solely to close off a malicious peer that
/// holds the stream open and streams arbitrary filler (see [`drain_trailer_bounded`]).
const MAX_TRAILER_DRAIN: u64 = 64 * 1024;

/// Drain and DISCARD up to `cap` trailer bytes from `reader` (the leftover after a range's last
/// frame), so the mux stream closes cleanly WITHOUT buffering an unbounded trailer into memory.
///
/// A previous implementation did `stream.read_to_end(&mut Vec::new())`, which has no length bound: a
/// peer that serves a valid complete range then keeps the stream open and streams filler forces the
/// client to buffer all of it until OOM (MEDIUM #179). This reads into a small fixed scratch buffer
/// and stops once `cap` bytes have been seen (or at EOF / error), never growing an unbounded `Vec`.
/// Returns the number of trailer bytes drained (capped at `cap`).
pub async fn drain_trailer_bounded<R: AsyncRead + Unpin>(reader: &mut R, cap: u64) -> u64 {
    let mut scratch = [0u8; 4096];
    let mut drained: u64 = 0;
    while drained < cap {
        let want = ((cap - drained) as usize).min(scratch.len());
        match reader.read(&mut scratch[..want]).await {
            Ok(0) => break, // EOF — stream ended cleanly
            Ok(n) => drained += n as u64,
            Err(_) => break, // treat a read error as end-of-drain (stream will be dropped)
        }
    }
    drained
}

/// A pooled per-peer [`DigPeer`] client, shared behind a mutex so many range fetches to the SAME peer
/// reuse ONE mTLS session (opening a cheap fresh yamux stream each) instead of re-handshaking per
/// request. The `&mut self` [`DigPeer`] RPC receivers are serialized by the mutex.
type PooledConn = Arc<tokio::sync::Mutex<DigPeer>>;

/// The real [`RangeTransport`] over [`dig-peer`](dig_peer): connects to a provider as a
/// [`DigPeer`] — the one DIG Network peer client — over the FULL NAT-traversal ladder (direct →
/// UPnP/NAT-PMP/PCP → hole-punch → relay, IPv6-first), **reuses the client via a per-peer pool**, and
/// runs `dig.getAvailability` / `dig.fetchRange` over the mux'd mTLS session.
///
/// # Why DigPeer (#1283)
///
/// dig-download talks to peers through the shared [`DigPeer`] client rather than driving
/// [`dig_nat`] directly, so the whole ecosystem reaches peers ONE way. Every connection is
/// established through a [`PeerTarget`](dig_nat::PeerTarget) carrying the provider's `peer_id`, which
/// [`DigPeer::connect`] PINS the mTLS handshake to: a caller that means to reach provider X cannot be
/// answered by a different CA-valid peer (the impersonation footgun). The availability + range calls
/// are public-read (merkle-verified content), so they ride the mTLS channel unsealed (§5.4 exemption);
/// this transport therefore configures no [`SealingIdentity`](dig_peer::SealingIdentity).
///
/// # The NAT ladder on the fetch leg (#1305)
///
/// Discovery (dig-dht lookups) already rides the full ladder via a live [`dig_nat::NatRuntime`]; the
/// content byte-download must too, or a fully-NAT'd peer would DISCOVER a provider it can never FETCH
/// from (a non-Direct-reachable holder reachable only via hole-punch/relay). This transport connects
/// via [`DigPeer::connect_with_runtime`], composing exactly the tiers whose live handles the injected
/// [`NatRuntime`](dig_nat::NatRuntime) carries: an empty runtime ([`new`](Self::new)) is Direct-only; a node's real runtime
/// ([`new_with_runtime`](Self::new_with_runtime)) unlocks hole-punch + relay. dig-node builds the SAME
/// shared `NatRuntime` it uses for the DHT-side dial and hands it here.
///
/// A download fans many ranges across a few providers; without pooling every range fetch paid a full
/// NAT-traversal + mTLS handshake (LOW #179). The pool keeps one [`DigPeer`] per `peer_id` and opens a
/// new mux stream per request over the reused mTLS session; a client that errors is evicted so the
/// next request re-dials. For `fetch_range` the per-peer lock is held only while opening the (owned)
/// range stream, then released before the bytes are read, so concurrent ranges to the same peer still
/// stream in parallel.
///
/// The network dial is the only part not exercised by the in-memory tests (it needs real sockets +
/// certs); the reassembly + provider→target mapping are pure and unit-tested. dig-node constructs one
/// of these with its [`NodeCert`](dig_nat::NodeCert) (its CA-signed mTLS identity, minted by dig-tls's
/// `NodeCert::load_or_generate`) + [`NatConfig`](dig_nat::NatConfig) + its live [`NatRuntime`](dig_nat::NatRuntime) and
/// hands it to the [`Downloader`](crate::Downloader) — see the implementers' note in the crate docs.
pub struct NatRangeTransport {
    node: std::sync::Arc<dig_nat::NodeCert>,
    config: dig_nat::NatConfig,
    network_id: String,
    /// The live traversal handles (relay reservation / hole-punch coordinator / mapped port) the
    /// full-ladder dial composes each connect from. An empty runtime yields a Direct-only dial; a
    /// node's real runtime unlocks the hole-punch + relay tiers (#1305). Shared (`Arc`) so it can be
    /// the SAME runtime the node's DHT-side dial uses.
    runtime: Arc<dig_nat::NatRuntime>,
    /// Per-peer connection pool keyed by provider `peer_id` (the 64-hex string).
    pool: tokio::sync::Mutex<HashMap<String, PooledConn>>,
}

impl NatRangeTransport {
    /// Build a transport that dials providers on `network_id`, presenting `node` (this peer's
    /// CA-signed mTLS identity) and using `config` to select the traversal methods + timeouts.
    ///
    /// This uses an EMPTY [`NatRuntime`](dig_nat::NatRuntime), so the dial composes the **Direct** tier only — suitable for
    /// a fully-reachable node or a test. A NAT'd node that must reach non-Direct providers over
    /// hole-punch/relay MUST use [`new_with_runtime`](Self::new_with_runtime) with its live runtime.
    pub fn new(
        node: std::sync::Arc<dig_nat::NodeCert>,
        config: dig_nat::NatConfig,
        network_id: impl Into<String>,
    ) -> Self {
        Self::new_with_runtime(
            node,
            config,
            network_id,
            Arc::new(dig_nat::NatRuntime::default()),
        )
    }

    /// Build a transport that dials over the **FULL** NAT-traversal ladder using the live handles in
    /// `runtime` (#1305). Mirrors the node's DHT-side [`dig_nat::connect_with_runtime`] path so the
    /// content-fetch leg reaches providers via hole-punch + relay, not just direct. dig-node passes the
    /// SAME shared [`NatRuntime`](dig_nat::NatRuntime) it built for its DHT transport.
    pub fn new_with_runtime(
        node: std::sync::Arc<dig_nat::NodeCert>,
        config: dig_nat::NatConfig,
        network_id: impl Into<String>,
        runtime: Arc<dig_nat::NatRuntime>,
    ) -> Self {
        NatRangeTransport {
            node,
            config,
            network_id: network_id.into(),
            runtime,
            pool: tokio::sync::Mutex::new(HashMap::new()),
        }
    }

    /// Every way to reach `provider`, in dial order: each resolvable candidate address as a
    /// [`dig_nat::PeerTarget`] — **IPv6 first, then IPv4** (§5.2) — followed by a relay-only target
    /// reached purely by identity.
    ///
    /// Each entry carries the candidate's rendered address so a failed dial can name WHICH address it
    /// tried. Unresolvable candidates are logged and skipped rather than aborting the provider: one
    /// malformed v6 candidate must never hide a working v4 one (#836).
    pub fn provider_dial_targets(
        &self,
        provider: &ProviderRecord,
    ) -> Result<Vec<(String, dig_nat::PeerTarget)>, DownloadError> {
        let peer_id = provider.provider_peer_id().ok_or_else(|| {
            DownloadError::transport(&provider.provider_peer_id, "malformed provider peer_id")
        })?;
        let mut targets = Vec::new();
        for candidate in crate::addr::dial_candidates(provider) {
            match crate::addr::candidate_socket(candidate) {
                Ok(socket) => targets.push((
                    socket.to_string(),
                    dig_nat::PeerTarget::with_addr(peer_id, socket, self.network_id.clone()),
                )),
                Err(e) => tracing::warn!(
                    peer = %crate::error::hex64_or_sentinel(&provider.provider_peer_id, "peer-id"),
                    candidate = %crate::addr::display(candidate),
                    error = %e,
                    "skipping unusable provider candidate address"
                ),
            }
        }
        targets.push((
            "relay-only".to_string(),
            dig_nat::PeerTarget::relay_only(peer_id, self.network_id.clone()),
        ));
        Ok(targets)
    }

    /// Build a [`dig_nat::PeerTarget`] from a provider record: its `peer_id` + the most-direct
    /// dialable candidate address (falling back to relay-only reachability by identity).
    ///
    /// This is the FIRST of [`provider_dial_targets`](Self::provider_dial_targets); dialing uses the
    /// full ordered list so a failing candidate falls through to the next.
    pub fn provider_to_target(
        &self,
        provider: &ProviderRecord,
    ) -> Result<dig_nat::PeerTarget, DownloadError> {
        let (_, target) = self
            .provider_dial_targets(provider)?
            .into_iter()
            .next()
            .expect("dial targets always include the relay-only fallback");
        Ok(target)
    }

    /// Connect to a provider as a [`DigPeer`] (fresh `peer_id`-pinned mTLS connection over the FULL
    /// NAT-traversal ladder). Composes exactly the tiers whose live handles this transport's
    /// [`NatRuntime`](dig_nat::NatRuntime) carries — Direct always, plus hole-punch/relay when the node
    /// injected them (#1305). The [`PeerTarget`](dig_nat::PeerTarget) carries the provider's `peer_id`,
    /// which [`DigPeer::connect_with_runtime`] pins so a different CA-valid peer cannot impersonate the
    /// intended provider (#1283).
    ///
    /// Every candidate address is tried in order (IPv6 first, then IPv4, then relay-only, §5.2) and
    /// each failure is logged with the address that produced it, so an unreachable v6 candidate falls
    /// through to a working v4 one instead of failing the whole holder (#836).
    async fn connect(&self, provider: &ProviderRecord) -> Result<DigPeer, DownloadError> {
        let mut last_error = None;
        for (addr, target) in self.provider_dial_targets(provider)? {
            match DigPeer::connect_with_runtime(&target, &self.node, &self.config, &self.runtime)
                .await
            {
                Ok(peer) => return Ok(peer),
                Err(e) => {
                    tracing::debug!(
                        peer = %crate::error::hex64_or_sentinel(&provider.provider_peer_id, "peer-id"),
                        candidate = %addr,
                        error = %e,
                        "provider dial candidate failed; trying the next address"
                    );
                    last_error = Some(format!("dial {addr}: {e}"));
                }
            }
        }
        Err(DownloadError::transport(
            &provider.provider_peer_id,
            last_error.unwrap_or_else(|| "no dialable candidate address".to_string()),
        ))
    }

    /// Get the pooled connection for `provider`, dialing (and caching) a fresh one if none is pooled.
    /// Reuses the existing mTLS session across requests; a broken connection is evicted via
    /// [`evict`](Self::evict) so the next call re-dials.
    async fn pooled_conn(&self, provider: &ProviderRecord) -> Result<PooledConn, DownloadError> {
        let key = provider.provider_peer_id.clone();
        if let Some(conn) = self.pool.lock().await.get(&key).cloned() {
            return Ok(conn);
        }
        // Dial OUTSIDE the pool lock (a handshake can be slow); race-insert, reusing a connection a
        // concurrent caller may have inserted first so we never hold two sessions to one peer.
        let fresh = Arc::new(tokio::sync::Mutex::new(self.connect(provider).await?));
        let mut pool = self.pool.lock().await;
        Ok(pool.entry(key).or_insert(fresh).clone())
    }

    /// Drop `provider`'s pooled connection so the next request re-dials (called after a stream error).
    async fn evict(&self, provider: &ProviderRecord) {
        self.pool.lock().await.remove(&provider.provider_peer_id);
    }
}

#[async_trait]
impl RangeTransport for NatRangeTransport {
    async fn query_availability(
        &self,
        provider: &ProviderRecord,
        items: Vec<AvailabilityItem>,
    ) -> Result<AvailabilityResponse, DownloadError> {
        let conn = self.pooled_conn(provider).await?;
        let res = {
            let mut guard = conn.lock().await;
            guard.get_availability(items).await
        };
        match res {
            Ok(resp) => Ok(resp),
            Err(e) => {
                // The pooled session is suspect — drop it so the next request re-dials.
                self.evict(provider).await;
                Err(DownloadError::transport(&provider.provider_peer_id, e))
            }
        }
    }

    async fn fetch_range(
        &self,
        provider: &ProviderRecord,
        req: &RangeRequest,
    ) -> Result<FetchedRange, DownloadError> {
        let conn = self.pooled_conn(provider).await?;
        // Hold the per-peer lock ONLY to open the (owned) range stream over the reused mTLS session;
        // release it before reading frames so concurrent ranges to the same peer stream in parallel.
        let stream = {
            let mut guard = conn.lock().await;
            guard.fetch_range(req).await
        };
        let mut stream = match stream {
            Ok(s) => s,
            Err(e) => {
                self.evict(provider).await;
                return Err(DownloadError::transport(&provider.provider_peer_id, e));
            }
        };
        let (bytes, meta) = assemble_range_stream(&mut stream, req.length)
            .await
            // ATTRIBUTE the reassembly error to this provider rather than WRAPPING it in a fresh
            // `Transport`. Wrapping flattened every typed variant the reassembler raises deliberately —
            // including `PagedPrologueUnsupported` — so a caller could never observe one, and the
            // `is_recoverable` arm for it was unreachable.
            .map_err(|e| e.attributed_to(&provider.provider_peer_id))?;
        // Drain any trailer so the mux stream closes cleanly — BOUNDED, so a peer that keeps the
        // stream open and streams filler after the last frame cannot exhaust our memory (MEDIUM
        // #179). Never read_to_end into an unbounded Vec.
        let _ = drain_trailer_bounded(&mut stream, MAX_TRAILER_DRAIN).await;
        Ok(FetchedRange {
            request_offset: req.offset,
            bytes,
            meta,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use dig_dht::{CandidateAddr, ProviderRecord};
    use dig_nat::PeerId;

    /// The generation root every conforming fixture frame is stamped with (64-hex, as the wire
    /// requires). Identity travels on EVERY frame since dig-nat 0.13, so it is named once here rather
    /// than re-spelled per fixture.
    fn test_root() -> String {
        "aa".repeat(32)
    }

    /// Encode a fixture frame, surfacing the dig-nat framing-ceiling refusal as a test failure.
    ///
    /// `RangeFrame::encode` became FALLIBLE in 0.13 (#1640): the encode side now refuses a frame a
    /// conforming decoder would have to reject. A fixture that trips it is a fixture bug, so the
    /// panic names the ceiling instead of silently disappearing into a `Result` nobody inspects.
    fn encode(frame: &RangeFrame) -> Vec<u8> {
        frame
            .encode()
            .expect("fixture frame must be within the dig-nat framing ceilings")
    }

    fn provider(peer: u8, host: &str, port: u16) -> ProviderRecord {
        ProviderRecord::new(
            &dig_dht::Key::from_bytes([0xAB; 32]),
            &PeerId::from_bytes([peer; 32]),
            vec![CandidateAddr::direct(host, port)],
            u64::MAX,
        )
    }

    #[test]
    fn provider_to_target_uses_direct_address() {
        let t = NatRangeTransport::new(
            fake_node_cert(),
            dig_nat::NatConfig::default(),
            "DIG_MAINNET",
        );
        let p = provider(1, "203.0.113.7", 9444);
        let target = t.provider_to_target(&p).unwrap();
        assert_eq!(
            target.direct_addr().unwrap().to_string(),
            "203.0.113.7:9444"
        );
        assert_eq!(target.network_id, "DIG_MAINNET");
    }

    #[test]
    fn new_with_runtime_builds_a_full_ladder_transport() {
        // #1305: the fetch leg must be constructible with a live NatRuntime (the same handle carrier
        // the node's DHT dial uses) so hole-punch/relay tiers compose. The dial itself needs real
        // sockets, so here we assert the runtime-injecting constructor yields a working transport
        // whose pure provider→target mapping is identical to the Direct-only `new`.
        let runtime = std::sync::Arc::new(dig_nat::NatRuntime::default());
        let t = NatRangeTransport::new_with_runtime(
            fake_node_cert(),
            dig_nat::NatConfig::default(),
            "DIG_MAINNET",
            runtime,
        );
        let p = provider(1, "203.0.113.7", 9444);
        let target = t.provider_to_target(&p).unwrap();
        assert_eq!(
            target.direct_addr().unwrap().to_string(),
            "203.0.113.7:9444"
        );
        assert_eq!(target.network_id, "DIG_MAINNET");
    }

    #[test]
    fn provider_to_target_accepts_v4_mapped_v6_host() {
        // #836 regression: the e2e read leg died with "addr: invalid socket address syntax" because
        // the host+port were STRING-formatted before parsing, and an IPv6 literal needs brackets.
        let t = NatRangeTransport::new(
            fake_node_cert(),
            dig_nat::NatConfig::default(),
            "DIG_MAINNET",
        );
        let p = provider(1, "::ffff:172.31.79.22", 9444);
        let target = t
            .provider_to_target(&p)
            .expect("v4-mapped v6 host must resolve");
        assert_eq!(
            target.direct_addr().unwrap(),
            std::net::SocketAddr::new("::ffff:172.31.79.22".parse().unwrap(), 9444)
        );
    }

    #[test]
    fn provider_to_target_accepts_plain_v6_host() {
        let t = NatRangeTransport::new(
            fake_node_cert(),
            dig_nat::NatConfig::default(),
            "DIG_MAINNET",
        );
        let p = provider(1, "2001:db8::1", 9444);
        let target = t.provider_to_target(&p).expect("v6 host must resolve");
        assert_eq!(
            target.direct_addr().unwrap(),
            std::net::SocketAddr::new("2001:db8::1".parse().unwrap(), 9444)
        );
    }

    #[test]
    fn unusable_first_candidate_falls_through_to_the_ipv4_one() {
        // #836 / §5.2: IPv6-first with IPv4 FALLBACK. A provider whose leading candidate is unusable
        // must still be dialed on its valid v4 candidate — previously the record's FIRST address was
        // the only one considered, so one bad candidate condemned the holder.
        let t = NatRangeTransport::new(
            fake_node_cert(),
            dig_nat::NatConfig::default(),
            "DIG_MAINNET",
        );
        let p = ProviderRecord::new(
            &dig_dht::Key::from_bytes([0xAB; 32]),
            &PeerId::from_bytes([3; 32]),
            vec![
                CandidateAddr::direct("not-an-ip-literal", 9444),
                CandidateAddr::direct("10.0.0.1", 9444),
            ],
            u64::MAX,
        );
        let target = t
            .provider_to_target(&p)
            .expect("the v4 candidate is dialable");
        assert_eq!(
            target.direct_addr().unwrap(),
            "10.0.0.1:9444".parse::<std::net::SocketAddr>().unwrap()
        );
    }

    #[test]
    fn dial_targets_order_v6_then_v4_then_relay() {
        let t = NatRangeTransport::new(
            fake_node_cert(),
            dig_nat::NatConfig::default(),
            "DIG_MAINNET",
        );
        let p = ProviderRecord::new(
            &dig_dht::Key::from_bytes([0xAB; 32]),
            &PeerId::from_bytes([4; 32]),
            vec![
                CandidateAddr::direct("172.31.79.22", 9444),
                CandidateAddr::direct("::ffff:172.31.79.22", 9444),
            ],
            u64::MAX,
        );
        let addrs: Vec<String> = t
            .provider_dial_targets(&p)
            .unwrap()
            .into_iter()
            .map(|(addr, _)| addr)
            .collect();
        assert_eq!(
            addrs,
            vec![
                "[::ffff:172.31.79.22]:9444",
                "172.31.79.22:9444",
                "relay-only"
            ]
        );
    }

    #[tokio::test]
    async fn connect_tries_every_candidate_before_failing() {
        // Both candidates are closed loopback ports: the dial must walk the whole list (v6 then v4
        // then relay-only) and report the LAST attempt, proving no early give-up.
        let t = NatRangeTransport::new(
            fake_node_cert(),
            dig_nat::NatConfig::default(),
            "DIG_MAINNET",
        );
        let p = ProviderRecord::new(
            &dig_dht::Key::from_bytes([0xAB; 32]),
            &PeerId::from_bytes([5; 32]),
            vec![
                CandidateAddr::direct("::1", 1),
                CandidateAddr::direct("127.0.0.1", 1),
            ],
            u64::MAX,
        );
        let err = t.connect(&p).await.expect_err("no listener is up");
        let reason = err.to_string();
        assert!(
            reason.contains("relay-only"),
            "the last attempt must be named: {reason}"
        );
    }

    #[test]
    fn provider_to_target_relay_only_without_address() {
        let t = NatRangeTransport::new(
            fake_node_cert(),
            dig_nat::NatConfig::default(),
            "DIG_MAINNET",
        );
        let p = ProviderRecord::new(
            &dig_dht::Key::from_bytes([0xAB; 32]),
            &PeerId::from_bytes([2; 32]),
            vec![CandidateAddr::relay_marker()],
            u64::MAX,
        );
        let target = t.provider_to_target(&p).unwrap();
        assert!(target.direct_addr().is_none());
    }

    #[tokio::test]
    async fn assemble_reassembles_ordered_frames() {
        // Two frames tiling a 6-byte range; first frame carries the metadata.
        let f0 = RangeFrame::data(0, b"ABC".to_vec())
            .with_identity(test_root(), 6, 2)
            .with_chunk_lens_page(0, vec![3, 3])
            .with_chunk_index(0)
            .with_inclusion_proof("proof");
        // The continuation frame starts on a chunk boundary, so it RESTATES the fixed-size identity
        // set + its own `chunk_index` and omits the once-per-stream prologue.
        let f1 = RangeFrame::data(3, b"DEF".to_vec())
            .with_complete(true)
            .with_identity(test_root(), 6, 2)
            .with_chunk_index(1);
        let mut wire = encode(&f0);
        wire.extend_from_slice(&encode(&f1));
        let mut cur = std::io::Cursor::new(wire);
        let (bytes, meta) = assemble_range_stream(&mut cur, 6).await.unwrap();
        assert_eq!(bytes, b"ABCDEF");
        assert_eq!(meta.total_length, Some(6));
        assert_eq!(meta.chunk_lens, Some(vec![3, 3]));
        assert_eq!(meta.chunk_index, Some(0));
        assert_eq!(meta.root, Some("aa".repeat(32)));
        assert_eq!(meta.inclusion_proof, Some("proof".into()));
    }

    /// A frame that STARTS beyond the requested window is a real protocol violation (its bytes can
    /// never belong to the range) and stays an error.
    #[tokio::test]
    async fn assemble_rejects_frame_starting_beyond_window() {
        // Deliberately IDENTITY-FREE: the frame is refused on its offset alone, before any metadata
        // is consulted, so attaching identity here would only obscure which field the rejection reads.
        let f = RangeFrame::data(8, vec![0u8; 4]).with_complete(true);
        let mut cur = std::io::Cursor::new(encode(&f));
        let err = assemble_range_stream(&mut cur, 5).await;
        assert!(matches!(err, Err(DownloadError::Transport { .. })));
    }

    /// The #836 metadata probe: `establish_commitment` asks for `length = 1` purely to obtain the
    /// first-frame metadata, and a chunk-granular server answers with a WHOLE chunk. The assembler
    /// must clip to the requested window and keep the metadata — erroring here discarded every
    /// holder and turned a healthy read into a 404.
    #[tokio::test]
    async fn assemble_clips_chunk_granular_frame_to_one_byte_probe() {
        let chunk = vec![0x5Au8; 4096];
        let f = RangeFrame::data(0, chunk)
            .with_complete(true)
            .with_identity(test_root(), 1_048_576, 256)
            .with_chunk_lens_page(0, vec![4096; 256])
            .with_chunk_index(0)
            .with_inclusion_proof("proof");
        let mut cur = std::io::Cursor::new(encode(&f));
        let (bytes, meta) = assemble_range_stream(&mut cur, 1).await.unwrap();
        assert_eq!(
            bytes,
            vec![0x5Au8],
            "clipped to exactly the requested window"
        );
        assert_eq!(meta.total_length, Some(1_048_576));
        assert_eq!(meta.chunk_lens, Some(vec![4096; 256]));
        assert_eq!(meta.chunk_index, Some(0));
        assert_eq!(meta.root, Some("aa".repeat(32)));
        assert_eq!(meta.inclusion_proof, Some("proof".into()));
    }

    /// Only the OVERSHOOTING tail is clipped: every earlier frame's bytes survive, in order.
    #[tokio::test]
    async fn assemble_clips_only_the_overshooting_last_frame() {
        let f0 = RangeFrame::data(0, b"ABC".to_vec())
            .with_identity(test_root(), 9, 2)
            .with_chunk_lens_page(0, vec![3, 6])
            .with_chunk_index(0);
        // Chunk-aligned continuation: identity restated, prologue not repeated.
        let f1 = RangeFrame::data(3, b"DEFGHI".to_vec())
            .with_complete(true)
            .with_identity(test_root(), 9, 2)
            .with_chunk_index(1);
        let mut wire = encode(&f0);
        wire.extend_from_slice(&encode(&f1));
        let mut cur = std::io::Cursor::new(wire);
        let (bytes, meta) = assemble_range_stream(&mut cur, 5).await.unwrap();
        assert_eq!(bytes, b"ABCDE");
        assert_eq!(meta.total_length, Some(9));
    }

    /// Once the requested window is full the assembler stops reading, even without a `complete`
    /// frame — it never buffers past `max_len`.
    #[tokio::test]
    async fn assemble_stops_once_the_window_is_full() {
        let f0 = RangeFrame::data(0, b"WXYZ".to_vec())
            .with_identity(test_root(), 8, 2)
            .with_chunk_lens_page(0, vec![4, 4])
            .with_chunk_index(0);
        // Chunk-aligned continuation: identity restated, prologue not repeated.
        let f1 = RangeFrame::data(4, b"nope".to_vec())
            .with_complete(true)
            .with_identity(test_root(), 8, 2)
            .with_chunk_index(1);
        let mut wire = encode(&f0);
        wire.extend_from_slice(&encode(&f1));
        let mut cur = std::io::Cursor::new(wire);
        let (bytes, _) = assemble_range_stream(&mut cur, 4).await.unwrap();
        assert_eq!(bytes, b"WXYZ");
    }

    #[tokio::test]
    async fn drain_trailer_is_bounded_by_cap() {
        // A "peer" that streams far more trailer than the cap: the drain must stop at the cap, never
        // buffering the whole thing (MEDIUM #179 — no unbounded read_to_end).
        let flood = vec![0u8; 1_000_000];
        let mut cur = std::io::Cursor::new(flood);
        let drained = drain_trailer_bounded(&mut cur, 64 * 1024).await;
        assert_eq!(drained, 64 * 1024, "drain must stop exactly at the cap");
        // The cursor still has bytes left (we did NOT read to end).
        assert!((cur.position() as usize) < 1_000_000);
    }

    #[tokio::test]
    async fn drain_trailer_stops_at_eof_below_cap() {
        // A well-behaved peer with a small (or empty) trailer: drain returns the actual count and
        // stops at EOF without waiting for the cap.
        let mut cur = std::io::Cursor::new(vec![0u8; 100]);
        assert_eq!(drain_trailer_bounded(&mut cur, 64 * 1024).await, 100);
        let mut empty = std::io::Cursor::new(Vec::<u8>::new());
        assert_eq!(drain_trailer_bounded(&mut empty, 64 * 1024).await, 0);
    }

    #[tokio::test]
    async fn assemble_stops_on_clean_eof() {
        // A single non-complete frame followed by EOF still yields the bytes.
        let f = RangeFrame::data(0, b"hi".to_vec())
            .with_identity(test_root(), 2, 1)
            .with_chunk_lens_page(0, vec![2])
            .with_chunk_index(0);
        let mut cur = std::io::Cursor::new(encode(&f));
        let (bytes, meta) = assemble_range_stream(&mut cur, 2).await.unwrap();
        assert_eq!(bytes, b"hi");
        assert_eq!(meta.total_length, Some(2));
    }

    /// #1640, from BOTH sides of the bound. A payload at exactly [`MAX_RANGE_FRAME_PAYLOAD`] is legal
    /// and must survive the real encode → decode → assemble path; one byte over must be REFUSED at the
    /// encode site rather than emitted for a decoder that is required to reject it.
    ///
    /// The fixture size is taken FROM the protocol constant, deliberately. #1640 hid for as long as it
    /// did because every fixture that touched this path was far below the ceiling — an 8-byte in-process
    /// mock and 20 KB / 27 KB e2e content — and a fixture that cannot exceed a bound can never detect an
    /// unbounded encoder. Testing only the at-bound case would be the same mistake in miniature: it
    /// confirms the ceiling is reachable without showing that anything stops one byte past it.
    ///
    /// Scope of the proof, stated honestly: the over-bound half is load-bearing against dig-nat 0.11,
    /// where `encode` returned a bare `Vec<u8>` and no ceiling existed at all. It does NOT distinguish
    /// 0.12 from 0.13 — the payload ceiling landed in 0.12.0 — so `dependency_tree.rs` carries the
    /// assertion that the resolved line is not a pre-0.12 one.
    #[tokio::test]
    async fn a_payload_at_the_ceiling_round_trips_and_one_byte_over_is_refused() {
        let ceiling = dig_nat::MAX_RANGE_FRAME_PAYLOAD;
        let at_ceiling = vec![0x7Eu8; ceiling];

        let f = RangeFrame::data(0, at_ceiling.clone())
            .with_complete(true)
            .with_identity(test_root(), ceiling as u64, 1)
            .with_chunk_lens_page(0, vec![ceiling as u64])
            .with_chunk_index(0);
        let wire = f
            .encode()
            .expect("a payload AT MAX_RANGE_FRAME_PAYLOAD is conforming and must encode");

        let mut cur = std::io::Cursor::new(wire);
        let (bytes, meta) = assemble_range_stream(&mut cur, ceiling as u64)
            .await
            .expect("a ceiling-sized frame decodes and assembles");
        assert_eq!(
            bytes, at_ceiling,
            "every byte of a ceiling-sized window survives the round trip"
        );
        assert_eq!(meta.total_length, Some(ceiling as u64));
        assert_eq!(meta.chunk_index, Some(0));

        let over = RangeFrame::data(0, vec![0x7Eu8; ceiling + 1]).with_complete(true);
        let err = over
            .encode()
            .expect_err("one byte past the ceiling has no conforming frame and must be refused");
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
    }

    /// A holder that declares a paged `chunk_count` but sends only its FIRST page then marks the stream
    /// complete has delivered an INCOMPLETE layout. The reader refuses it fail-closed at the assembler —
    /// it never surfaces the lone page as if it were the whole array — because `chunk_lens` is a DECRYPT
    /// input: a truncated array is not a degraded layout, it is one that decrypts every chunk to garbage.
    ///
    /// The fixture's `chunk_count` sits above [`MAX_CHUNK_LENS_PER_FRAME`], the sender's own paging
    /// threshold, so this is the genuinely-paged shape rather than a large-looking array that still fits
    /// one frame.
    #[tokio::test]
    async fn a_single_page_of_a_paged_prologue_is_refused_not_surfaced_as_the_whole_array() {
        let chunk_count = dig_nat::MAX_CHUNK_LENS_PER_FRAME + 952;
        let chunk_lens: Vec<u64> = (0..chunk_count).map(|i| 64 + (i as u64 % 7)).collect();
        let total_length: u64 = chunk_lens.iter().sum();
        let page0 = chunk_lens[..dig_nat::MAX_CHUNK_LENS_PER_FRAME].to_vec();

        let f = RangeFrame::data(0, b"AB".to_vec())
            .with_complete(true)
            .with_identity(test_root(), total_length, chunk_count as u64)
            .with_chunk_lens_page(0, page0)
            .with_chunk_index(0);
        let wire = f.encode().expect(
            "a first page of MAX_CHUNK_LENS_PER_FRAME entries is within the framing ceiling",
        );

        let mut cur = std::io::Cursor::new(wire);
        let err = assemble_range_stream(&mut cur, 2)
            .await
            .expect_err("a lone page of a paged prologue is not a complete layout");
        assert!(
            matches!(err, DownloadError::PagedPrologueUnsupported { .. }),
            "an incomplete prologue is refused fail-closed, never adopted; got {err:?}"
        );
        assert!(
            err.is_recoverable(),
            "the holder is skipped, not the download"
        );
    }

    #[test]
    fn source_tracker_backoff_and_recovery() {
        let mut t = SourceTracker::new(Duration::from_millis(100), Duration::from_secs(10));
        let now = Instant::now();
        assert!(t.is_available("p", now));
        t.record_failure("p", now);
        assert!(!t.is_available("p", now)); // inside backoff
        assert_eq!(t.failures("p"), 1);
        // After the backoff window it is schedulable again.
        assert!(t.is_available("p", now + Duration::from_millis(101)));
        // Success clears failures + backoff and counts a served range.
        t.record_success("p");
        assert!(t.is_available("p", now));
        assert_eq!(t.failures("p"), 0);
        assert_eq!(t.served("p"), 1);
    }

    #[test]
    fn source_tracker_backoff_is_exponential_and_capped() {
        let mut t = SourceTracker::new(Duration::from_millis(100), Duration::from_millis(250));
        let now = Instant::now();
        t.record_failure("p", now); // 100ms
        assert!(t.is_available("p", now + Duration::from_millis(150)));
        t.record_failure("p", now); // 200ms
        assert!(!t.is_available("p", now + Duration::from_millis(150)));
        t.record_failure("p", now); // 400ms → capped to 250ms
        assert!(t.is_available("p", now + Duration::from_millis(260)));
    }

    /// A real (but disposable) CA-signed [`dig_nat::NodeCert`] for the pure helpers under test — they
    /// never dial, so any validly-minted cert works. `NodeCert` has no public fields (only
    /// `generate_signed`/`load_or_generate`/`from_pem`), so it is minted from a BLS secret key
    /// deterministically derived from a fixed label (never a literal keypair — keeps CodeQL's
    /// hard-coded-crypto-value scan happy, matches dig-tls's own test convention).
    fn fake_node_cert() -> std::sync::Arc<dig_nat::NodeCert> {
        use sha2::{Digest, Sha256};
        let seed: [u8; 32] = Sha256::digest(b"dig-download/tests/fake-node-cert").into();
        let bls_sk = dig_tls::bls::SecretKey::from_seed(&seed);
        std::sync::Arc::new(dig_nat::NodeCert::generate_signed(&bls_sk).unwrap())
    }

    /// #1608 — the range assembly buffer is sized by a peer-DECLARED length, so its growth must be
    /// FALLIBLE: `Vec::resize` aborts the process through the uncatchable `handle_alloc_error`, which a
    /// peer must never be able to trigger. A frame that is SPARSE in a huge window (a high `offset`,
    /// a few payload bytes) reaches that path from ONE small frame.
    ///
    /// An ~18 EiB reservation fails on every host without touching a page, so this is deterministic
    /// rather than dependent on the CI host's memory or overcommit policy.
    #[tokio::test]
    async fn an_unsatisfiable_assembly_buffer_is_a_recoverable_error_not_an_abort() {
        // Deliberately IDENTITY-FREE: the reservation is sized from `offset + bytes.len()` against
        // `max_len`, so no metadata field participates. Stating identity here would suggest the
        // refusal depends on a declared length it does not read.
        let f = RangeFrame::data(u64::MAX - 4, vec![0xAB; 2]);
        let mut cur = std::io::Cursor::new(encode(&f));
        let err = assemble_range_stream(&mut cur, u64::MAX)
            .await
            .expect_err("an unsatisfiable window allocation is refused, not fatal");
        assert!(
            err.is_recoverable(),
            "and it is RECOVERABLE, so the scheduler re-fetches the range elsewhere: {err}"
        );
    }
    // ---- mid-stream identity revision (Obligation 2 of #1668) ----------------------------------
    //
    // The truthful CONTROL for this whole group is `assemble_reassembles_ordered_frames` above: a
    // conforming two-frame stream that RESTATES the identity set on its continuation frame and is
    // accepted. Without it a guard that rejected every multi-frame stream would satisfy every
    // rejection test below while breaking the reader outright.

    /// A conforming two-frame stream, and the ONE later-frame field each test varies from it.
    ///
    /// Built as a pair so every rejection differs from an ACCEPTED stream by exactly one field. A
    /// fixture assembled independently per test drifts, and then a rejection can no longer be
    /// attributed to the field under test.
    fn identity_pair() -> (RangeFrame, RangeFrame) {
        let first = RangeFrame::data(0, b"ABC".to_vec())
            .with_identity(test_root(), 6, 2)
            .with_chunk_lens_page(0, vec![3, 3])
            .with_chunk_index(0)
            .with_inclusion_proof("proof");
        let second = RangeFrame::data(3, b"DEF".to_vec())
            .with_complete(true)
            .with_identity(test_root(), 6, 2)
            .with_chunk_index(1);
        (first, second)
    }

    /// Assemble a two-frame stream and return the rejection reason, panicking if it was ACCEPTED.
    async fn reject_reason(first: &RangeFrame, second: &RangeFrame) -> String {
        let mut wire = encode(first);
        wire.extend_from_slice(&encode(second));
        let mut cur = std::io::Cursor::new(wire);
        match assemble_range_stream(&mut cur, 6).await {
            Err(DownloadError::Transport { reason, .. }) => reason,
            other => panic!("a revised identity must be rejected; got {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_later_frame_revising_the_root_is_rejected() {
        let (first, mut second) = identity_pair();
        second.root = Some("bb".repeat(32));
        let reason = reject_reason(&first, &second).await;
        assert!(
            reason.contains("root changed mid-stream"),
            "must name the revised field, not fail generically; got {reason}"
        );
    }

    #[tokio::test]
    async fn a_later_frame_revising_the_total_length_is_rejected() {
        let (first, mut second) = identity_pair();
        second.total_length = Some(7);
        let reason = reject_reason(&first, &second).await;
        assert!(
            reason.contains("total_length changed mid-stream"),
            "got {reason}"
        );
    }

    /// A revised `chunk_count` is the case NOTHING beneath this reader catches: dig-nat's
    /// `ChunkLensAssembler` is constructed with one count and never sees a later frame's declaration,
    /// so if this check is absent the revision is simply invisible.
    #[tokio::test]
    async fn a_later_frame_revising_the_chunk_count_is_rejected() {
        let (first, mut second) = identity_pair();
        second.chunk_count = Some(3);
        let reason = reject_reason(&first, &second).await;
        assert!(
            reason.contains("chunk_count changed mid-stream"),
            "got {reason}"
        );
    }

    /// Revising DOWNWARD, to pin the guard from BOTH sides.
    ///
    /// A check written as "the count may not grow" passes every test above and is bypassed by this one.
    /// The property is that the declared shape may not CHANGE, in either direction.
    #[tokio::test]
    async fn a_later_frame_revising_the_chunk_count_downward_is_also_rejected() {
        let (first, mut second) = identity_pair();
        second.chunk_count = Some(1);
        let reason = reject_reason(&first, &second).await;
        assert!(
            reason.contains("chunk_count changed mid-stream"),
            "a revision is a revision in EITHER direction; got {reason}"
        );
    }

    /// An identity field the first frame left UNSTATED, arriving later.
    ///
    /// The commitment binds to the first frame, so a holder that withholds a value from that frame and
    /// supplies it afterwards has revised the declaration the reader actually bound to. A guard written
    /// only as "the values must match" accepts this, because there is nothing to compare against.
    #[tokio::test]
    async fn an_identity_field_appearing_only_on_a_later_frame_is_rejected() {
        let (mut first, second) = identity_pair();
        first.chunk_count = None;
        let reason = reject_reason(&first, &second).await;
        assert!(
            reason.contains("appears only on a later frame"),
            "got {reason}"
        );
    }

    /// A `chunk_lens` page RESTATING entries an earlier page already filled — rejected whether or not it
    /// agrees.
    ///
    /// The guard is stated over `chunk_lens_offset`, not over "is this the first frame". The reader never
    /// compares a restated page, it refuses the restatement, so "frame 1 said A, frame 5 said B" cannot be
    /// expressed at all rather than merely being unpersuasive. The page HERE is byte-identical to the
    /// first frame's, so agreement cannot be what saves it.
    #[tokio::test]
    async fn a_later_frame_restating_an_identical_chunk_lens_page_is_rejected() {
        let (first, second) = identity_pair();
        let second = second.with_chunk_lens_page(0, vec![3, 3]);
        let reason = reject_reason(&first, &second).await;
        assert!(
            reason.contains("must ADVANCE, never restate"),
            "an identical restatement is still a restatement; got {reason}"
        );
    }

    /// An UNSTAMPED later page. Absent `chunk_lens_offset` means "begins at 0" per the wire contract, so
    /// this restates ground the first frame's page already covered.
    ///
    /// Separate from the test above because the two reach the rule by different routes: that one states an
    /// offset, this one omits it. A guard that only compared a PRESENT offset would let this through.
    #[tokio::test]
    async fn a_later_frame_carrying_an_unstamped_chunk_lens_page_is_rejected() {
        let (first, mut second) = identity_pair();
        second.chunk_lens = Some(vec![3, 3]);
        second.chunk_lens_offset = None;
        let reason = reject_reason(&first, &second).await;
        assert!(
            reason.contains("offset 0 re-covers entries below 2"),
            "an unstamped page begins at 0, which is already filled; got {reason}"
        );
    }

    /// A page that OVERLAPS rather than exactly repeating — the off-by-one variant of the same class.
    ///
    /// A guard written as "reject a page at an offset already seen" passes both tests above and is
    /// bypassed here: offset 1 was never itself the start of a page, yet entry 1 is already filled. The
    /// rule compares against the frontier, so partial overlap is caught the same way a duplicate is.
    #[tokio::test]
    async fn a_later_frame_whose_chunk_lens_page_partially_overlaps_is_rejected() {
        let (first, second) = identity_pair();
        let second = second.with_chunk_lens_page(1, vec![9]);
        let reason = reject_reason(&first, &second).await;
        assert!(
            reason.contains("offset 1 re-covers entries below 2"),
            "a partially overlapping page is a restatement too; got {reason}"
        );
    }

    // ---- paged-prologue reassembly (#1668) ------------------------------------------------------

    /// The `chunk_lens` entries of a `count`-entry resource, DISTINCT per index (`64 + i%7`).
    ///
    /// A uniform `vec![64; count]` would hide a page placed at the wrong offset or a page truncated by
    /// one entry — every slot looks identical — so the ceiling test that #1640 taught us to write needs
    /// entries that differ, and a reassembled array that equals this one proves each page landed exactly.
    fn distinct_lens(count: usize) -> Vec<u64> {
        (0..count).map(|i| 64 + (i % 7) as u64).collect()
    }

    /// The three frames of a paged prologue for a resource ABOVE the single-frame ceiling.
    ///
    /// `chunk_count = 2*2048 + 1 = 4097` needs exactly three pages — [0,2048), [2048,4096), [4096,4097)
    /// — so it exercises full pages AND a short final page, and sits above `MAX_CHUNK_LENS_PER_FRAME`
    /// where a reader that snapshots the layout from frame 1 alone could never read it. The first frame
    /// carries the only data bytes (`b"ABC"`); the two later frames are prologue-only (zero data), which
    /// is what makes the termination guard's "an accepted page is progress" exemption load-bearing.
    fn paged_prologue_frames() -> (Vec<u64>, RangeFrame, RangeFrame, RangeFrame) {
        let full = distinct_lens(4097);
        let total: u64 = full.iter().sum();
        let first = RangeFrame::data(0, b"ABC".to_vec())
            .with_identity(test_root(), total, 4097)
            .with_chunk_lens_page(0, full[0..2048].to_vec())
            .with_chunk_index(0)
            .with_inclusion_proof("proof");
        let second = RangeFrame::data(0, Vec::new())
            .with_identity(test_root(), total, 4097)
            .with_chunk_lens_page(2048, full[2048..4096].to_vec());
        let third = RangeFrame::data(0, Vec::new())
            .with_complete(true)
            .with_identity(test_root(), total, 4097)
            .with_chunk_lens_page(4096, full[4096..4097].to_vec());
        (full, first, second, third)
    }

    /// A paged prologue spanning THREE frames is reassembled into one array, and adopted only once the
    /// last page has landed. This is the capability #1668 adds: a resource above the single-frame layout
    /// ceiling now reads end-to-end instead of being refused.
    #[tokio::test]
    async fn a_paged_prologue_is_reassembled_into_the_full_chunk_lens_array() {
        let (full, first, second, third) = paged_prologue_frames();
        let mut wire = encode(&first);
        wire.extend_from_slice(&encode(&second));
        wire.extend_from_slice(&encode(&third));
        let mut cur = std::io::Cursor::new(wire);

        // A one-byte window (the metadata probe) MUST keep reading until every prologue page lands,
        // even though the byte window fills on the first frame.
        let (bytes, meta) = assemble_range_stream(&mut cur, 3)
            .await
            .expect("a conforming paged prologue reassembles");
        assert_eq!(bytes, b"ABC", "the data window is clipped and preserved");
        assert_eq!(meta.chunk_count, Some(4097));
        assert_eq!(
            meta.chunk_lens,
            Some(full),
            "the reassembled array equals the full, per-entry-distinct layout"
        );
    }

    /// FAIL-CLOSED: the SAME stream missing its last page yields NO layout. A partial `chunk_lens` sums
    /// short of `total_length` and would decrypt every chunk to garbage, so an incomplete prologue is a
    /// RECOVERABLE refusal (the holder is skipped) — never an adopted partial array (SPEC.md §2.2).
    #[tokio::test]
    async fn an_incomplete_paged_prologue_is_refused_not_adopted() {
        let (_full, first, mut second, _third) = paged_prologue_frames();
        // End the stream after the SECOND page (2 of 3 pages) by marking it complete.
        second.complete = true;
        let mut wire = encode(&first);
        wire.extend_from_slice(&encode(&second));
        let mut cur = std::io::Cursor::new(wire);

        let err = assemble_range_stream(&mut cur, 3)
            .await
            .expect_err("a prologue short of chunk_count must not be adopted");
        assert!(
            matches!(
                err,
                DownloadError::PagedPrologueUnsupported {
                    chunk_count: 4097,
                    ..
                }
            ),
            "an incomplete layout is refused fail-closed; got {err:?}"
        );
        assert!(
            err.is_recoverable(),
            "one holder's short prologue skips the holder, not the download"
        );
    }

    /// A declared `chunk_count` above `MAX_RESOURCE_CHUNK_COUNT` is refused BEFORE the assembler
    /// allocates its array — a peer-declared count is never allowed to become an allocation this host
    /// cannot survive. The refusal is recoverable, so the scheduler routes around the hostile holder.
    #[tokio::test]
    async fn a_chunk_count_above_the_resource_ceiling_is_refused_before_allocation() {
        let oversized = dig_nat::MAX_RESOURCE_CHUNK_COUNT as u64 + 1;
        let first = RangeFrame::data(0, b"A".to_vec())
            .with_identity(test_root(), 64, oversized)
            .with_chunk_lens_page(0, vec![64; 2048])
            .with_chunk_index(0);
        let mut cur = std::io::Cursor::new(encode(&first));

        let err = assemble_range_stream(&mut cur, 1)
            .await
            .expect_err("an over-ceiling chunk_count is refused pre-allocation");
        assert!(
            err.is_recoverable(),
            "a refused-before-allocation layout skips the holder, not the download; got {err:?}"
        );
    }

    /// A MISALIGNED later page (offset not a multiple of `MAX_CHUNK_LENS_PER_FRAME`) is rejected by the
    /// assembler's own placement rules — defense in depth beyond the identity frontier guard. Recoverable,
    /// so the holder is skipped.
    #[tokio::test]
    async fn a_misaligned_prologue_page_is_rejected() {
        let full = distinct_lens(4097);
        let total: u64 = full.iter().sum();
        let first = RangeFrame::data(0, b"ABC".to_vec())
            .with_identity(test_root(), total, 4097)
            .with_chunk_lens_page(0, full[0..2048].to_vec())
            .with_chunk_index(0);
        // Offset 2049 is not a page-aligned multiple of 2048, so the assembler refuses it even though it
        // advances the identity frontier past 2048.
        let second = RangeFrame::data(0, Vec::new())
            .with_complete(true)
            .with_identity(test_root(), total, 4097)
            .with_chunk_lens_page(2049, full[2049..4097].to_vec());
        let mut wire = encode(&first);
        wire.extend_from_slice(&encode(&second));
        let mut cur = std::io::Cursor::new(wire);

        let err = assemble_range_stream(&mut cur, 3)
            .await
            .expect_err("a misaligned page must be rejected");
        assert!(
            matches!(&err, DownloadError::Transport { reason, .. } if reason.contains("chunk_lens prologue rejected")),
            "the assembler's placement rule names the rejection; got {err:?}"
        );
        assert!(err.is_recoverable(), "a hostile page skips the holder");
    }

    // ---- termination against a holder that streams without progressing --------------------------

    /// A holder streaming EMPTY non-final frames must be refused, not read forever.
    ///
    /// Every loop exit depends on the window filling or the holder setting `complete`, so a frame that
    /// contributes no bytes and sets neither satisfies every other check — including the identity
    /// re-check, because omitting identity is conforming by design — and advances nothing. Sustained on a
    /// few dozen bytes per frame it pins the job while it still holds the staging claim, which makes the
    /// staging path permanently GC-exempt and permanently un-downloadable.
    ///
    /// The test is bounded so a REGRESSION fails instead of hanging the suite: an unbounded reader would
    /// otherwise consume the fixture and block, and a test that hangs reports nothing.
    #[tokio::test]
    async fn a_holder_streaming_empty_non_final_frames_is_refused() {
        let first = RangeFrame::data(0, b"AB".to_vec())
            .with_identity(test_root(), 64, 2)
            .with_chunk_lens_page(0, vec![32, 32])
            .with_chunk_index(0);
        // A frame carrying nothing, declaring nothing, and not completing — every field a hostile holder
        // is free to omit.
        let empty = RangeFrame::data(0, Vec::new());
        let mut wire = encode(&first);
        for _ in 0..64 {
            wire.extend_from_slice(&encode(&empty));
        }
        let mut cur = std::io::Cursor::new(wire);

        let outcome = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            assemble_range_stream(&mut cur, 64),
        )
        .await
        .expect("the reader must REFUSE a non-progressing stream, not consume it");
        let Err(DownloadError::Transport { reason, .. }) = outcome else {
            panic!("a stream that cannot progress must be an error; got {outcome:?}");
        };
        assert!(
            reason.contains("cannot progress"),
            "and must say why; got {reason}"
        );
    }

    /// The same guard, reached by a holder RE-SENDING bytes it already sent.
    ///
    /// This is the variant that slips past a rule aimed at the empty payload: the frame carries real bytes,
    /// so a check on `bytes.is_empty()` accepts it, yet re-writing an already-written prefix advances the
    /// assembled length by nothing and loops just as forever. The rule is therefore stated over the
    /// frontier — the CLASS of frame that does not extend the prefix — not over the empty instance of it.
    #[tokio::test]
    async fn a_holder_resending_an_already_written_prefix_is_refused() {
        let first = RangeFrame::data(0, b"AB".to_vec())
            .with_identity(test_root(), 64, 2)
            .with_chunk_lens_page(0, vec![32, 32])
            .with_chunk_index(0);
        let resend = RangeFrame::data(0, b"AB".to_vec());
        let mut wire = encode(&first);
        for _ in 0..64 {
            wire.extend_from_slice(&encode(&resend));
        }
        let mut cur = std::io::Cursor::new(wire);

        let outcome = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            assemble_range_stream(&mut cur, 64),
        )
        .await
        .expect("a re-sent prefix advances nothing and must be refused, not read forever");
        assert!(
            matches!(outcome, Err(DownloadError::Transport { .. })),
            "got {outcome:?}"
        );
    }

    /// The at-bound side: a frame that advances by ONE byte is progress and must be accepted.
    ///
    /// Without this the guard could be "reject any frame that does not fill the window" and both tests
    /// above would still pass, while every real chunk-granular multi-frame holder broke. Progress is
    /// progress however small.
    #[tokio::test]
    async fn a_frame_advancing_the_window_by_one_byte_is_accepted() {
        let first = RangeFrame::data(0, b"A".to_vec())
            .with_identity(test_root(), 3, 1)
            .with_chunk_lens_page(0, vec![3])
            .with_chunk_index(0);
        let second = RangeFrame::data(1, b"B".to_vec());
        let third = RangeFrame::data(2, b"C".to_vec()).with_complete(true);
        let mut wire = encode(&first);
        wire.extend_from_slice(&encode(&second));
        wire.extend_from_slice(&encode(&third));
        let mut cur = std::io::Cursor::new(wire);

        let (bytes, _) = assemble_range_stream(&mut cur, 3)
            .await
            .expect("one byte at a time is slow, not hostile");
        assert_eq!(bytes, b"ABC");
    }

    /// ATTRIBUTING and WRAPPING are not interchangeable: only one preserves the variant.
    ///
    /// `fetch_range` chooses between these two on a single line, and that call site is NOT covered by a
    /// test — it needs real sockets and certificates, so it is one of the few genuinely untestable spots
    /// in this crate. What is pinned here instead is the DIFFERENCE the choice makes, so a future edit that
    /// swaps back to wrapping has a test stating exactly what it destroys.
    #[tokio::test]
    async fn wrapping_a_typed_error_loses_the_variant_that_attributing_keeps() {
        let peer = "ab".repeat(32);
        // Built twice rather than cloned: `DownloadError` is not `Clone` (the derive existed only for the
        // removed re-adoption retry), and the two calls need separate owned values.
        let typed = || DownloadError::PagedPrologueUnsupported {
            provider: String::new(),
            chunk_count: 4,
            delivered: 4,
        };

        // Both halves matter and a weaker assertion misses one: dropping the variant's arm from
        // `attributed_to` leaves it falling through to the catch-all, which PRESERVES the variant while
        // silently failing to stamp the peer. Asserting only the variant would stay green on that.
        match typed().attributed_to(&peer) {
            DownloadError::PagedPrologueUnsupported { provider, .. } => assert_eq!(
                provider, peer,
                "attribution must fill the provider in, not merely keep the variant"
            ),
            other => panic!("attribution must leave the variant alone; got {other:?}"),
        }
        assert!(
            matches!(
                DownloadError::transport(&peer, typed()),
                DownloadError::Transport { .. }
            ),
            "wrapping flattens it to Transport, so `is_recoverable` can no longer tell it apart and the \
             stable error catalogue promises a variant no caller can ever match"
        );
    }

    // ---- omit-tolerance: a TERSE holder is conforming --------------------------------------------
    //
    // `SPEC.md` 2.2 states normatively that a later frame OMITTING an identity field asserts nothing and
    // must be accepted. Nothing tested it: `assemble_reassembles_ordered_frames` and both `identity_pair()`
    // frames all call `.with_identity(...)`, so no fixture anywhere fed a later frame that leaves one out.
    // Inverting the tolerant arm to an `Err` therefore left the whole suite green — vacuous, and exactly
    // the one-sided pinning the rewind rule already avoids.

    /// Assemble a two-frame stream whose continuation omits ONE identity field, and require success.
    ///
    /// Takes the field out of an otherwise-conforming frame, so the only difference from the accepted
    /// control is the omission under test.
    async fn assemble_with_terse_continuation(
        strip: impl FnOnce(&mut RangeFrame),
    ) -> Result<(Vec<u8>, RangeMeta), DownloadError> {
        let (first, mut second) = identity_pair();
        strip(&mut second);
        let mut wire = encode(&first);
        wire.extend_from_slice(&encode(&second));
        let mut cur = std::io::Cursor::new(wire);
        assemble_range_stream(&mut cur, 6).await
    }

    #[tokio::test]
    async fn a_later_frame_omitting_the_root_is_accepted() {
        let (bytes, meta) = assemble_with_terse_continuation(|f| f.root = None)
            .await
            .expect("a terse continuation asserts nothing and must be accepted");
        assert_eq!(bytes, b"ABCDEF", "and its bytes still land in the window");
        assert_eq!(
            meta.root,
            Some(test_root()),
            "the stream's identity stays the FIRST frame's declaration"
        );
    }

    #[tokio::test]
    async fn a_later_frame_omitting_the_total_length_is_accepted() {
        let (bytes, meta) = assemble_with_terse_continuation(|f| f.total_length = None)
            .await
            .expect("a terse continuation asserts nothing and must be accepted");
        assert_eq!(bytes, b"ABCDEF");
        assert_eq!(meta.total_length, Some(6));
    }

    #[tokio::test]
    async fn a_later_frame_omitting_the_chunk_count_is_accepted() {
        let (bytes, meta) = assemble_with_terse_continuation(|f| f.chunk_count = None)
            .await
            .expect("a terse continuation asserts nothing and must be accepted");
        assert_eq!(bytes, b"ABCDEF");
        assert_eq!(meta.chunk_count, Some(2));
    }

    /// A continuation frame carrying NO metadata at all — the maximally terse conforming holder.
    ///
    /// The three tests above each omit one field; this omits every one at once, which is what a holder
    /// that treats the identity set as first-frame-only actually sends. A guard that tolerated single
    /// omissions but tripped on the combination would pass all three and fail here.
    #[tokio::test]
    async fn a_later_frame_omitting_every_identity_field_is_accepted() {
        let (bytes, _) = assemble_with_terse_continuation(|f| {
            f.root = None;
            f.total_length = None;
            f.chunk_count = None;
            f.chunk_index = None;
        })
        .await
        .expect("a bare continuation frame is conforming");
        assert_eq!(bytes, b"ABCDEF");
    }

    #[tokio::test]
    async fn a_later_frame_restating_the_inclusion_proof_is_rejected() {
        let (first, second) = identity_pair();
        let second = second.with_inclusion_proof("proof");
        let reason = reject_reason(&first, &second).await;
        assert!(reason.contains("restates inclusion_proof"), "got {reason}");
    }

    /// `chunk_index` is per-FRAME, not per-resource, so it may advance but never rewind — frames arrive
    /// in ascending byte offset. Treating it as invariant would reject every conforming multi-frame
    /// stream, which is why the control test above matters.
    #[tokio::test]
    async fn a_later_frame_rewinding_the_chunk_index_is_rejected() {
        let (mut first, mut second) = identity_pair();
        first.chunk_index = Some(1);
        second.chunk_index = Some(0);
        let reason = reject_reason(&first, &second).await;
        assert!(reason.contains("rewinds below"), "got {reason}");
    }

    /// The other side of the rewind rule: an index that does NOT go backwards is accepted.
    ///
    /// A guard written as "reject any later chunk_index" passes the rewind test above and is caught only
    /// here, so the bound is pinned from both sides. A repeated index is legal — a frame that does not
    /// begin a new chunk restates the chunk it is inside.
    #[tokio::test]
    async fn a_later_frame_repeating_its_chunk_index_is_accepted() {
        let (first, mut second) = identity_pair();
        second.chunk_index = Some(0);
        let mut wire = encode(&first);
        wire.extend_from_slice(&encode(&second));
        let mut cur = std::io::Cursor::new(wire);
        let (bytes, _) = assemble_range_stream(&mut cur, 6)
            .await
            .expect("an equal chunk_index is not a rewind");
        assert_eq!(bytes, b"ABCDEF");
    }
}