decdn_bao_range/

lib.rs

//! Iroh-blobs-free bao verified-range helpers (#823, #915), shared by
//! `decdn-cache` (origin range import + client serve) and `decdn-client`
//! (client receive). Depends only on `bao-tree`, so linking it from the CLI's
//! pull path keeps the iroh-blobs-free invariant (#578).
//!
//! [ADR 037 §Origin-tier pull-through](https://github.com/decdn/decdn/blob/main/adr/037-regional-proxy-warming.md),
//! [ADR 038 §Wire format](https://github.com/decdn/decdn/blob/main/adr/038-bao-verified-range-streaming.md).
//!
//! An opaque origin backend (S3/R2/B2/HTTP/fs) serves raw bytes by hash with no
//! bao tree, so a node that wants only a byte range `[a, b)` of a large blob
//! historically had to pull the **whole** blob to verify any of it against the
//! content address. When the origin also publishes the blob's pre-order bao
//! outboard at the sibling key `{H}.obao4`, this module verifies the fetched
//! range against the root `H` using that (untrusted) outboard and produces the
//! bao interleaved encoding `iroh-blobs`' `import_bao_bytes` consumes — so the
//! node imports a verified **partial** blob without whole-blob origin egress.
//!
//! The outboard is **self-validating against `H`**: a tampered range, a tampered
//! outboard, or a wrong wanted-root all fail, because every leaf must hash to its
//! anchored parent and the spine must combine to the `H` the node already wanted
//! ([`encode_ranges_validated`] checks `ParentHashMismatch`/`LeafHashMismatch`).
//! There is no trusted-origin assumption — the origin stays a dumb byte store.

use bao_tree::io::outboard::PreOrderMemOutboard;
use bao_tree::io::sync::encode_ranges_validated;
use bao_tree::{BaoTree, BlockSize, ChunkNum, ChunkRanges};
use bytes::Bytes;
use positioned_io::ReadAt;

#[cfg(feature = "test-util")]
pub mod conformance;
pub mod origin_seed;
pub mod ranged_store;
pub mod streaming;
pub use origin_seed::{EncodedOutboard, OBAO4_SUFFIX, encode_outboard, shard_prefix};
pub use ranged_store::{RangedFuture, RangedStore, RangedStoreError};

/// `iroh-blobs`' canonical on-disk block size (16 KiB chunk groups,
/// `from_chunk_log(4)`). Declared from the `bao-tree` primitive rather than
/// re-exported from `iroh-blobs`, because this leaf crate must stay
/// iroh-blobs-free (#578) so the CLI's pull path can link it. The value is
/// **identical by construction** to `iroh_blobs::store::IROH_BLOCK_SIZE` (which
/// is also `BlockSize::from_chunk_log(4)`); `decdn-cache` carries a lock-step
/// test asserting the two stay equal, so a divergence across an iroh-blobs bump
/// — which ADR 038 treats as a protocol-contract change — fails the build there.
pub const IROH_BLOCK_SIZE: BlockSize = BlockSize::from_chunk_log(4);

/// Bytes per chunk group, derived from [`IROH_BLOCK_SIZE`] (`2^chunk_log` 1 KiB
/// chunks) — the granularity an origin range must snap to, because a bao proof
/// anchors whole chunk groups to the root.
///
/// Public so a caller that must snap an offset to a group boundary WITHOUT
/// knowing the blob's size can do so from the shared constant rather than
/// re-deriving it (the resumable client fetch, #1120: it picks its resume offset
/// before the signed response reveals `total_bytes`). Prefer [`align_range`] when
/// the size is known — it also bound-checks.
pub const CHUNK_GROUP_BYTES: u64 = 1u64 << (IROH_BLOCK_SIZE.chunk_log() + 10);

