1use flate2::read::ZlibDecoder;
15use flate2::write::ZlibEncoder;
16use flate2::Compression;
17use hashtree_config::{Config, StorageBackend};
18use hashtree_core::store::{Store, StoreError, StoreStats};
19use hashtree_core::types::Hash;
20use hashtree_core::{Cid, DirEntry, HashTree, HashTreeConfig, LinkType};
21use hashtree_fs::FsBlobStore;
22#[cfg(feature = "lmdb")]
23use hashtree_lmdb::LmdbBlobStore;
24use sha1::{Digest, Sha1};
25use std::collections::{BTreeMap, HashMap};
26use std::io::{Read, Write};
27use std::path::Path;
28use std::sync::Arc;
29use tokio::runtime::{Handle, Runtime};
30use tracing::{debug, info, warn};
31
32use super::object::{parse_tree, GitObject, ObjectId, ObjectType};
33use super::progress::{RepoTreeBuildPhase, RepoTreeBuildProgress};
34use super::refs::{validate_ref_name, Ref};
35use super::{Error, Result};
36
37type BoxFuture<'a, T> = std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send + 'a>>;
39
40#[derive(Default)]
41struct RefDirectory {
42 files: BTreeMap<String, String>,
43 dirs: BTreeMap<String, RefDirectory>,
44}
45
46impl RefDirectory {
47 fn insert(&mut self, parts: &[&str], value: String) {
48 let Some((name, rest)) = parts.split_first() else {
49 return;
50 };
51
52 if rest.is_empty() {
53 self.files.insert((*name).to_string(), value);
54 } else {
55 self.dirs
56 .entry((*name).to_string())
57 .or_default()
58 .insert(rest, value);
59 }
60 }
61}
62
63enum RuntimeExecutor {
65 Owned(Runtime),
66 Handle(Handle),
67}
68
69impl RuntimeExecutor {
70 fn block_on<F: std::future::Future>(&self, f: F) -> F::Output {
71 match self {
72 RuntimeExecutor::Owned(rt) => rt.block_on(f),
73 RuntimeExecutor::Handle(handle) => tokio::task::block_in_place(|| handle.block_on(f)),
74 }
75 }
76}
77
78pub enum LocalStore {
80 Fs(FsBlobStore),
81 #[cfg(feature = "lmdb")]
82 Lmdb(LmdbBlobStore),
83}
84
85impl LocalStore {
86 pub(crate) fn new_for_backend<P: AsRef<Path>>(
87 path: P,
88 backend: StorageBackend,
89 max_bytes: u64,
90 ) -> std::result::Result<Self, StoreError> {
91 let path = path.as_ref();
92 #[cfg(feature = "lmdb")]
93 {
94 return Self::new_for_backend_with_openers(
95 path,
96 backend,
97 max_bytes,
98 Self::open_fs_store,
99 Self::open_lmdb_store,
100 );
101 }
102
103 #[cfg(not(feature = "lmdb"))]
104 match backend {
105 StorageBackend::Fs => Self::open_fs_store(path, max_bytes),
106 #[cfg(not(feature = "lmdb"))]
107 StorageBackend::Lmdb => {
108 warn!(
109 "LMDB backend requested but lmdb feature not enabled, using filesystem storage"
110 );
111 Self::open_fs_store(path, max_bytes)
112 }
113 }
114 }
115
116 fn open_fs_store(path: &Path, max_bytes: u64) -> std::result::Result<Self, StoreError> {
117 if max_bytes > 0 {
118 Ok(LocalStore::Fs(FsBlobStore::with_max_bytes(
119 path, max_bytes,
120 )?))
121 } else {
122 Ok(LocalStore::Fs(FsBlobStore::new(path)?))
123 }
124 }
125
126 #[cfg(feature = "lmdb")]
127 fn open_lmdb_store(path: &Path, max_bytes: u64) -> std::result::Result<Self, StoreError> {
128 if max_bytes > 0 {
129 Ok(LocalStore::Lmdb(LmdbBlobStore::with_max_bytes(
130 path, max_bytes,
131 )?))
132 } else {
133 Ok(LocalStore::Lmdb(LmdbBlobStore::new(path)?))
134 }
135 }
136
137 #[cfg(feature = "lmdb")]
138 fn new_for_backend_with_openers<FS, LMDB>(
139 path: &Path,
140 backend: StorageBackend,
141 max_bytes: u64,
142 fs_open: FS,
143 lmdb_open: LMDB,
144 ) -> std::result::Result<Self, StoreError>
145 where
146 FS: Fn(&Path, u64) -> std::result::Result<Self, StoreError>,
147 LMDB: Fn(&Path, u64) -> std::result::Result<Self, StoreError>,
148 {
149 match backend {
150 StorageBackend::Fs => fs_open(path, max_bytes),
151 StorageBackend::Lmdb => match lmdb_open(path, max_bytes) {
152 Ok(store) => Ok(store),
153 Err(err) if should_fallback_from_lmdb_error(&err) => {
154 warn!(
155 path = %path.display(),
156 "LMDB backend is unsupported in this environment, falling back to filesystem storage"
157 );
158 fs_open(path, max_bytes)
159 }
160 Err(err) => Err(err),
161 },
162 }
163 }
164
165 pub fn new<P: AsRef<Path>>(path: P) -> std::result::Result<Self, StoreError> {
167 Self::new_with_max_bytes(path, 0)
168 }
169
170 pub fn new_with_max_bytes<P: AsRef<Path>>(
172 path: P,
173 max_bytes: u64,
174 ) -> std::result::Result<Self, StoreError> {
175 let config = Config::load_or_default();
176 Self::new_for_backend(path, config.storage.backend, max_bytes)
177 }
178
179 pub fn list(&self) -> std::result::Result<Vec<Hash>, StoreError> {
181 match self {
182 LocalStore::Fs(store) => store.list(),
183 #[cfg(feature = "lmdb")]
184 LocalStore::Lmdb(store) => store.list(),
185 }
186 }
187
188 pub fn get_sync(&self, hash: &Hash) -> std::result::Result<Option<Vec<u8>>, StoreError> {
190 match self {
191 LocalStore::Fs(store) => store.get_sync(hash),
192 #[cfg(feature = "lmdb")]
193 LocalStore::Lmdb(store) => store.get_sync(hash),
194 }
195 }
196}
197
198#[cfg(feature = "lmdb")]
199pub(crate) fn should_fallback_from_lmdb_error(error: &StoreError) -> bool {
200 matches!(
201 error,
202 StoreError::Io(io_error) if io_error.raw_os_error() == Some(libc::ENOSYS)
203 )
204}
205
206#[async_trait::async_trait]
207impl Store for LocalStore {
208 async fn put(&self, hash: Hash, data: Vec<u8>) -> std::result::Result<bool, StoreError> {
209 match self {
210 LocalStore::Fs(store) => store.put(hash, data).await,
211 #[cfg(feature = "lmdb")]
212 LocalStore::Lmdb(store) => store.put(hash, data).await,
213 }
214 }
215
216 async fn get(&self, hash: &Hash) -> std::result::Result<Option<Vec<u8>>, StoreError> {
217 match self {
218 LocalStore::Fs(store) => store.get(hash).await,
219 #[cfg(feature = "lmdb")]
220 LocalStore::Lmdb(store) => store.get(hash).await,
221 }
222 }
223
224 async fn has(&self, hash: &Hash) -> std::result::Result<bool, StoreError> {
225 match self {
226 LocalStore::Fs(store) => store.has(hash).await,
227 #[cfg(feature = "lmdb")]
228 LocalStore::Lmdb(store) => store.has(hash).await,
229 }
230 }
231
232 async fn delete(&self, hash: &Hash) -> std::result::Result<bool, StoreError> {
233 match self {
234 LocalStore::Fs(store) => store.delete(hash).await,
235 #[cfg(feature = "lmdb")]
236 LocalStore::Lmdb(store) => store.delete(hash).await,
237 }
238 }
239
240 fn set_max_bytes(&self, max: u64) {
241 match self {
242 LocalStore::Fs(store) => store.set_max_bytes(max),
243 #[cfg(feature = "lmdb")]
244 LocalStore::Lmdb(store) => store.set_max_bytes(max),
245 }
246 }
247
248 fn max_bytes(&self) -> Option<u64> {
249 match self {
250 LocalStore::Fs(store) => store.max_bytes(),
251 #[cfg(feature = "lmdb")]
252 LocalStore::Lmdb(store) => store.max_bytes(),
253 }
254 }
255
256 async fn stats(&self) -> StoreStats {
257 match self {
258 LocalStore::Fs(store) => match store.stats() {
259 Ok(stats) => StoreStats {
260 count: stats.count as u64,
261 bytes: stats.total_bytes,
262 pinned_count: stats.pinned_count as u64,
263 pinned_bytes: stats.pinned_bytes,
264 },
265 Err(_) => StoreStats::default(),
266 },
267 #[cfg(feature = "lmdb")]
268 LocalStore::Lmdb(store) => match store.stats() {
269 Ok(stats) => StoreStats {
270 count: stats.count as u64,
271 bytes: stats.total_bytes,
272 pinned_count: 0,
273 pinned_bytes: 0,
274 },
275 Err(_) => StoreStats::default(),
276 },
277 }
278 }
279
280 async fn evict_if_needed(&self) -> std::result::Result<u64, StoreError> {
281 match self {
282 LocalStore::Fs(store) => store.evict_if_needed().await,
283 #[cfg(feature = "lmdb")]
284 LocalStore::Lmdb(store) => store.evict_if_needed().await,
285 }
286 }
287}
288
289pub struct GitStorage {
291 store: Arc<LocalStore>,
292 tree: HashTree<LocalStore>,
293 runtime: RuntimeExecutor,
294 objects: std::sync::RwLock<HashMap<String, Vec<u8>>>,
296 refs: std::sync::RwLock<HashMap<String, String>>,
297 root_cid: std::sync::RwLock<Option<Cid>>,
299}
300
301impl GitStorage {
302 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
304 let config = Config::load_or_default();
305 let max_size_bytes = config
306 .storage
307 .max_size_gb
308 .saturating_mul(1024 * 1024 * 1024);
309 Self::open_with_max_bytes(path, max_size_bytes)
310 }
311
312 pub fn open_with_max_bytes(path: impl AsRef<Path>, max_size_bytes: u64) -> Result<Self> {
314 let config = Config::load_or_default();
315 Self::open_with_backend_and_max_bytes(path, config.storage.backend, max_size_bytes)
316 }
317
318 pub fn open_with_backend_and_max_bytes(
319 path: impl AsRef<Path>,
320 backend: StorageBackend,
321 max_size_bytes: u64,
322 ) -> Result<Self> {
323 let runtime = match Handle::try_current() {
324 Ok(handle) => RuntimeExecutor::Handle(handle),
325 Err(_) => {
326 let rt = Runtime::new()
327 .map_err(|e| Error::StorageError(format!("tokio runtime: {}", e)))?;
328 RuntimeExecutor::Owned(rt)
329 }
330 };
331
332 let store_path = path.as_ref().join("blobs");
333 let store = Arc::new(
334 LocalStore::new_for_backend(&store_path, backend, max_size_bytes)
335 .map_err(|e| Error::StorageError(format!("local store: {}", e)))?,
336 );
337
338 let tree = HashTree::new(HashTreeConfig::new(store.clone()));
340
341 Ok(Self {
342 store,
343 tree,
344 runtime,
345 objects: std::sync::RwLock::new(HashMap::new()),
346 refs: std::sync::RwLock::new(HashMap::new()),
347 root_cid: std::sync::RwLock::new(None),
348 })
349 }
350
351 pub fn evict_if_needed(&self) -> Result<u64> {
353 self.runtime
354 .block_on(self.store.evict_if_needed())
355 .map_err(|e| Error::StorageError(format!("evict: {}", e)))
356 }
357
358 fn write_object(&self, obj: &GitObject) -> Result<ObjectId> {
360 let oid = obj.id();
361 let key = oid.to_hex();
362
363 let loose = obj.to_loose_format();
364 let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
365 encoder.write_all(&loose)?;
366 let compressed = encoder.finish()?;
367
368 let mut objects = self
369 .objects
370 .write()
371 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
372 objects.insert(key, compressed);
373
374 if let Ok(mut root) = self.root_cid.write() {
376 *root = None;
377 }
378
379 Ok(oid)
380 }
381
382 pub fn write_raw_object(&self, obj_type: ObjectType, content: &[u8]) -> Result<ObjectId> {
384 let obj = GitObject::new(obj_type, content.to_vec());
385 self.write_object(&obj)
386 }
387
388 #[allow(dead_code)]
390 fn read_object(&self, oid: &ObjectId) -> Result<GitObject> {
391 let key = oid.to_hex();
392 let objects = self
393 .objects
394 .read()
395 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
396 let compressed = objects
397 .get(&key)
398 .ok_or_else(|| Error::ObjectNotFound(key.clone()))?;
399
400 let mut decoder = ZlibDecoder::new(compressed.as_slice());
401 let mut data = Vec::new();
402 decoder.read_to_end(&mut data)?;
403
404 GitObject::from_loose_format(&data)
405 }
406
407 pub fn write_ref(&self, name: &str, target: &Ref) -> Result<()> {
409 validate_ref_name(name)?;
410
411 let value = match target {
412 Ref::Direct(oid) => oid.to_hex(),
413 Ref::Symbolic(target) => format!("ref: {}", target),
414 };
415
416 let mut refs = self
417 .refs
418 .write()
419 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
420 refs.insert(name.to_string(), value);
421
422 if let Ok(mut root) = self.root_cid.write() {
424 *root = None;
425 }
426
427 Ok(())
428 }
429
430 #[allow(dead_code)]
432 pub fn read_ref(&self, name: &str) -> Result<Option<Ref>> {
433 let refs = self
434 .refs
435 .read()
436 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
437
438 match refs.get(name) {
439 Some(value) => {
440 if let Some(target) = value.strip_prefix("ref: ") {
441 Ok(Some(Ref::Symbolic(target.to_string())))
442 } else {
443 let oid = ObjectId::from_hex(value)
444 .ok_or_else(|| Error::StorageError(format!("invalid ref: {}", value)))?;
445 Ok(Some(Ref::Direct(oid)))
446 }
447 }
448 None => Ok(None),
449 }
450 }
451
452 #[allow(dead_code)]
454 pub fn list_refs(&self) -> Result<HashMap<String, String>> {
455 let refs = self
456 .refs
457 .read()
458 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
459 Ok(refs.clone())
460 }
461
462 pub fn delete_ref(&self, name: &str) -> Result<bool> {
464 let mut refs = self
465 .refs
466 .write()
467 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
468 let existed = refs.remove(name).is_some();
469
470 if let Ok(mut root) = self.root_cid.write() {
472 *root = None;
473 }
474
475 Ok(existed)
476 }
477
478 pub fn import_compressed_object(&self, oid: &str, compressed_data: Vec<u8>) -> Result<()> {
481 let mut objects = self
482 .objects
483 .write()
484 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
485 objects.insert(oid.to_string(), compressed_data);
486
487 if let Ok(mut root) = self.root_cid.write() {
489 *root = None;
490 }
491
492 Ok(())
493 }
494
495 pub fn import_ref(&self, name: &str, value: &str) -> Result<()> {
497 let mut refs = self
498 .refs
499 .write()
500 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
501 refs.insert(name.to_string(), value.to_string());
502
503 if let Ok(mut root) = self.root_cid.write() {
505 *root = None;
506 }
507
508 Ok(())
509 }
510
511 #[cfg(test)]
513 pub fn has_ref(&self, name: &str) -> Result<bool> {
514 let refs = self
515 .refs
516 .read()
517 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
518 Ok(refs.contains_key(name))
519 }
520
521 #[cfg(test)]
523 pub fn object_count(&self) -> Result<usize> {
524 let objects = self
525 .objects
526 .read()
527 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
528 Ok(objects.len())
529 }
530
531 #[allow(dead_code)]
533 pub fn get_root_cid(&self) -> Result<Option<Cid>> {
534 let root = self
535 .root_cid
536 .read()
537 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
538 Ok(root.clone())
539 }
540
541 #[allow(dead_code)]
543 pub fn default_branch(&self) -> Result<Option<String>> {
544 let refs = self
545 .refs
546 .read()
547 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
548
549 if let Some(head) = refs.get("HEAD") {
550 if let Some(target) = head.strip_prefix("ref: ") {
551 return Ok(Some(target.to_string()));
552 }
553 }
554 Ok(None)
555 }
556
557 fn get_commit_tree(
559 &self,
560 commit_oid: &str,
561 objects: &HashMap<String, Vec<u8>>,
562 ) -> Option<String> {
563 let compressed = objects.get(commit_oid)?;
564
565 let mut decoder = ZlibDecoder::new(&compressed[..]);
567 let mut decompressed = Vec::new();
568 decoder.read_to_end(&mut decompressed).ok()?;
569
570 let null_pos = decompressed.iter().position(|&b| b == 0)?;
572 let content = &decompressed[null_pos + 1..];
573
574 let content_str = std::str::from_utf8(content).ok()?;
576 let first_line = content_str.lines().next()?;
577 first_line
578 .strip_prefix("tree ")
579 .map(|tree_hash| tree_hash.to_string())
580 }
581
582 fn get_object_content(
584 &self,
585 oid: &str,
586 objects: &HashMap<String, Vec<u8>>,
587 ) -> Option<(ObjectType, Vec<u8>)> {
588 let compressed = objects.get(oid)?;
589 Self::parse_compressed_object(compressed)
590 }
591
592 fn parse_compressed_object(compressed: &[u8]) -> Option<(ObjectType, Vec<u8>)> {
593 let mut decoder = ZlibDecoder::new(compressed);
594 let mut decompressed = Vec::new();
595 decoder.read_to_end(&mut decompressed).ok()?;
596
597 let null_pos = decompressed.iter().position(|&b| b == 0)?;
599 let header = std::str::from_utf8(&decompressed[..null_pos]).ok()?;
600 let obj_type = if header.starts_with("blob") {
601 ObjectType::Blob
602 } else if header.starts_with("tree") {
603 ObjectType::Tree
604 } else if header.starts_with("commit") {
605 ObjectType::Commit
606 } else if header.starts_with("tag") {
607 ObjectType::Tag
608 } else {
609 return None;
610 };
611 let content = decompressed[null_pos + 1..].to_vec();
612 Some((obj_type, content))
613 }
614
615 async fn load_base_compressed_object<S: Store>(
616 &self,
617 oid: &str,
618 base_tree: &HashTree<S>,
619 base_objects_cid: &Cid,
620 ) -> Result<Option<Vec<u8>>> {
621 if oid.len() != 40 {
622 return Err(Error::ObjectNotFound(oid.to_string()));
623 }
624
625 let path = format!("{}/{}", &oid[..2], &oid[2..]);
626 let Some(object_cid) = base_tree
627 .resolve_path(base_objects_cid, &path)
628 .await
629 .map_err(|e| Error::StorageError(format!("resolve {} in base objects: {}", oid, e)))?
630 else {
631 return Ok(None);
632 };
633
634 base_tree
635 .get(&object_cid, None)
636 .await
637 .map_err(|e| Error::StorageError(format!("read {} from base objects: {}", oid, e)))
638 }
639
640 async fn get_object_content_from_base<S: Store>(
641 &self,
642 oid: &str,
643 base_tree: &HashTree<S>,
644 base_objects_cid: &Cid,
645 ) -> Result<Option<(ObjectType, Vec<u8>)>> {
646 let Some(compressed) = self
647 .load_base_compressed_object(oid, base_tree, base_objects_cid)
648 .await?
649 else {
650 return Ok(None);
651 };
652
653 Ok(Self::parse_compressed_object(&compressed))
654 }
655
656 async fn get_tree_entries_from_sources<S: Store>(
657 &self,
658 tree_oid: &str,
659 objects: &HashMap<String, Vec<u8>>,
660 base_tree: Option<&HashTree<S>>,
661 base_objects_cid: Option<&Cid>,
662 ) -> Result<Vec<super::object::TreeEntry>> {
663 let object = match self.get_object_content(tree_oid, objects) {
664 Some(object) => Some(object),
665 None => {
666 if let (Some(base_tree), Some(base_objects_cid)) = (base_tree, base_objects_cid) {
667 self.get_object_content_from_base(tree_oid, base_tree, base_objects_cid)
668 .await?
669 } else {
670 None
671 }
672 }
673 }
674 .ok_or_else(|| Error::ObjectNotFound(tree_oid.to_string()))?;
675
676 if object.0 != ObjectType::Tree {
677 return Err(Error::InvalidObjectType(format!(
678 "expected tree, got {:?}",
679 object.0
680 )));
681 }
682
683 parse_tree(&object.1)
684 }
685
686 fn tree_entry_to_dir_entry(entry: &hashtree_core::TreeEntry) -> DirEntry {
687 DirEntry::from_cid(
688 &entry.name,
689 &Cid {
690 hash: entry.hash,
691 key: entry.key,
692 },
693 )
694 .with_size(entry.size)
695 .with_link_type(entry.link_type)
696 }
697
698 async fn import_object_from_base<S: Store>(
699 &self,
700 oid: &str,
701 objects: &mut HashMap<String, Vec<u8>>,
702 base_tree: &HashTree<S>,
703 base_objects_cid: &Cid,
704 ) -> Result<bool> {
705 if objects.contains_key(oid) {
706 return Ok(true);
707 }
708 let Some(compressed) = self
709 .load_base_compressed_object(oid, base_tree, base_objects_cid)
710 .await?
711 else {
712 return Ok(false);
713 };
714
715 objects.insert(oid.to_string(), compressed);
716 Ok(true)
717 }
718
719 fn seed_missing_object_from_base_boxed<'a, S: Store + 'a>(
720 &'a self,
721 oid: &'a str,
722 objects: &'a mut HashMap<String, Vec<u8>>,
723 base_tree: &'a HashTree<S>,
724 base_objects_cid: &'a Cid,
725 ) -> BoxFuture<'a, Result<bool>> {
726 Box::pin(self.import_object_from_base(oid, objects, base_tree, base_objects_cid))
727 }
728
729 fn peel_tag_target(&self, oid: &str, objects: &HashMap<String, Vec<u8>>) -> Option<String> {
730 let (obj_type, content) = self.get_object_content(oid, objects)?;
731 if obj_type != ObjectType::Tag {
732 return Some(oid.to_string());
733 }
734
735 let target = std::str::from_utf8(&content)
736 .ok()?
737 .lines()
738 .find_map(|line| line.strip_prefix("object "))
739 .map(str::trim)?
740 .to_string();
741
742 match self.get_object_content(&target, objects)?.0 {
743 ObjectType::Tag => self.peel_tag_target(&target, objects),
744 _ => Some(target),
745 }
746 }
747
748 fn build_info_refs_content(
749 &self,
750 refs: &HashMap<String, String>,
751 objects: &HashMap<String, Vec<u8>>,
752 ) -> String {
753 let mut lines = Vec::new();
754
755 for (name, value) in refs {
756 if name == "HEAD" {
757 continue;
758 }
759
760 let oid = value.trim().to_string();
761 lines.push((name.clone(), oid.clone()));
762
763 if name.starts_with("refs/tags/") {
764 if let Some(peeled) = self.peel_tag_target(&oid, objects) {
765 if peeled != oid {
766 lines.push((format!("{}^{{}}", name), peeled));
767 }
768 }
769 }
770 }
771
772 lines.sort_by(|a, b| a.0.cmp(&b.0));
773
774 let mut content = String::new();
775 for (name, oid) in lines {
776 content.push_str(&oid);
777 content.push('\t');
778 content.push_str(&name);
779 content.push('\n');
780 }
781 content
782 }
783
784 async fn build_info_dir(
785 &self,
786 refs: &HashMap<String, String>,
787 objects: &HashMap<String, Vec<u8>>,
788 ) -> Result<Cid> {
789 let info_refs = self.build_info_refs_content(refs, objects);
790 let (info_refs_cid, info_refs_size) = self
791 .tree
792 .put(info_refs.as_bytes())
793 .await
794 .map_err(|e| Error::StorageError(format!("put info/refs: {}", e)))?;
795
796 self.tree
797 .put_directory(vec![
798 DirEntry::from_cid("refs", &info_refs_cid).with_size(info_refs_size)
799 ])
800 .await
801 .map_err(|e| Error::StorageError(format!("put info dir: {}", e)))
802 }
803
804 pub fn build_tree(&self) -> Result<Cid> {
806 self.build_tree_with_base_objects::<LocalStore>(None, None, None)
807 }
808
809 pub fn build_tree_with_progress(&self, progress: &RepoTreeBuildProgress) -> Result<Cid> {
810 self.build_tree_with_base_objects_internal::<LocalStore>(None, None, None, Some(progress))
811 }
812
813 pub fn validate_root_contains_direct_refs(&self, root_cid: &Cid) -> Result<()> {
814 let direct_refs: Vec<String> = {
815 let refs = self
816 .refs
817 .read()
818 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
819 refs.values()
820 .filter(|value| {
821 !value.starts_with("ref: ")
822 && value.len() == 40
823 && value.bytes().all(|byte| byte.is_ascii_hexdigit())
824 })
825 .cloned()
826 .collect()
827 };
828
829 if direct_refs.is_empty() {
830 return Ok(());
831 }
832
833 let objects_dir = self
834 .runtime
835 .block_on(self.tree.resolve_path(root_cid, ".git/objects"))
836 .map_err(|e| Error::StorageError(format!("resolve .git/objects: {}", e)))?;
837 if objects_dir.is_none() {
838 return Err(Error::StorageError(
839 "built root is missing .git/objects".to_string(),
840 ));
841 }
842
843 for oid in direct_refs {
844 let object_path = format!(".git/objects/{}/{}", &oid[..2], &oid[2..]);
845 let object_cid = self
846 .runtime
847 .block_on(self.tree.resolve_path(root_cid, &object_path))
848 .map_err(|e| Error::StorageError(format!("resolve {}: {}", object_path, e)))?;
849 if object_cid.is_none() {
850 return Err(Error::ObjectNotFound(oid));
851 }
852 }
853
854 Ok(())
855 }
856
857 pub fn build_tree_with_base_objects<S: Store>(
858 &self,
859 base_tree: Option<&HashTree<S>>,
860 base_root: Option<&Cid>,
861 base_commit_sha: Option<&str>,
862 ) -> Result<Cid> {
863 self.build_tree_with_base_objects_internal(base_tree, base_root, base_commit_sha, None)
864 }
865
866 pub fn build_tree_with_base_objects_with_progress<S: Store>(
867 &self,
868 base_tree: Option<&HashTree<S>>,
869 base_root: Option<&Cid>,
870 base_commit_sha: Option<&str>,
871 progress: &RepoTreeBuildProgress,
872 ) -> Result<Cid> {
873 self.build_tree_with_base_objects_internal(
874 base_tree,
875 base_root,
876 base_commit_sha,
877 Some(progress),
878 )
879 }
880
881 fn build_tree_with_base_objects_internal<S: Store>(
882 &self,
883 base_tree: Option<&HashTree<S>>,
884 base_root: Option<&Cid>,
885 base_commit_sha: Option<&str>,
886 progress: Option<&RepoTreeBuildProgress>,
887 ) -> Result<Cid> {
888 if let Ok(root) = self.root_cid.read() {
890 if let Some(ref cid) = *root {
891 if let Some(progress) = progress {
892 progress.mark_done();
893 }
894 return Ok(cid.clone());
895 }
896 }
897
898 if let Err(err) = self.evict_if_needed() {
899 debug!("pre-build eviction skipped: {}", err);
900 }
901
902 let objects = self
903 .objects
904 .read()
905 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
906 let refs = self
907 .refs
908 .read()
909 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
910
911 let (default_branch, commit_sha) = if let Some(head) = refs.get("HEAD") {
913 let branch = head.strip_prefix("ref: ").map(String::from);
914 let sha = branch.as_ref().and_then(|b| refs.get(b)).cloned();
915 (branch, sha)
916 } else {
917 let mut branch_info: Option<(String, String)> = None;
919 for (ref_name, sha) in refs.iter() {
920 if ref_name.starts_with("refs/heads/") {
921 branch_info = Some((ref_name.clone(), sha.clone()));
922 break;
923 }
924 }
925 match branch_info {
926 Some((branch, sha)) => (Some(branch), Some(sha)),
927 None => (None, None),
928 }
929 };
930
931 let tree_sha = commit_sha
933 .as_ref()
934 .and_then(|sha| self.get_commit_tree(sha, &objects));
935
936 let mut objects_clone = objects.clone();
938
939 let base_objects_cid = if let (Some(base_tree), Some(base_root)) = (base_tree, base_root) {
940 self.runtime
941 .block_on(base_tree.resolve_path(base_root, ".git/objects"))
942 .map_err(|e| Error::StorageError(format!("resolve base .git/objects: {}", e)))?
943 } else {
944 None
945 };
946
947 let base_tree_sha =
948 if let (Some(base_tree), Some(base_objects_cid), Some(base_commit_sha)) =
949 (base_tree, base_objects_cid.as_ref(), base_commit_sha)
950 {
951 self.runtime.block_on(async {
952 let Some((obj_type, content)) = self
953 .get_object_content_from_base(base_commit_sha, base_tree, base_objects_cid)
954 .await?
955 else {
956 return Ok(None);
957 };
958
959 if obj_type != ObjectType::Commit {
960 return Err(Error::InvalidObjectType(format!(
961 "expected commit, got {:?}",
962 obj_type
963 )));
964 }
965
966 let content = std::str::from_utf8(&content).map_err(|_| {
967 Error::InvalidObjectFormat("commit: invalid utf-8".to_string())
968 })?;
969 Ok::<Option<String>, Error>(
970 content
971 .lines()
972 .find_map(|line| line.strip_prefix("tree "))
973 .map(str::trim)
974 .map(str::to_string),
975 )
976 })?
977 } else {
978 None
979 };
980
981 let base_root_entries = if let (Some(base_tree), Some(base_root)) = (base_tree, base_root) {
982 Some(
983 self.runtime
984 .block_on(base_tree.list_directory(base_root))
985 .map_err(|e| Error::StorageError(format!("list base root: {}", e)))?,
986 )
987 } else {
988 None
989 };
990
991 let root_cid = loop {
992 let build_result = self.runtime.block_on(async {
993 let objects_cid = self
995 .build_objects_dir_with_base(
996 &objects_clone,
997 base_tree,
998 base_objects_cid.as_ref(),
999 progress,
1000 )
1001 .await?;
1002
1003 if let Some(progress) = progress {
1005 progress.start_phase(RepoTreeBuildPhase::Refs, Some(refs.len()));
1006 }
1007 let refs_cid = self.build_refs_dir(&refs, progress).await?;
1008
1009 if let Some(progress) = progress {
1011 progress.start_phase(RepoTreeBuildPhase::Info, Some(1));
1012 }
1013 let info_cid = self.build_info_dir(&refs, &objects_clone).await?;
1014 if let Some(progress) = progress {
1015 progress.increment_processed();
1016 }
1017
1018 if let Some(progress) = progress {
1021 progress.start_phase(RepoTreeBuildPhase::Head, Some(1));
1022 }
1023 let head_content = refs.get("HEAD")
1024 .map(|h| if h.ends_with('\n') { h.clone() } else { format!("{}\n", h) })
1025 .or_else(|| default_branch.as_ref().map(|b| format!("ref: {}\n", b)))
1026 .unwrap_or_else(|| "ref: refs/heads/main\n".to_string());
1027 debug!("HEAD content: {:?}", head_content);
1028 let (head_cid, head_size) = self.tree.put(head_content.as_bytes()).await
1029 .map_err(|e| Error::StorageError(format!("put HEAD: {}", e)))?;
1030 debug!("HEAD hash: {}", hex::encode(head_cid.hash));
1031 if let Some(progress) = progress {
1032 progress.increment_processed();
1033 }
1034
1035 let mut git_entries = vec![
1037 DirEntry::from_cid("HEAD", &head_cid).with_size(head_size),
1038 DirEntry::from_cid("info", &info_cid).with_link_type(LinkType::Dir),
1039 DirEntry::from_cid("objects", &objects_cid).with_link_type(LinkType::Dir),
1040 DirEntry::from_cid("refs", &refs_cid).with_link_type(LinkType::Dir),
1041 ];
1042
1043 if let Some(ref branch) = default_branch {
1045 let config = format!(
1046 "[core]\n\trepositoryformatversion = 0\n\tfilemode = true\n\tbare = true\n[init]\n\tdefaultBranch = {}\n",
1047 branch.trim_start_matches("refs/heads/")
1048 );
1049 let (config_cid, config_size) = self.tree.put(config.as_bytes()).await
1050 .map_err(|e| Error::StorageError(format!("put config: {}", e)))?;
1051 git_entries.push(DirEntry::from_cid("config", &config_cid).with_size(config_size));
1052 }
1053
1054 if let Some(ref tree_oid) = tree_sha {
1056 if let Some(progress) = progress {
1057 progress.start_phase(RepoTreeBuildPhase::Index, None);
1058 }
1059 let index_result = if let (Some(base_tree), Some(base_objects_cid), Some(base_tree_sha)) =
1060 (base_tree, base_objects_cid.as_ref(), base_tree_sha.as_deref())
1061 {
1062 self.build_index_file_with_base(
1063 tree_oid,
1064 &objects_clone,
1065 base_tree,
1066 base_objects_cid,
1067 base_tree_sha,
1068 base_root_entries.as_ref(),
1069 progress,
1070 )
1071 .await
1072 } else {
1073 self.build_index_file(tree_oid, &objects_clone, progress)
1074 };
1075 match index_result {
1076 Ok(index_data) => {
1077 let (index_cid, index_size) = self.tree.put(&index_data).await
1078 .map_err(|e| Error::StorageError(format!("put index: {}", e)))?;
1079 git_entries.push(DirEntry::from_cid("index", &index_cid).with_size(index_size));
1080 info!("Added git index file ({} bytes)", index_data.len());
1081 }
1082 Err(e) => {
1083 debug!("Failed to build git index file: {} - continuing without index", e);
1084 }
1085 }
1086 }
1087
1088 if let Some(progress) = progress {
1089 progress.start_phase(RepoTreeBuildPhase::GitDir, Some(1));
1090 }
1091 let git_cid = self.tree.put_directory(git_entries).await
1092 .map_err(|e| Error::StorageError(format!("put .git: {}", e)))?;
1093 if let Some(progress) = progress {
1094 progress.increment_processed();
1095 }
1096
1097 let mut root_entries = vec![DirEntry::from_cid(".git", &git_cid).with_link_type(LinkType::Dir)];
1100
1101 if let Some(ref tree_oid) = tree_sha {
1103 if let Some(progress) = progress {
1104 progress.start_phase(RepoTreeBuildPhase::WorkingTree, None);
1105 }
1106 let working_tree_entries = if let (Some(base_tree), Some(base_objects_cid), Some(base_tree_sha)) =
1107 (base_tree, base_objects_cid.as_ref(), base_tree_sha.as_deref())
1108 {
1109 self.build_working_tree_entries_with_base(
1110 tree_oid,
1111 &objects_clone,
1112 base_tree,
1113 base_objects_cid,
1114 base_tree_sha,
1115 base_root_entries.as_ref(),
1116 progress,
1117 )
1118 .await?
1119 } else {
1120 self.build_working_tree_entries(tree_oid, &objects_clone, progress).await?
1121 };
1122 root_entries.extend(working_tree_entries);
1123 info!("Added {} working tree entries to root", root_entries.len() - 1);
1124 }
1125
1126 root_entries.sort_by(|a, b| a.name.cmp(&b.name));
1128
1129 if let Some(progress) = progress {
1130 progress.start_phase(RepoTreeBuildPhase::Root, Some(1));
1131 }
1132 let root_cid = self.tree.put_directory(root_entries).await
1133 .map_err(|e| Error::StorageError(format!("put root: {}", e)))?;
1134 if let Some(progress) = progress {
1135 progress.increment_processed();
1136 }
1137
1138 info!("Built hashtree root: {} (encrypted: {}) (.git dir: {})",
1139 hex::encode(root_cid.hash),
1140 root_cid.key.is_some(),
1141 hex::encode(git_cid.hash));
1142
1143 Ok::<Cid, Error>(root_cid)
1144 });
1145
1146 match build_result {
1147 Ok(root_cid) => break root_cid,
1148 Err(Error::ObjectNotFound(oid))
1149 if base_tree.is_some() && base_objects_cid.is_some() =>
1150 {
1151 let imported =
1152 self.runtime
1153 .block_on(self.seed_missing_object_from_base_boxed(
1154 &oid,
1155 &mut objects_clone,
1156 base_tree.expect("checked is_some"),
1157 base_objects_cid.as_ref().expect("checked is_some"),
1158 ))?;
1159 if imported {
1160 continue;
1161 }
1162 return Err(Error::ObjectNotFound(oid).into());
1163 }
1164 Err(err) => return Err(err.into()),
1165 }
1166 };
1167
1168 if let Ok(mut root) = self.root_cid.write() {
1170 *root = Some(root_cid.clone());
1171 }
1172 if let Some(progress) = progress {
1173 progress.mark_done();
1174 }
1175
1176 Ok(root_cid)
1177 }
1178
1179 async fn build_working_tree_entries(
1181 &self,
1182 tree_oid: &str,
1183 objects: &HashMap<String, Vec<u8>>,
1184 progress: Option<&RepoTreeBuildProgress>,
1185 ) -> Result<Vec<DirEntry>> {
1186 let mut entries = Vec::new();
1187
1188 let (obj_type, content) = self
1190 .get_object_content(tree_oid, objects)
1191 .ok_or_else(|| Error::ObjectNotFound(tree_oid.to_string()))?;
1192
1193 if obj_type != ObjectType::Tree {
1194 return Err(Error::InvalidObjectType(format!(
1195 "expected tree, got {:?}",
1196 obj_type
1197 )));
1198 }
1199
1200 let tree_entries = parse_tree(&content)?;
1202
1203 for entry in tree_entries {
1204 let oid_hex = entry.oid.to_hex();
1205
1206 if entry.is_tree() {
1207 let sub_entries = self
1209 .build_working_tree_entries_boxed(&oid_hex, objects, progress)
1210 .await?;
1211
1212 let dir_cid =
1214 self.tree.put_directory(sub_entries).await.map_err(|e| {
1215 Error::StorageError(format!("put dir {}: {}", entry.name, e))
1216 })?;
1217
1218 entries
1220 .push(DirEntry::from_cid(&entry.name, &dir_cid).with_link_type(LinkType::Dir));
1221 if let Some(progress) = progress {
1222 progress.record_working_dir(false);
1223 }
1224 } else {
1225 if let Some((ObjectType::Blob, blob_content)) =
1227 self.get_object_content(&oid_hex, objects)
1228 {
1229 let (cid, size) = self.tree.put(&blob_content).await.map_err(|e| {
1231 Error::StorageError(format!("put blob {}: {}", entry.name, e))
1232 })?;
1233
1234 entries.push(DirEntry::from_cid(&entry.name, &cid).with_size(size));
1236 if let Some(progress) = progress {
1237 progress.record_working_file(false);
1238 }
1239 }
1240 }
1241 }
1242
1243 entries.sort_by(|a, b| a.name.cmp(&b.name));
1245
1246 Ok(entries)
1247 }
1248
1249 fn build_working_tree_entries_boxed<'a>(
1251 &'a self,
1252 tree_oid: &'a str,
1253 objects: &'a HashMap<String, Vec<u8>>,
1254 progress: Option<&'a RepoTreeBuildProgress>,
1255 ) -> BoxFuture<'a, Result<Vec<DirEntry>>> {
1256 Box::pin(self.build_working_tree_entries(tree_oid, objects, progress))
1257 }
1258
1259 fn build_working_tree_entries_with_base_recursive_boxed<'a, S: Store + 'a>(
1260 &'a self,
1261 tree_oid: &'a str,
1262 objects: &'a HashMap<String, Vec<u8>>,
1263 base_tree: &'a HashTree<S>,
1264 base_objects_cid: &'a Cid,
1265 old_tree_oid: Option<&'a str>,
1266 old_dir_entries: Option<&'a Vec<hashtree_core::TreeEntry>>,
1267 progress: Option<&'a RepoTreeBuildProgress>,
1268 ) -> BoxFuture<'a, Result<Vec<DirEntry>>> {
1269 Box::pin(async move {
1270 let tree_entries = self
1271 .get_tree_entries_from_sources(
1272 tree_oid,
1273 objects,
1274 Some(base_tree),
1275 Some(base_objects_cid),
1276 )
1277 .await?;
1278 let old_tree_entries = if let Some(old_tree_oid) = old_tree_oid {
1279 Some(
1280 self.get_tree_entries_from_sources(
1281 old_tree_oid,
1282 objects,
1283 Some(base_tree),
1284 Some(base_objects_cid),
1285 )
1286 .await?,
1287 )
1288 } else {
1289 None
1290 };
1291
1292 let mut old_tree_entry_map = old_tree_entries
1293 .unwrap_or_default()
1294 .into_iter()
1295 .map(|entry| (entry.name.clone(), entry))
1296 .collect::<HashMap<_, _>>();
1297 let mut old_dir_entry_map = old_dir_entries
1298 .cloned()
1299 .unwrap_or_default()
1300 .into_iter()
1301 .map(|entry| (entry.name.clone(), entry))
1302 .collect::<HashMap<_, _>>();
1303
1304 let mut entries = Vec::new();
1305 for entry in tree_entries {
1306 let oid_hex = entry.oid.to_hex();
1307 let old_tree_entry = old_tree_entry_map.remove(&entry.name);
1308 let old_dir_entry = old_dir_entry_map.remove(&entry.name);
1309
1310 if entry.is_tree() {
1311 if let (Some(old_tree_entry), Some(old_dir_entry)) =
1312 (old_tree_entry.as_ref(), old_dir_entry.as_ref())
1313 {
1314 if old_tree_entry.is_tree()
1315 && old_tree_entry.oid == entry.oid
1316 && old_dir_entry.link_type == LinkType::Dir
1317 {
1318 entries.push(Self::tree_entry_to_dir_entry(old_dir_entry));
1319 if let Some(progress) = progress {
1320 progress.record_working_dir(true);
1321 }
1322 continue;
1323 }
1324 }
1325
1326 let old_subtree_oid = old_tree_entry
1327 .as_ref()
1328 .filter(|old| old.is_tree())
1329 .map(|old| old.oid.to_hex());
1330 let old_subdir_entries = if let Some(old_dir_entry) = old_dir_entry.as_ref() {
1331 if old_dir_entry.link_type == LinkType::Dir {
1332 Some(
1333 base_tree
1334 .list_directory(&Cid {
1335 hash: old_dir_entry.hash,
1336 key: old_dir_entry.key,
1337 })
1338 .await
1339 .map_err(|e| {
1340 Error::StorageError(format!(
1341 "list base working dir {}: {}",
1342 entry.name, e
1343 ))
1344 })?,
1345 )
1346 } else {
1347 None
1348 }
1349 } else {
1350 None
1351 };
1352
1353 let sub_entries = self
1354 .build_working_tree_entries_with_base_recursive_boxed(
1355 &oid_hex,
1356 objects,
1357 base_tree,
1358 base_objects_cid,
1359 old_subtree_oid.as_deref(),
1360 old_subdir_entries.as_ref(),
1361 progress,
1362 )
1363 .await?;
1364 let dir_cid = self.tree.put_directory(sub_entries).await.map_err(|e| {
1365 Error::StorageError(format!("put dir {}: {}", entry.name, e))
1366 })?;
1367 entries.push(
1368 DirEntry::from_cid(&entry.name, &dir_cid).with_link_type(LinkType::Dir),
1369 );
1370 if let Some(progress) = progress {
1371 progress.record_working_dir(false);
1372 }
1373 continue;
1374 }
1375
1376 if let (Some(old_tree_entry), Some(old_dir_entry)) =
1377 (old_tree_entry.as_ref(), old_dir_entry.as_ref())
1378 {
1379 if !old_tree_entry.is_tree()
1380 && old_tree_entry.oid == entry.oid
1381 && old_dir_entry.link_type != LinkType::Dir
1382 {
1383 entries.push(Self::tree_entry_to_dir_entry(old_dir_entry));
1384 if let Some(progress) = progress {
1385 progress.record_working_file(true);
1386 }
1387 continue;
1388 }
1389 }
1390
1391 let blob_content = match self.get_object_content(&oid_hex, objects) {
1392 Some((ObjectType::Blob, blob_content)) => blob_content,
1393 Some((obj_type, _)) => {
1394 return Err(Error::InvalidObjectType(format!(
1395 "expected blob, got {:?}",
1396 obj_type
1397 )));
1398 }
1399 None => match self
1400 .get_object_content_from_base(&oid_hex, base_tree, base_objects_cid)
1401 .await?
1402 {
1403 Some((ObjectType::Blob, blob_content)) => blob_content,
1404 Some((obj_type, _)) => {
1405 return Err(Error::InvalidObjectType(format!(
1406 "expected blob, got {:?}",
1407 obj_type
1408 )));
1409 }
1410 None => return Err(Error::ObjectNotFound(oid_hex)),
1411 },
1412 };
1413
1414 let (cid, size) =
1415 self.tree.put(&blob_content).await.map_err(|e| {
1416 Error::StorageError(format!("put blob {}: {}", entry.name, e))
1417 })?;
1418 entries.push(DirEntry::from_cid(&entry.name, &cid).with_size(size));
1419 if let Some(progress) = progress {
1420 progress.record_working_file(false);
1421 }
1422 }
1423
1424 entries.sort_by(|a, b| a.name.cmp(&b.name));
1425 Ok(entries)
1426 })
1427 }
1428
1429 async fn build_working_tree_entries_with_base<S: Store>(
1430 &self,
1431 tree_oid: &str,
1432 objects: &HashMap<String, Vec<u8>>,
1433 base_tree: &HashTree<S>,
1434 base_objects_cid: &Cid,
1435 base_tree_oid: &str,
1436 base_root_entries: Option<&Vec<hashtree_core::TreeEntry>>,
1437 progress: Option<&RepoTreeBuildProgress>,
1438 ) -> Result<Vec<DirEntry>> {
1439 let root_entries = base_root_entries.map(|entries| {
1440 entries
1441 .iter()
1442 .filter(|entry| entry.name != ".git")
1443 .cloned()
1444 .collect::<Vec<_>>()
1445 });
1446
1447 self.build_working_tree_entries_with_base_recursive_boxed(
1448 tree_oid,
1449 objects,
1450 base_tree,
1451 base_objects_cid,
1452 Some(base_tree_oid),
1453 root_entries.as_ref(),
1454 progress,
1455 )
1456 .await
1457 }
1458
1459 async fn build_objects_prefix_dir(
1460 &self,
1461 prefix: &str,
1462 old_entries: Option<Vec<hashtree_core::TreeEntry>>,
1463 new_objects: &[(String, Vec<u8>)],
1464 progress: Option<&RepoTreeBuildProgress>,
1465 ) -> Result<Cid> {
1466 let mut sub_entries: BTreeMap<String, DirEntry> = old_entries
1467 .unwrap_or_default()
1468 .into_iter()
1469 .map(|entry| (entry.name.clone(), Self::tree_entry_to_dir_entry(&entry)))
1470 .collect();
1471
1472 for (suffix, data) in new_objects {
1473 let (cid, size) = self.tree.put(data).await.map_err(|e| {
1474 Error::StorageError(format!("put object {}{}: {}", prefix, suffix, e))
1475 })?;
1476 sub_entries.insert(
1477 suffix.clone(),
1478 DirEntry::from_cid(suffix, &cid).with_size(size),
1479 );
1480 if let Some(progress) = progress {
1481 progress.record_object_blob();
1482 }
1483 }
1484
1485 let cid = self
1486 .tree
1487 .put_directory(sub_entries.into_values().collect())
1488 .await
1489 .map_err(|e| Error::StorageError(format!("put objects/{}: {}", prefix, e)))?;
1490 if let Some(progress) = progress {
1491 progress.record_object_prefix();
1492 }
1493 Ok(cid)
1494 }
1495
1496 async fn build_objects_dir_with_base<S: Store>(
1499 &self,
1500 objects: &HashMap<String, Vec<u8>>,
1501 base_tree: Option<&HashTree<S>>,
1502 base_objects_cid: Option<&Cid>,
1503 progress: Option<&RepoTreeBuildProgress>,
1504 ) -> Result<Cid> {
1505 if let Some(progress) = progress {
1506 progress.start_phase(RepoTreeBuildPhase::Objects, Some(objects.len()));
1507 }
1508
1509 let mut buckets: BTreeMap<String, Vec<(String, Vec<u8>)>> = BTreeMap::new();
1510 for (oid, data) in objects {
1511 let prefix = &oid[..2];
1512 let suffix = &oid[2..];
1513 buckets
1514 .entry(prefix.to_string())
1515 .or_default()
1516 .push((suffix.to_string(), data.clone()));
1517 }
1518
1519 let mut top_entries: BTreeMap<String, DirEntry> = BTreeMap::new();
1520 let mut merged_prefixes = std::collections::HashSet::new();
1521
1522 if let (Some(base_tree), Some(base_objects_cid)) = (base_tree, base_objects_cid) {
1523 for entry in base_tree
1524 .list_directory(base_objects_cid)
1525 .await
1526 .map_err(|e| Error::StorageError(format!("list base objects dir: {}", e)))?
1527 {
1528 if entry.name == "info" {
1529 top_entries.insert(entry.name.clone(), Self::tree_entry_to_dir_entry(&entry));
1530 continue;
1531 }
1532
1533 let Some(delta_objects) = buckets.get(&entry.name) else {
1534 top_entries.insert(entry.name.clone(), Self::tree_entry_to_dir_entry(&entry));
1535 continue;
1536 };
1537
1538 let prefix_cid = Cid {
1539 hash: entry.hash,
1540 key: entry.key,
1541 };
1542 let old_prefix_entries = if entry.link_type == LinkType::Dir {
1543 Some(base_tree.list_directory(&prefix_cid).await.map_err(|e| {
1544 Error::StorageError(format!("list base objects/{} dir: {}", entry.name, e))
1545 })?)
1546 } else {
1547 None
1548 };
1549 let merged_cid = self
1550 .build_objects_prefix_dir(
1551 &entry.name,
1552 old_prefix_entries,
1553 delta_objects,
1554 progress,
1555 )
1556 .await?;
1557 top_entries.insert(
1558 entry.name.clone(),
1559 DirEntry::from_cid(&entry.name, &merged_cid).with_link_type(LinkType::Dir),
1560 );
1561 merged_prefixes.insert(entry.name);
1562 }
1563 }
1564
1565 for (prefix, objs) in &buckets {
1566 if merged_prefixes.contains(prefix) {
1567 continue;
1568 }
1569 let sub_cid = self
1570 .build_objects_prefix_dir(prefix, None, objs, progress)
1571 .await?;
1572 top_entries.insert(
1573 prefix.clone(),
1574 DirEntry::from_cid(prefix, &sub_cid).with_link_type(LinkType::Dir),
1575 );
1576 }
1577
1578 if !top_entries.contains_key("info") {
1579 let (packs_cid, packs_size) = self
1580 .tree
1581 .put(b"")
1582 .await
1583 .map_err(|e| Error::StorageError(format!("put objects/info/packs: {}", e)))?;
1584 let info_cid = self
1585 .tree
1586 .put_directory(vec![
1587 DirEntry::from_cid("packs", &packs_cid).with_size(packs_size)
1588 ])
1589 .await
1590 .map_err(|e| Error::StorageError(format!("put objects/info: {}", e)))?;
1591 top_entries.insert(
1592 "info".to_string(),
1593 DirEntry::from_cid("info", &info_cid).with_link_type(LinkType::Dir),
1594 );
1595 }
1596
1597 let entry_count = top_entries.len();
1598 let cid = self
1599 .tree
1600 .put_directory(top_entries.into_values().collect())
1601 .await
1602 .map_err(|e| Error::StorageError(format!("put objects dir: {}", e)))?;
1603
1604 debug!(
1605 "Built objects dir with {} entries: {}",
1606 entry_count,
1607 hex::encode(cid.hash)
1608 );
1609 Ok(cid)
1610 }
1611
1612 async fn build_refs_dir(
1614 &self,
1615 refs: &HashMap<String, String>,
1616 progress: Option<&RepoTreeBuildProgress>,
1617 ) -> Result<Cid> {
1618 let mut root = RefDirectory::default();
1619
1620 for (ref_name, value) in refs {
1621 let parts: Vec<&str> = ref_name.split('/').collect();
1622 if parts.len() >= 3 && parts[0] == "refs" {
1623 root.insert(&parts[1..], value.clone());
1624 }
1625 }
1626
1627 let mut ref_entries = self
1628 .build_ref_entries_recursive(&root, "refs", progress)
1629 .await?;
1630
1631 if ref_entries.is_empty() {
1632 let empty_cid = self
1634 .tree
1635 .put_directory(vec![])
1636 .await
1637 .map_err(|e| Error::StorageError(format!("put empty refs: {}", e)))?;
1638 return Ok(empty_cid);
1639 }
1640
1641 ref_entries.sort_by(|a, b| a.name.cmp(&b.name));
1642
1643 let refs_cid = self
1644 .tree
1645 .put_directory(ref_entries)
1646 .await
1647 .map_err(|e| Error::StorageError(format!("put refs dir: {}", e)))?;
1648 debug!("refs dir -> {}", hex::encode(refs_cid.hash));
1649 Ok(refs_cid)
1650 }
1651
1652 fn build_ref_entries_recursive<'a>(
1653 &'a self,
1654 dir: &'a RefDirectory,
1655 prefix: &'a str,
1656 progress: Option<&'a RepoTreeBuildProgress>,
1657 ) -> BoxFuture<'a, Result<Vec<DirEntry>>> {
1658 Box::pin(async move {
1659 let mut entries = Vec::new();
1660
1661 for (name, value) in &dir.files {
1662 let (cid, size) = self
1663 .tree
1664 .put(value.as_bytes())
1665 .await
1666 .map_err(|e| Error::StorageError(format!("put ref: {}", e)))?;
1667 debug!("{}/{} -> blob {}", prefix, name, hex::encode(cid.hash));
1668 entries.push(DirEntry::from_cid(name, &cid).with_size(size));
1669 if let Some(progress) = progress {
1670 progress.increment_processed();
1671 }
1672 }
1673
1674 for (name, child) in &dir.dirs {
1675 let child_prefix = format!("{prefix}/{name}");
1676 let child_entries = self
1677 .build_ref_entries_recursive(child, &child_prefix, progress)
1678 .await?;
1679 let child_cid =
1680 self.tree.put_directory(child_entries).await.map_err(|e| {
1681 Error::StorageError(format!("put {child_prefix} dir: {}", e))
1682 })?;
1683 debug!("{} dir -> {}", child_prefix, hex::encode(child_cid.hash));
1684 entries.push(DirEntry::from_cid(name, &child_cid).with_link_type(LinkType::Dir));
1685 }
1686
1687 entries.sort_by(|a, b| a.name.cmp(&b.name));
1688 Ok(entries)
1689 })
1690 }
1691
1692 fn build_index_file(
1695 &self,
1696 tree_oid: &str,
1697 objects: &HashMap<String, Vec<u8>>,
1698 progress: Option<&RepoTreeBuildProgress>,
1699 ) -> Result<Vec<u8>> {
1700 let mut entries: Vec<(String, [u8; 20], u32, u32)> = Vec::new(); self.collect_tree_entries_for_index(tree_oid, objects, "", &mut entries, progress)?;
1703
1704 self.build_index_bytes(entries)
1705 }
1706
1707 async fn build_index_file_with_base<S: Store>(
1708 &self,
1709 tree_oid: &str,
1710 objects: &HashMap<String, Vec<u8>>,
1711 base_tree: &HashTree<S>,
1712 base_objects_cid: &Cid,
1713 base_tree_oid: &str,
1714 base_root_entries: Option<&Vec<hashtree_core::TreeEntry>>,
1715 progress: Option<&RepoTreeBuildProgress>,
1716 ) -> Result<Vec<u8>> {
1717 let mut entries: Vec<(String, [u8; 20], u32, u32)> = Vec::new();
1718 let root_entries = base_root_entries.map(|entries| {
1719 entries
1720 .iter()
1721 .filter(|entry| entry.name != ".git")
1722 .cloned()
1723 .collect::<Vec<_>>()
1724 });
1725
1726 self.collect_tree_entries_for_index_with_base_boxed(
1727 tree_oid,
1728 objects,
1729 base_tree,
1730 base_objects_cid,
1731 Some(base_tree_oid),
1732 root_entries.as_ref(),
1733 "",
1734 &mut entries,
1735 progress,
1736 )
1737 .await?;
1738
1739 self.build_index_bytes(entries)
1740 }
1741
1742 fn build_index_bytes(&self, mut entries: Vec<(String, [u8; 20], u32, u32)>) -> Result<Vec<u8>> {
1743 entries.sort_by(|a, b| a.0.cmp(&b.0));
1745
1746 let entry_count = entries.len() as u32;
1747 debug!("Building git index with {} entries", entry_count);
1748
1749 let mut index_data = Vec::new();
1751
1752 index_data.extend_from_slice(b"DIRC");
1754 index_data.extend_from_slice(&2u32.to_be_bytes()); index_data.extend_from_slice(&entry_count.to_be_bytes());
1756
1757 let now_sec = std::time::SystemTime::now()
1759 .duration_since(std::time::UNIX_EPOCH)
1760 .unwrap_or_default()
1761 .as_secs() as u32;
1762
1763 for (path, sha1, mode, size) in &entries {
1764 let entry_start = index_data.len();
1765
1766 index_data.extend_from_slice(&now_sec.to_be_bytes());
1768 index_data.extend_from_slice(&0u32.to_be_bytes());
1769 index_data.extend_from_slice(&now_sec.to_be_bytes());
1771 index_data.extend_from_slice(&0u32.to_be_bytes());
1772 index_data.extend_from_slice(&0u32.to_be_bytes());
1774 index_data.extend_from_slice(&0u32.to_be_bytes());
1775 index_data.extend_from_slice(&mode.to_be_bytes());
1777 index_data.extend_from_slice(&0u32.to_be_bytes());
1779 index_data.extend_from_slice(&0u32.to_be_bytes());
1780 index_data.extend_from_slice(&size.to_be_bytes());
1782 index_data.extend_from_slice(sha1);
1784 let path_len = std::cmp::min(path.len(), 0xFFF) as u16;
1786 index_data.extend_from_slice(&path_len.to_be_bytes());
1787 index_data.extend_from_slice(path.as_bytes());
1789 index_data.push(0); let entry_len = index_data.len() - entry_start;
1793 let padding = (8 - (entry_len % 8)) % 8;
1794 index_data.extend(std::iter::repeat_n(0, padding));
1795 }
1796
1797 let mut hasher = Sha1::new();
1799 hasher.update(&index_data);
1800 let checksum = hasher.finalize();
1801 index_data.extend_from_slice(&checksum);
1802
1803 debug!(
1804 "Built git index: {} bytes, {} entries",
1805 index_data.len(),
1806 entry_count
1807 );
1808 Ok(index_data)
1809 }
1810
1811 fn collect_tree_entries_for_index(
1813 &self,
1814 tree_oid: &str,
1815 objects: &HashMap<String, Vec<u8>>,
1816 prefix: &str,
1817 entries: &mut Vec<(String, [u8; 20], u32, u32)>,
1818 progress: Option<&RepoTreeBuildProgress>,
1819 ) -> Result<()> {
1820 let (obj_type, content) = self
1821 .get_object_content(tree_oid, objects)
1822 .ok_or_else(|| Error::ObjectNotFound(tree_oid.to_string()))?;
1823
1824 if obj_type != ObjectType::Tree {
1825 return Err(Error::InvalidObjectType(format!(
1826 "expected tree, got {:?}",
1827 obj_type
1828 )));
1829 }
1830
1831 let tree_entries = parse_tree(&content)?;
1832
1833 for entry in tree_entries {
1834 let path = if prefix.is_empty() {
1835 entry.name.clone()
1836 } else {
1837 format!("{}/{}", prefix, entry.name)
1838 };
1839
1840 let oid_hex = entry.oid.to_hex();
1841
1842 if entry.is_tree() {
1843 self.collect_tree_entries_for_index(&oid_hex, objects, &path, entries, progress)?;
1845 } else {
1846 if let Some((ObjectType::Blob, blob_content)) =
1848 self.get_object_content(&oid_hex, objects)
1849 {
1850 let mut sha1_bytes = [0u8; 20];
1852 if let Ok(bytes) = hex::decode(&oid_hex) {
1853 if bytes.len() == 20 {
1854 sha1_bytes.copy_from_slice(&bytes);
1855 }
1856 }
1857
1858 let mode = entry.mode;
1860 let size = blob_content.len() as u32;
1861
1862 entries.push((path, sha1_bytes, mode, size));
1863 if let Some(progress) = progress {
1864 progress.record_index_entry();
1865 }
1866 }
1867 }
1868 }
1869
1870 Ok(())
1871 }
1872
1873 fn collect_tree_entries_for_index_with_base_boxed<'a, S: Store + 'a>(
1874 &'a self,
1875 tree_oid: &'a str,
1876 objects: &'a HashMap<String, Vec<u8>>,
1877 base_tree: &'a HashTree<S>,
1878 base_objects_cid: &'a Cid,
1879 old_tree_oid: Option<&'a str>,
1880 old_dir_entries: Option<&'a Vec<hashtree_core::TreeEntry>>,
1881 prefix: &'a str,
1882 entries: &'a mut Vec<(String, [u8; 20], u32, u32)>,
1883 progress: Option<&'a RepoTreeBuildProgress>,
1884 ) -> BoxFuture<'a, Result<()>> {
1885 Box::pin(async move {
1886 let tree_entries = self
1887 .get_tree_entries_from_sources(
1888 tree_oid,
1889 objects,
1890 Some(base_tree),
1891 Some(base_objects_cid),
1892 )
1893 .await?;
1894 let old_tree_entries = if let Some(old_tree_oid) = old_tree_oid {
1895 Some(
1896 self.get_tree_entries_from_sources(
1897 old_tree_oid,
1898 objects,
1899 Some(base_tree),
1900 Some(base_objects_cid),
1901 )
1902 .await?,
1903 )
1904 } else {
1905 None
1906 };
1907 let mut old_tree_entry_map = old_tree_entries
1908 .unwrap_or_default()
1909 .into_iter()
1910 .map(|entry| (entry.name.clone(), entry))
1911 .collect::<HashMap<_, _>>();
1912 let mut old_dir_entry_map = old_dir_entries
1913 .cloned()
1914 .unwrap_or_default()
1915 .into_iter()
1916 .map(|entry| (entry.name.clone(), entry))
1917 .collect::<HashMap<_, _>>();
1918
1919 for entry in tree_entries {
1920 let path = if prefix.is_empty() {
1921 entry.name.clone()
1922 } else {
1923 format!("{}/{}", prefix, entry.name)
1924 };
1925 let oid_hex = entry.oid.to_hex();
1926 let old_tree_entry = old_tree_entry_map.remove(&entry.name);
1927 let old_dir_entry = old_dir_entry_map.remove(&entry.name);
1928
1929 if entry.is_tree() {
1930 let old_subtree_oid = old_tree_entry
1931 .as_ref()
1932 .filter(|old| old.is_tree())
1933 .map(|old| old.oid.to_hex());
1934 let old_subdir_entries = if let Some(old_dir_entry) = old_dir_entry.as_ref() {
1935 if old_dir_entry.link_type == LinkType::Dir {
1936 Some(
1937 base_tree
1938 .list_directory(&Cid {
1939 hash: old_dir_entry.hash,
1940 key: old_dir_entry.key,
1941 })
1942 .await
1943 .map_err(|e| {
1944 Error::StorageError(format!(
1945 "list base working dir {}: {}",
1946 path, e
1947 ))
1948 })?,
1949 )
1950 } else {
1951 None
1952 }
1953 } else {
1954 None
1955 };
1956
1957 self.collect_tree_entries_for_index_with_base_boxed(
1958 &oid_hex,
1959 objects,
1960 base_tree,
1961 base_objects_cid,
1962 old_subtree_oid.as_deref(),
1963 old_subdir_entries.as_ref(),
1964 &path,
1965 entries,
1966 progress,
1967 )
1968 .await?;
1969 continue;
1970 }
1971
1972 let mut sha1_bytes = [0u8; 20];
1973 if let Ok(bytes) = hex::decode(&oid_hex) {
1974 if bytes.len() == 20 {
1975 sha1_bytes.copy_from_slice(&bytes);
1976 }
1977 }
1978
1979 let size = if let (Some(old_tree_entry), Some(old_dir_entry)) =
1980 (old_tree_entry.as_ref(), old_dir_entry.as_ref())
1981 {
1982 if !old_tree_entry.is_tree()
1983 && old_tree_entry.oid == entry.oid
1984 && old_dir_entry.link_type != LinkType::Dir
1985 {
1986 old_dir_entry.size as u32
1987 } else {
1988 self.blob_size_from_sources(
1989 &oid_hex,
1990 objects,
1991 Some(base_tree),
1992 Some(base_objects_cid),
1993 )
1994 .await? as u32
1995 }
1996 } else {
1997 self.blob_size_from_sources(
1998 &oid_hex,
1999 objects,
2000 Some(base_tree),
2001 Some(base_objects_cid),
2002 )
2003 .await? as u32
2004 };
2005
2006 entries.push((path, sha1_bytes, entry.mode, size));
2007 if let Some(progress) = progress {
2008 progress.record_index_entry();
2009 }
2010 }
2011
2012 Ok(())
2013 })
2014 }
2015
2016 async fn blob_size_from_sources<S: Store>(
2017 &self,
2018 oid: &str,
2019 objects: &HashMap<String, Vec<u8>>,
2020 base_tree: Option<&HashTree<S>>,
2021 base_objects_cid: Option<&Cid>,
2022 ) -> Result<usize> {
2023 let object = match self.get_object_content(oid, objects) {
2024 Some(object) => Some(object),
2025 None => {
2026 if let (Some(base_tree), Some(base_objects_cid)) = (base_tree, base_objects_cid) {
2027 self.get_object_content_from_base(oid, base_tree, base_objects_cid)
2028 .await?
2029 } else {
2030 None
2031 }
2032 }
2033 }
2034 .ok_or_else(|| Error::ObjectNotFound(oid.to_string()))?;
2035
2036 if object.0 != ObjectType::Blob {
2037 return Err(Error::InvalidObjectType(format!(
2038 "expected blob, got {:?}",
2039 object.0
2040 )));
2041 }
2042
2043 Ok(object.1.len())
2044 }
2045
2046 pub fn store(&self) -> &Arc<LocalStore> {
2048 &self.store
2049 }
2050
2051 #[allow(dead_code)]
2053 pub fn hashtree(&self) -> &HashTree<LocalStore> {
2054 &self.tree
2055 }
2056
2057 #[allow(dead_code)]
2059 pub fn push_to_file_servers(
2060 &self,
2061 blossom: &hashtree_blossom::BlossomClient,
2062 ) -> Result<(usize, usize)> {
2063 let hashes = self
2064 .store
2065 .list()
2066 .map_err(|e| Error::StorageError(format!("list hashes: {}", e)))?;
2067
2068 info!("Pushing {} blobs to file servers", hashes.len());
2069
2070 let mut uploaded = 0;
2071 let mut existed = 0;
2072
2073 self.runtime.block_on(async {
2074 for hash in &hashes {
2075 let hex_hash = hex::encode(hash);
2076 let data = match self.store.get_sync(hash) {
2077 Ok(Some(d)) => d,
2078 _ => continue,
2079 };
2080
2081 match blossom.upload_if_missing(&data).await {
2082 Ok((_, true)) => {
2083 debug!("Uploaded {}", &hex_hash[..12]);
2084 uploaded += 1;
2085 }
2086 Ok((_, false)) => {
2087 existed += 1;
2088 }
2089 Err(e) => {
2090 debug!("Failed to upload {}: {}", &hex_hash[..12], e);
2091 }
2092 }
2093 }
2094 });
2095
2096 info!(
2097 "Upload complete: {} new, {} already existed",
2098 uploaded, existed
2099 );
2100 Ok((uploaded, existed))
2101 }
2102
2103 #[allow(dead_code)]
2105 pub fn clear(&self) -> Result<()> {
2106 let mut objects = self
2107 .objects
2108 .write()
2109 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
2110 let mut refs = self
2111 .refs
2112 .write()
2113 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
2114 let mut root = self
2115 .root_cid
2116 .write()
2117 .map_err(|e| Error::StorageError(format!("lock: {}", e)))?;
2118
2119 objects.clear();
2120 refs.clear();
2121 *root = None;
2122 Ok(())
2123 }
2124}
2125
2126#[cfg(test)]
2127mod tests {
2128 use super::*;
2129 use hashtree_core::store::Store;
2130 use hashtree_core::{sha256, LinkType};
2131 use std::io::{Read, Write};
2132 use std::net::{TcpListener, TcpStream};
2133 use std::path::Path;
2134 use std::process::{Child, Command, Stdio};
2135 #[cfg(feature = "lmdb")]
2136 use std::sync::atomic::{AtomicUsize, Ordering};
2137 use std::time::{Duration, Instant};
2138 use tempfile::TempDir;
2139
2140 fn create_test_storage() -> (GitStorage, TempDir) {
2141 let temp_dir = TempDir::new().unwrap();
2142 let storage = GitStorage::open(temp_dir.path()).unwrap();
2143 (storage, temp_dir)
2144 }
2145
2146 fn create_test_storage_with_limit(max_size_bytes: u64) -> (GitStorage, TempDir) {
2147 let temp_dir = TempDir::new().unwrap();
2148 let storage = GitStorage::open_with_backend_and_max_bytes(
2149 temp_dir.path(),
2150 StorageBackend::Fs,
2151 max_size_bytes,
2152 )
2153 .unwrap();
2154 (storage, temp_dir)
2155 }
2156
2157 fn local_total_bytes(storage: &GitStorage) -> u64 {
2158 match storage.store().as_ref() {
2159 LocalStore::Fs(store) => store.stats().unwrap().total_bytes,
2160 #[cfg(feature = "lmdb")]
2161 LocalStore::Lmdb(store) => store.stats().unwrap().total_bytes,
2162 }
2163 }
2164
2165 fn write_test_commit(storage: &GitStorage) -> ObjectId {
2166 let blob_oid = storage
2167 .write_raw_object(ObjectType::Blob, b"hello from hashtree\n")
2168 .unwrap();
2169
2170 let mut tree_content = Vec::new();
2171 tree_content.extend_from_slice(b"100644 README.md\0");
2172 tree_content.extend_from_slice(&hex::decode(blob_oid.to_hex()).unwrap());
2173 let tree_oid = storage
2174 .write_raw_object(ObjectType::Tree, &tree_content)
2175 .unwrap();
2176
2177 let commit_content = format!(
2178 "tree {}\nauthor Test User <test@example.com> 0 +0000\ncommitter Test User <test@example.com> 0 +0000\n\nInitial commit\n",
2179 tree_oid.to_hex()
2180 );
2181 storage
2182 .write_raw_object(ObjectType::Commit, commit_content.as_bytes())
2183 .unwrap()
2184 }
2185
2186 fn export_tree_to_fs<S: Store>(
2187 runtime: &RuntimeExecutor,
2188 tree: &HashTree<S>,
2189 cid: &Cid,
2190 dst: &Path,
2191 ) {
2192 std::fs::create_dir_all(dst).unwrap();
2193 let entries = runtime.block_on(tree.list_directory(cid)).unwrap();
2194 for entry in entries {
2195 let entry_cid = Cid {
2196 hash: entry.hash,
2197 key: entry.key,
2198 };
2199 let path = dst.join(&entry.name);
2200 match entry.link_type {
2201 LinkType::Dir => export_tree_to_fs(runtime, tree, &entry_cid, &path),
2202 LinkType::Blob | LinkType::File => {
2203 let data = runtime
2204 .block_on(tree.get(&entry_cid, None))
2205 .unwrap()
2206 .unwrap();
2207 if let Some(parent) = path.parent() {
2208 std::fs::create_dir_all(parent).unwrap();
2209 }
2210 std::fs::write(path, data).unwrap();
2211 }
2212 }
2213 }
2214 }
2215
2216 fn spawn_http_server(root: &Path, port: u16) -> Child {
2217 Command::new("python3")
2218 .args([
2219 "-m",
2220 "http.server",
2221 &port.to_string(),
2222 "--bind",
2223 "127.0.0.1",
2224 ])
2225 .current_dir(root)
2226 .stdout(Stdio::null())
2227 .stderr(Stdio::null())
2228 .spawn()
2229 .expect("spawn python http server")
2230 }
2231
2232 fn wait_for_http_server(server: &mut Child, port: u16, path: &str) {
2233 let deadline = Instant::now() + Duration::from_secs(5);
2234
2235 loop {
2236 if let Some(status) = server.try_wait().expect("check http server status") {
2237 panic!("python http server exited before becoming ready: {status}");
2238 }
2239
2240 if let Ok(mut stream) = TcpStream::connect(("127.0.0.1", port)) {
2241 stream
2242 .set_read_timeout(Some(Duration::from_millis(200)))
2243 .expect("set read timeout");
2244 stream
2245 .set_write_timeout(Some(Duration::from_millis(200)))
2246 .expect("set write timeout");
2247 let request =
2248 format!("GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n");
2249 if stream.write_all(request.as_bytes()).is_ok() {
2250 let mut response = String::new();
2251 if stream.read_to_string(&mut response).is_ok()
2252 && response.starts_with("HTTP/1.0 200")
2253 {
2254 return;
2255 }
2256 }
2257 }
2258
2259 if Instant::now() >= deadline {
2260 panic!("python http server did not become ready on port {port}");
2261 }
2262 std::thread::sleep(Duration::from_millis(50));
2263 }
2264 }
2265
2266 #[test]
2267 fn test_import_ref() {
2268 let (storage, _temp) = create_test_storage();
2269
2270 storage
2272 .import_ref("refs/heads/main", "abc123def456")
2273 .unwrap();
2274
2275 assert!(storage.has_ref("refs/heads/main").unwrap());
2277
2278 let refs = storage.list_refs().unwrap();
2280 assert_eq!(
2281 refs.get("refs/heads/main"),
2282 Some(&"abc123def456".to_string())
2283 );
2284 }
2285
2286 #[cfg(feature = "lmdb")]
2287 #[test]
2288 fn test_local_store_falls_back_to_fs_when_lmdb_open_returns_enosys() {
2289 let temp_dir = TempDir::new().unwrap();
2290 let fs_calls = AtomicUsize::new(0);
2291 let lmdb_calls = AtomicUsize::new(0);
2292
2293 let store = LocalStore::new_for_backend_with_openers(
2294 temp_dir.path(),
2295 StorageBackend::Lmdb,
2296 0,
2297 |path, max_bytes| {
2298 fs_calls.fetch_add(1, Ordering::SeqCst);
2299 LocalStore::open_fs_store(path, max_bytes)
2300 },
2301 |_path, _max_bytes| {
2302 lmdb_calls.fetch_add(1, Ordering::SeqCst);
2303 Err(StoreError::Io(std::io::Error::from_raw_os_error(
2304 libc::ENOSYS,
2305 )))
2306 },
2307 )
2308 .unwrap();
2309
2310 assert!(matches!(store, LocalStore::Fs(_)));
2311 assert_eq!(lmdb_calls.load(Ordering::SeqCst), 1);
2312 assert_eq!(fs_calls.load(Ordering::SeqCst), 1);
2313 }
2314
2315 #[cfg(feature = "lmdb")]
2316 #[test]
2317 fn test_local_store_does_not_fallback_on_unrelated_lmdb_errors() {
2318 let temp_dir = TempDir::new().unwrap();
2319 let fs_calls = AtomicUsize::new(0);
2320
2321 let result = LocalStore::new_for_backend_with_openers(
2322 temp_dir.path(),
2323 StorageBackend::Lmdb,
2324 0,
2325 |path, max_bytes| {
2326 fs_calls.fetch_add(1, Ordering::SeqCst);
2327 LocalStore::open_fs_store(path, max_bytes)
2328 },
2329 |_path, _max_bytes| {
2330 Err(StoreError::Io(std::io::Error::from_raw_os_error(
2331 libc::EACCES,
2332 )))
2333 },
2334 );
2335
2336 assert!(
2337 matches!(result, Err(StoreError::Io(io_error)) if io_error.raw_os_error() == Some(libc::EACCES))
2338 );
2339 assert_eq!(fs_calls.load(Ordering::SeqCst), 0);
2340 }
2341
2342 #[test]
2343 fn test_import_multiple_refs_preserves_all() {
2344 let (storage, _temp) = create_test_storage();
2345
2346 storage.import_ref("refs/heads/main", "sha_main").unwrap();
2348 storage.import_ref("refs/heads/dev", "sha_dev").unwrap();
2349 storage
2350 .import_ref("refs/heads/feature", "sha_feature")
2351 .unwrap();
2352
2353 assert!(storage.has_ref("refs/heads/main").unwrap());
2355 assert!(storage.has_ref("refs/heads/dev").unwrap());
2356 assert!(storage.has_ref("refs/heads/feature").unwrap());
2357
2358 storage
2360 .write_ref(
2361 "refs/heads/new-branch",
2362 &Ref::Direct(
2363 ObjectId::from_hex("0123456789abcdef0123456789abcdef01234567").unwrap(),
2364 ),
2365 )
2366 .unwrap();
2367
2368 let refs = storage.list_refs().unwrap();
2370 assert_eq!(refs.len(), 4);
2371 assert!(refs.contains_key("refs/heads/main"));
2372 assert!(refs.contains_key("refs/heads/dev"));
2373 assert!(refs.contains_key("refs/heads/feature"));
2374 assert!(refs.contains_key("refs/heads/new-branch"));
2375 }
2376
2377 #[test]
2378 fn test_import_compressed_object() {
2379 let (storage, _temp) = create_test_storage();
2380
2381 let fake_compressed = vec![0x78, 0x9c, 0x01, 0x02, 0x03]; storage
2385 .import_compressed_object("abc123def456", fake_compressed.clone())
2386 .unwrap();
2387
2388 assert_eq!(storage.object_count().unwrap(), 1);
2390 }
2391
2392 #[test]
2393 fn test_write_ref_overwrites_imported() {
2394 let (storage, _temp) = create_test_storage();
2395
2396 storage.import_ref("refs/heads/main", "old_sha").unwrap();
2398
2399 storage
2401 .write_ref(
2402 "refs/heads/main",
2403 &Ref::Direct(
2404 ObjectId::from_hex("0123456789abcdef0123456789abcdef01234567").unwrap(),
2405 ),
2406 )
2407 .unwrap();
2408
2409 let refs = storage.list_refs().unwrap();
2411 assert_eq!(
2412 refs.get("refs/heads/main"),
2413 Some(&"0123456789abcdef0123456789abcdef01234567".to_string())
2414 );
2415 }
2416
2417 #[test]
2418 fn test_delete_ref_preserves_others() {
2419 let (storage, _temp) = create_test_storage();
2420
2421 storage.import_ref("refs/heads/main", "sha_main").unwrap();
2423 storage.import_ref("refs/heads/dev", "sha_dev").unwrap();
2424
2425 storage.delete_ref("refs/heads/dev").unwrap();
2427
2428 assert!(storage.has_ref("refs/heads/main").unwrap());
2430 assert!(!storage.has_ref("refs/heads/dev").unwrap());
2431 }
2432
2433 #[test]
2434 fn test_clear_removes_all() {
2435 let (storage, _temp) = create_test_storage();
2436
2437 storage.import_ref("refs/heads/main", "sha_main").unwrap();
2439 storage
2440 .import_compressed_object("obj1", vec![1, 2, 3])
2441 .unwrap();
2442
2443 storage.clear().unwrap();
2445
2446 assert!(!storage.has_ref("refs/heads/main").unwrap());
2448 assert_eq!(storage.object_count().unwrap(), 0);
2449 }
2450
2451 #[test]
2452 fn test_evict_if_needed_respects_configured_limit() {
2453 let (storage, _temp) = create_test_storage_with_limit(1_024);
2454
2455 storage
2456 .write_raw_object(ObjectType::Blob, &vec![b'a'; 900])
2457 .unwrap();
2458 storage
2459 .write_raw_object(ObjectType::Blob, &vec![b'b'; 900])
2460 .unwrap();
2461 storage
2462 .write_ref(
2463 "refs/heads/main",
2464 &Ref::Direct(
2465 ObjectId::from_hex("0123456789abcdef0123456789abcdef01234567").unwrap(),
2466 ),
2467 )
2468 .unwrap();
2469
2470 storage.build_tree().unwrap();
2471
2472 let before = local_total_bytes(&storage);
2473 assert!(before > 1_024);
2474
2475 let freed = storage.evict_if_needed().unwrap();
2476 assert!(freed > 0);
2477
2478 let after = local_total_bytes(&storage);
2479 assert!(after <= 1_024);
2480 }
2481
2482 #[test]
2483 fn test_build_tree_evicts_stale_blobs_before_writing_new_tree() {
2484 let max_size_bytes = 16 * 1024;
2485 let (storage, _temp) = create_test_storage_with_limit(max_size_bytes);
2486
2487 let stale_blobs = vec![
2488 vec![b'x'; 7 * 1024],
2489 vec![b'y'; 7 * 1024],
2490 vec![b'z'; 7 * 1024],
2491 ];
2492 let stale_hashes: Vec<Hash> = stale_blobs.iter().map(|blob| sha256(blob)).collect();
2493
2494 for (hash, blob) in stale_hashes.iter().zip(stale_blobs) {
2495 storage
2496 .runtime
2497 .block_on(storage.store().put(*hash, blob))
2498 .unwrap();
2499 }
2500
2501 let before = local_total_bytes(&storage);
2502 assert!(before > max_size_bytes);
2503
2504 let commit_oid = write_test_commit(&storage);
2505 storage
2506 .write_ref("refs/heads/main", &Ref::Direct(commit_oid))
2507 .unwrap();
2508 storage
2509 .write_ref("HEAD", &Ref::Symbolic("refs/heads/main".to_string()))
2510 .unwrap();
2511
2512 storage.build_tree().unwrap();
2513
2514 let evicted_stale = stale_hashes
2515 .iter()
2516 .filter(|hash| !storage.runtime.block_on(storage.store().has(hash)).unwrap())
2517 .count();
2518
2519 assert!(
2520 evicted_stale > 0,
2521 "expected build_tree preflight eviction to remove stale blobs before writing"
2522 );
2523 }
2524
2525 #[test]
2526 fn test_build_tree_progress_tracks_repo_tree_work() {
2527 let (storage, _temp) = create_test_storage();
2528 let commit_oid = write_test_commit(&storage);
2529 storage
2530 .write_ref("refs/heads/main", &Ref::Direct(commit_oid))
2531 .unwrap();
2532 storage
2533 .write_ref("HEAD", &Ref::Symbolic("refs/heads/main".to_string()))
2534 .unwrap();
2535
2536 let progress = RepoTreeBuildProgress::new();
2537 storage.build_tree_with_progress(&progress).unwrap();
2538
2539 let snapshot = progress.snapshot();
2540 assert_eq!(snapshot.phase, RepoTreeBuildPhase::Done);
2541 assert_eq!(snapshot.object_blobs, 3);
2542 assert_eq!(snapshot.files, 1);
2543 assert_eq!(snapshot.dirs, 0);
2544 assert_eq!(snapshot.reused, 0);
2545 assert_eq!(
2546 snapshot.format_for_label("Building repo tree"),
2547 " Building repo tree: done (3 object blobs, 1 files, 0 dirs, 0 reused)"
2548 );
2549 }
2550
2551 #[test]
2552 fn test_build_tree_adds_dumb_http_metadata() {
2553 let (storage, _temp) = create_test_storage();
2554 let commit_oid = write_test_commit(&storage);
2555 let tag_content = format!(
2556 "object {}\ntype commit\ntag v1.0.0\ntagger Test User <test@example.com> 0 +0000\n\nrelease\n",
2557 commit_oid.to_hex()
2558 );
2559 let tag_oid = storage
2560 .write_raw_object(ObjectType::Tag, tag_content.as_bytes())
2561 .unwrap();
2562
2563 storage
2564 .write_ref("refs/heads/main", &Ref::Direct(commit_oid))
2565 .unwrap();
2566 storage
2567 .write_ref("refs/tags/v1.0.0", &Ref::Direct(tag_oid))
2568 .unwrap();
2569 storage
2570 .write_ref("HEAD", &Ref::Symbolic("refs/heads/main".to_string()))
2571 .unwrap();
2572
2573 let root_cid = storage.build_tree().unwrap();
2574
2575 let info_refs_cid = storage
2576 .runtime
2577 .block_on(storage.tree.resolve_path(&root_cid, ".git/info/refs"))
2578 .unwrap()
2579 .expect("info/refs exists");
2580 let info_refs = storage
2581 .runtime
2582 .block_on(storage.tree.get(&info_refs_cid, None))
2583 .unwrap()
2584 .unwrap();
2585 let info_refs = String::from_utf8(info_refs).unwrap();
2586
2587 assert_eq!(
2588 info_refs,
2589 format!(
2590 "{commit}\trefs/heads/main\n{tag}\trefs/tags/v1.0.0\n{commit}\trefs/tags/v1.0.0^{{}}\n",
2591 commit = commit_oid.to_hex(),
2592 tag = tag_oid.to_hex()
2593 )
2594 );
2595
2596 let packs_cid = storage
2597 .runtime
2598 .block_on(
2599 storage
2600 .tree
2601 .resolve_path(&root_cid, ".git/objects/info/packs"),
2602 )
2603 .unwrap()
2604 .expect("objects/info/packs exists");
2605 let packs = storage
2606 .runtime
2607 .block_on(storage.tree.get(&packs_cid, None))
2608 .unwrap()
2609 .unwrap();
2610 assert!(packs.is_empty(), "objects/info/packs should be empty");
2611 }
2612
2613 #[test]
2614 fn test_build_tree_materializes_loose_refs_at_git_paths() {
2615 let (storage, _temp) = create_test_storage();
2616 let commit_oid = write_test_commit(&storage);
2617
2618 storage
2619 .write_ref("refs/heads/master", &Ref::Direct(commit_oid))
2620 .unwrap();
2621 storage
2622 .write_ref("refs/heads/codex/meshrouter-prod", &Ref::Direct(commit_oid))
2623 .unwrap();
2624 storage
2625 .write_ref("refs/tags/v1.0.0", &Ref::Direct(commit_oid))
2626 .unwrap();
2627 storage
2628 .write_ref("HEAD", &Ref::Symbolic("refs/heads/master".to_string()))
2629 .unwrap();
2630
2631 let root_cid = storage.build_tree().unwrap();
2632
2633 for path in [
2634 ".git/refs/heads/master",
2635 ".git/refs/heads/codex/meshrouter-prod",
2636 ".git/refs/tags/v1.0.0",
2637 ] {
2638 let ref_cid = storage
2639 .runtime
2640 .block_on(storage.tree.resolve_path(&root_cid, path))
2641 .unwrap()
2642 .unwrap_or_else(|| panic!("{path} should exist"));
2643 let ref_value = storage
2644 .runtime
2645 .block_on(storage.tree.get(&ref_cid, None))
2646 .unwrap()
2647 .unwrap();
2648 assert_eq!(
2649 String::from_utf8(ref_value).unwrap(),
2650 commit_oid.to_hex(),
2651 "{path} should contain the ref target",
2652 );
2653 }
2654 }
2655
2656 #[test]
2657 fn test_materialized_tree_supports_static_http_clone_from_git_dir() {
2658 let (storage, _temp) = create_test_storage();
2659 let commit_oid = write_test_commit(&storage);
2660 storage
2661 .write_ref("refs/heads/main", &Ref::Direct(commit_oid))
2662 .unwrap();
2663 storage
2664 .write_ref("HEAD", &Ref::Symbolic("refs/heads/main".to_string()))
2665 .unwrap();
2666
2667 let root_cid = storage.build_tree().unwrap();
2668 let export_dir = TempDir::new().unwrap();
2669 let repo_dir = export_dir.path().join("repo");
2670 export_tree_to_fs(&storage.runtime, &storage.tree, &root_cid, &repo_dir);
2671
2672 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
2673 let port = listener.local_addr().unwrap().port();
2674 drop(listener);
2675
2676 let mut server = spawn_http_server(export_dir.path(), port);
2677 wait_for_http_server(&mut server, port, "/repo/.git/HEAD");
2678
2679 let clone_dir = TempDir::new().unwrap();
2680 let clone_path = clone_dir.path().join("clone");
2681 let output = Command::new("git")
2682 .current_dir(clone_dir.path())
2683 .args([
2684 "clone",
2685 &format!("http://127.0.0.1:{port}/repo/.git", port = port),
2686 clone_path.to_str().unwrap(),
2687 ])
2688 .output()
2689 .unwrap();
2690
2691 let _ = server.kill();
2692 let _ = server.wait();
2693
2694 assert!(
2695 output.status.success(),
2696 "git clone failed: {}",
2697 String::from_utf8_lossy(&output.stderr)
2698 );
2699 assert_eq!(
2700 std::fs::read_to_string(clone_path.join("README.md")).unwrap(),
2701 "hello from hashtree\n"
2702 );
2703 }
2704
2705 #[test]
2706 fn test_validate_root_contains_direct_refs_rejects_missing_tip_object() {
2707 let (storage, _temp) = create_test_storage();
2708 let commit_oid = write_test_commit(&storage);
2709 storage
2710 .write_ref("refs/heads/main", &Ref::Direct(commit_oid))
2711 .unwrap();
2712 storage
2713 .write_ref("HEAD", &Ref::Symbolic("refs/heads/main".to_string()))
2714 .unwrap();
2715
2716 let empty_objects_dir = storage
2717 .runtime
2718 .block_on(storage.tree.put_directory(vec![]))
2719 .unwrap();
2720 let refs_dir = storage
2721 .runtime
2722 .block_on(storage.tree.put_directory(vec![]))
2723 .unwrap();
2724 let info_dir = storage
2725 .runtime
2726 .block_on(storage.tree.put_directory(vec![]))
2727 .unwrap();
2728 let git_dir = storage
2729 .runtime
2730 .block_on(storage.tree.put_directory(vec![
2731 DirEntry::from_cid("HEAD", &info_dir).with_size(0),
2732 DirEntry::from_cid("info", &info_dir).with_link_type(LinkType::Dir),
2733 DirEntry::from_cid("objects", &empty_objects_dir).with_link_type(LinkType::Dir),
2734 DirEntry::from_cid("refs", &refs_dir).with_link_type(LinkType::Dir),
2735 ]))
2736 .unwrap();
2737 let root_cid = storage
2738 .runtime
2739 .block_on(storage.tree.put_directory(vec![
2740 DirEntry::from_cid(".git", &git_dir).with_link_type(LinkType::Dir),
2741 ]))
2742 .unwrap();
2743
2744 let err = storage
2745 .validate_root_contains_direct_refs(&root_cid)
2746 .expect_err("missing ref-tip object should fail validation");
2747 assert!(
2748 err.to_string().contains(&commit_oid.to_hex()),
2749 "validation error should mention missing commit oid: {}",
2750 err
2751 );
2752 }
2753}