decdn_cache/

ranged_store.rs

//! [`NodeRangedStore`]: the node-side [`RangedStore`] backend over the
//! iroh-blobs cache (#1621).
//!
//! A thin, per-blob adapter over [`CacheEngine`]'s
//! present/missing/admit/export API, adding no new storage
//! logic. `finalize` relies on the node's own auto-promotion: once every
//! chunk of a blob is present the store already reports it complete, so
//! finalize is just that check, erroring [`RangedStoreError::Incomplete`]
//! otherwise.

use bytes::Bytes;
use decdn_bao_range::{AlignedRange, RangedFuture, RangedStore, RangedStoreError};
use iroh_blobs::Hash;

use crate::engine::CacheEngine;
use crate::serve_store::{EncodeStream, PresentRangeWatch, ServeStore};

/// Node-side [`RangedStore`]: a per-blob view over the iroh-blobs cache.
#[derive(Debug)]
pub struct NodeRangedStore {
    engine: CacheEngine,
    hash: Hash,
    total_bytes: u64,
}

impl NodeRangedStore {
    /// Wrap `engine`'s view of `hash` (a `total_bytes`-byte blob) as a
    /// [`RangedStore`]. `engine` is cheap to clone (`Arc`-backed), so this
    /// holds it by value rather than behind an extra `Arc`.
    #[must_use]
    pub const fn new(engine: CacheEngine, hash: Hash, total_bytes: u64) -> Self {
        Self {
            engine,
            hash,
            total_bytes,
        }
    }

    /// The blob hash this store is scoped to.
    #[must_use]
    pub const fn hash(&self) -> Hash {
        self.hash
    }

    /// The wrapped [`CacheEngine`] handle. Lets a node-crate wrapper (#1621 Task
    /// 10, `NodeAdmitStore`) reach the engine's `admit_bao_stream` without
    /// duplicating the `(engine, hash, total_bytes)` triple it already holds.
    #[must_use]
    pub const fn engine(&self) -> &CacheEngine {
        &self.engine
    }

    /// Does a deny/blacklist/evict/quarantine gate currently refuse this hash? A
    /// pure in-memory check ([`CacheEngine::refuses`] reads in-memory sets and
    /// never awaits a store hop), so the serve encoder can fold takedown into its per-leaf
    /// presence check without a store round trip. Mirrors the guard
    /// [`CacheEngine::present_ranges`] applies before it answers presence.
    #[must_use]
    pub fn refuses(&self) -> bool {
        self.engine.refuses(self.hash)
    }
}

fn backend<E: std::error::Error + Send + Sync + 'static>(e: E) -> RangedStoreError {
    RangedStoreError::Backend(Box::new(e))
}

impl RangedStore for NodeRangedStore {
    fn total_bytes(&self) -> u64 {
        self.total_bytes
    }

    fn present_ranges(&self) -> RangedFuture<'_, bao_tree::ChunkRanges> {
        Box::pin(async move {
            let p = self
                .engine
                .present_ranges(self.hash)
                .await
                .map_err(backend)?;
            Ok(p.chunk_ranges().clone())
        })
    }

    fn missing_ranges(
        &self,
        byte_offset: u64,
        byte_len: u64,
    ) -> RangedFuture<'_, bao_tree::ChunkRanges> {
        Box::pin(async move {
            // Validate + align locally so a start at or past the blob end is a
            // typed `Alignment` error, not an opaque backend fault (a bad start is
            // an argument error, and the client backend classifies it the same
            // way). An end past the blob clamps to it rather than erroring: the caller here is a pull leg driving a request whose
            // claimed size can grow, so its gap computation must clamp exactly as
            // the serve tiers do, or a signed `ok: true` response then faults
            // fetching its own pull range.
            let aligned =
                decdn_bao_range::align_range_clamped(byte_offset, byte_len, self.total_bytes)?;
            let present = self
                .engine
                .present_ranges(self.hash)
                .await
                .map_err(backend)?;
            Ok(aligned.chunk_ranges().clone() - present.chunk_ranges())
        })
    }

    fn admit(&self, range: AlignedRange, bao_bytes: Bytes) -> RangedFuture<'_, ()> {
        Box::pin(async move {
            self.engine
                .admit_bao(self.hash, range.chunk_ranges().clone(), bao_bytes)
                .await
                .map_err(backend)
        })
    }

    fn read(&self, byte_offset: u64, byte_len: u64) -> RangedFuture<'_, Bytes> {
        Box::pin(async move {
            // Reject a start at or past the blob end as a typed `Alignment` error
            // before touching the store. An end past the blob clamps instead of
            // erroring; `read` is byte-exact, so the clamped content length, not
            // the raw, possibly-overshooting `byte_len`, is what gets forwarded
            // to the backend.
            let aligned =
                decdn_bao_range::align_range_clamped(byte_offset, byte_len, self.total_bytes)?;
            // `byte_len == 0` keeps its "to end" meaning: `export_range` resolves
            // that dynamically against the store's own reported size, which a
            // partial blob may not have yet. Only a non-zero, explicit `byte_len`
            // clamps to the blob's end here.
            let clamped_len = if byte_len == 0 {
                0
            } else {
                aligned
                    .blob_size()
                    .saturating_sub(byte_offset)
                    .min(byte_len)
            };
            self.engine
                .export_range(self.hash, byte_offset, clamped_len)
                .await
                .map_err(backend)
        })
    }

    fn is_complete(&self) -> RangedFuture<'_, bool> {
        Box::pin(async move {
            Ok(self
                .engine
                .present_ranges(self.hash)
                .await
                .map_err(backend)?
                .is_complete())
        })
    }

    fn finalize(&self) -> RangedFuture<'_, ()> {
        Box::pin(async move {
            if self
                .engine
                .present_ranges(self.hash)
                .await
                .map_err(backend)?
                .is_complete()
            {
                Ok(())
            } else {
                Err(RangedStoreError::Incomplete)
            }
        })
    }
}

impl ServeStore for NodeRangedStore {
    fn observe(&self) -> RangedFuture<'_, PresentRangeWatch> {
        Box::pin(async move {
            self.engine
                .observe_present_ranges(self.hash)
                .await
                .map_err(backend)
        })
    }

    fn encode_range(&self, byte_offset: u64, byte_len: u64) -> RangedFuture<'_, EncodeStream> {
        Box::pin(async move {
            let stream = self
                .engine
                .export_bao_range_stream(self.hash, byte_offset, byte_len, self.total_bytes)
                .await
                .map_err(backend)?;
            Ok(Box::pin(futures_util::StreamExt::map(stream, |item| {
                item.map_err(backend)
            })) as EncodeStream)
        })
    }
}

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