/// Exact byte length of the **header-less** bao interleaved encoding a serving
/// node emits for `chunk_ranges` of a `total_bytes`-byte blob — the sum of every
/// proof node (64 bytes each) and data leaf (≤ chunk-group bytes) in response
/// (pre-order) order ([ADR 038 §Wire format](https://github.com/decdn/decdn/blob/main/adr/038-bao-verified-range-streaming.md)).
///
/// This is the byte count that travels on `cdn/client/v1`: the wire carries the
/// response-format stream (no 8-byte size header — `total_bytes` is already the
/// signed `StreamResponse` field), so this excludes the header. It is computed
/// from the same [`BaoTree`] + [`ChunkRanges`] the serve-side encoder walks, so
/// it equals the emitted length byte-for-byte. The receiver uses it as the
/// voucher-cadence / overrun / short-delivery bound, because paid bytes are wire
/// bytes — content **plus** proof (ADR 038 §Payment metering) — not content
/// bytes. `chunk_ranges` should be the [`AlignedRange::chunk_ranges`] both sides
/// derive from [`align_range`], keeping encoder and decoder in lock-step.
///
/// Cost: `O(log groups)` for the group-aligned contiguous range [`align_range`]
/// produces (a closed-form count over the tree's shape, see
/// `aligned_wire_size`); every other range shape falls back to walking the
/// encoder's node set, which is `O(groups in range)`. The closed form is what
/// keeps a receiver's wire bound cheap against a peer-claimed or ceiling-sized
/// `total_bytes` in the terabytes, where a walk would visit tens of millions of
/// groups before the first byte arrives.
#[must_use]
pub fn bao_encoded_size(total_bytes: u64, chunk_ranges: &ChunkRanges) -> u64 {
    // A 0-byte blob (#1054) has no proof and no data — zero wire bytes for ANY
    // requested range. An empty range set likewise encodes to nothing. Short-
    // circuit both before touching the tree: `bao-tree`'s pre-order chunk iterator
    // `debug_assert!`s `!ranges.is_empty()` (and would otherwise mis-walk an empty
    // set in release), and a 0-byte `BaoTree` walked over a non-empty range like
    // `ChunkRanges::all()` is a degenerate the callers never intend to bill for.
    if total_bytes == 0 || chunk_ranges.is_empty() {
        return 0;
    }
    if let Some(size) = aligned_wire_size(total_bytes, chunk_ranges) {
        return size;
    }
    walked_wire_size(total_bytes, chunk_ranges)
}

/// [`bao_encoded_size`] by walking exactly the node set the serve-side encoder
/// emits: `encode_ranges` / `encode_ranges_validated` iterate
/// `ranges_pre_order_chunks_iter_ref(ranges, 0)` over the real
/// [`IROH_BLOCK_SIZE`] tree (sync.rs), so the same walk yields the same Parent
/// (64 B) + Leaf (data) sequence — and thus the same byte count — that travels on
/// the wire. (The `ResponseIterRef` convenience walks a block-size-zero tree and
/// descends to 1 KiB chunks, over-counting parents on partial groups, so it must
/// NOT be used here.) [`align_range`]'s ranges are already clamped to the blob,
/// so the encoder's `truncate_ranges` step is a no-op and is not reachable here
/// (its module is private upstream).
///
/// The reference the closed-form [`aligned_wire_size`] is checked against, and
/// the fallback for range shapes it does not cover. `total_bytes` must be
/// non-zero and `chunk_ranges` non-empty.
fn walked_wire_size(total_bytes: u64, chunk_ranges: &ChunkRanges) -> u64 {
    let tree = BaoTree::new(total_bytes, IROH_BLOCK_SIZE);
    tree.ranges_pre_order_chunks_iter_ref(chunk_ranges.as_ref(), 0)
        .map(|chunk| u64::try_from(chunk.without_ranges().size()).unwrap_or(u64::MAX))
        .fold(0u64, u64::saturating_add)
}

/// Proof bytes per parent node on the wire: two 32-byte child hashes.
const PARENT_WIRE_BYTES: u64 = 64;

