1use std::{
5 fs::{self, OpenOptions},
6 path::{Path, PathBuf},
7};
8
9use fs2::FileExt;
10use heddle_format::compression::{header_uncompressed_size, is_compressed};
11use tracing::{debug, instrument, trace};
12
13use super::{
14 FsStore,
15 fs_io::{list_hashes_from_dir, read_file_bytes, read_file_header},
16 fs_paths::{
17 action_path, actions_dir, blobs_dir, hash_path, redaction_path, redactions_dir,
18 state_attachment_index_lock_path, state_attachment_index_path, state_attachment_path,
19 state_attachments_dir, state_path, state_visibility_dir, state_visibility_path, states_dir,
20 trees_dir,
21 },
22};
23use crate::{
24 object::{
25 Action, ActionId, Blob, ContentHash, State, StateAttachment, StateAttachmentId, StateId,
26 Tree,
27 },
28 store::{
29 HeddleError, ObjectStore, Result, codec,
30 pack::{ObjectType, PackManager, PackObjectId},
31 },
32};
33
34const BLOB_HEADER_PEEK: usize = 13;
42
43fn validate_loaded_tree(tree: Tree) -> Result<Tree> {
44 tree.validate()?;
45 Ok(tree)
46}
47
48fn validate_blob_bytes(data: &[u8], hash: ContentHash) -> Result<()> {
49 let mut hasher = ContentHash::typed_hasher("blob", data.len() as u64);
50 hasher.update(data);
51 let found = ContentHash::from_bytes(hasher.finalize().into());
52 if found != hash {
53 return Err(HeddleError::Corruption {
54 expected: hash,
55 found,
56 });
57 }
58
59 Ok(())
60}
61
62fn validate_tree_serialized(data: &[u8], hash: ContentHash) -> Result<Tree> {
63 let tree = codec::decode_tree_serialized(data)?;
64 let tree = validate_loaded_tree(tree)?;
65 let found = tree.hash();
66 if found != hash {
67 return Err(HeddleError::Corruption {
68 expected: hash,
69 found,
70 });
71 }
72
73 Ok(tree)
74}
75
76fn validate_loaded_state(requested_id: &StateId, mut state: State) -> Result<State> {
77 let computed = state.id();
78 if computed != *requested_id {
79 return Err(HeddleError::InvalidObject(format!(
80 "state id mismatch: requested {requested_id}, computed {computed}"
81 )));
82 }
83 state.state_id = computed;
84 Ok(state)
85}
86
87fn validate_state_serialized(data: &[u8], id: StateId) -> Result<State> {
88 let state: State = rmp_serde::from_slice(data)?;
89 validate_loaded_state(&id, state)
90}
91
92fn validate_loaded_action(requested_id: &ActionId, action: Action) -> Result<Action> {
93 let found_id = action.compute_id();
94 if found_id != *requested_id {
95 return Err(HeddleError::InvalidObject(format!(
96 "action id mismatch: requested {}, found {}",
97 requested_id, found_id
98 )));
99 }
100
101 Ok(action)
102}
103
104fn validate_action_serialized(data: &[u8], id: ActionId) -> Result<Action> {
105 let action: Action = rmp_serde::from_slice(data)?;
106 validate_loaded_action(&id, action)
107}
108
109impl FsStore {
110 fn with_state_attachment_index_lock<T>(
111 &self,
112 state: &StateId,
113 operation: impl FnOnce() -> Result<T>,
114 ) -> Result<T> {
115 let path = state_attachment_index_lock_path(&self.root, state);
116 if let Some(parent) = path.parent() {
117 fs::create_dir_all(parent)?;
118 }
119 let file = OpenOptions::new()
120 .create(true)
121 .truncate(false)
122 .read(true)
123 .write(true)
124 .open(path)?;
125 file.lock_exclusive()?;
126 let result = operation();
127 file.unlock()?;
128 result
129 }
130
131 fn rebuild_state_attachment_index(&self, state: &StateId) -> Result<Vec<StateAttachmentId>> {
132 #[cfg(test)]
133 fs::write(
134 state_attachment_index_path(&self.root, state).with_extension("rebuild-marker"),
135 b"rebuilt",
136 )?;
137 let mut ids = Vec::new();
138 let dir = state_attachments_dir(&self.root, state);
139 if let Ok(entries) = fs::read_dir(dir) {
140 for entry in entries {
141 let attachment: StateAttachment = rmp_serde::from_slice(&fs::read(entry?.path())?)?;
142 if attachment.state_id != *state {
143 return Err(HeddleError::InvalidObject(
144 "state attachment stored under wrong state".to_string(),
145 ));
146 }
147 ids.push(attachment.id());
148 }
149 }
150 if let Ok(manager) = self.pack_manager().read() {
151 for pack_id in manager.list_all_ids()? {
152 let PackObjectId::Hash(hash) = pack_id else {
153 continue;
154 };
155 let Some((ObjectType::StateAttachment, bytes)) =
156 manager.get_hashed_object(&hash)?
157 else {
158 continue;
159 };
160 let attachment: StateAttachment = rmp_serde::from_slice(&bytes)?;
161 if attachment.state_id == *state {
162 ids.push(attachment.id());
163 }
164 }
165 }
166 ids.sort();
167 ids.dedup();
168 let path = state_attachment_index_path(&self.root, state);
169 self.write_loose_object_atomic(&path, &rmp_serde::to_vec_named(&ids)?)?;
170 Ok(ids)
171 }
172}
173
174fn validate_and_list_pack(reader: &crate::store::pack::PackReader) -> Result<Vec<PackObjectId>> {
182 let ids = reader.list_ids();
183 for id in &ids {
184 let Some((obj_type, data)) = reader.get_object_bytes(id)? else {
185 continue;
186 };
187 validate_pack_entry(id, obj_type, data.as_ref())?;
188 }
189 Ok(ids)
190}
191
192fn state_entries_from_pack(
193 reader: &crate::store::pack::PackReader,
194 ids: &[PackObjectId],
195) -> Result<Vec<(StateId, Vec<u8>)>> {
196 let mut states = Vec::new();
197 for id in ids {
198 let PackObjectId::StateId(change_id) = id else {
199 continue;
200 };
201 let Some((obj_type, data)) = reader.get_object(id)? else {
202 continue;
203 };
204 if obj_type != ObjectType::State {
205 return Err(HeddleError::InvalidObject(format!(
206 "pack id {} is indexed as {:?}, expected State",
207 change_id.to_string_full(),
208 obj_type
209 )));
210 }
211 validate_state_serialized(&data, *change_id)?;
212 states.push((*change_id, data));
213 }
214 Ok(states)
215}
216
217fn attachment_entries_from_pack(
218 reader: &crate::store::pack::PackReader,
219 ids: &[PackObjectId],
220) -> Result<Vec<StateAttachment>> {
221 let mut attachments = Vec::new();
222 for id in ids {
223 let Some((ObjectType::StateAttachment, data)) = reader.get_object(id)? else {
224 continue;
225 };
226 attachments.push(rmp_serde::from_slice(&data)?);
227 }
228 Ok(attachments)
229}
230
231fn validate_pack_entry(id: &PackObjectId, obj_type: ObjectType, data: &[u8]) -> Result<()> {
232 match (id, obj_type) {
233 (PackObjectId::Hash(hash), ObjectType::Blob) => validate_blob_bytes(data, *hash),
234 (PackObjectId::Hash(hash), ObjectType::Tree) => {
235 validate_tree_serialized(data, *hash).map(|_| ())
236 }
237 (PackObjectId::Hash(hash), ObjectType::Action) => {
238 validate_action_serialized(data, ActionId::from_hash(*hash)).map(|_| ())
239 }
240 (PackObjectId::StateId(change_id), ObjectType::State) => {
241 validate_state_serialized(data, *change_id).map(|_| ())
242 }
243 (PackObjectId::Hash(hash), ObjectType::StateAttachment) => {
244 let attachment: StateAttachment = rmp_serde::from_slice(data)?;
245 if attachment.id().as_hash() != hash {
246 return Err(HeddleError::InvalidObject(
247 "state attachment pack id mismatch".to_string(),
248 ));
249 }
250 Ok(())
251 }
252 _ => Err(HeddleError::InvalidObject(format!(
253 "unsupported native pack object: {:?} {:?}",
254 id, obj_type
255 ))),
256 }
257}
258
259impl FsStore {
260 fn cache_recent_blob(&self, hash: ContentHash, blob: &Blob) {
262 if blob.content().len() > super::fs_store::RECENT_BLOB_CACHE_MAX_BYTES {
263 return;
264 }
265 if let Ok(mut cache) = self.recent_blobs.write() {
266 cache.insert(hash, blob.clone());
267 }
268 }
269
270 fn try_get_blob_once(&self, hash: &ContentHash) -> Result<Option<Blob>> {
273 if let Ok(mut cache) = self.recent_blobs.write()
276 && let Some(blob) = cache.get(hash)
277 {
278 trace!("Found blob in recent object cache");
279 return Ok(Some(blob.clone()));
280 }
281
282 if let Ok(manager) = self.pack_manager().read()
283 && let Some((obj_type, data)) = manager.get_hashed_object(hash)?
284 && obj_type == ObjectType::Blob
285 {
286 trace!("Found blob in packfile");
287 let blob = Blob::new(data);
295 self.cache_recent_blob(*hash, &blob);
296 return Ok(Some(blob));
297 }
298
299 let path = hash_path(&blobs_dir(&self.root), hash);
300 match read_file_bytes(&path)? {
301 Some(data) => {
302 trace!(size = data.as_slice().len(), "Blob data read");
303 let content = codec::decode_blob_content(data.as_slice())?;
304 let blob = Blob::new(content);
305 if blob.hash() != *hash {
312 return Err(HeddleError::Corruption {
313 expected: *hash,
314 found: blob.hash(),
315 });
316 }
317 self.cache_recent_blob(*hash, &blob);
318 Ok(Some(blob))
319 }
320 None => Ok(None),
321 }
322 }
323
324 fn loose_or_packed(
329 &self,
330 loose_path: &Path,
331 in_pack: impl FnOnce(&PackManager) -> bool,
332 ) -> Result<bool> {
333 if loose_path.exists() {
334 return Ok(true);
335 }
336 if let Ok(manager) = self.pack_manager().read() {
337 return Ok(in_pack(&manager));
338 }
339 Ok(false)
340 }
341
342 fn try_has_blob_once(&self, hash: &ContentHash) -> Result<bool> {
343 if let Ok(cache) = self.recent_blobs.read()
348 && cache.contains(hash)
349 {
350 return Ok(true);
351 }
352 let path = hash_path(&blobs_dir(&self.root), hash);
353 self.loose_or_packed(&path, |m| m.has_object(hash))
354 }
355
356 fn try_get_blob_size_once(&self, hash: &ContentHash) -> Result<Option<u64>> {
371 if let Ok(mut cache) = self.recent_blobs.write()
372 && let Some(blob) = cache.get(hash)
373 {
374 return Ok(Some(blob.content().len() as u64));
375 }
376
377 let path = hash_path(&blobs_dir(&self.root), hash);
378 if let Some((header, file_len)) = read_file_header(&path, BLOB_HEADER_PEEK)? {
379 if let Some(size) = header_uncompressed_size(&header) {
380 return Ok(Some(size));
381 }
382 return Ok(Some(file_len));
385 }
386
387 if let Ok(manager) = self.pack_manager().read()
388 && let Some(size) = manager.get_hashed_object_size(hash)?
389 {
390 return Ok(Some(size));
391 }
392 Ok(None)
393 }
394
395 fn try_get_tree_once(&self, hash: &ContentHash) -> Result<Option<Tree>> {
396 if let Ok(mut cache) = self.recent_trees.write()
400 && let Some(tree) = cache.get(hash)
401 {
402 trace!("Found tree in recent object cache");
403 return Ok(Some(tree.clone()));
404 }
405
406 let path = hash_path(&trees_dir(&self.root), hash);
410 if path.exists()
411 && let Some(data) = read_file_bytes(&path)?
412 {
413 trace!(size = data.as_slice().len(), "Tree data read");
414 let tree = validate_loaded_tree(codec::decode_tree(data.as_slice())?)?;
415 if tree.hash() != *hash {
416 return Err(HeddleError::Corruption {
417 expected: *hash,
418 found: tree.hash(),
419 });
420 }
421 if let Ok(mut cache) = self.recent_trees.write() {
422 cache.insert(*hash, tree.clone());
423 }
424 return Ok(Some(tree));
425 }
426
427 if let Ok(manager) = self.pack_manager().read()
428 && let Some((obj_type, data)) = manager.get_hashed_object(hash)?
429 && obj_type == ObjectType::Tree
430 {
431 trace!("Found tree in packfile");
432 let tree = validate_loaded_tree(codec::decode_tree_serialized(&data)?)?;
433 if tree.hash() != *hash {
434 return Err(HeddleError::Corruption {
435 expected: *hash,
436 found: tree.hash(),
437 });
438 }
439 if let Ok(mut cache) = self.recent_trees.write() {
440 cache.insert(*hash, tree.clone());
441 }
442 return Ok(Some(tree));
443 }
444 Ok(None)
445 }
446
447 fn try_get_tree_serialized_once(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
448 let path = hash_path(&trees_dir(&self.root), hash);
449 if path.exists()
450 && let Some(data) = read_file_bytes(&path)?
451 {
452 return Ok(Some(codec::decode_tree_body(data.as_slice())?));
453 }
454
455 if let Ok(manager) = self.pack_manager().read()
456 && let Some((obj_type, data)) = manager.get_hashed_object(hash)?
457 && obj_type == ObjectType::Tree
458 {
459 return Ok(Some(data));
460 }
461
462 Ok(None)
463 }
464
465 fn try_has_tree_once(&self, hash: &ContentHash) -> Result<bool> {
466 if let Ok(cache) = self.recent_trees.read()
469 && cache.contains(hash)
470 {
471 return Ok(true);
472 }
473 let path = hash_path(&trees_dir(&self.root), hash);
474 self.loose_or_packed(&path, |m| m.has_object(hash))
475 }
476
477 fn try_get_state_once(&self, id: &StateId) -> Result<Option<State>> {
478 if let Ok(mut cache) = self.recent_states.write()
482 && let Some(state) = cache.get(id)
483 {
484 trace!("Found state in recent object cache");
485 return Ok(Some(state.clone()));
486 }
487
488 let path = state_path(&self.root, id);
489 if let Some(data) = read_file_bytes(&path)? {
490 trace!(size = data.as_slice().len(), "State read from loose object");
491 let state = validate_loaded_state(id, codec::decode_state(data.as_slice())?)?;
492 if let Ok(mut cache) = self.recent_states.write() {
493 cache.insert(*id, state.clone());
494 }
495 return Ok(Some(state));
496 }
497
498 if let Ok(manager) = self.pack_manager().read()
499 && let Some((obj_type, data)) = manager.get_object(&PackObjectId::StateId(*id))?
500 && obj_type == ObjectType::State
501 {
502 trace!("Found state in packfile");
503 let state = validate_loaded_state(id, rmp_serde::from_slice(&data)?)?;
504 if let Ok(mut cache) = self.recent_states.write() {
505 cache.insert(*id, state.clone());
506 }
507 return Ok(Some(state));
508 }
509
510 Ok(None)
511 }
512
513 fn try_has_state_once(&self, id: &StateId) -> Result<bool> {
514 if let Ok(cache) = self.recent_states.read()
517 && cache.contains(id)
518 {
519 return Ok(true);
520 }
521 let path = state_path(&self.root, id);
522 self.loose_or_packed(&path, |m| m.has_object_id(&PackObjectId::StateId(*id)))
523 }
524}
525
526impl ObjectStore for FsStore {
527 fn clear_recent_caches(&self) {
528 self.clear_recent_object_caches();
529 }
530
531 fn get_blob_bytes(&self, hash: &ContentHash) -> Result<Option<bytes::Bytes>> {
538 if let Ok(manager) = self.pack_manager().read()
539 && let Some((obj_type, data)) = manager.get_hashed_object_bytes(hash)?
540 && obj_type == crate::store::pack::ObjectType::Blob
541 {
542 return Ok(Some(data));
543 }
544 Ok(self
545 .get_blob(hash)?
546 .map(|blob| bytes::Bytes::from(blob.into_content())))
547 }
548
549 #[instrument(skip(self), fields(hash = %hash.short()))]
550 fn get_blob(&self, hash: &ContentHash) -> Result<Option<Blob>> {
551 if let Some(blob) = self.try_get_blob_once(hash)? {
552 return Ok(Some(blob));
553 }
554 if self.reload_packs_if_stale()?
559 && let Some(blob) = self.try_get_blob_once(hash)?
560 {
561 return Ok(Some(blob));
562 }
563 trace!("Blob not found");
564 match &self.external_source {
565 Some(source) => source.get_blob(hash),
566 None => Ok(None),
567 }
568 }
569
570 #[instrument(skip(self, blob), fields(size = blob.content().len()))]
571 fn put_blob(&self, blob: &Blob) -> Result<ContentHash> {
572 let hash = blob.hash();
573 let path = hash_path(&blobs_dir(&self.root), &hash);
574
575 if !path.exists() {
576 let data = codec::encode_blob_content(blob.content(), &self.compression)?;
577 trace!(compressed_size = data.len(), "Writing blob");
578 self.write_loose_object_atomic(&path, &data)?;
579 } else {
580 trace!("Blob already exists, skipping write");
581 }
582 self.cache_recent_blob(hash, blob);
583
584 Ok(hash)
585 }
586
587 #[instrument(skip(self, blob), fields(hash = %hash.short()))]
588 fn put_blob_with_hash(&self, blob: &Blob, hash: ContentHash) -> Result<ContentHash> {
589 if blob.hash() != hash {
590 return Err(HeddleError::Corruption {
591 expected: hash,
592 found: blob.hash(),
593 });
594 }
595
596 let path = hash_path(&blobs_dir(&self.root), &hash);
597
598 if !path.exists() {
599 let data = codec::encode_blob_content(blob.content(), &self.compression)?;
600 trace!(
601 compressed_size = data.len(),
602 "Writing blob with precomputed hash"
603 );
604 self.write_loose_object_atomic(&path, &data)?;
605 }
606 self.cache_recent_blob(hash, blob);
607
608 Ok(hash)
609 }
610
611 #[instrument(skip(self, data), fields(hash = %hash.short(), size = data.len()))]
612 fn put_blob_bytes_with_hash(&self, data: &[u8], hash: ContentHash) -> Result<ContentHash> {
613 validate_blob_bytes(data, hash)?;
614
615 let path = hash_path(&blobs_dir(&self.root), &hash);
616 if !path.exists() {
617 trace!(
618 size = data.len(),
619 "Writing raw blob bytes with precomputed hash"
620 );
621 self.write_loose_object_atomic(&path, data)?;
622 }
623 self.cache_recent_blob(hash, &Blob::from_slice(data));
624
625 Ok(hash)
626 }
627
628 #[instrument(skip(self), fields(hash = %hash.short()))]
629 fn has_blob(&self, hash: &ContentHash) -> Result<bool> {
630 if ObjectStore::has_blob_locally(self, hash)? {
631 return Ok(true);
632 }
633 match &self.external_source {
634 Some(source) => Ok(source.get_blob(hash)?.is_some()),
635 None => Ok(false),
636 }
637 }
638
639 fn has_blob_locally(&self, hash: &ContentHash) -> Result<bool> {
640 if self.try_has_blob_once(hash)? {
641 return Ok(true);
642 }
643 Ok(self.reload_packs_if_stale()? && self.try_has_blob_once(hash)?)
644 }
645
646 fn loose_blob_path(&self, hash: &ContentHash) -> Option<PathBuf> {
663 let path = hash_path(&blobs_dir(&self.root), hash);
664 if let Ok(verified) = self.verified_loose_blobs.read()
669 && verified.contains(hash)
670 && path.exists()
671 {
672 return Some(path);
673 }
674
675 let (header, _) = read_file_header(&path, BLOB_HEADER_PEEK).ok().flatten()?;
688 if is_compressed(&header) {
689 return None;
690 }
691 let bytes = read_file_bytes(&path).ok().flatten()?;
692 let actual = ContentHash::compute_typed("blob", bytes.as_slice());
693 if actual != *hash {
694 return None;
698 }
699 if let Ok(mut verified) = self.verified_loose_blobs.write() {
700 verified.insert(*hash, ());
701 }
702 Some(path)
703 }
704
705 #[instrument(skip(self), fields(hash = %hash.short()))]
718 fn promote_to_loose_uncompressed(&self, hash: &ContentHash) -> Result<bool> {
719 let path = hash_path(&blobs_dir(&self.root), hash);
720
721 if !self.try_has_blob_once(hash)?
726 && let Some(source) = &self.external_source
727 && source.get_blob(hash)?.is_some()
728 {
729 return Ok(false);
730 }
731
732 if let Some((header, _)) = read_file_header(&path, 9)?
734 && !is_compressed(&header)
735 {
736 trace!("Blob already loose+uncompressed; skipping promotion");
737 return Ok(false);
738 }
739
740 let blob = self.get_blob(hash)?.ok_or_else(|| {
744 HeddleError::NotFound(format!(
745 "blob {} not found in store; cannot promote to loose-uncompressed",
746 hash
747 ))
748 })?;
749
750 debug!(
763 size = blob.content().len(),
764 "Promoting blob to loose-uncompressed canonical store"
765 );
766 self.write_loose_object_cache(&path, blob.content())?;
767 if let Ok(mut verified) = self.verified_loose_blobs.write() {
768 verified.insert(*hash, ());
769 }
770 Ok(true)
771 }
772
773 #[instrument(skip(self), fields(hash = %hash.short()))]
774 fn blob_size(&self, hash: &ContentHash) -> Result<Option<u64>> {
775 if let Some(size) = self.try_get_blob_size_once(hash)? {
776 return Ok(Some(size));
777 }
778 if self.reload_packs_if_stale()?
782 && let Some(size) = self.try_get_blob_size_once(hash)?
783 {
784 return Ok(Some(size));
785 }
786 match &self.external_source {
787 Some(source) => Ok(source
788 .get_blob(hash)?
789 .map(|blob| blob.content().len() as u64)),
790 None => Ok(None),
791 }
792 }
793
794 #[instrument(skip(self), fields(hash = %hash.short()))]
795 fn get_tree(&self, hash: &ContentHash) -> Result<Option<Tree>> {
796 if let Some(tree) = self.try_get_tree_once(hash)? {
797 return Ok(Some(tree));
798 }
799 if self.reload_packs_if_stale()?
800 && let Some(tree) = self.try_get_tree_once(hash)?
801 {
802 return Ok(Some(tree));
803 }
804 trace!("Tree not found");
805 match &self.external_source {
806 Some(source) => source.get_tree(hash),
807 None => Ok(None),
808 }
809 }
810
811 #[instrument(skip(self), fields(hash = %hash.short()))]
812 fn get_tree_serialized(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
813 if let Some(data) = self.try_get_tree_serialized_once(hash)? {
814 return Ok(Some(data));
815 }
816 if self.reload_packs_if_stale()?
817 && let Some(data) = self.try_get_tree_serialized_once(hash)?
818 {
819 return Ok(Some(data));
820 }
821 match &self.external_source {
822 Some(source) => source
823 .get_tree(hash)?
824 .map(|tree| rmp_serde::to_vec_named(&tree))
825 .transpose()
826 .map_err(|error| HeddleError::InvalidObject(error.to_string())),
827 None => Ok(None),
828 }
829 }
830
831 #[instrument(skip(self, tree), fields(entry_count = tree.entries().len()))]
832 fn put_tree(&self, tree: &Tree) -> Result<ContentHash> {
833 let hash = tree.hash();
834 let path = hash_path(&trees_dir(&self.root), &hash);
835
836 if !ObjectStore::has_tree_locally(self, &hash)? {
841 let (_, data) = codec::encode_tree(tree, &self.compression)?;
842 trace!(compressed_size = data.len(), "Writing tree");
843 self.write_loose_object_atomic(&path, &data)?;
844 } else {
845 trace!("Tree already exists, skipping write");
846 }
847 if let Ok(mut cache) = self.recent_trees.write() {
848 cache.insert(hash, tree.clone());
849 }
850
851 Ok(hash)
852 }
853
854 #[instrument(skip(self, data), fields(hash = %hash.short(), size = data.len()))]
855 fn put_tree_serialized(&self, data: &[u8], hash: ContentHash) -> Result<ContentHash> {
856 let tree = validate_tree_serialized(data, hash)?;
857
858 let path = hash_path(&trees_dir(&self.root), &hash);
859 let should_write = match read_file_bytes(&path)? {
860 Some(existing) => codec::decode_tree_body(existing.as_slice())? != data,
861 None => true,
862 };
863 if should_write {
864 trace!(size = data.len(), "Writing raw serialized tree");
865 self.write_loose_object_atomic(&path, data)?;
866 }
867 if let Ok(mut cache) = self.recent_trees.write() {
868 cache.insert(hash, tree);
869 }
870
871 Ok(hash)
872 }
873
874 #[instrument(skip(self), fields(hash = %hash.short()))]
875 fn has_tree(&self, hash: &ContentHash) -> Result<bool> {
876 if ObjectStore::has_tree_locally(self, hash)? {
877 return Ok(true);
878 }
879 match &self.external_source {
880 Some(source) => Ok(source.get_tree(hash)?.is_some()),
881 None => Ok(false),
882 }
883 }
884
885 fn has_tree_locally(&self, hash: &ContentHash) -> Result<bool> {
886 if self.try_has_tree_once(hash)? {
887 return Ok(true);
888 }
889 Ok(self.reload_packs_if_stale()? && self.try_has_tree_once(hash)?)
890 }
891
892 #[instrument(skip(self), fields(id = %id.short()))]
893 fn get_state(&self, id: &StateId) -> Result<Option<State>> {
894 if let Some(state) = self.try_get_state_once(id)? {
895 return Ok(Some(state));
896 }
897 if self.reload_packs_if_stale()?
898 && let Some(state) = self.try_get_state_once(id)?
899 {
900 return Ok(Some(state));
901 }
902 trace!("State not found");
903 match &self.external_source {
904 Some(source) => source.get_state(id),
905 None => Ok(None),
906 }
907 }
908
909 #[instrument(skip(self, state), fields(id = %state.id().short()))]
910 fn put_state(&self, state: &State) -> Result<()> {
911 let state_id = state.id();
912 let path = state_path(&self.root, &state_id);
913 let data = codec::encode_state(state, &self.compression)?;
914 trace!(compressed_size = data.len(), "Writing state");
915 self.write_loose_object_atomic(&path, &data)?;
916 if let Ok(mut cache) = self.recent_states.write() {
917 let mut cached = state.clone();
918 cached.state_id = state_id;
919 cache.insert(state_id, cached);
920 }
921 Ok(())
922 }
923
924 #[instrument(skip(self, data), fields(id = %id.short(), size = data.len()))]
925 fn put_state_serialized(&self, data: &[u8], id: StateId) -> Result<()> {
926 let state = validate_state_serialized(data, id)?;
927 let path = state_path(&self.root, &id);
928 trace!(size = data.len(), "Writing raw serialized state");
929 self.write_loose_object_atomic(&path, data)?;
930 if let Ok(mut cache) = self.recent_states.write() {
931 cache.insert(id, state);
932 }
933 Ok(())
934 }
935
936 #[instrument(skip(self), fields(id = %id.short()))]
937 fn has_state(&self, id: &StateId) -> Result<bool> {
938 if self.try_has_state_once(id)? {
939 return Ok(true);
940 }
941 if self.reload_packs_if_stale()? && self.try_has_state_once(id)? {
942 return Ok(true);
943 }
944 match &self.external_source {
945 Some(source) => Ok(source.get_state(id)?.is_some()),
946 None => Ok(false),
947 }
948 }
949
950 #[instrument(skip(self))]
951 fn list_states(&self) -> Result<Vec<StateId>> {
952 self.reload_packs_if_stale()?;
953
954 let dir = states_dir(&self.root);
955 let mut states = Vec::new();
956 if dir.exists() {
957 for entry in fs::read_dir(&dir)? {
958 let entry = entry?;
959 let path = entry.path();
960 if let Some(name) = path.file_stem()
961 && let Some(name_str) = name.to_str()
962 && let Ok(id) = StateId::parse(name_str)
963 {
964 states.push(id);
965 }
966 }
967 }
968 if let Ok(manager) = self.pack_manager().read() {
969 for id in manager.list_all_ids()? {
970 if let PackObjectId::StateId(change_id) = id
971 && !states.contains(&change_id)
972 {
973 states.push(change_id);
974 }
975 }
976 }
977 if let Some(source) = &self.external_source {
978 for id in source.list_states()? {
979 if !states.contains(&id) {
980 states.push(id);
981 }
982 }
983 }
984 debug!(count = states.len(), "Listed states");
985 Ok(states)
986 }
987
988 fn get_state_attachment(
989 &self,
990 state: &StateId,
991 id: &StateAttachmentId,
992 ) -> Result<Option<StateAttachment>> {
993 let path = state_attachment_path(&self.root, state, id);
994 let bytes = if let Some(bytes) = read_file_bytes(&path)? {
995 bytes.as_slice().to_vec()
996 } else if let Ok(manager) = self.pack_manager().read()
997 && let Some((ObjectType::StateAttachment, bytes)) =
998 manager.get_hashed_object(id.as_hash())?
999 {
1000 bytes
1001 } else {
1002 return Ok(None);
1003 };
1004 let attachment: StateAttachment = rmp_serde::from_slice(&bytes)?;
1005 if attachment.state_id != *state || attachment.id() != *id {
1006 return Err(HeddleError::InvalidObject(
1007 "state attachment address does not match content".to_string(),
1008 ));
1009 }
1010 Ok(Some(attachment))
1011 }
1012
1013 fn put_state_attachment(&self, attachment: &StateAttachment) -> Result<StateAttachmentId> {
1014 let id = attachment.id();
1015 self.with_state_attachment_index_lock(&attachment.state_id, || {
1016 let index_path = state_attachment_index_path(&self.root, &attachment.state_id);
1017 let mut ids: Vec<StateAttachmentId> = match read_file_bytes(&index_path)? {
1018 Some(bytes) => rmp_serde::from_slice(bytes.as_slice())?,
1019 None => self.rebuild_state_attachment_index(&attachment.state_id)?,
1020 };
1021 if !ids.contains(&id) {
1022 ids.push(id);
1023 ids.sort();
1024 self.write_loose_object_atomic(&index_path, &rmp_serde::to_vec_named(&ids)?)?;
1025 }
1026 let path = state_attachment_path(&self.root, &attachment.state_id, &id);
1027 self.write_loose_object_atomic(&path, &rmp_serde::to_vec_named(attachment)?)?;
1028 Ok(id)
1029 })
1030 }
1031
1032 fn list_state_attachments(&self, state: &StateId) -> Result<Vec<StateAttachment>> {
1033 self.with_state_attachment_index_lock(state, || {
1034 let index_path = state_attachment_index_path(&self.root, state);
1035 let mut ids: Vec<StateAttachmentId> = match read_file_bytes(&index_path)? {
1036 Some(bytes) => rmp_serde::from_slice(bytes.as_slice())?,
1037 None => self.rebuild_state_attachment_index(state)?,
1038 };
1039 let mut attachments = Vec::new();
1040 let mut stale = false;
1041 for id in &ids {
1042 match self.get_state_attachment(state, id)? {
1043 Some(attachment) => attachments.push(attachment),
1044 None => stale = true,
1045 }
1046 }
1047 if stale {
1048 ids = self.rebuild_state_attachment_index(state)?;
1049 attachments.clear();
1050 for id in ids {
1051 let attachment = self.get_state_attachment(state, &id)?.ok_or_else(|| {
1052 HeddleError::InvalidObject(format!(
1053 "rebuilt state attachment index references missing {id}"
1054 ))
1055 })?;
1056 attachments.push(attachment);
1057 }
1058 }
1059 Ok(attachments)
1060 })
1061 }
1062
1063 #[instrument(skip(self), fields(id = %id))]
1064 fn get_action(&self, id: &ActionId) -> Result<Option<Action>> {
1065 let path = action_path(&self.root, id);
1066 if !path.exists()
1067 && let Ok(manager) = self.pack_manager().read()
1068 && let Some((obj_type, data)) = manager.get_hashed_object(id.as_hash())?
1069 && obj_type == ObjectType::Action
1070 {
1071 trace!("Found action in packfile");
1072 let action = validate_loaded_action(id, rmp_serde::from_slice(&data)?)?;
1073 return Ok(Some(action));
1074 }
1075 match read_file_bytes(&path)? {
1076 Some(data) => {
1077 trace!(size = data.as_slice().len(), "Action data read");
1078 let action = validate_loaded_action(id, codec::decode_action(data.as_slice())?)?;
1079 Ok(Some(action))
1080 }
1081 None => {
1082 trace!("Action not found");
1083 Ok(None)
1084 }
1085 }
1086 }
1087
1088 #[instrument(skip(self, action))]
1089 fn put_action(&self, action: &mut Action) -> Result<ActionId> {
1090 let id = action.id();
1091 let path = action_path(&self.root, &id);
1092
1093 if !path.exists() {
1094 let (_, data) = codec::encode_action(action, &self.compression)?;
1095 trace!(id = %id, compressed_size = data.len(), "Writing action");
1096 self.write_loose_object_atomic(&path, &data)?;
1097 }
1098
1099 Ok(id)
1100 }
1101
1102 #[instrument(skip(self))]
1103 fn list_actions(&self) -> Result<Vec<ActionId>> {
1104 let dir = actions_dir(&self.root);
1105 let mut actions = Vec::new();
1106 if dir.exists() {
1107 for entry in fs::read_dir(&dir)? {
1108 let entry = entry?;
1109 let path = entry.path();
1110 if let Some(name) = path.file_stem()
1111 && let Some(name_str) = name.to_str()
1112 && let Ok(hash) = ContentHash::from_hex(name_str)
1113 {
1114 actions.push(ActionId::from_hash(hash));
1115 }
1116 }
1117 }
1118 if let Ok(manager) = self.pack_manager().read() {
1119 for id in manager.list_all_ids()? {
1120 if let PackObjectId::Hash(hash) = id
1121 && !actions.iter().any(|action_id| action_id.as_hash() == &hash)
1122 && let Some((obj_type, _)) = manager.get_hashed_object(&hash)?
1123 && obj_type == ObjectType::Action
1124 {
1125 actions.push(ActionId::from_hash(hash));
1126 }
1127 }
1128 }
1129 debug!(count = actions.len(), "Listed actions");
1130 Ok(actions)
1131 }
1132
1133 #[instrument(skip(self))]
1134 fn list_blobs(&self) -> Result<Vec<ContentHash>> {
1135 let dir = blobs_dir(&self.root);
1136 let mut blobs = list_hashes_from_dir(&dir)?;
1137 if let Ok(manager) = self.pack_manager().read() {
1138 for id in manager.list_all_ids()? {
1139 if let PackObjectId::Hash(hash) = id
1140 && !blobs.contains(&hash)
1141 && let Some((obj_type, _)) = manager.get_hashed_object(&hash)?
1142 && obj_type == ObjectType::Blob
1143 {
1144 blobs.push(hash);
1145 }
1146 }
1147 }
1148 Ok(blobs)
1149 }
1150
1151 #[instrument(skip(self))]
1152 fn list_trees(&self) -> Result<Vec<ContentHash>> {
1153 let dir = trees_dir(&self.root);
1154 let mut trees = list_hashes_from_dir(&dir)?;
1155 if let Ok(manager) = self.pack_manager().read() {
1156 for id in manager.list_all_ids()? {
1157 if let PackObjectId::Hash(hash) = id
1158 && !trees.contains(&hash)
1159 && let Some((obj_type, _)) = manager.get_hashed_object(&hash)?
1160 && obj_type == ObjectType::Tree
1161 {
1162 trees.push(hash);
1163 }
1164 }
1165 }
1166 Ok(trees)
1167 }
1168
1169 #[instrument(skip(self))]
1170 fn pack_objects(&self, aggressive: bool) -> Result<(u64, u64)> {
1171 self.pack_objects_impl(aggressive)
1172 }
1173
1174 #[instrument(skip(self), fields(id = ?id))]
1175 fn get_pack_object(&self, id: &PackObjectId) -> Result<Option<(ObjectType, Vec<u8>)>> {
1176 if let Ok(manager) = self.pack_manager().read()
1177 && let Some((obj_type, data)) = manager.get_object(id)?
1178 {
1179 return Ok(Some((obj_type, data)));
1180 }
1181
1182 match id {
1183 PackObjectId::Hash(hash) => {
1184 if let Some(blob) = self.get_blob(hash)? {
1185 return Ok(Some((ObjectType::Blob, blob.content().to_vec())));
1186 }
1187 if let Some(tree) = self.get_tree(hash)? {
1188 return Ok(Some((ObjectType::Tree, rmp_serde::to_vec_named(&tree)?)));
1189 }
1190 if let Some(action) = self.get_action(&ActionId::from_hash(*hash))? {
1191 return Ok(Some((
1192 ObjectType::Action,
1193 rmp_serde::to_vec_named(&action)?,
1194 )));
1195 }
1196 Ok(None)
1197 }
1198 PackObjectId::StateId(change_id) => {
1199 if let Some(state) = self.get_state(change_id)? {
1200 Ok(Some((ObjectType::State, rmp_serde::to_vec_named(&state)?)))
1201 } else {
1202 Ok(None)
1203 }
1204 }
1205 }
1206 }
1207
1208 #[instrument(skip(self, pack_data, index_data))]
1209 fn install_pack(&self, pack_data: &[u8], index_data: &[u8]) -> Result<Vec<PackObjectId>> {
1210 let reader = crate::store::pack::PackReader::from_slice(pack_data, index_data)?;
1211 let ids = validate_and_list_pack(&reader)?;
1212 let state_entries = state_entries_from_pack(&reader, &ids)?;
1213 self.install_pack_files(pack_data, index_data)?;
1214 for (id, data) in state_entries {
1215 self.put_state_serialized(&data, id)?;
1216 }
1217 for id in &ids {
1218 let Some((ObjectType::StateAttachment, data)) = reader.get_object(id)? else {
1219 continue;
1220 };
1221 let attachment: StateAttachment = rmp_serde::from_slice(&data)?;
1222 self.put_state_attachment(&attachment)?;
1223 }
1224 self.clear_recent_object_caches();
1225 Ok(ids)
1226 }
1227
1228 #[instrument(skip(self, blobs), fields(count = blobs.len()))]
1229 fn put_blobs_packed(&self, blobs: Vec<(crate::object::ContentHash, Vec<u8>)>) -> Result<()> {
1230 self.put_blobs_packed_impl(blobs)
1231 }
1232
1233 #[instrument(skip(self))]
1234 fn install_pack_streaming(
1235 &self,
1236 pack_path: &std::path::Path,
1237 index_path: &std::path::Path,
1238 ) -> Result<Vec<PackObjectId>> {
1239 let ids = {
1245 let reader = crate::store::pack::PackReader::open(pack_path, index_path)?;
1246 validate_and_list_pack(&reader)?
1247 };
1248 let state_entries = {
1249 let reader = crate::store::pack::PackReader::open(pack_path, index_path)?;
1250 state_entries_from_pack(&reader, &ids)?
1251 };
1252 let attachment_entries = {
1253 let reader = crate::store::pack::PackReader::open(pack_path, index_path)?;
1254 attachment_entries_from_pack(&reader, &ids)?
1255 };
1256 self.install_pack_files_streaming(pack_path, index_path)?;
1257 for (id, data) in state_entries {
1258 self.put_state_serialized(&data, id)?;
1259 }
1260 for attachment in attachment_entries {
1261 self.put_state_attachment(&attachment)?;
1262 }
1263 Ok(ids)
1264 }
1265
1266 #[instrument(skip(self))]
1267 fn prune_loose_objects(&self) -> Result<(u64, u64)> {
1268 self.prune_loose_objects_impl()
1269 }
1270
1271 #[instrument(skip(self))]
1272 fn begin_snapshot_write_batch(&self) -> Result<()> {
1273 self.begin_snapshot_write_batch_impl()
1274 }
1275
1276 #[instrument(skip(self))]
1277 fn flush_snapshot_write_batch(&self) -> Result<()> {
1278 self.flush_snapshot_write_batch_impl()
1279 }
1280
1281 #[instrument(skip(self))]
1282 fn abort_snapshot_write_batch(&self) {
1283 self.abort_snapshot_write_batch_impl();
1284 }
1285
1286 fn has_redactions_for_blob(&self, blob: &ContentHash) -> Result<bool> {
1287 Ok(redaction_path(&self.root, blob).exists())
1288 }
1289
1290 fn get_redactions_bytes_for_blob(&self, blob: &ContentHash) -> Result<Option<Vec<u8>>> {
1291 let path = redaction_path(&self.root, blob);
1292 match fs::read(&path) {
1293 Ok(bytes) => Ok(Some(bytes)),
1294 Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
1295 Err(err) => Err(HeddleError::Io(err)),
1296 }
1297 }
1298
1299 fn put_redactions_bytes_for_blob(&self, blob: &ContentHash, bytes: &[u8]) -> Result<()> {
1300 let dir = redactions_dir(&self.root);
1301 if !dir.exists() {
1302 crate::fs_atomic::create_dir_all_durable(&dir)?;
1303 }
1304 let path = redaction_path(&self.root, blob);
1305 crate::fs_atomic::write_file_atomic(&path, bytes)?;
1306 Ok(())
1307 }
1308
1309 fn list_blobs_with_redactions(&self) -> Result<Vec<ContentHash>> {
1310 let dir = redactions_dir(&self.root);
1311 if !dir.exists() {
1312 return Ok(Vec::new());
1313 }
1314 let mut out = Vec::new();
1315 for entry in fs::read_dir(&dir)? {
1316 let entry = entry?;
1317 let path = entry.path();
1318 if path.extension().and_then(|e| e.to_str()) != Some("bin") {
1319 continue;
1320 }
1321 let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else {
1322 continue;
1323 };
1324 if let Ok(hash) = ContentHash::from_hex(stem) {
1325 out.push(hash);
1326 }
1327 }
1328 Ok(out)
1329 }
1330
1331 fn has_state_visibility_for_state(&self, state: &StateId) -> Result<bool> {
1332 Ok(state_visibility_path(&self.root, state).exists())
1333 }
1334
1335 fn get_state_visibility_bytes_for_state(&self, state: &StateId) -> Result<Option<Vec<u8>>> {
1336 let path = state_visibility_path(&self.root, state);
1337 match fs::read(&path) {
1338 Ok(bytes) => Ok(Some(bytes)),
1339 Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
1340 Err(err) => Err(HeddleError::Io(err)),
1341 }
1342 }
1343
1344 fn put_state_visibility_bytes_for_state(&self, state: &StateId, bytes: &[u8]) -> Result<()> {
1345 let dir = state_visibility_dir(&self.root);
1346 if !dir.exists() {
1347 crate::fs_atomic::create_dir_all_durable(&dir)?;
1348 }
1349 let path = state_visibility_path(&self.root, state);
1350 crate::fs_atomic::write_file_atomic(&path, bytes)?;
1351 Ok(())
1352 }
1353
1354 fn list_states_with_visibility(&self) -> Result<Vec<StateId>> {
1355 let dir = state_visibility_dir(&self.root);
1356 if !dir.exists() {
1357 return Ok(Vec::new());
1358 }
1359 let mut out = Vec::new();
1360 for entry in fs::read_dir(&dir)? {
1361 let entry = entry?;
1362 let path = entry.path();
1363 if path.extension().and_then(|e| e.to_str()) != Some("bin") {
1364 continue;
1365 }
1366 let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else {
1367 continue;
1368 };
1369 if let Ok(state) = StateId::parse(stem) {
1370 out.push(state);
1371 }
1372 }
1373 Ok(out)
1374 }
1375}
1376
1377#[cfg(test)]
1378mod state_attachment_tests {
1379 use std::sync::Arc;
1380
1381 use chrono::Utc;
1382
1383 use super::*;
1384 use crate::{
1385 object::{Attribution, Principal, StateAttachmentBody},
1386 store::{CompressionConfig, pack::PackBuilder},
1387 };
1388
1389 fn fixture(store: &FsStore) -> (State, StateAttachment) {
1390 let tree = store.put_tree(&Tree::new()).unwrap();
1391 let attribution = Attribution::human(Principal::new("Test", "test@example.com"));
1392 let state = State::new(tree, vec![], attribution.clone());
1393 store.put_state(&state).unwrap();
1394 let attachment = StateAttachment {
1395 state_id: state.id(),
1396 body: StateAttachmentBody::Context(ContentHash::compute(b"context")),
1397 attribution,
1398 created_at: Utc::now(),
1399 supersedes: None,
1400 };
1401 (state, attachment)
1402 }
1403
1404 #[test]
1405 fn concurrent_attachment_writes_keep_every_index_entry() {
1406 let temp = tempfile::TempDir::new().unwrap();
1407 let store = Arc::new(FsStore::new(temp.path()));
1408 let (state, base) = fixture(&store);
1409 let mut threads = Vec::new();
1410 for byte in 0..16u8 {
1411 let store = Arc::clone(&store);
1412 let mut attachment = base.clone();
1413 attachment.body = StateAttachmentBody::Context(ContentHash::compute(&[byte]));
1414 threads.push(std::thread::spawn(move || {
1415 store.put_state_attachment(&attachment).unwrap();
1416 }));
1417 }
1418 for thread in threads {
1419 thread.join().unwrap();
1420 }
1421 assert_eq!(store.list_state_attachments(&state.id()).unwrap().len(), 16);
1422 }
1423
1424 #[test]
1425 fn missing_index_rebuilds_from_loose_objects() {
1426 let temp = tempfile::TempDir::new().unwrap();
1427 let store = FsStore::new(temp.path());
1428 let (state, attachment) = fixture(&store);
1429 store.put_state_attachment(&attachment).unwrap();
1430 fs::remove_file(state_attachment_index_path(&store.root, &state.id())).unwrap();
1431 assert_eq!(
1432 store.list_state_attachments(&state.id()).unwrap(),
1433 vec![attachment]
1434 );
1435 }
1436
1437 #[test]
1438 fn packed_attachment_uses_state_index_for_lookup() {
1439 let temp = tempfile::TempDir::new().unwrap();
1440 let store = FsStore::new(temp.path());
1441 let (state, attachment) = fixture(&store);
1442 let mut builder = PackBuilder::new(CompressionConfig::default());
1443 builder.add(
1444 *attachment.id().as_hash(),
1445 ObjectType::StateAttachment,
1446 rmp_serde::to_vec_named(&attachment).unwrap(),
1447 );
1448 let (pack, index, _) = builder.build().unwrap();
1449 store.install_pack(&pack, &index).unwrap();
1450 fs::remove_file(state_attachment_path(
1451 &store.root,
1452 &state.id(),
1453 &attachment.id(),
1454 ))
1455 .unwrap();
1456 let rebuild_marker =
1457 state_attachment_index_path(&store.root, &state.id()).with_extension("rebuild-marker");
1458 let _ = fs::remove_file(&rebuild_marker);
1459 assert_eq!(
1460 store.list_state_attachments(&state.id()).unwrap(),
1461 vec![attachment.clone()]
1462 );
1463 assert_eq!(
1464 store.list_state_attachments(&state.id()).unwrap(),
1465 vec![attachment]
1466 );
1467 assert!(!rebuild_marker.exists());
1468 }
1469}