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    /// Rediscovers the current lease holder's advertised client endpoint
42    /// from the root record — peers with storage credentials can rebuild
43    /// their endpoint preference after an HA takeover without any static
44    /// configuration.
45    ///
46    /// # Errors
47    /// Returns an error when the root store cannot be read.
48    pub fn lease_holder_endpoint(&self, db: &str) -> Result<Option<String>, StoreError> {
49        Ok(self
50            .index_root(db)?
51            .and_then(|root| (!root.owner_endpoint.is_empty()).then_some(root.owner_endpoint)))
52    }
53
54    /// Loads the segment for one index order of a published root, through
55    /// the cache.
56    ///
57    /// # Errors
58    /// Returns an error when the blob cannot be loaded.
59    pub fn segment(
60        &self,
61        root: &DbRoot,
62        order: IndexOrder,
63    ) -> Result<Option<Arc<[u8]>>, StoreError> {
64        let Some(roots) = &root.roots else {
65            return Ok(None);
66        };
67        let slot = match order {
68            IndexOrder::Eavt => 0,
69            IndexOrder::Aevt => 1,
70            IndexOrder::Avet => 2,
71            IndexOrder::Vaet => 3,
72        };
73        self.cache.get_or_load(self.store.as_ref(), &roots[slot])
74    }
75
76    /// Decodes a segment's length-prefixed key entries.
77    ///
78    /// # Errors
79    /// Returns [`StoreError::CorruptBlob`]-free decode failures as `None`
80    /// entries never occur; malformed framing yields an error.
81    pub fn segment_keys(bytes: &[u8]) -> Result<Vec<Vec<u8>>, StoreError> {
82        let mut keys = Vec::new();
83        let mut input = bytes;
84        while !input.is_empty() {
85            let (len_bytes, rest) = input
86                .split_at_checked(8)
87                .ok_or_else(|| StoreError::Io(std::io::Error::other("truncated segment")))?;
88            let len = usize::try_from(u64::from_be_bytes(len_bytes.try_into().unwrap_or_default()))
89                .map_err(|_| StoreError::Io(std::io::Error::other("segment key too large")))?;
90            let (key, rest) = rest
91                .split_at_checked(len)
92                .ok_or_else(|| StoreError::Io(std::io::Error::other("truncated segment key")))?;
93            keys.push(key.to_vec());
94            input = rest;
95        }
96        Ok(keys)
97    }
98}
corium_peer/segment.rs

corium_peer/
segment.rs
