Skip to main content

objects/store/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Backend-neutral object storage abstractions and concrete implementations.
3
4use std::{path::PathBuf, sync::Arc};
5
6use crate::object::{
7    Action, ActionId, Blob, ContentHash, State, StateAttachment, StateAttachmentId, StateId, Tree,
8};
9
10pub mod actor_presence;
11pub mod agent_task;
12pub mod codec;
13pub mod fs;
14pub mod liveness;
15#[cfg(any(test, feature = "memory-backend"))]
16pub mod memory;
17pub use heddle_pack::store::pack;
18pub mod shallow;
19mod snapshot_commit;
20pub mod source;
21pub mod store_compliance;
22pub mod writer_lease;
23
24pub use actor_presence::{
25    ActorChainNode, ActorPresence, ActorPresenceStatus, ActorPresenceStore, AgentUsageSummary,
26    ContextQueryEntry, generate_actor_session_id,
27};
28pub use agent_task::{
29    AGENT_TASK_SCHEMA_VERSION, AgentTaskRecord, AgentTaskStatus, AgentTaskStore,
30    generate_agent_task_id, validate_task_id,
31};
32pub use fs::{
33    DEFAULT_PACK_INSTALL_INTENT_TTL_SECS, FsRepackOperation, FsStore, PackInstallIntent,
34    PackInstallMetricsSnapshot, PackInstallPhase, PackInstallRecoverReport,
35    install_pack_bytes_journaled, pack_install_metrics_reset, pack_install_metrics_snapshot,
36    recover_pack_install_intents, recover_pack_install_intents_with_ttl,
37};
38pub use heddle_format::compression::{CompressionConfig, CompressionError, compress, decompress};
39pub use liveness::{
40    AGENT_LEASE_DURATION, Liveness, current_boot_id, process_alive, reservation_liveness_at,
41};
42#[cfg(any(test, feature = "memory-backend"))]
43pub use memory::InMemoryStore;
44pub use pack::{
45    CancellationToken as RepackCancellationToken, LoadMonitor as RepackLoadMonitor, PackBuilder,
46    PackObjectId, PackReader, PackStats, RepackContext, RepackError, RepackHandle, RepackInventory,
47    RepackOperation, RepackOutcome, RepackPolicy, RepackReason, RepackReport, RepackResourceLimits,
48    RepackSchedule, RepackScheduler, StreamingPackBuilder, SyncData,
49};
50pub use shallow::ShallowInfo;
51#[doc(hidden)]
52pub use snapshot_commit::{
53    SNAPSHOT_COMMIT_ARTIFACT_SCHEMA, SnapshotCommitArtifact, SnapshotCommitDescriptor,
54    SnapshotPackManager,
55};
56#[cfg(feature = "async-source")]
57pub use source::AsyncObjectSource;
58pub use source::ObjectSource;
59pub use writer_lease::{
60    WriterLease, WriterLeaseAuthOutcome, WriterLeaseDraft, WriterLeaseGrant,
61    WriterLeaseReserveOutcome, WriterLeaseStatus, WriterLeaseStore, generate_writer_lease_id,
62    generate_writer_lease_token,
63};
64
65/// Read-only objects whose authoritative representation lives outside the
66/// native Heddle object directory. Git-overlay repositories use this seam to
67/// translate objects directly from `.git` without importing a second copy.
68pub trait ExternalObjectSource: Send + Sync {
69    fn get_blob(&self, hash: &ContentHash) -> Result<Option<Blob>>;
70    fn get_tree(&self, hash: &ContentHash) -> Result<Option<Tree>>;
71    fn get_state(&self, id: &StateId) -> Result<Option<State>>;
72    fn list_states(&self) -> Result<Vec<StateId>>;
73}
74
75pub use crate::error::{HeddleError as StoreError, HeddleError, Result};
76
77/// Static-dispatch enum over the concrete object stores Heddle ships.
78///
79/// This is the default `S` for [`Repository`](crate) so the store backend
80/// remains compile-time-monomorphized — no vtable. Each [`ObjectStore`] method
81/// `match`-dispatches to the inner variant, so the compiler inlines through
82/// the enum to the concrete backend's implementation (including its overridden
83/// default methods).
84///
85/// Sealed by construction: only the variants enumerated here are valid
86/// stores. Heddle is the sole implementer (heddle#259 / #283) — `AnyStore`
87/// is not a public extension point.
88#[derive(Clone)]
89pub enum AnyStore {
90    Fs(FsStore),
91}
92
93/// Forward an [`ObjectStore`] call to the active [`AnyStore`] variant.
94///
95/// Every arm calls the *same* method on the inner concrete store, so a
96/// backend's override of a defaulted trait method (e.g. `FsStore::blob_size`)
97/// is preserved rather than falling back to the trait default.
98macro_rules! any_store_dispatch {
99    ($self:ident, $method:ident ( $($arg:expr),* )) => {
100        match $self {
101            AnyStore::Fs(inner) => inner.$method($($arg),*),
102        }
103    };
104}
105
106impl ObjectStore for AnyStore {
107    fn get_blob(&self, hash: &ContentHash) -> Result<Option<Blob>> {
108        match self {
109            AnyStore::Fs(inner) => ObjectStore::get_blob(inner, hash),
110        }
111    }
112    fn put_blob(&self, blob: &Blob) -> Result<ContentHash> {
113        any_store_dispatch!(self, put_blob(blob))
114    }
115    fn get_blob_bytes(&self, hash: &ContentHash) -> Result<Option<bytes::Bytes>> {
116        match self {
117            AnyStore::Fs(inner) => ObjectStore::get_blob_bytes(inner, hash),
118        }
119    }
120    fn blob_size(&self, hash: &ContentHash) -> Result<Option<u64>> {
121        any_store_dispatch!(self, blob_size(hash))
122    }
123    fn loose_blob_path(&self, hash: &ContentHash) -> Option<PathBuf> {
124        any_store_dispatch!(self, loose_blob_path(hash))
125    }
126    fn promote_to_loose_uncompressed(&self, hash: &ContentHash) -> Result<bool> {
127        any_store_dispatch!(self, promote_to_loose_uncompressed(hash))
128    }
129    fn clear_recent_caches(&self) {
130        any_store_dispatch!(self, clear_recent_caches())
131    }
132    fn put_blob_with_hash(&self, blob: &Blob, hash: ContentHash) -> Result<ContentHash> {
133        any_store_dispatch!(self, put_blob_with_hash(blob, hash))
134    }
135    fn has_blob(&self, hash: &ContentHash) -> Result<bool> {
136        any_store_dispatch!(self, has_blob(hash))
137    }
138    fn has_blob_locally(&self, hash: &ContentHash) -> Result<bool> {
139        any_store_dispatch!(self, has_blob_locally(hash))
140    }
141    fn get_tree(&self, hash: &ContentHash) -> Result<Option<Tree>> {
142        match self {
143            AnyStore::Fs(inner) => ObjectStore::get_tree(inner, hash),
144        }
145    }
146    fn get_tree_serialized(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
147        match self {
148            AnyStore::Fs(inner) => ObjectStore::get_tree_serialized(inner, hash),
149        }
150    }
151    fn put_tree(&self, tree: &Tree) -> Result<ContentHash> {
152        any_store_dispatch!(self, put_tree(tree))
153    }
154    fn has_tree(&self, hash: &ContentHash) -> Result<bool> {
155        any_store_dispatch!(self, has_tree(hash))
156    }
157    fn has_tree_locally(&self, hash: &ContentHash) -> Result<bool> {
158        any_store_dispatch!(self, has_tree_locally(hash))
159    }
160    fn get_state(&self, id: &StateId) -> Result<Option<State>> {
161        match self {
162            AnyStore::Fs(inner) => ObjectStore::get_state(inner, id),
163        }
164    }
165    fn put_state(&self, state: &State) -> Result<()> {
166        any_store_dispatch!(self, put_state(state))
167    }
168    fn has_state(&self, id: &StateId) -> Result<bool> {
169        any_store_dispatch!(self, has_state(id))
170    }
171    fn list_states(&self) -> Result<Vec<StateId>> {
172        any_store_dispatch!(self, list_states())
173    }
174    fn get_state_attachment(
175        &self,
176        state: &StateId,
177        id: &StateAttachmentId,
178    ) -> Result<Option<StateAttachment>> {
179        any_store_dispatch!(self, get_state_attachment(state, id))
180    }
181    fn put_state_attachment(&self, attachment: &StateAttachment) -> Result<StateAttachmentId> {
182        any_store_dispatch!(self, put_state_attachment(attachment))
183    }
184    fn list_state_attachments(&self, state: &StateId) -> Result<Vec<StateAttachment>> {
185        any_store_dispatch!(self, list_state_attachments(state))
186    }
187    fn get_action(&self, id: &ActionId) -> Result<Option<Action>> {
188        any_store_dispatch!(self, get_action(id))
189    }
190    fn put_action(&self, action: &mut Action) -> Result<ActionId> {
191        any_store_dispatch!(self, put_action(action))
192    }
193    fn list_actions(&self) -> Result<Vec<ActionId>> {
194        any_store_dispatch!(self, list_actions())
195    }
196    fn list_blobs(&self) -> Result<Vec<ContentHash>> {
197        any_store_dispatch!(self, list_blobs())
198    }
199    fn list_trees(&self) -> Result<Vec<ContentHash>> {
200        any_store_dispatch!(self, list_trees())
201    }
202    fn put_blob_bytes_with_hash(&self, data: &[u8], hash: ContentHash) -> Result<ContentHash> {
203        any_store_dispatch!(self, put_blob_bytes_with_hash(data, hash))
204    }
205    fn put_tree_serialized(&self, data: &[u8], hash: ContentHash) -> Result<ContentHash> {
206        match self {
207            AnyStore::Fs(inner) => ObjectStore::put_tree_serialized(inner, data, hash),
208        }
209    }
210    fn put_state_serialized(&self, data: &[u8], id: StateId) -> Result<()> {
211        any_store_dispatch!(self, put_state_serialized(data, id))
212    }
213    fn put_action_serialized(&self, data: &[u8], id: ActionId) -> Result<()> {
214        any_store_dispatch!(self, put_action_serialized(data, id))
215    }
216    fn get_pack_object(
217        &self,
218        id: &pack::PackObjectId,
219    ) -> Result<Option<(pack::ObjectType, Vec<u8>)>> {
220        any_store_dispatch!(self, get_pack_object(id))
221    }
222    fn put_blobs_packed(&self, blobs: Vec<(ContentHash, Vec<u8>)>) -> Result<()> {
223        any_store_dispatch!(self, put_blobs_packed(blobs))
224    }
225    fn put_snapshot_objects_packed(
226        &self,
227        blobs: Vec<(ContentHash, Vec<u8>)>,
228        tree: &Tree,
229        state: &State,
230    ) -> Result<()> {
231        any_store_dispatch!(self, put_snapshot_objects_packed(blobs, tree, state))
232    }
233    fn put_snapshot_objects_and_attachments_packed(
234        &self,
235        blobs: Vec<(ContentHash, Vec<u8>)>,
236        tree: &Tree,
237        state: &State,
238        attachments: Vec<StateAttachment>,
239    ) -> Result<()> {
240        any_store_dispatch!(
241            self,
242            put_snapshot_objects_and_attachments_packed(blobs, tree, state, attachments)
243        )
244    }
245
246    fn install_pack(&self, pack_data: &[u8], index_data: &[u8]) -> Result<Vec<pack::PackObjectId>> {
247        any_store_dispatch!(self, install_pack(pack_data, index_data))
248    }
249    fn install_pack_streaming(
250        &self,
251        pack_path: &std::path::Path,
252        index_path: &std::path::Path,
253    ) -> Result<Vec<pack::PackObjectId>> {
254        any_store_dispatch!(self, install_pack_streaming(pack_path, index_path))
255    }
256    fn pack_objects(&self, delta_search: bool) -> Result<(u64, u64)> {
257        any_store_dispatch!(self, pack_objects(delta_search))
258    }
259    fn prune_loose_objects(&self) -> Result<(u64, u64)> {
260        any_store_dispatch!(self, prune_loose_objects())
261    }
262    fn discard_corrupt_clone_packs(&self) -> Result<usize> {
263        any_store_dispatch!(self, discard_corrupt_clone_packs())
264    }
265    fn begin_snapshot_write_batch(&self) -> Result<()> {
266        any_store_dispatch!(self, begin_snapshot_write_batch())
267    }
268    fn flush_snapshot_write_batch(&self) -> Result<()> {
269        any_store_dispatch!(self, flush_snapshot_write_batch())
270    }
271    fn abort_snapshot_write_batch(&self) {
272        any_store_dispatch!(self, abort_snapshot_write_batch())
273    }
274    fn has_redactions_for_blob(&self, blob: &ContentHash) -> Result<bool> {
275        any_store_dispatch!(self, has_redactions_for_blob(blob))
276    }
277    fn get_redactions_bytes_for_blob(&self, blob: &ContentHash) -> Result<Option<Vec<u8>>> {
278        any_store_dispatch!(self, get_redactions_bytes_for_blob(blob))
279    }
280    fn put_redactions_bytes_for_blob(&self, blob: &ContentHash, bytes: &[u8]) -> Result<()> {
281        any_store_dispatch!(self, put_redactions_bytes_for_blob(blob, bytes))
282    }
283    fn list_blobs_with_redactions(&self) -> Result<Vec<ContentHash>> {
284        any_store_dispatch!(self, list_blobs_with_redactions())
285    }
286    fn has_state_visibility_for_state(&self, state: &StateId) -> Result<bool> {
287        any_store_dispatch!(self, has_state_visibility_for_state(state))
288    }
289    fn get_state_visibility_bytes_for_state(&self, state: &StateId) -> Result<Option<Vec<u8>>> {
290        any_store_dispatch!(self, get_state_visibility_bytes_for_state(state))
291    }
292    fn put_state_visibility_bytes_for_state(&self, state: &StateId, bytes: &[u8]) -> Result<()> {
293        any_store_dispatch!(self, put_state_visibility_bytes_for_state(state, bytes))
294    }
295    fn list_states_with_visibility(&self) -> Result<Vec<StateId>> {
296        any_store_dispatch!(self, list_states_with_visibility())
297    }
298}
299
300impl AnyStore {
301    /// Attach a read-only external source to the local filesystem store.
302    pub fn set_external_source(&mut self, source: Arc<dyn ExternalObjectSource>) {
303        match self {
304            Self::Fs(store) => store.set_external_source(source),
305        }
306    }
307
308    /// Internal repository seam for the local authoritative snapshot artifact.
309    /// Kept off `ObjectStore` so third-party backends do not acquire a Heddle
310    /// filesystem recovery contract.
311    #[doc(hidden)]
312    pub fn snapshot_commit_descriptors(&self) -> Result<Vec<SnapshotCommitDescriptor>> {
313        match self {
314            Self::Fs(store) => store.snapshot_commit_descriptors_impl(),
315        }
316    }
317
318    /// O(1) lookup for the authoritative snapshot pack associated with a
319    /// pushed state. The filesystem store maintains this index as packs are
320    /// installed so hosted Push does not scan historical snapshot artifacts.
321    #[doc(hidden)]
322    pub fn snapshot_commit_descriptor_for_state(
323        &self,
324        state: &StateId,
325    ) -> Result<Option<SnapshotCommitDescriptor>> {
326        match self {
327            Self::Fs(store) => store.snapshot_commit_descriptor_for_state_impl(state),
328        }
329    }
330
331    /// Install a structured snapshot closure and its commit artifact through
332    /// the filesystem store's single durable pack barrier.
333    #[doc(hidden)]
334    pub fn put_committed_snapshot_objects_packed(
335        &self,
336        blobs: Vec<(ContentHash, Vec<u8>)>,
337        tree: &Tree,
338        state: &State,
339        attachments: Vec<StateAttachment>,
340        artifact: SnapshotCommitArtifact,
341    ) -> Result<SnapshotCommitDescriptor> {
342        match self {
343            Self::Fs(store) => store.put_committed_snapshot_objects_packed_impl(
344                blobs,
345                tree,
346                state,
347                attachments,
348                artifact,
349            ),
350        }
351    }
352}
353
354/// Trait for object storage backends.
355pub trait ObjectStore: Send + Sync {
356    fn get_blob(&self, hash: &ContentHash) -> Result<Option<Blob>>;
357    fn put_blob(&self, blob: &Blob) -> Result<ContentHash>;
358
359    /// Zero-copy variant of `get_blob`. Returns a [`bytes::Bytes`]
360    /// view of the blob's content, which for `FsStore` reads is a
361    /// slice into the pack file's mmap when the entry is non-delta
362    /// and uncompressed — no allocation, no memcpy.
363    ///
364    /// Default impl wraps `get_blob`'s `Vec<u8>` in a `Bytes` (one
365    /// Arc allocation, no body copy) so backends without a native
366    /// fast path still satisfy the contract. The mount's hot read
367    /// path goes through this method instead of `get_blob` so the
368    /// pack-mmap fast path lights up automatically.
369    fn get_blob_bytes(&self, hash: &ContentHash) -> Result<Option<bytes::Bytes>> {
370        Ok(self
371            .get_blob(hash)?
372            .map(|blob| bytes::Bytes::from(blob.into_content())))
373    }
374
375    /// Return the *uncompressed* byte length of the blob identified by
376    /// `hash`, or `Ok(None)` when the blob is not in the store.
377    ///
378    /// The contract is "size without paying for content": backends are
379    /// expected to honour this with a header read or index lookup
380    /// rather than a full decompression. This is the hot path for
381    /// directory listings (`ls -l` over a thread mount) where loading
382    /// every blob just to learn its size would dominate.
383    ///
384    /// The default implementation falls back to `get_blob` so backends
385    /// without a cheap size accessor still satisfy the contract; native
386    /// stores (`FsStore`, `InMemoryStore`) override this with a
387    /// header- or hashmap-only path.
388    fn blob_size(&self, hash: &ContentHash) -> Result<Option<u64>> {
389        Ok(self.get_blob(hash)?.map(|blob| blob.content().len() as u64))
390    }
391
392    /// Filesystem path of the loose blob whose on-disk bytes are
393    /// byte-identical to the blob's *uncompressed* content, suitable
394    /// for `hard_link`/`clonefile` materialization without going
395    /// through `get_blob`.
396    ///
397    /// Returns `None` when the blob is missing, is only available via
398    /// a packfile, is stored compressed (the on-disk bytes wouldn't
399    /// match what a worktree consumer needs to read), or the backend
400    /// doesn't expose stable filesystem paths (e.g. `InMemoryStore`). The
401    /// default impl returns `None` so non-`FsStore` backends silently fall
402    /// through to the bytes path.
403    fn loose_blob_path(&self, _hash: &ContentHash) -> Option<PathBuf> {
404        None
405    }
406
407    /// Ensure the blob identified by `hash` is materialized as an
408    /// uncompressed loose file at the canonical loose path so that
409    /// `loose_blob_path` returns `Some(path)` on a subsequent call.
410    ///
411    /// This is the "warm canonical store" path that lets the
412    /// hardlink-first materializer keep its 5–10× wall-clock and
413    /// storage-allocation wins after `pack_objects + prune_loose_objects`
414    /// has moved everything into a packfile. Without this, the lazy
415    /// hardlink path silently degrades to `fs::write(decompressed)` on
416    /// every materialize, because `loose_blob_path` returns `None` for
417    /// pack-only and compressed-loose blobs.
418    ///
419    /// Cost-amortization: the first promotion of a blob pays
420    /// `decompress + atomic write`. Every subsequent materialize of
421    /// the same blob — into the same worktree on `goto`, or into a
422    /// sibling worktree on `delegate` — is a single `link(2)`. Net
423    /// win for any N > 1 materializations; break-even at N == 1.
424    ///
425    /// Pack invariants are preserved: this method does not remove the
426    /// pack-resident copy. The blob lives in both pack and loose-
427    /// uncompressed until the next `prune_loose_objects` cycle, at
428    /// which point the loose mirror is discarded and a future
429    /// materialize re-promotes on demand.
430    ///
431    /// Idempotent: a blob that's already loose-and-uncompressed is a
432    /// no-op fast path. A blob that's loose-but-compressed is
433    /// rewritten in place (atomically) with the uncompressed bytes.
434    /// A blob that's pack-resident is decompressed out of the pack
435    /// and written loose without touching the pack.
436    ///
437    /// Returns `Ok(true)` when the call did real work (a write
438    /// happened), `Ok(false)` when it was a no-op (blob was already
439    /// loose+uncompressed), and `Err` when the blob isn't in the
440    /// store at all. The default impl returns `Ok(false)` for
441    /// backends that don't expose loose paths (`InMemoryStore`), since the
442    /// hardlink path is fundamentally inapplicable there.
443    fn promote_to_loose_uncompressed(&self, _hash: &ContentHash) -> Result<bool> {
444        Ok(false)
445    }
446
447    /// Drop any in-memory caches of decompressed blobs / trees /
448    /// states. The next access to any object pays full I/O +
449    /// decompression cost. No-op for stores that don't cache
450    /// (`InMemoryStore` is already the source of truth).
451    ///
452    /// Exposed primarily for benchmarks that want to measure the
453    /// true cold-cache path without rebuilding the store from
454    /// scratch. Production callers don't need to invoke this.
455    fn clear_recent_caches(&self) {}
456
457    fn put_blob_with_hash(&self, blob: &Blob, hash: ContentHash) -> Result<ContentHash> {
458        if blob.hash() != hash {
459            return Err(HeddleError::InvalidObject("blob hash mismatch".to_string()));
460        }
461        self.put_blob(blob)
462    }
463
464    fn has_blob(&self, hash: &ContentHash) -> Result<bool>;
465    /// Return whether the blob is owned by this store, excluding any configured
466    /// read-through source. Snapshot builders use this to ensure a new native
467    /// state owns its complete object closure.
468    fn has_blob_locally(&self, hash: &ContentHash) -> Result<bool> {
469        self.has_blob(hash)
470    }
471    fn get_tree(&self, hash: &ContentHash) -> Result<Option<Tree>>;
472    fn put_tree(&self, tree: &Tree) -> Result<ContentHash>;
473    fn has_tree(&self, hash: &ContentHash) -> Result<bool>;
474    /// Return whether the tree is owned by this store, excluding any configured
475    /// read-through source.
476    fn has_tree_locally(&self, hash: &ContentHash) -> Result<bool> {
477        self.has_tree(hash)
478    }
479    fn get_state(&self, id: &StateId) -> Result<Option<State>>;
480    fn put_state(&self, state: &State) -> Result<()>;
481    fn has_state(&self, id: &StateId) -> Result<bool>;
482    fn list_states(&self) -> Result<Vec<StateId>>;
483    fn get_state_attachment(
484        &self,
485        _state: &StateId,
486        _id: &StateAttachmentId,
487    ) -> Result<Option<StateAttachment>> {
488        Ok(None)
489    }
490    fn put_state_attachment(&self, _attachment: &StateAttachment) -> Result<StateAttachmentId> {
491        Err(HeddleError::InvalidObject(
492            "object store does not support state attachments".to_string(),
493        ))
494    }
495    fn list_state_attachments(&self, _state: &StateId) -> Result<Vec<StateAttachment>> {
496        Ok(Vec::new())
497    }
498    fn get_action(&self, id: &ActionId) -> Result<Option<Action>>;
499    fn put_action(&self, action: &mut Action) -> Result<ActionId>;
500    fn list_actions(&self) -> Result<Vec<ActionId>>;
501    fn list_blobs(&self) -> Result<Vec<ContentHash>>;
502    fn list_trees(&self) -> Result<Vec<ContentHash>>;
503
504    fn put_blob_bytes_with_hash(&self, data: &[u8], hash: ContentHash) -> Result<ContentHash> {
505        self.put_blob_with_hash(&Blob::from_slice(data), hash)
506    }
507
508    /// Return the raw rmp-encoded tree body for `hash`.
509    ///
510    /// This is a migration seam, not a runtime compatibility reader: callers
511    /// that need current tree semantics should use [`ObjectStore::get_tree`].
512    /// Backends with direct raw storage override this so one-shot migrations can
513    /// canonicalize older tree encodings without reintroducing fallback decode
514    /// into the durable `Tree` type.
515    fn get_tree_serialized(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
516        Ok(self
517            .get_tree(hash)?
518            .map(|tree| rmp_serde::to_vec(&tree))
519            .transpose()?)
520    }
521
522    fn put_tree_serialized(&self, data: &[u8], hash: ContentHash) -> Result<ContentHash> {
523        let tree: Tree = rmp_serde::from_slice(data)?;
524        tree.validate()?;
525        if tree.hash() != hash {
526            return Err(HeddleError::Corruption {
527                expected: hash,
528                found: tree.hash(),
529            });
530        }
531        self.put_tree(&tree)
532    }
533
534    fn put_state_serialized(&self, data: &[u8], id: StateId) -> Result<()> {
535        let state: State = rmp_serde::from_slice(data)?;
536        let found = state.id();
537        if found != id {
538            return Err(HeddleError::InvalidObject(format!(
539                "state id mismatch: expected {id}, computed {found}"
540            )));
541        }
542        self.put_state(&state)
543    }
544
545    fn put_action_serialized(&self, data: &[u8], id: ActionId) -> Result<()> {
546        let mut action: Action = rmp_serde::from_slice(data)?;
547        let found_id = action.compute_id();
548        if found_id != id {
549            return Err(HeddleError::InvalidObject(format!(
550                "action id mismatch: expected {}, found {}",
551                id, found_id
552            )));
553        }
554        let stored_id = self.put_action(&mut action)?;
555        if stored_id != id {
556            return Err(HeddleError::InvalidObject(format!(
557                "action id mismatch after write: expected {}, found {}",
558                id, stored_id
559            )));
560        }
561        Ok(())
562    }
563
564    fn get_pack_object(
565        &self,
566        id: &pack::PackObjectId,
567    ) -> Result<Option<(pack::ObjectType, Vec<u8>)>> {
568        match id {
569            pack::PackObjectId::Hash(hash) => {
570                if let Some(blob) = self.get_blob(hash)? {
571                    return Ok(Some((pack::ObjectType::Blob, blob.content().to_vec())));
572                }
573                if let Some(tree) = self.get_tree(hash)? {
574                    return Ok(Some((
575                        pack::ObjectType::Tree,
576                        rmp_serde::to_vec_named(&tree)?,
577                    )));
578                }
579                if let Some(action) = self.get_action(&ActionId::from_hash(*hash))? {
580                    return Ok(Some((
581                        pack::ObjectType::Action,
582                        rmp_serde::to_vec_named(&action)?,
583                    )));
584                }
585                Ok(None)
586            }
587            pack::PackObjectId::StateId(change_id) => {
588                if let Some(state) = self.get_state(change_id)? {
589                    Ok(Some((
590                        pack::ObjectType::State,
591                        rmp_serde::to_vec_named(&state)?,
592                    )))
593                } else {
594                    Ok(None)
595                }
596            }
597        }
598    }
599
600    /// Bulk-write a batch of blobs as a single durable unit. The default
601    /// implementation falls back to per-blob writes; backends that
602    /// support packfiles (i.e. `FsStore`) override this to install one
603    /// packfile + index — two fsyncs total instead of N. Used by the
604    /// snapshot hot path so writing 1000 small files takes ~one fsync,
605    /// not 1000.
606    ///
607    /// Blobs already present in the store are skipped on the way in
608    /// (the caller would otherwise duplicate them in the pack).
609    fn put_blobs_packed(&self, blobs: Vec<(ContentHash, Vec<u8>)>) -> Result<()> {
610        for (hash, data) in blobs {
611            if !self.has_blob(&hash)? {
612                self.put_blob_bytes_with_hash(&data, hash)?;
613            }
614        }
615        Ok(())
616    }
617
618    /// Durably install a snapshot's newly-authored immutable object closure as
619    /// one storage batch. Pack-capable backends override this to share one pack
620    /// installation across blobs, the root tree, and the state; other backends
621    /// preserve the same ordering through their ordinary object methods.
622    fn put_snapshot_objects_packed(
623        &self,
624        blobs: Vec<(ContentHash, Vec<u8>)>,
625        tree: &Tree,
626        state: &State,
627    ) -> Result<()> {
628        self.put_blobs_packed(blobs)?;
629        self.put_tree(tree)?;
630        self.put_state(state)
631    }
632
633    /// Snapshot closure variant that also durably installs immutable authored
634    /// attachments. The separate method preserves the existing backend API;
635    /// pack-capable stores override it to share the snapshot pack barrier.
636    fn put_snapshot_objects_and_attachments_packed(
637        &self,
638        blobs: Vec<(ContentHash, Vec<u8>)>,
639        tree: &Tree,
640        state: &State,
641        attachments: Vec<StateAttachment>,
642    ) -> Result<()> {
643        self.put_snapshot_objects_packed(blobs, tree, state)?;
644        for attachment in attachments {
645            self.put_state_attachment(&attachment)?;
646        }
647        Ok(())
648    }
649    fn install_pack(&self, pack_data: &[u8], index_data: &[u8]) -> Result<Vec<pack::PackObjectId>> {
650        let reader = pack::PackReader::from_slice(pack_data, index_data)?;
651        let ids = reader.list_ids()?;
652        for id in &ids {
653            let Some((obj_type, data)) = reader.get_object(id)? else {
654                continue;
655            };
656            match (id, obj_type) {
657                (pack::PackObjectId::Hash(hash), pack::ObjectType::Blob) => {
658                    self.put_blob_bytes_with_hash(&data, *hash)?;
659                }
660                (pack::PackObjectId::Hash(hash), pack::ObjectType::Tree) => {
661                    self.put_tree_serialized(&data, *hash)?;
662                }
663                (pack::PackObjectId::Hash(hash), pack::ObjectType::Action) => {
664                    self.put_action_serialized(&data, ActionId::from_hash(*hash))?;
665                }
666                (pack::PackObjectId::StateId(change_id), pack::ObjectType::State) => {
667                    self.put_state_serialized(&data, *change_id)?;
668                }
669                (_, pack::ObjectType::TimelineOperation) => {
670                    return Err(HeddleError::InvalidObject(
671                        "timeline operations belong in the timeline pack store".to_string(),
672                    ));
673                }
674                _ => {
675                    return Err(HeddleError::InvalidObject(format!(
676                        "unsupported native pack object: {:?} {:?}",
677                        id, obj_type
678                    )));
679                }
680            }
681        }
682        Ok(ids)
683    }
684
685    /// Install a pack and its index from on-disk files
686    /// (typically produced by `StreamingPackBuilder`). The default
687    /// impl reads both files fully and delegates to `install_pack`,
688    /// so any backend that doesn't override this still works (at the
689    /// cost of giving back the bounded-memory promise). Real fs-
690    /// backed stores override this to `rename(2)` both files into the
691    /// pack directory without ever loading them.
692    ///
693    /// On success, the source files at `pack_path`/`index_path` may
694    /// have been moved or removed depending on the backend; callers
695    /// shouldn't continue to rely on them.
696    ///
697    /// Returns the ids of the installed objects — the same set
698    /// `install_pack` reports for the equivalent byte-buffer install,
699    /// so callers (e.g. native sync) read the installed ids off the
700    /// install result instead of tracking them out-of-band.
701    fn install_pack_streaming(
702        &self,
703        pack_path: &std::path::Path,
704        index_path: &std::path::Path,
705    ) -> Result<Vec<pack::PackObjectId>> {
706        let pack_data = std::fs::read(pack_path).map_err(StoreError::from)?;
707        let index_data = std::fs::read(index_path).map_err(StoreError::from)?;
708        let ids = self.install_pack(&pack_data, &index_data)?;
709        // Default impl: clean up the staged files. Override
710        // implementations that move/rename should not call super and
711        // should manage the file lifecycle themselves.
712        let _ = std::fs::remove_file(pack_path);
713        let _ = std::fs::remove_file(index_path);
714        Ok(ids)
715    }
716
717    fn pack_objects(&self, delta_search: bool) -> Result<(u64, u64)> {
718        let _ = delta_search;
719        Ok((0, 0))
720    }
721
722    fn prune_loose_objects(&self) -> Result<(u64, u64)> {
723        Ok((0, 0))
724    }
725
726    /// Remove only pack/index pairs that fail checksum, index, or object-hash
727    /// validation so a clone repair pull advertises their objects as missing.
728    fn discard_corrupt_clone_packs(&self) -> Result<usize> {
729        Ok(0)
730    }
731
732    fn begin_snapshot_write_batch(&self) -> Result<()> {
733        Ok(())
734    }
735
736    fn flush_snapshot_write_batch(&self) -> Result<()> {
737        Ok(())
738    }
739
740    fn abort_snapshot_write_batch(&self) {}
741
742    /// Whether the store holds any redaction record for the given blob.
743    ///
744    /// Redactions live in a sidecar (`<heddle_dir>/redactions/`) that is
745    /// structurally outside the content-addressed object graph so GC
746    /// can't reach them. The wire layer needs a cheap probe to decide
747    /// whether to ship a redaction for a blob in the closure, so this
748    /// is a separate method rather than a `get_*` + null check.
749    ///
750    /// Default impl returns `Ok(false)` — stores that don't model
751    /// redactions silently report "no redactions," which is the
752    /// correct behaviour for purely in-memory or remote-shim stores.
753    fn has_redactions_for_blob(&self, _blob: &ContentHash) -> Result<bool> {
754        Ok(false)
755    }
756
757    /// Return the raw rmp-encoded `RedactionsBlob` bytes for the given
758    /// blob, or `Ok(None)` if no redaction record exists. The bytes
759    /// are byte-identical to what was written by `put_redactions_bytes_for_blob`
760    /// (or by `Repository::put_redaction`); this is the wire-transfer
761    /// payload, not a re-serialized view.
762    ///
763    /// Default impl returns `Ok(None)`.
764    fn get_redactions_bytes_for_blob(&self, _blob: &ContentHash) -> Result<Option<Vec<u8>>> {
765        Ok(None)
766    }
767
768    /// Persist the rmp-encoded `RedactionsBlob` bytes for the given
769    /// blob. Receiver-side replay calls this after signature
770    /// verification so the bytes land in the same sidecar that the
771    /// sender's `Repository::put_redaction` writes to.
772    ///
773    /// Default impl returns an "unsupported" error — stores that don't
774    /// model redactions (e.g. read-only shims) refuse rather than
775    /// silently dropping the record.
776    fn put_redactions_bytes_for_blob(&self, _blob: &ContentHash, _bytes: &[u8]) -> Result<()> {
777        Err(HeddleError::InvalidObject(
778            "this object store does not support persisting redactions".to_string(),
779        ))
780    }
781
782    /// List every blob that has at least one redaction record. Used by
783    /// the GC pin guard and by sync to enumerate redactions for the
784    /// state closure. Order is unspecified; callers that need stable
785    /// ordering should sort.
786    ///
787    /// Default impl returns `Ok(vec![])`.
788    fn list_blobs_with_redactions(&self) -> Result<Vec<ContentHash>> {
789        Ok(Vec::new())
790    }
791
792    /// Whether the store holds any state-visibility record for `state`.
793    ///
794    /// Like redactions, state-visibility records live in a sidecar outside
795    /// the content-addressed object graph and cannot ride native packs.
796    /// Sync uses this probe while enumerating a state closure so a non-public
797    /// state can advertise the sidecar that must travel out-of-pack.
798    ///
799    /// Default impl returns `Ok(false)` for stores that do not model this
800    /// sidecar.
801    fn has_state_visibility_for_state(&self, _state: &StateId) -> Result<bool> {
802        Ok(false)
803    }
804
805    /// Return the raw rmp-encoded `StateVisibilityBlob` bytes for `state`,
806    /// or `Ok(None)` if no sidecar exists. The bytes are the wire-transfer
807    /// payload for state visibility.
808    ///
809    /// Default impl returns `Ok(None)`.
810    fn get_state_visibility_bytes_for_state(&self, _state: &StateId) -> Result<Option<Vec<u8>>> {
811        Ok(None)
812    }
813
814    /// Persist raw `StateVisibilityBlob` bytes for `state`.
815    ///
816    /// Default impl returns an "unsupported" error so stores that do not
817    /// model the sidecar refuse instead of dropping it.
818    fn put_state_visibility_bytes_for_state(&self, _state: &StateId, _bytes: &[u8]) -> Result<()> {
819        Err(HeddleError::InvalidObject(
820            "this object store does not support persisting state visibility".to_string(),
821        ))
822    }
823
824    /// List every state with at least one state-visibility record.
825    ///
826    /// Default impl returns `Ok(vec![])`.
827    fn list_states_with_visibility(&self) -> Result<Vec<StateId>> {
828        Ok(Vec::new())
829    }
830}
831
832#[cfg(test)]
833mod any_store_tests {
834    use tempfile::TempDir;
835
836    use super::*;
837    use crate::object::{Attribution, Operation, Principal};
838
839    fn fs_any_store() -> (TempDir, AnyStore) {
840        let temp = TempDir::new().unwrap();
841        let store = FsStore::new(temp.path().join(".heddle"));
842        store.init().unwrap();
843        (temp, AnyStore::Fs(store))
844    }
845
846    /// Drive every `ObjectStore` method through the `AnyStore::Fs` dispatch arm
847    /// so the enum's match-dispatch is exercised end-to-end. This is the
848    /// coverage seam for heddle#283: each arm forwards to the inner concrete
849    /// store, and a missing arm would fail to compile or silently fall back to
850    /// a trait default.
851    #[test]
852    fn fs_variant_dispatches_every_object_store_method() {
853        let (_temp, store) = fs_any_store();
854
855        // ── Blobs ──
856        let blob = Blob::from("any-store dispatch blob");
857        let blob_hash = store.put_blob(&blob).unwrap();
858        assert_eq!(
859            ObjectStore::get_blob(&store, &blob_hash)
860                .unwrap()
861                .unwrap()
862                .content(),
863            blob.content()
864        );
865        assert!(store.has_blob(&blob_hash).unwrap());
866        assert_eq!(
867            ObjectStore::get_blob_bytes(&store, &blob_hash)
868                .unwrap()
869                .unwrap()
870                .as_ref(),
871            blob.content()
872        );
873        assert_eq!(
874            store.blob_size(&blob_hash).unwrap().unwrap(),
875            blob.content().len() as u64
876        );
877        assert!(store.loose_blob_path(&blob_hash).is_some());
878        store.promote_to_loose_uncompressed(&blob_hash).unwrap();
879        assert!(store.list_blobs().unwrap().contains(&blob_hash));
880
881        let bytes_blob = Blob::from("put-with-hash blob");
882        let bytes_hash = bytes_blob.hash();
883        assert_eq!(
884            store.put_blob_with_hash(&bytes_blob, bytes_hash).unwrap(),
885            bytes_hash
886        );
887        let raw_blob = Blob::from("raw bytes blob");
888        let raw_hash = raw_blob.hash();
889        assert_eq!(
890            store
891                .put_blob_bytes_with_hash(raw_blob.content(), raw_hash)
892                .unwrap(),
893            raw_hash
894        );
895
896        // ── Trees ──
897        let tree = Tree::new();
898        let tree_hash = store.put_tree(&tree).unwrap();
899        assert!(ObjectStore::get_tree(&store, &tree_hash).unwrap().is_some());
900        assert!(store.has_tree(&tree_hash).unwrap());
901        assert!(store.list_trees().unwrap().contains(&tree_hash));
902        let tree2 = Tree::new();
903        let tree2_bytes = rmp_serde::to_vec_named(&tree2).unwrap();
904        assert_eq!(
905            store
906                .put_tree_serialized(&tree2_bytes, tree2.hash())
907                .unwrap(),
908            tree2.hash()
909        );
910
911        // ── States ──
912        let attribution =
913            Attribution::human(Principal::new("AnyStore Test", "anystore@example.com"));
914        let state = State::new(tree_hash, vec![], attribution.clone());
915        let state_id = state.id();
916        store.put_state(&state).unwrap();
917        assert!(ObjectStore::get_state(&store, &state_id).unwrap().is_some());
918        assert!(store.has_state(&state_id).unwrap());
919        assert!(store.list_states().unwrap().contains(&state_id));
920        let state2 = State::new(tree2.hash(), vec![], attribution.clone());
921        let state2_bytes = rmp_serde::to_vec_named(&state2).unwrap();
922        store
923            .put_state_serialized(&state2_bytes, state2.id())
924            .unwrap();
925
926        // ── Actions ──
927        let mut action = Action::new(
928            None,
929            StateId::from_bytes([3; 32]),
930            Operation::Snapshot,
931            "any-store action",
932            attribution,
933        );
934        let action_id = store.put_action(&mut action).unwrap();
935        assert!(store.get_action(&action_id).unwrap().is_some());
936        assert!(store.list_actions().unwrap().contains(&action_id));
937        let action_bytes = rmp_serde::to_vec_named(&action).unwrap();
938        store
939            .put_action_serialized(&action_bytes, action_id)
940            .unwrap();
941
942        // ── Packs ──
943        let packed = Blob::from("packed-via-any-store");
944        let packed_hash = packed.hash();
945        store
946            .put_blobs_packed(vec![(packed_hash, packed.into_content())])
947            .unwrap();
948        assert!(
949            store
950                .get_pack_object(&pack::PackObjectId::Hash(packed_hash))
951                .unwrap()
952                .is_some()
953        );
954        store.pack_objects(false).unwrap();
955        store.prune_loose_objects().unwrap();
956        // install_pack / install_pack_streaming need valid packfile inputs;
957        // exercising the dispatch arm with bogus data is enough — we only
958        // assert the call routes through the enum, not the backend behaviour.
959        let _ = store.install_pack(&[], &[]);
960        let _ = store.install_pack_streaming(
961            std::path::Path::new("/nonexistent/pack"),
962            std::path::Path::new("/nonexistent/idx"),
963        );
964
965        // ── Snapshot write batch ──
966        store.begin_snapshot_write_batch().unwrap();
967        store.flush_snapshot_write_batch().unwrap();
968        store.begin_snapshot_write_batch().unwrap();
969        store.abort_snapshot_write_batch();
970
971        // ── Redactions ──
972        let redaction = b"any-store redaction bytes";
973        store
974            .put_redactions_bytes_for_blob(&blob_hash, redaction)
975            .unwrap();
976        assert!(store.has_redactions_for_blob(&blob_hash).unwrap());
977        assert_eq!(
978            store
979                .get_redactions_bytes_for_blob(&blob_hash)
980                .unwrap()
981                .as_deref(),
982            Some(redaction.as_slice())
983        );
984        assert!(
985            store
986                .list_blobs_with_redactions()
987                .unwrap()
988                .contains(&blob_hash)
989        );
990
991        // ── State visibility ──
992        let state_visibility = b"any-store state visibility bytes";
993        store
994            .put_state_visibility_bytes_for_state(&state_id, state_visibility)
995            .unwrap();
996        assert!(store.has_state_visibility_for_state(&state_id).unwrap());
997        assert_eq!(
998            store
999                .get_state_visibility_bytes_for_state(&state_id)
1000                .unwrap()
1001                .as_deref(),
1002            Some(state_visibility.as_slice())
1003        );
1004        assert!(
1005            store
1006                .list_states_with_visibility()
1007                .unwrap()
1008                .contains(&state_id)
1009        );
1010
1011        // ── Caches ──
1012        store.clear_recent_caches();
1013    }
1014}