/// [`bao_encoded_size`] in closed form for ONE contiguous chunk range that starts
/// on a chunk-group boundary and ends on one (or at the blob's end) — the shape
/// [`align_range`] produces. `None` for any other shape (a sub-group start or
/// end, several ranges, an open-ended range), which the caller walks instead.
///
/// What the encoder emits for such a range, per [`walked_wire_size`]'s iterator
/// over the [`IROH_BLOCK_SIZE`]-shifted tree, whose leaves are whole chunk groups:
///
/// - the data of every group in the range — one leaf per group (the iterator's
///   leaf split into 1 KiB halves only applies to a block-size-zero tree);
/// - one parent for every internal node of the shifted tree whose group span
///   intersects the range ([`intersecting_parents`]).
///
/// The shifted tree is bao's left-full binary tree over `n` groups: a lone group
/// is a leaf, otherwise the left child is the full tree over the largest power of
/// two below `n` and the right child the tree over the rest. Every internal node
/// has two children, so a full tree over `n` groups has exactly `n - 1` parents.
fn aligned_wire_size(total_bytes: u64, chunk_ranges: &ChunkRanges) -> Option<u64> {
    let group_chunks: u64 = 1u64 << IROH_BLOCK_SIZE.chunk_log();
    let &[start, end] = chunk_ranges.boundaries() else {
        return None;
    };
    let (start, end) = (start.0, end.0);
    let total_chunks = total_bytes.div_ceil(1024);
    let start_on_group = start % group_chunks == 0;
    let end_on_group = end % group_chunks == 0 || end == total_chunks;
    if !start_on_group || !end_on_group || end <= start || end > total_chunks {
        return None;
    }
    let groups = total_chunks.div_ceil(group_chunks);
    let first_group = start / group_chunks;
    let end_group = end.div_ceil(group_chunks);

    let data_start = start.saturating_mul(1024);
    let data_end = end.saturating_mul(1024).min(total_bytes);
    let data = data_end.saturating_sub(data_start);

    let parents = intersecting_parents(groups, 0, first_group, end_group);
    Some(data.saturating_add(parents.saturating_mul(PARENT_WIRE_BYTES)))
}

/// Internal nodes of bao's left-full tree over `n` groups at `base` whose group
/// span intersects `[a, b)`. A node either lies entirely outside the range (adds
/// nothing), entirely inside it (adds every parent of its full subtree, `n - 1`),
/// or straddles a boundary (adds itself and recurses) — so the recursion visits
/// only the two boundary paths, `O(log n)` nodes.
fn intersecting_parents(n: u64, base: u64, a: u64, b: u64) -> u64 {
    let span_end = base.saturating_add(n);
    if n <= 1 || b <= base || a >= span_end {
        return 0;
    }
    if a <= base && span_end <= b {
        return n - 1;
    }
    // Largest power of two strictly below `n` (`n >= 2` here): the left child's
    // group count.
    let left = (n - 1).isolate_highest_one();
    1u64.saturating_add(intersecting_parents(left, base, a, b))
        .saturating_add(intersecting_parents(
            n - left,
            base.saturating_add(left),
            a,
            b,
        ))
}

/// Errors from preparing a verified origin range pull.
#[derive(Debug, thiserror::Error)]
pub enum RangeVerifyError {
    /// `byte_offset` is at or past a non-empty blob's end, or (for the empty
    /// blob) the request is not `(0, 0)`. [`align_range`] also raises this when
    /// `byte_offset + byte_len` runs past the blob or overflows, rejecting the
    /// request; [`align_range_clamped`] clamps that same case to the blob's end
    /// instead of raising it.
    #[error("range [{offset}, +{len}) is out of bounds for a {blob_size}-byte blob")]
    RangeOutOfBounds {
        /// Requested start offset.
        offset: u64,
        /// Requested length as supplied (`0` denotes "to end"), reported verbatim.
        len: u64,
        /// Known blob size from the signed `total_bytes` / manifest entry.
        blob_size: u64,
    },
    /// The supplied outboard is not the length a `blob_size`-byte blob's
    /// pre-order outboard must be — the origin served a malformed `{H}.obao4`.
    #[error("outboard is {got} bytes, expected {expected} for a {blob_size}-byte blob")]
    OutboardSize {
        /// Length a correct pre-order outboard would have.
        expected: usize,
        /// Length actually supplied.
        got: usize,
        /// Known blob size.
        blob_size: u64,
    },
    /// `range_data` is not exactly the aligned fetch window's length — the origin
    /// served a 206 whose body is shorter or longer than `[fetch_start, fetch_end)`
    /// (e.g. a corrupt/extra tail). Rejected rather than silently ignoring the
    /// surplus (or short-reading the deficit) inside the offset-mapped reader.
    #[error(
        "range_data is {got} bytes, expected {expected} for the aligned fetch window \
         of a {blob_size}-byte blob"
    )]
    RangeDataSize {
        /// Aligned fetch-window length the range body must have (`fetch_len`).
        expected: u64,
        /// Length actually supplied.
        got: usize,
        /// Known blob size.
        blob_size: u64,
    },
    /// The fetched range or the outboard did not verify against the wanted root
    /// `H` — a tampered range, a tampered/foreign outboard, or a wrong root.
    /// Carries the bao codec's typed cause so a caller can distinguish an
    /// integrity failure (`ParentHashMismatch` / `LeafHashMismatch` — do not
    /// retry) from a truncated origin response (`Io` short read — retryable).
    #[error("range failed bao verification against the content root")]
    Verification {
        /// Underlying bao codec rejection.
        #[source]
        source: bao_tree::io::EncodeError,
    },
}

