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

corium_peer/
segment.rs
