snix_castore/blobstore/blob_meta/
mod.rs

1use std::sync::Arc;
2
3use crate::{B3Digest, proto};
4
5mod bytes_stream;
6
7/// Describes information to assemble a Blob.
8///
9/// Currently stores only a list of non-empty chunk digests and offsets,
10/// but might gain bao support and other fields in the future.
11///
12/// (See <https://snix.dev/docs/components/castore/blobstore-chunking-verified-streaming>)
13#[derive(Clone, Debug, PartialEq, Eq)]
14pub struct BlobMeta {
15    chunk_metas: Arc<[ChunkMeta]>,
16}
17
18impl BlobMeta {
19    /// Returns the length of the blob.
20    pub fn blob_len(&self) -> u64 {
21        self.chunk_metas.last().map(|cm| cm.end).unwrap_or_default()
22    }
23
24    /// Constructs from an iterator of digests and chunk sizes.
25    ///
26    /// NOTE: The empty chunk is skipped silently, as there's no point fetching it.
27    pub fn from_digests_and_sizes<T: IntoIterator<Item = (B3Digest, u64)>>(iter: T) -> Self {
28        let mut blob_len: u64 = 0;
29        let chunk_metas = iter
30            .into_iter()
31            .filter(|(_, chunk_size)| *chunk_size != 0)
32            .map(|(chunk_digest, chunk_size)| {
33                blob_len += chunk_size;
34                ChunkMeta {
35                    digest: chunk_digest,
36                    end: blob_len,
37                }
38            })
39            .collect();
40
41        Self { chunk_metas }
42    }
43
44    /// For a given offset into the blob, returns the index of the first chunk
45    /// holding the requested bytes, as well as the number of bytes to skip inside the first chunk.
46    fn chunk_idx_from(&self, offset: u64) -> (usize, u64) {
47        assert!(
48            offset <= self.blob_len(),
49            "illegal to seek past end of blob ({offset} in a blob of len {})",
50            self.blob_len()
51        );
52
53        let partition_point = self.chunk_metas.partition_point(|e| e.end <= offset);
54
55        // get the start
56        let start_offset = if partition_point == 0 {
57            0
58        } else {
59            self.chunk_metas[partition_point - 1].end
60        };
61
62        (partition_point, offset - start_offset)
63    }
64
65    /// For a given length into the blob, returns the index of the last chunk holding the requested bytes,
66    /// and how many bytes of that chunk are still in range (`chunk[..size]`)
67    fn chunk_idx_until(&self, size: u64) -> (usize, u64) {
68        if size == 0 {
69            return (0, 0);
70        }
71
72        let (idx, skip) = self.chunk_idx_from(size - 1);
73        (idx, skip + 1)
74    }
75}
76
77/// Records information about a chunk part of a [BlobMeta]
78#[derive(Clone, Debug, PartialEq, Eq)]
79struct ChunkMeta {
80    /// Digest of that specific chunk
81    digest: B3Digest,
82    /// Total length in the blob so far, with this chunk included.
83    end: u64,
84}
85
86impl From<BlobMeta> for proto::BlobMeta {
87    fn from(value: BlobMeta) -> Self {
88        let mut prev_end = 0;
89
90        Self {
91            chunks: value
92                .chunk_metas
93                .iter()
94                .map(|ChunkMeta { digest, end }| {
95                    let size = end - prev_end;
96                    prev_end = *end;
97                    proto::blob_meta::ChunkMeta {
98                        digest: (*digest).into(),
99                        size,
100                    }
101                })
102                .collect(),
103        }
104    }
105}
106
107/// Constructs by parsing a [proto::BlobMeta] message.
108///
109/// This will reject a [proto::BlobMeta] message without any chunks.
110/// This is to prevent accidentially passing a `StatBlobResponse` message which
111/// historically omitted chunks in the single-chunk case.
112///
113/// It will also reject chunks with a size of zero
114///
115/// To explicitly create a [BlobMeta] for the empty blob,
116/// use [BlobMeta::from_digests_and_sizes] with an empty iterator instead.
117impl TryFrom<proto::BlobMeta> for BlobMeta {
118    type Error = DecodeError;
119
120    fn try_from(value: proto::BlobMeta) -> Result<Self, Self::Error> {
121        if value.chunks.is_empty() {
122            return Err(DecodeError::NoChunks);
123        }
124
125        // We iterate until a possible error.
126        // If there's no error we can return the complete BlobMeta.
127        let mut err = None;
128        let blob_meta =
129            Self::from_digests_and_sizes(value.chunks.into_iter().enumerate().map_while(
130                |(idx, proto::blob_meta::ChunkMeta { size, digest })| {
131                    if size == 0 {
132                        err = Some(DecodeError::EmptyChunk(idx));
133                        return None;
134                    }
135                    match B3Digest::try_from(digest) {
136                        Ok(digest) => Some((digest, size)),
137                        Err(_) => {
138                            err = Some(DecodeError::InvalidDigest(idx));
139                            // stop iteration
140                            None
141                        }
142                    }
143                },
144            ));
145
146        match err {
147            Some(err) => Err(err),
148            None => Ok(blob_meta),
149        }
150    }
151}
152
153#[derive(thiserror::Error, Debug, PartialEq, Eq)]
154pub enum DecodeError {
155    #[error("refusing to construct with empty chunk at idx {0}")]
156    EmptyChunk(usize),
157    #[error("refusing to construct with zero chunks")]
158    NoChunks,
159    #[error("invalid digest len at idx {0}")]
160    InvalidDigest(usize),
161}
162
163#[cfg(test)]
164mod test {
165    use std::sync::LazyLock;
166
167    use crate::{chunkstore::Chunk, proto};
168
169    use super::{BlobMeta, DecodeError};
170
171    const CHUNK_1: Chunk = Chunk::from_static(b"ab");
172    const CHUNK_2: Chunk = Chunk::from_static(b"c");
173
174    static BLOB_META: LazyLock<BlobMeta> = LazyLock::new(|| {
175        BlobMeta::from_digests_and_sizes([
176            (CHUNK_1.digest(), CHUNK_1.len() as u64),
177            (CHUNK_2.digest(), CHUNK_2.len() as u64),
178            (CHUNK_1.digest(), CHUNK_1.len() as u64),
179        ])
180    });
181
182    #[rstest::rstest]
183    #[case::start(0, 0, 0)]
184    #[case::one(1, 0, 1)]
185    #[case::two(2, 1, 0)]
186    #[case::three(3, 2, 0)]
187    #[case::four(4, 2, 1)]
188    fn chunk_idx_from(
189        #[case] offset: u64,
190        #[case] exp_start_idx: usize,
191        #[case] exp_skip_in_chunk: u64,
192    ) {
193        let (start_idx, skip_in_chunk) = BLOB_META.chunk_idx_from(offset);
194        assert_eq!(exp_skip_in_chunk, skip_in_chunk, "skip_in_chunk to match");
195        assert_eq!(exp_start_idx, start_idx, "start_idx to match");
196    }
197
198    #[test]
199    /// Calling chunk_idx_from for a offset equal to the blob len should return a start_idx equal to the chunk_meta len.
200    /// (So the range (start_idx..chunk_meta.len()) is empty)
201    fn chunk_idx_from_end() {
202        let (start_idx, skip_in_chunk) = BLOB_META.chunk_idx_from(BLOB_META.blob_len());
203        assert_eq!(
204            BLOB_META.chunk_metas.len(),
205            start_idx,
206            "chunk_idx should be 3"
207        );
208        assert_eq!(0, skip_in_chunk, "skip_in_chunk should be zero");
209    }
210
211    #[test]
212    #[should_panic]
213    /// Calling chunk_idx_from for a offset past the blob len should panic
214    fn chunk_digests_past_end() {
215        let _ = BLOB_META.chunk_idx_from(BLOB_META.blob_len() + 1);
216    }
217
218    #[test]
219    fn chunk_digests_empty() {
220        let blob_meta = BlobMeta::from_digests_and_sizes([]);
221        let (chunk_idx, skip_in_chunk) = blob_meta.chunk_idx_from(0);
222        assert_eq!(0, skip_in_chunk, "skip_in_chunk should be zero");
223        assert_eq!(0, chunk_idx, "chunk_idx should be 0");
224    }
225
226    #[test]
227    /// Decoding a BlobMeta without any chunks should fail
228    fn decode_blob_meta_empty() {
229        let blob_meta = proto::BlobMeta { chunks: vec![] };
230        let err = BlobMeta::try_from(blob_meta).expect_err("should fail");
231        assert_eq!(DecodeError::NoChunks, err);
232    }
233
234    #[test]
235    /// Decoding a BlobMeta with a wrong digest len should fail
236    fn decode_blob_meta_wrong_digest() {
237        let blob_meta = proto::BlobMeta {
238            chunks: vec![proto::blob_meta::ChunkMeta {
239                digest: "abcd".into(),
240                size: 42,
241            }],
242        };
243
244        let err = BlobMeta::try_from(blob_meta).expect_err("should fail");
245        assert_eq!(DecodeError::InvalidDigest(0), err);
246    }
247
248    #[test]
249    /// Decoding a BlobMeta with an empty chunk should fail
250    fn decode_blob_meta_empty_chunk() {
251        let blob_meta = proto::BlobMeta {
252            chunks: vec![proto::blob_meta::ChunkMeta {
253                digest: Chunk::from_static(b"").digest().into(),
254                size: 0,
255            }],
256        };
257
258        let err = BlobMeta::try_from(blob_meta).expect_err("should fail");
259        assert_eq!(DecodeError::EmptyChunk(0), err);
260    }
261}
snix_castore/blobstore/blob_meta/mod.rs

snix_castore/blobstore/blob_meta/
mod.rs
