decdn_cache/
serve_store.rs

1//! [`ServeStore`]: node-only extension of [`RangedStore`] for progressive
2//! serve-while-filling (#1621).
3//!
4//! `ServeStore` extends `bao_range::RangedStore` but is defined here, in
5//! `cache`, rather than in the `bao-range` leaf crate: its two methods speak
6//! iroh-blobs / `futures_util::Stream` types directly (a live present-ranges
7//! watch and a re-encoded wire-bao stream), and `bao-range` stays free of an
8//! iroh-blobs dependency (ADR 038). The node serve-miss serve path
9//! (`crates/node/src/handlers/client/serve_encoder.rs`) consumes it, observing
10//! the store's present ranges as the blob fills; `ServeStore` is the typed seam
11//! that keeps those iroh-blobs types out of `bao-range`.
12
13use bytes::Bytes;
14use decdn_bao_range::{RangedFuture, RangedStore, RangedStoreError};
15use std::pin::Pin;
16
17/// A live watch of which chunk ranges of a blob are present. The first item
18/// is the current bitfield; further items arrive as the blob fills. Mirrors
19/// `iroh_blobs::api::blobs::ObserveProgress::stream()` — NEVER
20/// `await_completion`, which hangs on a partial blob.
21pub type PresentRangeWatch =
22    Pin<Box<dyn futures_util::Stream<Item = bao_tree::ChunkRanges> + Send>>;
23
24/// A stream of wire-format bao chunks for a re-encoded byte range —
25/// interleaved proof pairs and chunk-group data in tree order, **header-less**
26/// (no 8-byte size prefix; the signed `StreamResponse.total_bytes` is the
27/// authoritative size, ADR 038). Each item is fallible with the store's
28/// [`RangedStoreError`].
29pub type EncodeStream =
30    Pin<Box<dyn futures_util::Stream<Item = Result<Bytes, RangedStoreError>> + Send>>;
31
32/// Node-only extension of [`RangedStore`]: progressive serve-while-filling
33/// primitives over an already-open per-blob store.
34///
35/// `observe` watches which ranges are present as a partial blob fills;
36/// `encode_range` re-encodes whatever is currently held in a byte span into
37/// the `cdn/client/v1` header-less interleaved wire bao (ADR 038), the same
38/// wire shape [`RangedStore::read`] does not produce.
39pub trait ServeStore: RangedStore {
40    /// Watch which chunk ranges are present, from now until the blob is
41    /// complete.
42    ///
43    /// **Precondition:** the blob must already be materialized (at least its
44    /// first chunk group admitted, so the store reports `Partial`). A blob the
45    /// store has never seen has no defined current bitfield to watch, so this
46    /// returns `Err` rather than an empty watch. A serve driver must therefore
47    /// open the watch AFTER the first admit lands (or tolerate a transient
48    /// `Err` and retry), not the instant it kicks off an upstream pull.
49    fn observe(&self) -> RangedFuture<'_, PresentRangeWatch>;
50
51    /// Re-encode the currently-present bytes of
52    /// `[byte_offset, byte_offset + byte_len)` into header-less wire bao.
53    fn encode_range(&self, byte_offset: u64, byte_len: u64) -> RangedFuture<'_, EncodeStream>;
54}
decdn_cache/serve_store.rs

decdn_cache/
serve_store.rs
