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::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 ([`crate::error::Error::is_damaged_object`]) 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 ([`crate::error::Error::is_damaged_object`]) 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) if err.is_damaged_object() => {
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::error::Error;
166    use crate::objects::ObjectKind;
167    use crate::odb::store::{MemoryStore, WritableObjectStore};
168    use crate::odb::WriteOptions;
169
170    #[test]
171    fn for_each_stops_after_break_across_layers() {
172        let store_a = MemoryStore::new(HashAlgo::Sha1);
173        let store_b = MemoryStore::new(HashAlgo::Sha1);
174        store_a
175            .write(ObjectKind::Blob, b"a", WriteOptions::default())
176            .unwrap();
177        store_b
178            .write(ObjectKind::Blob, b"b", WriteOptions::default())
179            .unwrap();
180        let composite = CompositeStore::new(
181            vec![
182                Arc::new(store_a) as Arc<dyn ObjectStore>,
183                Arc::new(store_b) as Arc<dyn ObjectStore>,
184            ],
185            HashAlgo::Sha1,
186        );
187        let mut calls = 0usize;
188        composite
189            .for_each_object(&mut |_| {
190                calls += 1;
191                ControlFlow::Break(())
192            })
193            .unwrap();
194        assert_eq!(calls, 1);
195    }
196
197    #[test]
198    fn first_hit_read_and_deduped_for_each() {
199        let store_a = MemoryStore::new(HashAlgo::Sha1);
200        let store_b = MemoryStore::new(HashAlgo::Sha1);
201        let oid_a = store_a
202            .write(ObjectKind::Blob, b"only-a", WriteOptions::default())
203            .unwrap();
204        let oid_b = store_b
205            .write(ObjectKind::Blob, b"only-b", WriteOptions::default())
206            .unwrap();
207        let a = Arc::new(store_a) as Arc<dyn ObjectStore>;
208        let b = Arc::new(store_b) as Arc<dyn ObjectStore>;
209        let composite = CompositeStore::new(vec![a, b], HashAlgo::Sha1);
210        assert_eq!(
211            composite.read(&oid_a).unwrap().expect("a").data,
212            b"only-a".as_slice()
213        );
214        assert_eq!(
215            composite.read(&oid_b).unwrap().expect("b").data,
216            b"only-b".as_slice()
217        );
218        let mut ids = Vec::new();
219        composite
220            .for_each_object(&mut |oid| {
221                ids.push(*oid);
222                ControlFlow::Continue(())
223            })
224            .unwrap();
225        ids.sort_by_key(|o| o.to_hex());
226        let mut expected = vec![oid_a, oid_b];
227        expected.sort_by_key(|o| o.to_hex());
228        assert_eq!(ids, expected);
229    }
230
231    /// Holds a damaged copy of every object in `inner`: reads fail with `make_err`.
232    #[derive(Debug)]
233    struct Rotten {
234        inner: MemoryStore,
235        make_err: fn() -> Error,
236    }
237
238    impl ObjectStore for Rotten {
239        fn hash_algo(&self) -> HashAlgo {
240            self.inner.hash_algo()
241        }
242
243        fn read(&self, oid: &ObjectId) -> Result<Option<Object>> {
244            match self.inner.read(oid)? {
245                Some(_) => Err((self.make_err)()),
246                None => Ok(None),
247            }
248        }
249
250        fn read_into(&self, oid: &ObjectId, buf: &mut Vec<u8>) -> Result<Option<ObjectKind>> {
251            buf.clear();
252            buf.extend_from_slice(b"garbage from a half-inflated copy");
253            self.read(oid).map(|obj| obj.map(|o| o.kind))
254        }
255
256        fn read_info(&self, oid: &ObjectId) -> Result<Option<ObjectInfo>> {
257            self.read(oid).map(|obj| obj.map(|_| unreachable!()))
258        }
259
260        fn for_each_object(&self, f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>) -> Result<()> {
261            self.inner.for_each_object(f)
262        }
263    }
264
265    fn rotten_with(data: &[u8], make_err: fn() -> Error) -> (Arc<dyn ObjectStore>, ObjectId) {
266        let inner = MemoryStore::new(HashAlgo::Sha1);
267        let oid = inner
268            .write(ObjectKind::Blob, data, WriteOptions::default())
269            .unwrap();
270        (Arc::new(Rotten { inner, make_err }), oid)
271    }
272
273    fn corrupt() -> Error {
274        Error::CorruptObject("checksum mismatch".to_owned())
275    }
276
277    fn zlib() -> Error {
278        Error::Zlib("invalid stored block lengths".to_owned())
279    }
280
281    fn assert_every_read_fails(
282        composite: &CompositeStore,
283        oid: &ObjectId,
284        is_expected: fn(&Error) -> bool,
285    ) {
286        let err = composite.read(oid).unwrap_err();
287        assert!(is_expected(&err), "read: {err:?}");
288        let err = composite.read_into(oid, &mut Vec::new()).unwrap_err();
289        assert!(is_expected(&err), "read_into: {err:?}");
290        let err = composite.read_info(oid).unwrap_err();
291        assert!(is_expected(&err), "read_info: {err:?}");
292        let Err(err) = composite.open_stream(oid) else {
293            panic!("open_stream returned Ok");
294        };
295        assert!(is_expected(&err), "open_stream: {err:?}");
296    }
297
298    #[test]
299    fn corrupt_copy_with_no_good_copy_is_reported_not_missing() {
300        let (rotten, oid) = rotten_with(b"precious data", corrupt);
301        let empty = Arc::new(MemoryStore::new(HashAlgo::Sha1)) as Arc<dyn ObjectStore>;
302        let composite = CompositeStore::new(vec![rotten, empty], HashAlgo::Sha1);
303        assert_every_read_fails(&composite, &oid, |e| matches!(e, Error::CorruptObject(_)));
304    }
305
306    #[test]
307    fn zlib_failure_with_no_good_copy_is_reported_not_missing() {
308        let empty = Arc::new(MemoryStore::new(HashAlgo::Sha1)) as Arc<dyn ObjectStore>;
309        let (rotten, oid) = rotten_with(b"precious data", zlib);
310        let composite = CompositeStore::new(vec![empty, rotten], HashAlgo::Sha1);
311        assert_every_read_fails(&composite, &oid, |e| matches!(e, Error::Zlib(_)));
312    }
313
314    #[test]
315    fn first_corruption_error_wins_when_every_copy_is_damaged() {
316        let (zlib_store, oid) = rotten_with(b"precious data", zlib);
317        let (corrupt_store, _) = rotten_with(b"precious data", corrupt);
318        let composite = CompositeStore::new(vec![zlib_store, corrupt_store], HashAlgo::Sha1);
319        assert_every_read_fails(&composite, &oid, |e| matches!(e, Error::Zlib(_)));
320    }
321
322    #[test]
323    fn good_copy_in_later_store_beats_corrupt_copy() {
324        for make_err in [corrupt as fn() -> Error, zlib] {
325            let (rotten, oid) = rotten_with(b"precious data", make_err);
326            let good = MemoryStore::new(HashAlgo::Sha1);
327            good.write(ObjectKind::Blob, b"precious data", WriteOptions::default())
328                .unwrap();
329            let composite = CompositeStore::new(
330                vec![rotten, Arc::new(good) as Arc<dyn ObjectStore>],
331                HashAlgo::Sha1,
332            );
333
334            assert_eq!(
335                composite.read(&oid).unwrap().expect("good copy").data,
336                b"precious data"
337            );
338            let mut buf = Vec::new();
339            assert_eq!(
340                composite.read_into(&oid, &mut buf).unwrap(),
341                Some(ObjectKind::Blob)
342            );
343            assert_eq!(buf, b"precious data");
344            let info = composite.read_info(&oid).unwrap().expect("good copy");
345            assert_eq!((info.kind, info.size), (ObjectKind::Blob, 13));
346            let mut stream = composite.open_stream(&oid).unwrap().expect("good copy");
347            let mut streamed = Vec::new();
348            std::io::Read::read_to_end(&mut stream.reader, &mut streamed).unwrap();
349            assert_eq!(streamed, b"precious data");
350        }
351    }
352
353    #[test]
354    fn other_errors_still_stop_the_lookup() {
355        let (rotten, oid) = rotten_with(b"precious data", || {
356            Error::Io(std::io::Error::other("backend down"))
357        });
358        let good = MemoryStore::new(HashAlgo::Sha1);
359        good.write(ObjectKind::Blob, b"precious data", WriteOptions::default())
360            .unwrap();
361        let composite = CompositeStore::new(
362            vec![rotten, Arc::new(good) as Arc<dyn ObjectStore>],
363            HashAlgo::Sha1,
364        );
365        assert!(matches!(composite.read(&oid), Err(Error::Io(_))));
366    }
367
368    #[test]
369    fn contains_delegates_without_reading_payload() {
370        let store_a = MemoryStore::new(HashAlgo::Sha1);
371        let oid = store_a
372            .write(ObjectKind::Blob, b"x", WriteOptions::default())
373            .unwrap();
374        let composite = CompositeStore::new(
375            vec![Arc::new(store_a) as Arc<dyn ObjectStore>],
376            HashAlgo::Sha1,
377        );
378        assert!(composite.contains(&oid).unwrap());
379        assert!(!composite
380            .contains(&ObjectId::from_hex("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef").unwrap())
381            .unwrap());
382    }
383
384    /// Records the batches it is asked about.
385    #[derive(Debug)]
386    struct BatchLog {
387        inner: MemoryStore,
388        batches: std::sync::Mutex<Vec<Vec<ObjectId>>>,
389    }
390
391    impl ObjectStore for BatchLog {
392        fn hash_algo(&self) -> HashAlgo {
393            self.inner.hash_algo()
394        }
395        fn read(&self, oid: &ObjectId) -> Result<Option<Object>> {
396            self.inner.read(oid)
397        }
398        fn read_info(&self, oid: &ObjectId) -> Result<Option<ObjectInfo>> {
399            self.inner.read_info(oid)
400        }
401        fn contains_many(&self, oids: &[ObjectId]) -> Result<Vec<bool>> {
402            self.batches
403                .lock()
404                .unwrap_or_else(|e| e.into_inner())
405                .push(oids.to_vec());
406            self.inner.contains_many(oids)
407        }
408        fn for_each_object(&self, f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>) -> Result<()> {
409            self.inner.for_each_object(f)
410        }
411    }
412
413    #[test]
414    fn contains_many_asks_later_layers_only_about_misses() {
415        let first = MemoryStore::new(HashAlgo::Sha1);
416        let in_first = first
417            .write(ObjectKind::Blob, b"first", WriteOptions::default())
418            .unwrap();
419        let second = Arc::new(BatchLog {
420            inner: MemoryStore::new(HashAlgo::Sha1),
421            batches: std::sync::Mutex::new(Vec::new()),
422        });
423        let in_second = second
424            .inner
425            .write(ObjectKind::Blob, b"second", WriteOptions::default())
426            .unwrap();
427        let absent = ObjectId::from_hex("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef").unwrap();
428        let composite = CompositeStore::new(
429            vec![
430                Arc::new(first) as Arc<dyn ObjectStore>,
431                second.clone() as Arc<dyn ObjectStore>,
432            ],
433            HashAlgo::Sha1,
434        );
435        assert_eq!(
436            composite
437                .contains_many(&[absent, in_first, in_second])
438                .unwrap(),
439            vec![false, true, true]
440        );
441        assert_eq!(
442            *second.batches.lock().unwrap(),
443            vec![vec![absent, in_second]],
444            "one batch for the second layer, without ids the first layer holds"
445        );
446
447        second.batches.lock().unwrap().clear();
448        assert_eq!(composite.contains_many(&[in_first]).unwrap(), vec![true]);
449        assert!(second.batches.lock().unwrap().is_empty());
450    }
451}
grit_lib/odb/store/composite.rs

grit_lib/odb/store/
composite.rs
