triblespace_core/blob/
memoryblobstore.rs

1use crate::blob::encodings::UnknownBlob;
2use crate::blob::Blob;
3use crate::blob::BlobEncoding;
4use crate::blob::IntoBlob;
5use crate::patch::{Entry, IdentitySchema, PATCH};
6use crate::repo::BlobStore;
7use crate::repo::BlobStoreGet;
8use crate::repo::BlobStoreKeep;
9use crate::repo::BlobStoreList;
10use crate::repo::BlobStorePut;
11use crate::inline::encodings::hash::Handle;
12use crate::inline::Inline;
13use crate::inline::INLINE_LEN;
14
15use std::convert::Infallible;
16use std::error::Error;
17use std::fmt::Debug;
18use std::fmt::{self};
19use std::iter::FromIterator;
20
21use super::TryFromBlob;
22
23/// In-memory blob storage keyed by content-hash handle.
24///
25/// Internally a [`PATCH`] mapping the 32-byte raw handle to a
26/// [`Blob<UnknownBlob>`]. Writes go through `&mut self` (the
27/// type system enforces single-writer); [`reader`] hands out
28/// owned snapshots that are independent of the original
29/// store. PATCH's structural sharing makes those snapshots
30/// O(1) clones — the writer keeps mutating the canonical
31/// PATCH, readers each hold a pinned Arc-clone.
32///
33/// [`reader`]: BlobStore::reader
34pub struct MemoryBlobStore {
35    blobs: PATCH<INLINE_LEN, IdentitySchema, Blob<UnknownBlob>>,
36}
37
38impl Debug for MemoryBlobStore {
39    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
40        write!(f, "MemoryBlobStore")
41    }
42}
43
44#[derive(Debug)]
45/// Snapshot view into a [`MemoryBlobStore`]. Independent from
46/// the source store — subsequent writes to the store are not
47/// visible to a reader produced earlier; call [`reader`] again
48/// to pick them up.
49///
50/// `Clone` is O(1) (PATCH structural sharing). The reader is
51/// `Send + Sync` and freely composes through `find!` /
52/// `pattern!` / `and!` / `or!`.
53///
54/// [`reader`]: BlobStore::reader
55pub struct MemoryBlobStoreReader {
56    blobs: PATCH<INLINE_LEN, IdentitySchema, Blob<UnknownBlob>>,
57}
58
59impl Clone for MemoryBlobStoreReader {
60    fn clone(&self) -> Self {
61        MemoryBlobStoreReader {
62            blobs: self.blobs.clone(),
63        }
64    }
65}
66
67impl PartialEq for MemoryBlobStoreReader {
68    fn eq(&self, other: &Self) -> bool {
69        self.blobs == other.blobs
70    }
71}
72
73impl Eq for MemoryBlobStoreReader {}
74
75impl MemoryBlobStoreReader {
76    fn new(blobs: PATCH<INLINE_LEN, IdentitySchema, Blob<UnknownBlob>>) -> Self {
77        MemoryBlobStoreReader { blobs }
78    }
79
80    /// Number of blobs in this snapshot.
81    pub fn len(&self) -> usize {
82        self.blobs.len() as usize
83    }
84
85    /// True iff the snapshot is empty.
86    pub fn is_empty(&self) -> bool {
87        self.len() == 0
88    }
89
90    /// Iterator over `(handle, blob)` pairs in this snapshot.
91    /// Iteration order is unspecified.
92    pub fn iter(&self) -> MemoryBlobStoreIter {
93        let for_iter = self.blobs.clone();
94        let lookup = for_iter.clone();
95        MemoryBlobStoreIter {
96            keys: for_iter.into_iter(),
97            lookup,
98        }
99    }
100}
101
102impl Clone for MemoryBlobStore {
103    fn clone(&self) -> Self {
104        MemoryBlobStore {
105            blobs: self.blobs.clone(),
106        }
107    }
108}
109
110impl PartialEq for MemoryBlobStore {
111    fn eq(&self, other: &Self) -> bool {
112        self.blobs == other.blobs
113    }
114}
115
116impl Eq for MemoryBlobStore {}
117
118impl Default for MemoryBlobStore {
119    fn default() -> Self {
120        Self::new()
121    }
122}
123
124impl MemoryBlobStore {
125    /// Creates a new [`MemoryBlobStore`] with no blobs.
126    pub fn new() -> MemoryBlobStore {
127        MemoryBlobStore {
128            blobs: PATCH::new(),
129        }
130    }
131
132    /// Inserts `blob` into the store and returns its handle.
133    ///
134    /// O(1) over the handle computation — the handle was hashed once
135    /// at `Blob::new` and cached in the blob; this method reuses it.
136    /// Idempotent at the PATCH level: re-inserting the same handle is
137    /// a no-op, which matches the content-addressed semantics
138    /// (same handle ⇒ same bytes).
139    pub fn insert<S>(&mut self, blob: Blob<S>) -> Inline<Handle<S>>
140    where
141        S: BlobEncoding,
142        Handle<S>: crate::inline::InlineEncoding,
143    {
144        let handle: Inline<Handle<S>> = blob.get_handle();
145        let unknown_handle: Inline<Handle<UnknownBlob>> = handle.transmute();
146        let blob: Blob<UnknownBlob> = blob.transmute::<UnknownBlob>();
147        let entry = Entry::with_value(&unknown_handle.raw, blob);
148        self.blobs.insert(&entry);
149        handle
150    }
151
152    /// Number of distinct blobs in the store.
153    pub fn len(&self) -> usize {
154        self.blobs.len() as usize
155    }
156
157    /// True iff the store contains no blobs.
158    pub fn is_empty(&self) -> bool {
159        self.len() == 0
160    }
161
162/// Structurally merge `other` into this store, consuming `other`.
163    ///
164    /// Handle bytes match by content-addressing — duplicate keys
165    /// collapse via PATCH's union semantics (idempotent). Faster
166    /// than per-blob `BlobStorePut::put`: PATCH's `union` is a
167    /// structural merge — cost is bounded by the size of the
168    /// non-overlapping subtrees, not the total blob count.
169    pub fn union(&mut self, other: Self) {
170        self.blobs.union(other.blobs);
171    }
172
173    /// Drops any blobs that are not referenced by one of the provided tribles.
174    pub fn keep<I>(&mut self, handles: I)
175    where
176        I: IntoIterator<Item = Inline<Handle<UnknownBlob>>>,
177    {
178        let mut surviving = PATCH::new();
179        for handle in handles {
180            if let Some(blob) = self.blobs.get(&handle.raw) {
181                let entry = Entry::with_value(&handle.raw, blob.clone());
182                surviving.insert(&entry);
183            }
184        }
185        self.blobs = surviving;
186    }
187}
188
189impl BlobStoreKeep for MemoryBlobStore {
190    fn keep<I>(&mut self, handles: I)
191    where
192        I: IntoIterator<Item = Inline<Handle<UnknownBlob>>>,
193    {
194        MemoryBlobStore::keep(self, handles);
195    }
196}
197
198impl FromIterator<(Inline<Handle<UnknownBlob>>, Blob<UnknownBlob>)> for MemoryBlobStore {
199    fn from_iter<I: IntoIterator<Item = (Inline<Handle<UnknownBlob>>, Blob<UnknownBlob>)>>(
200        iter: I,
201    ) -> Self {
202        let mut store = MemoryBlobStore::new();
203        for (handle, blob) in iter {
204            let entry = Entry::with_value(&handle.raw, blob);
205            store.blobs.insert(&entry);
206        }
207        store
208    }
209}
210
211impl IntoIterator for MemoryBlobStoreReader {
212    type Item = (Inline<Handle<UnknownBlob>>, Blob<UnknownBlob>);
213    type IntoIter = MemoryBlobStoreIter;
214    fn into_iter(self) -> Self::IntoIter {
215        self.iter()
216    }
217}
218
219#[derive(Debug)]
220pub enum MemoryStoreGetError<E: Error> {
221    /// This error occurs when a blob is requested that does not exist in the store.
222    NotFound(),
223    /// This error occurs when a blob is requested that exists, but cannot be converted to the requested type.
224    ConversionFailed(E),
225}
226
227impl<E: Error> fmt::Display for MemoryStoreGetError<E> {
228    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
229        match self {
230            MemoryStoreGetError::NotFound() => write!(f, "Blob not found in memory store"),
231            MemoryStoreGetError::ConversionFailed(e) => write!(f, "Blob conversion failed: {e}"),
232        }
233    }
234}
235
236impl<E: Error> Error for MemoryStoreGetError<E> {}
237
238/// Iterator returned by [`MemoryBlobStoreReader::iter`].
239///
240/// Yields `(Handle, Blob)` pairs. Owned snapshot via PATCH
241/// clones — does not borrow from the source reader.
242pub struct MemoryBlobStoreIter {
243    keys: crate::patch::PATCHIntoIterator<INLINE_LEN, IdentitySchema, Blob<UnknownBlob>>,
244    lookup: PATCH<INLINE_LEN, IdentitySchema, Blob<UnknownBlob>>,
245}
246
247impl Debug for MemoryBlobStoreIter {
248    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
249        f.debug_struct("MemoryBlobStoreIter").finish()
250    }
251}
252
253impl Iterator for MemoryBlobStoreIter {
254    type Item = (Inline<Handle<UnknownBlob>>, Blob<UnknownBlob>);
255
256    fn next(&mut self) -> Option<Self::Item> {
257        let key = self.keys.next()?;
258        let handle: Inline<Handle<UnknownBlob>> = Inline::new(key);
259        let blob = self
260            .lookup
261            .get(&key)
262            .cloned()
263            .expect("key from PATCH iterator must resolve in the same snapshot");
264        Some((handle, blob))
265    }
266}
267
268/// Adapter over [`MemoryBlobStoreIter`] that yields only blob handles.
269pub struct MemoryBlobStoreListIter {
270    inner: MemoryBlobStoreIter,
271}
272
273impl Iterator for MemoryBlobStoreListIter {
274    type Item = Result<Inline<Handle<UnknownBlob>>, Infallible>;
275
276    fn next(&mut self) -> Option<Self::Item> {
277        let (handle, _) = self.inner.next()?;
278        Some(Ok(handle))
279    }
280}
281
282impl BlobStoreList for MemoryBlobStoreReader {
283    type Iter<'a> = MemoryBlobStoreListIter;
284    type Err = Infallible;
285
286    fn blobs(&self) -> Self::Iter<'static> {
287        MemoryBlobStoreListIter { inner: self.iter() }
288    }
289}
290
291impl BlobStoreGet for MemoryBlobStoreReader {
292    type GetError<E: Error + Send + Sync + 'static> = MemoryStoreGetError<E>;
293
294    fn get<T, S>(
295        &self,
296        handle: Inline<Handle<S>>,
297    ) -> Result<T, Self::GetError<<T as TryFromBlob<S>>::Error>>
298    where
299        S: BlobEncoding,
300        T: TryFromBlob<S>,
301    {
302        let handle: Inline<Handle<UnknownBlob>> = handle.transmute();
303        let Some(blob) = self.blobs.get(&handle.raw) else {
304            return Err(MemoryStoreGetError::NotFound());
305        };
306        let blob: Blob<S> = blob.clone().transmute();
307        match blob.try_from_blob() {
308            Ok(value) => Ok(value),
309            Err(e) => Err(MemoryStoreGetError::ConversionFailed(e)),
310        }
311    }
312}
313
314impl crate::repo::BlobChildren for MemoryBlobStoreReader {}
315
316impl BlobStorePut for MemoryBlobStore {
317    type PutError = Infallible;
318
319    fn put<S, T>(&mut self, item: T) -> Result<Inline<Handle<S>>, Self::PutError>
320    where
321        S: BlobEncoding,
322        T: IntoBlob<S>,
323    {
324        let blob = item.to_blob();
325        let handle = blob.get_handle();
326        self.insert(blob);
327        Ok(handle)
328    }
329}
330
331impl BlobStore for MemoryBlobStore {
332    type Reader = MemoryBlobStoreReader;
333    type ReaderError = Infallible;
334
335    fn reader(&mut self) -> Result<Self::Reader, Self::ReaderError> {
336        Ok(MemoryBlobStoreReader::new(self.blobs.clone()))
337    }
338}
339
340#[cfg(test)]
341mod tests {
342    use crate::prelude::*;
343
344    use super::*;
345    use anybytes::Bytes;
346    use fake::faker::name::raw::Name;
347    use fake::locales::EN;
348    use fake::Fake;
349
350    use blobencodings::LongString;
351    use inlineencodings::Handle;
352
353    attributes! {
354        "5AD0FAFB1FECBC197A385EC20166899E" as description: Handle<LongString>;
355    }
356
357    #[test]
358    fn keep() {
359        use crate::repo::potential_handles;
360        use crate::trible::TribleSet;
361
362        let mut kb = TribleSet::new();
363        let mut blobs = MemoryBlobStore::new();
364        for _i in 0..200 {
365            kb += entity! {
366               description: blobs.put(Bytes::from_source(Name(EN).fake::<String>()).view().unwrap()).unwrap()
367            };
368        }
369        blobs.keep(potential_handles(&kb));
370    }
371
372    /// `MemoryBlobStoreReader` must be `Send + Sync` so it composes
373    /// through the parallel-iter ready `and!` / `or!` macros.
374    #[test]
375    fn reader_is_send_sync() {
376        fn assert_send_sync<T: Send + Sync>() {}
377        assert_send_sync::<MemoryBlobStoreReader>();
378    }
379
380    /// `reader()` returns an independent snapshot — writes after
381    /// the reader is produced are not visible to that reader.
382    #[test]
383    fn reader_is_a_pinned_snapshot() {
384        let mut store = MemoryBlobStore::new();
385        let blob_a: Inline<Handle<LongString>> =
386            store.put(Bytes::from_source("hello".to_string()).view().unwrap()).unwrap();
387        let snapshot = store.reader().unwrap();
388        assert_eq!(snapshot.len(), 1);
389
390        let _blob_b: Inline<Handle<LongString>> =
391            store.put(Bytes::from_source("world".to_string()).view().unwrap()).unwrap();
392        // The snapshot still has only the original blob.
393        assert_eq!(snapshot.len(), 1);
394        use anybytes::View;
395        let recovered: View<str> =
396            snapshot.get::<View<str>, LongString>(blob_a).unwrap();
397        assert_eq!(&*recovered, "hello");
398
399        // A fresh reader sees both.
400        let fresh = store.reader().unwrap();
401        assert_eq!(fresh.len(), 2);
402    }
403
404    /// `union` structurally merges two stores; handles round-trip.
405    #[test]
406    fn union_merges_and_preserves_handles() {
407        let mut a = MemoryBlobStore::new();
408        let h_hello: Inline<Handle<LongString>> = a
409            .put(Bytes::from_source("hello".to_string()).view().unwrap())
410            .unwrap();
411        let mut b = MemoryBlobStore::new();
412        let h_world: Inline<Handle<LongString>> = b
413            .put(Bytes::from_source("world".to_string()).view().unwrap())
414            .unwrap();
415        // Idempotent overlap: putting "hello" in b too — union should
416        // collapse the duplicate, not double-count.
417        let _h_hello_b: Inline<Handle<LongString>> = b
418            .put(Bytes::from_source("hello".to_string()).view().unwrap())
419            .unwrap();
420
421        a.union(b);
422        assert_eq!(a.reader().unwrap().len(), 2, "duplicates collapse via union");
423
424        use anybytes::View;
425        let recovered_hello: View<str> = a
426            .reader()
427            .unwrap()
428            .get::<View<str>, LongString>(h_hello)
429            .unwrap();
430        assert_eq!(&*recovered_hello, "hello");
431        let recovered_world: View<str> = a
432            .reader()
433            .unwrap()
434            .get::<View<str>, LongString>(h_world)
435            .unwrap();
436        assert_eq!(&*recovered_world, "world");
437    }
438}
triblespace_core/blob/memoryblobstore.rs

triblespace_core/blob/
memoryblobstore.rs