/// The chunk-group-aligned byte span a node must fetch from origin to serve a
/// requested `[byte_offset, byte_offset + byte_len)`, plus the chunk ranges to
/// import. A bao proof covers whole 16 KiB groups, so the fetch widens to the
/// enclosing group boundaries (the extra bytes are paid origin egress but never
/// re-served as client bytes — the request is still scoped, not whole-blob).
/// Fields are private and the only constructor is [`align_range`], so holding an
/// `AlignedRange` guarantees its invariants: `fetch_start <= fetch_end`, both
/// group-aligned (or `fetch_end == blob_size` for the final partial group), and
/// `chunk_ranges` exactly covers `[fetch_start, fetch_end)`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AlignedRange {
    fetch_start: u64,
    fetch_end: u64,
    chunk_ranges: ChunkRanges,
    blob_size: u64,
}

impl AlignedRange {
    /// Whole-blob size this range was aligned against — the signed `total_bytes`
    /// [`align_range`] was given. Held so [`wire_len`](Self::wire_len) and
    /// [`encode_verified_range`] cannot be handed a `blob_size` that disagrees
    /// with the `chunk_ranges`.
    #[must_use]
    pub const fn blob_size(&self) -> u64 {
        self.blob_size
    }

    /// Exact header-less bao **wire** byte count for this range (content **plus**
    /// interleaved proof, ADR 038) — the paid quantity, and the receiver's
    /// voucher-cadence / overrun / short-delivery bound.
    ///
    /// This is [`bao_encoded_size`] applied to this range's own `blob_size` and
    /// `chunk_ranges`, so the two can never drift: a mismatched `(total, ranges)`
    /// pair would otherwise silently yield a plausible-but-wrong payment bound
    /// with no runtime backstop — the one misuse seam the crypto never sees.
    #[must_use]
    pub fn wire_len(&self) -> u64 {
        bao_encoded_size(self.blob_size, &self.chunk_ranges)
    }

    /// First byte to fetch (the requested offset floored to a group boundary).
    #[must_use]
    pub const fn fetch_start(&self) -> u64 {
        self.fetch_start
    }

    /// One past the last byte to fetch (requested end ceiled to a group
    /// boundary, then clamped to the blob size for the final partial group).
    #[must_use]
    pub const fn fetch_end(&self) -> u64 {
        self.fetch_end
    }

    /// Chunk ranges (1 KiB `ChunkNum` units) the fetched bytes cover — passed to
    /// both [`encode_verified_range`] and `import_bao_bytes`.
    #[must_use]
    pub const fn chunk_ranges(&self) -> &ChunkRanges {
        &self.chunk_ranges
    }

    /// Number of bytes to fetch from origin for this range. Non-underflowing by
    /// construction — `align_range` guarantees `fetch_end >= fetch_start`.
    #[must_use]
    pub const fn fetch_len(&self) -> u64 {
        self.fetch_end - self.fetch_start
    }
}

/// Chunk-group-aligns `[byte_offset, end)` of a `blob_size`-byte blob into an
/// [`AlignedRange`]. Shared tail of [`align_range`] and [`align_range_clamped`]
/// once each has resolved its own `end` and validated its own `byte_offset`.
fn build_aligned_range(byte_offset: u64, end: u64, blob_size: u64) -> AlignedRange {
    let fetch_start = (byte_offset / CHUNK_GROUP_BYTES) * CHUNK_GROUP_BYTES;
    let fetch_end = end
        .div_ceil(CHUNK_GROUP_BYTES)
        .saturating_mul(CHUNK_GROUP_BYTES)
        .min(blob_size);
    let chunk_ranges =
        ChunkRanges::from(ChunkNum::full_chunks(fetch_start)..ChunkNum::chunks(fetch_end));
    AlignedRange {
        fetch_start,
        fetch_end,
        chunk_ranges,
        blob_size,
    }
}

