Skip to main content

objects/store/fs/
fs_impl.rs

1// SPDX-License-Identifier: Apache-2.0
2//! ObjectStore implementation for FsStore.
3
4use 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
34/// Bytes we read off disk to recover a blob's uncompressed size.
35/// Must cover the 9-byte modern header **plus** the 4-byte ZSTD
36/// magic that `header_uncompressed_size` uses to disambiguate
37/// modern from legacy (5-byte) headers — without the magic in the
38/// peek buffer the lookup silently returns the on-disk byte length
39/// instead of the recorded uncompressed size, which left `stat`
40/// reporting the compressed size of every loose blob.
41const 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
174/// Validate every entry in a pack against its tagged id (checksum
175/// validation) and return the installed id list. This is the shared
176/// validated core for both install seams: the byte-buffer install
177/// (`install_pack`) and the memory-bounded temp-file install
178/// (`install_pack_streaming`) both run their pack through here, so
179/// both apply the same checksum validation and report the same
180/// installed ids regardless of how the bytes reach the store.
181fn 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    /// Insert into the recent-blob cache when the payload fits the size gate.
261    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    /// Single-pass blob lookup. The wrapper in `ObjectStore::get_blob`
271    /// retries this once after a stale-reload on miss.
272    fn try_get_blob_once(&self, hash: &ContentHash) -> Result<Option<Blob>> {
273        // Cache first — avoid `path.exists()` / pack probes on warm hits.
274        // Write lock: LRU promotion mutates the order list.
275        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            // Step 2: skip the BLAKE3 re-hash. The pack reader already
288            // located this entry by its content-addressed key in the
289            // pack index — anything served here either matches or
290            // means the pack itself is corrupted in ways a per-read
291            // hash check can't recover from cleanly. For multi-MB
292            // blobs the verify was the dominant tail of the cold
293            // read (~3GB/s × 10MB ≈ 3.3ms per call).
294            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                // Loose blobs are bare bytes on disk: a half-written
306                // file or bit-rot inside the payload would slip past
307                // the path-is-the-hash invariant. Keep the verify on
308                // this path. Pack-resident reads above skip it because
309                // pack entries are framed with offset + length records
310                // that fail to parse if the pack is corrupt.
311                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    /// Shared body for `try_has_{blob,tree,state}_once`: object is
325    /// present iff the loose path exists or the pack manager
326    /// resolves it. Callers pass the loose path and the
327    /// pack-manager probe; the helper handles the lock.
328    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        // Keep `has_blob` coherent with cache-first `get_blob`. A pure
344        // existence check needs no LRU promotion, so take the *read*
345        // lock and use `contains` — concurrent `has_blob` calls in
346        // heddled/mount don't serialize on the exclusive write lock.
347        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    /// Header-only size lookup for a single attempt. Tries:
357    /// 1. The recent-blob cache (we already have the bytes in
358    ///    memory — `len()` is free).
359    /// 2. The loose blob: peek the 9-byte compression header. For a
360    ///    compressed blob the recorded uncompressed size lives in the
361    ///    header. For an uncompressed blob (no recognised header) the
362    ///    on-disk file length IS the blob size.
363    /// 3. Any loaded pack: the pack format records the uncompressed
364    ///    size as a varint right after the tagged id, so we can decode
365    ///    it without touching the body.
366    ///
367    /// Cost: one short read (typically 9 bytes) for loose blobs, or a
368    /// pure in-memory varint decode for packed blobs. *No*
369    /// decompression.
370    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            // No recognised compression header — the file is raw
383            // blob bytes. The on-disk length is the blob size.
384            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        // Cache first. The recent-object cache only ever holds trees we
397        // wrote or read this process, so a hit is authoritative for a
398        // read. Write lock: LRU promotion mutates the order list.
399        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        // Loose trees may be migration-promoted V2 shadows of an older packed
407        // V1 encoding at the same semantic tree hash. Prefer the loose copy
408        // when it exists, then fall through to pack lookup.
409        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        // Read-lock `contains`: an existence check needs no LRU
467        // promotion, so it must not serialize on the write lock.
468        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        // Cache first — avoid `path.exists()` / pack probes on warm hits.
479        // Write lock: LRU promotion mutates the order list. Put paths and
480        // successful reads below keep this coherent for the process.
481        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        // Read-lock `contains`: an existence check needs no LRU
515        // promotion, so it must not serialize on the write lock.
516        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    /// Zero-copy pack fast path. When the blob lives in a packfile
532    /// and is non-delta + uncompressed, returns a `Bytes::slice`
533    /// view of the pack's mmap — no decompression, no allocation,
534    /// no memcpy. Compressed pack entries, delta entries, and
535    /// loose blobs fall back to `get_blob` and wrap the result in a
536    /// `Bytes` (the `Vec` → `Bytes` conversion is itself zero-copy).
537    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        // Miss path: a sibling FsStore (e.g. the worktree's repo
555        // backing the same `.heddle/`) may have installed a new pack
556        // since we loaded ours. Cheap disk-count check first; full
557        // reload only when the count grew.
558        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    /// Loose blob path safe for clonefile/copy materialization.
647    ///
648    /// Returns `Some(path)` only when the loose file exists, is
649    /// stored uncompressed, *and* its bytes hash to the expected
650    /// content hash. Compressed blobs and pack-only blobs fall
651    /// through to `None`; so do *torn* cache-mirror files (the
652    /// `AtomicWriteMode::NoSync` write side may leave one if the
653    /// host crashed during a previous promote). On the torn case
654    /// the caller re-promotes from the authoritative pack copy.
655    ///
656    /// Verification is amortised: a hash that passes the check once
657    /// in this process is recorded in `verified_loose_blobs` and
658    /// subsequent calls skip the read+hash. So the cost on the
659    /// materialize hot path is at most one BLAKE3 over each unique
660    /// blob per process lifetime — negligible for tiny blobs,
661    /// bounded by working-set size for huge ones.
662    fn loose_blob_path(&self, hash: &ContentHash) -> Option<PathBuf> {
663        let path = hash_path(&blobs_dir(&self.root), hash);
664        // Fast path: this process already verified (or wrote) this
665        // hash's loose mirror in `promote_to_loose_uncompressed`.
666        // Trust without re-hashing — `path.exists()` is the only
667        // I/O we need.
668        if let Ok(verified) = self.verified_loose_blobs.read()
669            && verified.contains(hash)
670            && path.exists()
671        {
672            return Some(path);
673        }
674
675        // First-time-this-process check: peek the header to filter
676        // out compressed-loose files cheaply, then verify the
677        // body's hash matches what the caller expects. A torn-write
678        // (post-crash) cache mirror fails this and the caller
679        // re-promotes from the pack.
680        //
681        // Header peek must cover the 9-byte modern header **plus**
682        // the 4-byte ZSTD magic that `is_compressed` checks —
683        // peeking only 9 bytes makes `is_compressed` falsely
684        // return `false` on a properly-compressed blob, and we'd
685        // hand the caller the compressed file path. Same off-by-4
686        // we fixed in `BLOB_HEADER_PEEK`.
687        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            // Torn write or unrelated corruption. Leave the file on
695            // disk; the caller's `promote_to_loose_uncompressed`
696            // will overwrite it via the standard temp+rename path.
697            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    /// Promote a blob to its uncompressed-loose canonical path so
706    /// `loose_blob_path` returns `Some(path)` and hardlink-first
707    /// materialization fires.
708    ///
709    /// Three cases:
710    /// 1. Already loose+uncompressed: peek the header, no-op.
711    /// 2. Loose but compressed: read+decompress, atomically rewrite
712    ///    the canonical path with raw bytes.
713    /// 3. Pack-only: read out of the pack via `get_blob`, atomically
714    ///    write to the canonical loose path. Pack copy is left in
715    ///    place — the next prune cycle will discard the loose mirror
716    ///    and a future materialize will re-promote.
717    #[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        // An external Git-overlay blob must stay external: materialization can
722        // read its bytes through `get_blob_bytes`, but promotion would turn a
723        // read-through into an accidental native copy and violate the source-
724        // authority boundary.
725        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        // Idempotent fast path: already loose AND uncompressed.
733        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        // Either compressed-loose or pack-only. Reading via
741        // `get_blob` covers both: compressed-loose decompresses on
742        // the way out, pack-only reads from the loaded pack manager.
743        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        // Install the uncompressed bytes at the canonical loose path
751        // via the cache-mirror atomic-write variant: no fsync, just
752        // temp+rename. The fsync skip is what makes promotion fast
753        // (measured: ~5 ms/blob with `sync_data` vs ~0.2 ms without
754        // on macOS APFS); the safety comes from the read-side hash
755        // check in `loose_blob_path`. A torn write after a crash
756        // produces a file whose content hash doesn't match, so the
757        // next reader rejects it and re-promotes from the pack.
758        //
759        // Record the hash in this process's verified-blobs cache:
760        // we just wrote the bytes ourselves, so the subsequent read
761        // path can trust them without re-hashing.
762        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        // Sibling-store recovery, mirroring the read path: if a
779        // concurrent writer just installed a pack we don't know about,
780        // reload and retry once before reporting a miss.
781        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        // `put_tree` is an ownership boundary: a native state that references
837        // this tree must survive loss or pruning of an overlay read-through
838        // source. Descriptor-only states do not call this method; they retain
839        // their explicit external-source semantics.
840        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        // Validate + list ids through the same core as the byte-buffer
1240        // seam, but via an mmap-backed reader so the pack is never
1241        // copied into the heap — the memory-bounded promise survives.
1242        // Drop the reader (releasing the mmap) before the rename so
1243        // the file move isn't racing an open mapping.
1244        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}