1mod composite;
11mod files;
12#[cfg(test)]
13pub(crate) mod flaky;
14pub mod loose;
15mod midx;
16mod packs;
17
18pub use composite::CompositeStore;
19pub use files::FilesSource;
20pub use loose::LooseStore;
21pub use midx::{MidxObjects, MidxObjectsStatus};
22pub(crate) use packs::touch_pack_mtime;
23pub use packs::{PackFilter, PackedObjects};
24
25use std::io::{self, Cursor, Read};
26use std::ops::ControlFlow;
27use std::sync::{Arc, RwLock};
28
29use std::path::Path;
30
31use crate::error::{Error, Result};
32use crate::hash;
33use crate::objects::{
34 oid_hash_map, HashAlgo, Object, ObjectId, ObjectIdPrefix, ObjectInfo, ObjectKind, OidHashMap,
35};
36use crate::odb::WriteOptions;
37use crate::pack;
38
39type MemoryObjectEntry = (ObjectKind, Arc<[u8]>);
40
41#[derive(Debug)]
43pub(crate) enum BoundedRead {
44 Object(Object),
46 Larger,
48}
49
50#[derive(Debug)]
52pub(crate) enum LocalSizedRead {
53 Packed(Option<u64>),
56 Loose(BoundedRead),
58 Missing,
60}
61
62pub struct ObjectStream<'a> {
66 pub kind: ObjectKind,
68 pub size: u64,
70 pub reader: Box<dyn Read + Send + 'a>,
72}
73
74pub trait ObjectStore: Send + Sync + std::fmt::Debug {
79 fn hash_algo(&self) -> HashAlgo;
81
82 fn read(&self, oid: &ObjectId) -> Result<Option<Object>>;
88
89 fn read_into(&self, oid: &ObjectId, buf: &mut Vec<u8>) -> Result<Option<ObjectKind>> {
98 let Some(obj) = self.read(oid)? else {
99 return Ok(None);
100 };
101 buf.clear();
102 buf.extend_from_slice(&obj.data);
103 Ok(Some(obj.kind))
104 }
105
106 fn read_info(&self, oid: &ObjectId) -> Result<Option<ObjectInfo>>;
112
113 fn contains(&self, oid: &ObjectId) -> Result<bool> {
119 Ok(self.read_info(oid)?.is_some())
120 }
121
122 fn contains_many(&self, oids: &[ObjectId]) -> Result<Vec<bool>> {
135 oids.iter().map(|oid| self.contains(oid)).collect()
136 }
137
138 fn open_stream(&self, oid: &ObjectId) -> Result<Option<ObjectStream<'_>>> {
146 let Some(obj) = self.read(oid)? else {
147 return Ok(None);
148 };
149 let size = u64::try_from(obj.data.len()).map_err(|_| {
150 Error::CorruptObject(format!("object size overflow for {}", oid.to_hex()))
151 })?;
152 Ok(Some(ObjectStream {
153 kind: obj.kind,
154 size,
155 reader: Box::new(Cursor::new(obj.data)),
156 }))
157 }
158
159 fn for_each_object(&self, f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>) -> Result<()>;
165
166 fn lookup_prefix(&self, prefix: &str, limit: usize, out: &mut Vec<ObjectId>) -> Result<()> {
184 let prefix = ObjectIdPrefix::from_hex(prefix, self.hash_algo())?;
185 let mut count = 0usize;
186 self.for_each_object(&mut |oid| {
187 if prefix.matches(oid) {
188 out.push(*oid);
189 count += 1;
190 if limit != 0 && count >= limit {
191 return ControlFlow::Break(());
192 }
193 }
194 ControlFlow::Continue(())
195 })
196 }
197
198 fn refresh(&self) -> Result<bool> {
206 Ok(false)
207 }
208}
209
210pub trait WritableObjectStore: ObjectStore {
212 fn write(&self, kind: ObjectKind, data: &[u8], options: WriteOptions) -> Result<ObjectId>;
221
222 fn write_stream(
230 &self,
231 kind: ObjectKind,
232 size: u64,
233 reader: &mut dyn Read,
234 options: WriteOptions,
235 ) -> Result<ObjectId> {
236 let size_usize = usize::try_from(size).map_err(|_| {
237 Error::CorruptObject(format!("object size {size} exceeds address space"))
238 })?;
239 let mut buf = vec![0u8; size_usize];
240 reader.read_exact(&mut buf).map_err(Error::Io)?;
241 self.write(kind, &buf, options)
242 }
243
244 fn freshen(&self, oid: &ObjectId) -> Result<bool>;
252}
253
254pub(crate) fn local_object_materialized(objects_dir: &Path, oid: &ObjectId) -> bool {
256 if local_object_materialized_without_rescan(objects_dir, oid) {
257 return true;
258 }
259 if pack::reprepare_pack_directory_on_miss(objects_dir).ok() == Some(true) {
260 return object_in_local_packs(objects_dir, oid);
261 }
262 false
263}
264
265pub(crate) fn local_object_materialized_without_rescan(objects_dir: &Path, oid: &ObjectId) -> bool {
268 object_in_local_packs(objects_dir, oid) || oid.loose_path_in(objects_dir).is_file()
269}
270
271fn object_in_local_packs(objects_dir: &Path, oid: &ObjectId) -> bool {
272 let Ok(indexes) = pack::read_local_pack_indexes_cached(objects_dir) else {
273 return false;
274 };
275 for idx in &indexes {
276 if idx.is_promisor {
277 continue;
278 }
279 if idx.contains(oid).unwrap_or(false) {
280 return true;
281 }
282 }
283 false
284}
285
286pub(crate) fn local_object_stored_without_rescan(objects_dir: &Path, oid: &ObjectId) -> bool {
291 let Ok(indexes) = pack::read_local_pack_indexes_cached(objects_dir) else {
292 return oid.loose_path_in(objects_dir).is_file();
293 };
294 for idx in &indexes {
295 if idx.contains(oid).unwrap_or(false) {
296 return true;
297 }
298 }
299 oid.loose_path_in(objects_dir).is_file()
300}
301
302pub(crate) fn local_object_stored(objects_dir: &Path, oid: &ObjectId) -> bool {
304 if local_object_stored_without_rescan(objects_dir, oid) {
305 return true;
306 }
307 if pack::reprepare_pack_directory_on_miss(objects_dir).ok() == Some(true) {
308 return local_object_stored_without_rescan(objects_dir, oid);
309 }
310 false
311}
312
313#[derive(Debug)]
317pub struct MemoryStore {
318 algo: HashAlgo,
319 objects: RwLock<OidHashMap<MemoryObjectEntry>>,
320}
321
322impl MemoryStore {
323 #[must_use]
325 pub fn new(algo: HashAlgo) -> Self {
326 Self {
327 algo,
328 objects: RwLock::new(oid_hash_map()),
329 }
330 }
331
332 pub(crate) fn insert_object(&self, oid: ObjectId, kind: ObjectKind, data: Arc<[u8]>) {
334 if let Ok(mut guard) = self.objects.write() {
335 guard.entry(oid).or_insert((kind, data));
336 }
337 }
338}
339
340impl ObjectStore for MemoryStore {
341 fn hash_algo(&self) -> HashAlgo {
342 self.algo
343 }
344
345 fn read(&self, oid: &ObjectId) -> Result<Option<Object>> {
346 let guard = self
347 .objects
348 .read()
349 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
350 Ok(guard.get(oid).map(|(kind, data)| Object {
351 kind: *kind,
352 data: data.to_vec(),
353 }))
354 }
355
356 fn read_info(&self, oid: &ObjectId) -> Result<Option<ObjectInfo>> {
357 let guard = self
358 .objects
359 .read()
360 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
361 Ok(guard.get(oid).map(|(kind, data)| ObjectInfo {
362 kind: *kind,
363 size: u64::try_from(data.len()).unwrap_or(u64::MAX),
364 }))
365 }
366
367 fn contains(&self, oid: &ObjectId) -> Result<bool> {
368 let guard = self
369 .objects
370 .read()
371 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
372 Ok(guard.contains_key(oid))
373 }
374
375 fn contains_many(&self, oids: &[ObjectId]) -> Result<Vec<bool>> {
376 let guard = self
377 .objects
378 .read()
379 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
380 Ok(oids.iter().map(|oid| guard.contains_key(oid)).collect())
381 }
382
383 fn open_stream(&self, oid: &ObjectId) -> Result<Option<ObjectStream<'_>>> {
384 let guard = self
385 .objects
386 .read()
387 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
388 let Some((kind, data)) = guard.get(oid) else {
389 return Ok(None);
390 };
391 let size = u64::try_from(data.len()).map_err(|_| {
392 Error::CorruptObject(format!("object size overflow for {}", oid.to_hex()))
393 })?;
394 Ok(Some(ObjectStream {
395 kind: *kind,
396 size,
397 reader: Box::new(Cursor::new(Arc::clone(data))),
398 }))
399 }
400
401 fn for_each_object(&self, f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>) -> Result<()> {
402 let guard = self
403 .objects
404 .read()
405 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
406 for oid in guard.keys() {
407 if f(oid).is_break() {
408 break;
409 }
410 }
411 Ok(())
412 }
413
414 fn lookup_prefix(&self, prefix: &str, limit: usize, out: &mut Vec<ObjectId>) -> Result<()> {
415 let prefix = ObjectIdPrefix::from_hex(prefix, self.hash_algo())?;
416 let guard = self
417 .objects
418 .read()
419 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
420 let matches = guard.keys().filter(|oid| prefix.matches(oid)).copied();
421 if limit == 0 {
422 out.extend(matches);
423 } else {
424 out.extend(matches.take(limit));
425 }
426 Ok(())
427 }
428}
429
430impl WritableObjectStore for MemoryStore {
431 fn write(&self, kind: ObjectKind, data: &[u8], _options: WriteOptions) -> Result<ObjectId> {
432 let oid = hash::hash_object(self.algo, kind, data)?;
433 let mut guard = self
434 .objects
435 .write()
436 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
437 guard
438 .entry(oid)
439 .or_insert_with(|| (kind, Arc::from(data.to_owned())));
440 Ok(oid)
441 }
442
443 fn freshen(&self, oid: &ObjectId) -> Result<bool> {
444 let guard = self
445 .objects
446 .read()
447 .map_err(|e| Error::Io(io::Error::other(e.to_string())))?;
448 Ok(guard.contains_key(oid))
449 }
450}
451
452pub(crate) fn resolve_pending(
457 oids: &[ObjectId],
458 present: &mut [bool],
459 pending: &mut Vec<usize>,
460 probe: impl FnOnce(&[ObjectId]) -> Result<Vec<bool>>,
461) -> Result<()> {
462 if pending.is_empty() {
463 return Ok(());
464 }
465 let batch: Vec<ObjectId> = pending.iter().map(|&i| oids[i]).collect();
466 let mut found = probe(&batch)?.into_iter();
467 pending.retain(|&i| {
468 let hit = found.next().unwrap_or(false);
469 present[i] |= hit;
470 !hit
471 });
472 Ok(())
473}
474
475pub(crate) fn for_each_propagate_break(
477 store: &dyn ObjectStore,
478 f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>,
479) -> Result<bool> {
480 let mut stop = false;
481 store.for_each_object(&mut |oid| {
482 if stop {
483 return ControlFlow::Break(());
484 }
485 match f(oid) {
486 ControlFlow::Break(()) => {
487 stop = true;
488 ControlFlow::Break(())
489 }
490 ControlFlow::Continue(()) => ControlFlow::Continue(()),
491 }
492 })?;
493 Ok(stop)
494}
495
496#[cfg(test)]
497mod tests {
498 use super::*;
499 use std::io::Read;
500
501 #[test]
502 fn read_hit_and_miss() {
503 let store = MemoryStore::new(HashAlgo::Sha1);
504 let data = b"hello";
505 let oid = store
506 .write(ObjectKind::Blob, data, WriteOptions::default())
507 .unwrap();
508 let obj = store.read(&oid).unwrap().expect("hit");
509 assert_eq!(obj.kind, ObjectKind::Blob);
510 assert_eq!(obj.data, data);
511 assert!(store.read(&ObjectId::zero()).unwrap().is_none());
512 assert!(store.read_info(&ObjectId::zero()).unwrap().is_none());
513 assert!(!store.contains(&ObjectId::zero()).unwrap());
514 assert!(store.contains(&oid).unwrap());
515 assert_eq!(
516 store.contains_many(&[oid, ObjectId::zero(), oid]).unwrap(),
517 vec![true, false, true]
518 );
519 assert!(store.contains_many(&[]).unwrap().is_empty());
520 }
521
522 #[test]
523 fn resolve_pending_marks_hits_and_keeps_misses() {
524 let ids: Vec<ObjectId> = (1u8..=3)
525 .map(|b| ObjectId::from_bytes(&[b; 20]).unwrap())
526 .collect();
527 let mut present = vec![false; 3];
528 let mut pending = vec![0, 2];
529 resolve_pending(&ids, &mut present, &mut pending, |batch| {
530 assert_eq!(batch, [ids[0], ids[2]]);
531 Ok(vec![false, true])
532 })
533 .unwrap();
534 assert_eq!(present, [false, false, true]);
535 assert_eq!(pending, [0]);
536
537 resolve_pending(&ids, &mut present, &mut pending, |_| Ok(Vec::new())).unwrap();
538 assert_eq!(pending, [0], "a short answer counts as absent");
539
540 pending.clear();
541 resolve_pending(&ids, &mut present, &mut pending, |_| {
542 panic!("no probe when nothing is pending")
543 })
544 .unwrap();
545 }
546
547 #[test]
548 fn stream_four_mib_blob_in_sixty_four_kib_chunks() {
549 let store = MemoryStore::new(HashAlgo::Sha1);
550 let payload = vec![0xABu8; 4 * 1024 * 1024];
551 let oid = store
552 .write(ObjectKind::Blob, &payload, WriteOptions::default())
553 .unwrap();
554 let mut stream = store.open_stream(&oid).unwrap().expect("stream");
555 assert_eq!(stream.kind, ObjectKind::Blob);
556 assert_eq!(stream.size, payload.len() as u64);
557 let mut out = Vec::with_capacity(payload.len());
558 let mut chunk = [0u8; 64 * 1024];
559 loop {
560 let n = stream.reader.read(&mut chunk).unwrap();
561 if n == 0 {
562 break;
563 }
564 out.extend_from_slice(&chunk[..n]);
565 }
566 assert_eq!(out, payload);
567 }
568
569 #[test]
570 fn for_each_object_stops_on_break() {
571 let store = MemoryStore::new(HashAlgo::Sha1);
572 for i in 0u8..5 {
573 store
574 .write(ObjectKind::Blob, &[i], WriteOptions::default())
575 .unwrap();
576 }
577 let mut seen = 0usize;
578 store
579 .for_each_object(&mut |_| {
580 seen += 1;
581 ControlFlow::Break(())
582 })
583 .unwrap();
584 assert_eq!(seen, 1);
585 }
586
587 #[test]
588 fn lookup_prefix_unique_ambiguous_and_none() {
589 let store = MemoryStore::new(HashAlgo::Sha1);
590 let a = store
591 .write(ObjectKind::Blob, b"a", WriteOptions::default())
592 .unwrap();
593 let b = store
594 .write(ObjectKind::Blob, b"b", WriteOptions::default())
595 .unwrap();
596 let hex_a = a.to_hex();
597 let unique_prefix = &hex_a[..12];
598 let mut out = Vec::new();
599 store.lookup_prefix(unique_prefix, 0, &mut out).unwrap();
600 assert_eq!(out, vec![a]);
601
602 let mut ambiguous = Vec::new();
603 store.lookup_prefix("", 2, &mut ambiguous).unwrap();
604 assert_eq!(ambiguous.len(), 2);
605 assert!(ambiguous.contains(&a));
606 assert!(ambiguous.contains(&b));
607
608 let mut none = Vec::new();
609 store
610 .lookup_prefix("ffffffffffffffffffffffffffffffffffffffff", 0, &mut none)
611 .unwrap();
612 assert!(none.is_empty());
613 }
614
615 #[test]
616 fn lookup_prefix_odd_even_empty_and_full_length() {
617 let store = MemoryStore::new(HashAlgo::Sha1);
618 let ids: Vec<ObjectId> = [
619 "abcdef0123000000000000000000000000000000",
620 "abcdef0124000000000000000000000000000000",
621 "abd0000000000000000000000000000000000000",
622 ]
623 .iter()
624 .map(|hex| ObjectId::from_hex(hex).unwrap())
625 .collect();
626 for id in &ids {
627 store.insert_object(*id, ObjectKind::Blob, Arc::from(Vec::new()));
628 }
629 let lookup = |prefix: &str, limit: usize| {
630 let mut out = Vec::new();
631 store.lookup_prefix(prefix, limit, &mut out).unwrap();
632 out.sort();
633 out
634 };
635 assert_eq!(lookup("", 0), ids);
636 assert_eq!(lookup("ab", 1).len(), 1);
637 assert_eq!(lookup("abc", 0), ids[..2]);
638 assert_eq!(lookup("abcdef012", 0), ids[..2]);
639 assert_eq!(lookup("ABCDEF0124", 0), ids[1..2]);
640 assert_eq!(lookup(&ids[2].to_hex(), 0), ids[2..]);
641 assert!(lookup("abe", 0).is_empty());
642 let mut out = Vec::new();
643 assert!(matches!(
644 store.lookup_prefix("xyz", 0, &mut out),
645 Err(Error::InvalidObjectId(_))
646 ));
647 }
648
649 #[test]
650 fn write_idempotent() {
651 let store = MemoryStore::new(HashAlgo::Sha1);
652 let data = b"same";
653 let oid1 = store
654 .write(ObjectKind::Blob, data, WriteOptions::default())
655 .unwrap();
656 let oid2 = store
657 .write(ObjectKind::Blob, data, WriteOptions::default())
658 .unwrap();
659 assert_eq!(oid1, oid2);
660 let guard = store.objects.read().unwrap();
661 assert_eq!(guard.len(), 1);
662 }
663
664 #[test]
665 fn write_stream_default_and_freshen() {
666 let store = MemoryStore::new(HashAlgo::Sha1);
667 let payload = b"streamed";
668 let mut cursor = Cursor::new(*payload);
669 let oid = store
670 .write_stream(
671 ObjectKind::Blob,
672 payload.len() as u64,
673 &mut cursor,
674 WriteOptions::default(),
675 )
676 .unwrap();
677 assert_eq!(
678 store.read(&oid).unwrap().expect("written").data,
679 payload.as_slice()
680 );
681 assert!(store.freshen(&oid).unwrap());
682 assert!(!store.freshen(&ObjectId::zero()).unwrap());
683 }
684
685 #[test]
686 fn refresh_default_false() {
687 let store = MemoryStore::new(HashAlgo::Sha1);
688 assert!(!store.refresh().unwrap());
689 }
690
691 #[test]
692 fn trait_defaults_via_blanket_test_store() {
693 struct DefaultsOnly {
694 inner: MemoryStore,
695 }
696
697 impl std::fmt::Debug for DefaultsOnly {
698 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
699 self.inner.fmt(f)
700 }
701 }
702
703 impl ObjectStore for DefaultsOnly {
704 fn hash_algo(&self) -> HashAlgo {
705 self.inner.hash_algo()
706 }
707
708 fn read(&self, oid: &ObjectId) -> Result<Option<Object>> {
709 self.inner.read(oid)
710 }
711
712 fn read_info(&self, oid: &ObjectId) -> Result<Option<ObjectInfo>> {
713 self.inner.read_info(oid)
714 }
715
716 fn for_each_object(
717 &self,
718 f: &mut dyn FnMut(&ObjectId) -> ControlFlow<()>,
719 ) -> Result<()> {
720 self.inner.for_each_object(f)
721 }
722 }
723
724 let store = DefaultsOnly {
725 inner: MemoryStore::new(HashAlgo::Sha1),
726 };
727 let oid = WritableObjectStore::write(
728 &store.inner,
729 ObjectKind::Blob,
730 b"x",
731 WriteOptions::default(),
732 )
733 .unwrap();
734 assert!(ObjectStore::contains(&store, &oid).unwrap());
735 assert_eq!(
736 ObjectStore::contains_many(&store, &[ObjectId::zero(), oid]).unwrap(),
737 vec![false, true]
738 );
739 assert!(ObjectStore::open_stream(&store, &oid).unwrap().is_some());
740 let mut prefixed = Vec::new();
741 ObjectStore::lookup_prefix(&store, &oid.to_hex()[..6], 0, &mut prefixed).unwrap();
742 assert_eq!(prefixed, vec![oid]);
743 }
744}