/// Compute the chunk-group-aligned fetch span for a `[byte_offset, +byte_len)`
/// request against a `blob_size`-byte blob. `byte_len == 0` means "to end".
///
/// The empty (0-byte) blob is addressable only as the whole blob — `(0, 0)` —
/// and aligns to an empty [`AlignedRange`] (`fetch_start == fetch_end == 0`,
/// empty `chunk_ranges`, zero [`wire_len`](AlignedRange::wire_len)). This
/// preserves the pre-bao empty-bytes delivery on the bao path (#1054); the empty
/// stream is still proven against the empty root `blake3::hash(&[])` by the
/// receiver, never accepted for an arbitrary root.
///
/// # Errors
///
/// [`RangeVerifyError::RangeOutOfBounds`] if the offset is past the blob end, or
/// the explicit end overflows / exceeds the blob size. For a 0-byte blob, any
/// positive offset or explicit positive length is out of bounds. Use
/// [`align_range_clamped`] when an end past the blob should be served up to the
/// blob's end instead of refused.
pub fn align_range(
    byte_offset: u64,
    byte_len: u64,
    blob_size: u64,
) -> Result<AlignedRange, RangeVerifyError> {
    let oob = || RangeVerifyError::RangeOutOfBounds {
        offset: byte_offset,
        len: byte_len,
        blob_size,
    };
    // The empty blob is served only as its whole (empty) self: offset 0, "to
    // end". The construction below then yields an empty range. A non-empty blob
    // still rejects any offset at or past the end (reject, never clamp — ADR 005).
    if blob_size == 0 {
        if byte_offset != 0 || byte_len != 0 {
            return Err(oob());
        }
    } else if byte_offset >= blob_size {
        return Err(oob());
    }
    let end = if byte_len == 0 {
        blob_size
    } else {
        let e = byte_offset.checked_add(byte_len).ok_or_else(oob)?;
        if e > blob_size {
            return Err(oob());
        }
        e
    };
    Ok(build_aligned_range(byte_offset, end, blob_size))
}

/// [`align_range`], but an end past the blob clamps to the blob's end instead
/// of being refused: a claimed size is a hint, and the planner's bound can
/// overshoot it. Only `byte_offset >= blob_size` (for a non-empty blob) is
/// refused, with [`RangeVerifyError::RangeOutOfBounds`]: a start past the end
/// has no chunk group to anchor, clamped or not. `(0, 0)` means the whole
/// blob, and the empty blob is addressable only as `(0, 0)`, exactly as in
/// [`align_range`].
///
/// Both a node deciding what to serve and a client deciding what wire length to
/// expect call this, so the two always agree on the served range.
///
/// # Errors
///
/// [`RangeVerifyError::RangeOutOfBounds`] if `byte_offset` is at or past a
/// non-empty blob's end, or (for the empty blob) the request is not `(0, 0)`.
pub fn align_range_clamped(
    byte_offset: u64,
    byte_len: u64,
    blob_size: u64,
) -> Result<AlignedRange, RangeVerifyError> {
    let oob = || RangeVerifyError::RangeOutOfBounds {
        offset: byte_offset,
        len: byte_len,
        blob_size,
    };
    if blob_size == 0 {
        if byte_offset != 0 || byte_len != 0 {
            return Err(oob());
        }
    } else if byte_offset >= blob_size {
        return Err(oob());
    }
    let end = if byte_len == 0 {
        blob_size
    } else {
        byte_offset
            .checked_add(byte_len)
            .map_or(blob_size, |e| e.min(blob_size))
    };
    Ok(build_aligned_range(byte_offset, end, blob_size))
}

/// Maps absolute blob offsets onto a buffer that holds only `[base, base+len)` —
/// the bytes a ranged origin fetch returned. Reads outside the held span report
/// EOF, which surfaces as a verification short-read rather than reading zeros.
struct OffsetReadAt<'a> {
    base: u64,
    data: &'a [u8],
}

