grit_lib/odb/store/
composite.rs

1//! Ordered chain of [`ObjectStore`] backends: first hit wins on lookup; enumeration deduplicates ids.
2
3use std::collections::HashSet;
4use std::ops::ControlFlow;
5use std::sync::Arc;
6
7use crate::error::{Error, Result};
8use crate::objects::{HashAlgo, Object, ObjectId, ObjectInfo, ObjectKind};
9
10use super::{ObjectStore, ObjectStream};
11
12/// Read-only store that consults an ordered list of backends.
13///
14/// # Errors
15///
16/// Reads ([`ObjectStore::read`], `read_into`, `read_info`, `open_stream`) skip past a backend
17/// whose copy is damaged ([`Error::Zlib`] or [`Error::CorruptObject`]) and return a later
18/// backend's good copy. When no backend returns the object, the first such corruption error is
19/// returned instead of `Ok(None)`. Any other backend error is returned immediately.
20#[derive(Debug)]
21pub struct CompositeStore {
22    stores: Vec<Arc<dyn ObjectStore>>,
23    hash_algo: HashAlgo,
24}
25
26impl CompositeStore {
27    /// Build a composite from `stores` in lookup order (first match wins).
28    ///
29    /// `hash_algo` is used when every child is empty; otherwise the first store's algorithm applies.
30    #[must_use]
31    pub fn new(stores: Vec<Arc<dyn ObjectStore>>, hash_algo: HashAlgo) -> Self {
32        Self { stores, hash_algo }
33    }
34
35    /// Backends in lookup order (read-only).
36    #[must_use]
37    pub fn stores(&self) -> &[Arc<dyn ObjectStore>] {
38        &self.stores
39    }
40
41    /// Ask each child in order and return the first hit.
42    ///
43    /// A child that reports a damaged copy ([`Error::Zlib`] or [`Error::CorruptObject`]) does not
44    /// stop the lookup, since a later child may hold a good copy. If none does, the first such
45    /// error is returned so a damaged object is never reported as missing. Any other error stops
46    /// the lookup immediately.
47    fn first_hit<'s, T>(
48        &'s self,
49        mut lookup: impl FnMut(&'s dyn ObjectStore) -> Result<Option<T>>,
50    ) -> Result<Option<T>> {
51        let mut damaged = None;
52        for store in &self.stores {
53            match lookup(store.as_ref()) {
54                Ok(Some(found)) => return Ok(Some(found)),
55                Ok(None) => {}
56                Err(err @ (Error::Zlib(_) | Error::CorruptObject(_))) => {
57                    damaged.get_or_insert(err);
58                }
59                Err(err) => return Err(err),
60            }
61        }
62        damaged.map_or(Ok(None), Err)
63    }
64}
65
66impl ObjectStore for CompositeStore {
67    fn hash_algo(&self) -> HashAlgo {
68        self.stores
69            .first()
70            .map(|store| store.hash_algo())
71            .unwrap_or(self.hash_algo)
72    }
73
74    fn read(&self, oid: &ObjectId) -> Result<Option<Object>> {
75        self.first_hit(|store| store.read(oid))
76    }
77
78    fn read_into(&self, oid: &ObjectId, buf: &mut Vec<u8>) -> Result<Option<ObjectKind>> {
79        self.first_hit(|store| store.read_into(oid, buf))
80    }
81
82    fn read_info(&self, oid: &ObjectId) -> Result<Option<ObjectInfo>> {
83        self.first_hit(|store| store.read_info(oid))
84    }
85
86    fn contains(&self, oid: &ObjectId) -> Result<bool> {
87        for store in &self.stores {
88            if store.contains(oid)? {
89                return Ok(true);
90            }
91        }
92        Ok(false)
93    }
94
95    fn contains_many(&self, oids: &[ObjectId]) -> Result<Vec<bool>> {
96        let mut present = vec![false; oids.len()];
97        let mut pending: Vec<usize> = (0..oids.len()).collect();
98        for store in &self.stores {
99            super::resolve_pending(oids, &mut present, &mut pending, |batch| {
100                store.contains_many(batch)
101            })?;
102        }
103        Ok(present)
104    }
105
106    fn open_stream(&self, oid: &ObjectId) -> Result<Option<ObjectStream<'_>>> {
107        self.first_hit(|store| store.open_stream(oid))
108    }
109
110    fn for_each_object(&self, f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>) -> Result<()> {
111        let mut seen = HashSet::new();
112        for store in &self.stores {
113            let stop = super::for_each_propagate_break(store.as_ref(), &mut |oid| {
114                if !seen.insert(*oid) {
115                    return ControlFlow::Continue(());
116                }
117                f(oid)
118            })?;
119            if stop {
120                break;
121            }
122        }
123        Ok(())
124    }
125
126    fn lookup_prefix(&self, prefix: &str, limit: usize, out: &mut Vec<ObjectId>) -> Result<()> {
127        let mut seen: HashSet<ObjectId> = out.iter().copied().collect();
128        for store in &self.stores {
129            if limit != 0 && out.len() >= limit {
130                return Ok(());
131            }
132            let sub_limit = if limit == 0 {
133                0
134            } else {
135                limit.saturating_sub(out.len())
136            };
137            let mut scratch = Vec::new();
138            store.lookup_prefix(prefix, sub_limit, &mut scratch)?;
139            for oid in scratch {
140                if seen.insert(oid) {
141                    out.push(oid);
142                    if limit != 0 && out.len() >= limit {
143                        return Ok(());
144                    }
145                }
146            }
147        }
148        Ok(())
149    }
150
151    fn refresh(&self) -> Result<bool> {
152        let mut changed = false;
153        for store in &self.stores {
154            if store.refresh()? {
155                changed = true;
156            }
157        }
158        Ok(changed)
159    }
160}
161
162#[cfg(test)]
163mod tests {
164    use super::*;
165    use crate::objects::ObjectKind;
166    use crate::odb::store::{MemoryStore, WritableObjectStore};
167    use crate::odb::WriteOptions;
168
169    #[test]
170    fn for_each_stops_after_break_across_layers() {
171        let store_a = MemoryStore::new(HashAlgo::Sha1);
172        let store_b = MemoryStore::new(HashAlgo::Sha1);
173        store_a
174            .write(ObjectKind::Blob, b"a", WriteOptions::default())
175            .unwrap();
176        store_b
177            .write(ObjectKind::Blob, b"b", WriteOptions::default())
178            .unwrap();
179        let composite = CompositeStore::new(
180            vec![
181                Arc::new(store_a) as Arc<dyn ObjectStore>,
182                Arc::new(store_b) as Arc<dyn ObjectStore>,
183            ],
184            HashAlgo::Sha1,
185        );
186        let mut calls = 0usize;
187        composite
188            .for_each_object(&mut |_| {
189                calls += 1;
190                ControlFlow::Break(())
191            })
192            .unwrap();
193        assert_eq!(calls, 1);
194    }
195
196    #[test]
197    fn first_hit_read_and_deduped_for_each() {
198        let store_a = MemoryStore::new(HashAlgo::Sha1);
199        let store_b = MemoryStore::new(HashAlgo::Sha1);
200        let oid_a = store_a
201            .write(ObjectKind::Blob, b"only-a", WriteOptions::default())
202            .unwrap();
203        let oid_b = store_b
204            .write(ObjectKind::Blob, b"only-b", WriteOptions::default())
205            .unwrap();
206        let a = Arc::new(store_a) as Arc<dyn ObjectStore>;
207        let b = Arc::new(store_b) as Arc<dyn ObjectStore>;
208        let composite = CompositeStore::new(vec![a, b], HashAlgo::Sha1);
209        assert_eq!(
210            composite.read(&oid_a).unwrap().expect("a").data,
211            b"only-a".as_slice()
212        );
213        assert_eq!(
214            composite.read(&oid_b).unwrap().expect("b").data,
215            b"only-b".as_slice()
216        );
217        let mut ids = Vec::new();
218        composite
219            .for_each_object(&mut |oid| {
220                ids.push(*oid);
221                ControlFlow::Continue(())
222            })
223            .unwrap();
224        ids.sort_by_key(|o| o.to_hex());
225        let mut expected = vec![oid_a, oid_b];
226        expected.sort_by_key(|o| o.to_hex());
227        assert_eq!(ids, expected);
228    }
229
230    /// Holds a damaged copy of every object in `inner`: reads fail with `make_err`.
231    #[derive(Debug)]
232    struct Rotten {
233        inner: MemoryStore,
234        make_err: fn() -> Error,
235    }
236
237    impl ObjectStore for Rotten {
238        fn hash_algo(&self) -> HashAlgo {
239            self.inner.hash_algo()
240        }
241
242        fn read(&self, oid: &ObjectId) -> Result<Option<Object>> {
243            match self.inner.read(oid)? {
244                Some(_) => Err((self.make_err)()),
245                None => Ok(None),
246            }
247        }
248
249        fn read_into(&self, oid: &ObjectId, buf: &mut Vec<u8>) -> Result<Option<ObjectKind>> {
250            buf.clear();
251            buf.extend_from_slice(b"garbage from a half-inflated copy");
252            self.read(oid).map(|obj| obj.map(|o| o.kind))
253        }
254
255        fn read_info(&self, oid: &ObjectId) -> Result<Option<ObjectInfo>> {
256            self.read(oid).map(|obj| obj.map(|_| unreachable!()))
257        }
258
259        fn for_each_object(&self, f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>) -> Result<()> {
260            self.inner.for_each_object(f)
261        }
262    }
263
264    fn rotten_with(data: &[u8], make_err: fn() -> Error) -> (Arc<dyn ObjectStore>, ObjectId) {
265        let inner = MemoryStore::new(HashAlgo::Sha1);
266        let oid = inner
267            .write(ObjectKind::Blob, data, WriteOptions::default())
268            .unwrap();
269        (Arc::new(Rotten { inner, make_err }), oid)
270    }
271
272    fn corrupt() -> Error {
273        Error::CorruptObject("checksum mismatch".to_owned())
274    }
275
276    fn zlib() -> Error {
277        Error::Zlib("invalid stored block lengths".to_owned())
278    }
279
280    fn assert_every_read_fails(
281        composite: &CompositeStore,
282        oid: &ObjectId,
283        is_expected: fn(&Error) -> bool,
284    ) {
285        let err = composite.read(oid).unwrap_err();
286        assert!(is_expected(&err), "read: {err:?}");
287        let err = composite.read_into(oid, &mut Vec::new()).unwrap_err();
288        assert!(is_expected(&err), "read_into: {err:?}");
289        let err = composite.read_info(oid).unwrap_err();
290        assert!(is_expected(&err), "read_info: {err:?}");
291        let Err(err) = composite.open_stream(oid) else {
292            panic!("open_stream returned Ok");
293        };
294        assert!(is_expected(&err), "open_stream: {err:?}");
295    }
296
297    #[test]
298    fn corrupt_copy_with_no_good_copy_is_reported_not_missing() {
299        let (rotten, oid) = rotten_with(b"precious data", corrupt);
300        let empty = Arc::new(MemoryStore::new(HashAlgo::Sha1)) as Arc<dyn ObjectStore>;
301        let composite = CompositeStore::new(vec![rotten, empty], HashAlgo::Sha1);
302        assert_every_read_fails(&composite, &oid, |e| matches!(e, Error::CorruptObject(_)));
303    }
304
305    #[test]
306    fn zlib_failure_with_no_good_copy_is_reported_not_missing() {
307        let empty = Arc::new(MemoryStore::new(HashAlgo::Sha1)) as Arc<dyn ObjectStore>;
308        let (rotten, oid) = rotten_with(b"precious data", zlib);
309        let composite = CompositeStore::new(vec![empty, rotten], HashAlgo::Sha1);
310        assert_every_read_fails(&composite, &oid, |e| matches!(e, Error::Zlib(_)));
311    }
312
313    #[test]
314    fn first_corruption_error_wins_when_every_copy_is_damaged() {
315        let (zlib_store, oid) = rotten_with(b"precious data", zlib);
316        let (corrupt_store, _) = rotten_with(b"precious data", corrupt);
317        let composite = CompositeStore::new(vec![zlib_store, corrupt_store], HashAlgo::Sha1);
318        assert_every_read_fails(&composite, &oid, |e| matches!(e, Error::Zlib(_)));
319    }
320
321    #[test]
322    fn good_copy_in_later_store_beats_corrupt_copy() {
323        for make_err in [corrupt as fn() -> Error, zlib] {
324            let (rotten, oid) = rotten_with(b"precious data", make_err);
325            let good = MemoryStore::new(HashAlgo::Sha1);
326            good.write(ObjectKind::Blob, b"precious data", WriteOptions::default())
327                .unwrap();
328            let composite = CompositeStore::new(
329                vec![rotten, Arc::new(good) as Arc<dyn ObjectStore>],
330                HashAlgo::Sha1,
331            );
332
333            assert_eq!(
334                composite.read(&oid).unwrap().expect("good copy").data,
335                b"precious data"
336            );
337            let mut buf = Vec::new();
338            assert_eq!(
339                composite.read_into(&oid, &mut buf).unwrap(),
340                Some(ObjectKind::Blob)
341            );
342            assert_eq!(buf, b"precious data");
343            let info = composite.read_info(&oid).unwrap().expect("good copy");
344            assert_eq!((info.kind, info.size), (ObjectKind::Blob, 13));
345            let mut stream = composite.open_stream(&oid).unwrap().expect("good copy");
346            let mut streamed = Vec::new();
347            std::io::Read::read_to_end(&mut stream.reader, &mut streamed).unwrap();
348            assert_eq!(streamed, b"precious data");
349        }
350    }
351
352    #[test]
353    fn other_errors_still_stop_the_lookup() {
354        let (rotten, oid) = rotten_with(b"precious data", || {
355            Error::Io(std::io::Error::other("backend down"))
356        });
357        let good = MemoryStore::new(HashAlgo::Sha1);
358        good.write(ObjectKind::Blob, b"precious data", WriteOptions::default())
359            .unwrap();
360        let composite = CompositeStore::new(
361            vec![rotten, Arc::new(good) as Arc<dyn ObjectStore>],
362            HashAlgo::Sha1,
363        );
364        assert!(matches!(composite.read(&oid), Err(Error::Io(_))));
365    }
366
367    #[test]
368    fn contains_delegates_without_reading_payload() {
369        let store_a = MemoryStore::new(HashAlgo::Sha1);
370        let oid = store_a
371            .write(ObjectKind::Blob, b"x", WriteOptions::default())
372            .unwrap();
373        let composite = CompositeStore::new(
374            vec![Arc::new(store_a) as Arc<dyn ObjectStore>],
375            HashAlgo::Sha1,
376        );
377        assert!(composite.contains(&oid).unwrap());
378        assert!(!composite
379            .contains(&ObjectId::from_hex("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef").unwrap())
380            .unwrap());
381    }
382
383    /// Records the batches it is asked about.
384    #[derive(Debug)]
385    struct BatchLog {
386        inner: MemoryStore,
387        batches: std::sync::Mutex<Vec<Vec<ObjectId>>>,
388    }
389
390    impl ObjectStore for BatchLog {
391        fn hash_algo(&self) -> HashAlgo {
392            self.inner.hash_algo()
393        }
394        fn read(&self, oid: &ObjectId) -> Result<Option<Object>> {
395            self.inner.read(oid)
396        }
397        fn read_info(&self, oid: &ObjectId) -> Result<Option<ObjectInfo>> {
398            self.inner.read_info(oid)
399        }
400        fn contains_many(&self, oids: &[ObjectId]) -> Result<Vec<bool>> {
401            self.batches
402                .lock()
403                .unwrap_or_else(|e| e.into_inner())
404                .push(oids.to_vec());
405            self.inner.contains_many(oids)
406        }
407        fn for_each_object(&self, f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>) -> Result<()> {
408            self.inner.for_each_object(f)
409        }
410    }
411
412    #[test]
413    fn contains_many_asks_later_layers_only_about_misses() {
414        let first = MemoryStore::new(HashAlgo::Sha1);
415        let in_first = first
416            .write(ObjectKind::Blob, b"first", WriteOptions::default())
417            .unwrap();
418        let second = Arc::new(BatchLog {
419            inner: MemoryStore::new(HashAlgo::Sha1),
420            batches: std::sync::Mutex::new(Vec::new()),
421        });
422        let in_second = second
423            .inner
424            .write(ObjectKind::Blob, b"second", WriteOptions::default())
425            .unwrap();
426        let absent = ObjectId::from_hex("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef").unwrap();
427        let composite = CompositeStore::new(
428            vec![
429                Arc::new(first) as Arc<dyn ObjectStore>,
430                second.clone() as Arc<dyn ObjectStore>,
431            ],
432            HashAlgo::Sha1,
433        );
434        assert_eq!(
435            composite
436                .contains_many(&[absent, in_first, in_second])
437                .unwrap(),
438            vec![false, true, true]
439        );
440        assert_eq!(
441            *second.batches.lock().unwrap(),
442            vec![vec![absent, in_second]],
443            "one batch for the second layer, without ids the first layer holds"
444        );
445
446        second.batches.lock().unwrap().clear();
447        assert_eq!(composite.contains_many(&[in_first]).unwrap(), vec![true]);
448        assert!(second.batches.lock().unwrap().is_empty());
449    }
450}
grit_lib/odb/store/composite.rs

grit_lib/odb/store/
composite.rs
