corium_peer/
segment.rs

1//! Direct blob-store segment access for peers.
2//!
3//! Segments never travel over gRPC: peers with storage credentials read
4//! published index segments straight from the blob store through a local
5//! read-through cache (see `docs/design/protocol.md`). Blobs are immutable
6//! and content-addressed, so cache entries never invalidate.
7
8use std::sync::Arc;
9
10use corium_core::IndexOrder;
11use corium_store::{BlobStore, DbRoot, RootStore, SegmentCache, StoreError, db_root_name};
12
13/// Read-through segment source over a blob/root store.
14pub struct SegmentSource<S> {
15    store: Arc<S>,
16    cache: SegmentCache,
17}
18
19impl<S: BlobStore + RootStore> SegmentSource<S> {
20    /// Wraps a store with an empty cache.
21    #[must_use]
22    pub fn new(store: Arc<S>) -> Self {
23        Self {
24            store,
25            cache: SegmentCache::default(),
26        }
27    }
28
29    /// Reads the current published index root for `db`.
30    ///
31    /// # Errors
32    /// Returns an error when the root store cannot be read.
33    pub fn index_root(&self, db: &str) -> Result<Option<DbRoot>, StoreError> {
34        Ok(self
35            .store
36            .get_root(&db_root_name(db))?
37            .as_deref()
38            .and_then(DbRoot::decode))
39    }
40
41    /// Loads the segment for one index order of a published root, through
42    /// the cache.
43    ///
44    /// # Errors
45    /// Returns an error when the blob cannot be loaded.
46    pub fn segment(
47        &self,
48        root: &DbRoot,
49        order: IndexOrder,
50    ) -> Result<Option<Arc<[u8]>>, StoreError> {
51        let Some(roots) = &root.roots else {
52            return Ok(None);
53        };
54        let slot = match order {
55            IndexOrder::Eavt => 0,
56            IndexOrder::Aevt => 1,
57            IndexOrder::Avet => 2,
58            IndexOrder::Vaet => 3,
59        };
60        self.cache.get_or_load(self.store.as_ref(), &roots[slot])
61    }
62
63    /// Decodes a segment's length-prefixed key entries.
64    ///
65    /// # Errors
66    /// Returns [`StoreError::CorruptBlob`]-free decode failures as `None`
67    /// entries never occur; malformed framing yields an error.
68    pub fn segment_keys(bytes: &[u8]) -> Result<Vec<Vec<u8>>, StoreError> {
69        let mut keys = Vec::new();
70        let mut input = bytes;
71        while !input.is_empty() {
72            let (len_bytes, rest) = input
73                .split_at_checked(8)
74                .ok_or_else(|| StoreError::Io(std::io::Error::other("truncated segment")))?;
75            let len = usize::try_from(u64::from_be_bytes(len_bytes.try_into().unwrap_or_default()))
76                .map_err(|_| StoreError::Io(std::io::Error::other("segment key too large")))?;
77            let (key, rest) = rest
78                .split_at_checked(len)
79                .ok_or_else(|| StoreError::Io(std::io::Error::other("truncated segment key")))?;
80            keys.push(key.to_vec());
81            input = rest;
82        }
83        Ok(keys)
84    }
85}
corium_peer/segment.rs

corium_peer/
segment.rs