impl ReadAt for OffsetReadAt<'_> {
    fn read_at(&self, pos: u64, buf: &mut [u8]) -> std::io::Result<usize> {
        let Some(rel) = pos.checked_sub(self.base) else {
            return Ok(0);
        };
        let Ok(rel) = usize::try_from(rel) else {
            return Ok(0);
        };
        let Some(src) = self.data.get(rel..) else {
            return Ok(0);
        };
        let n = src.len().min(buf.len());
        let (Some(dst), Some(s)) = (buf.get_mut(..n), src.get(..n)) else {
            return Ok(0);
        };
        dst.copy_from_slice(s);
        Ok(n)
    }
}

/// Verify `range_data` (covering `aligned.fetch_start..aligned.fetch_end`) and
/// the supplied pre-order `outboard` against the wanted root `root`, returning
/// the bao interleaved encoding for `aligned.chunk_ranges` — ready for
/// `Blobs::import_bao_bytes(root, aligned.chunk_ranges, _)`.
///
/// The outboard is untrusted: verification anchors every chunk group to `root`,
/// so a tampered range, a foreign/tampered outboard, or a wrong `root` all fail
/// with [`RangeVerifyError::Verification`] — there is no trusted-origin path.
///
/// # Errors
///
/// - [`RangeVerifyError::RangeDataSize`] if `range_data` is not exactly
///   `aligned.fetch_len()` bytes (a corrupt/extra or truncated origin 206 body).
/// - [`RangeVerifyError::OutboardSize`] if `outboard` is the wrong length for the
///   aligned range's blob size (`aligned.blob_size()`).
/// - [`RangeVerifyError::Verification`] if the range/outboard do not verify
///   against `root`.
pub fn encode_verified_range(
    root: [u8; 32],
    aligned: &AlignedRange,
    range_data: &[u8],
    outboard: Bytes,
) -> Result<Bytes, RangeVerifyError> {
    let blob_size = aligned.blob_size();
    // Guard the one seam the crypto never sees: the offset-mapped reader below only
    // exposes `[fetch_start, fetch_end)`, so a `range_data` longer than the aligned
    // fetch window would have its tail silently dropped (and a shorter one would
    // surface only as an opaque verification short-read). Reject either up front so
    // a malformed origin 206 is a typed error, not a silent truncation.
    if u64::try_from(range_data.len()).ok() != Some(aligned.fetch_len()) {
        return Err(RangeVerifyError::RangeDataSize {
            expected: aligned.fetch_len(),
            got: range_data.len(),
            blob_size,
        });
    }
    let tree = BaoTree::new(blob_size, IROH_BLOCK_SIZE);
    let expected = usize::try_from(tree.outboard_size()).unwrap_or(usize::MAX);
    if outboard.len() != expected {
        return Err(RangeVerifyError::OutboardSize {
            expected,
            got: outboard.len(),
            blob_size,
        });
    }
    let ob = PreOrderMemOutboard {
        root: bao_tree::blake3::Hash::from(root),
        tree,
        data: outboard,
    };
    let reader = OffsetReadAt {
        base: aligned.fetch_start,
        data: range_data,
    };
    // bao "combined" encoding: an 8-byte little-endian size header precedes the
    // interleaved proof+data stream. `iroh-blobs`' `import_bao_reader` reads the
    // header first to frame the tree, then feeds the rest to its decoder, so the
    // helper owns producing the full importable byte string. Pre-size the buffer
    // to the header + fetched range (plus the ~0.4% proof overhead grows once);
    // `fetch_len` always fits `usize` when `range_data` does, so a failed
    // conversion just degrades to the default growth.
    let cap = usize::try_from(aligned.fetch_len())
        .unwrap_or(0)
        .saturating_add(8);
    let mut encoded = Vec::with_capacity(cap);
    encoded.extend_from_slice(&blob_size.to_le_bytes());
    encode_ranges_validated(&reader, &ob, aligned.chunk_ranges.as_ref(), &mut encoded)
        .map_err(|source| RangeVerifyError::Verification { source })?;
    Ok(Bytes::from(encoded))
}

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