prolly/prolly/
manifest.rs

1//! Named root manifest support.
2//!
3//! Content-addressed tree nodes identify immutable snapshots, but applications
4//! also need durable names such as `main`, `workspace/123/latest`, or
5//! `agent-run/abc/checkpoint/42`. The manifest layer records those names
6//! separately from node storage and provides compare-and-swap updates for
7//! concurrent writers.
8
9use serde::{Deserialize, Serialize};
10use std::sync::Arc;
11use std::time::Duration;
12
13use super::cid::Cid;
14use super::config::Config;
15use super::error::Error;
16use super::tree::Tree;
17
18const ROOT_MANIFEST_VERSION: u64 = 1;
19
20#[derive(Serialize, Deserialize)]
21struct RootManifestWire {
22    version: u64,
23    root: Option<Cid>,
24    config: Config,
25    #[serde(default, skip_serializing_if = "Option::is_none")]
26    created_at_millis: Option<u64>,
27    #[serde(default, skip_serializing_if = "Option::is_none")]
28    updated_at_millis: Option<u64>,
29}
30
31/// Durable named-root payload.
32///
33/// A root manifest stores the complete tree handle needed to reopen a named
34/// snapshot. The root CID alone is not enough because chunking and encoding
35/// config determine how the tree should be interpreted.
36#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
37pub struct RootManifest {
38    /// Root node CID, or `None` for the empty tree.
39    pub root: Option<Cid>,
40    /// Tree configuration associated with this root.
41    pub config: Config,
42    /// Optional Unix timestamp in milliseconds when this named root was created.
43    pub created_at_millis: Option<u64>,
44    /// Optional Unix timestamp in milliseconds when this named root was updated.
45    pub updated_at_millis: Option<u64>,
46}
47
48impl RootManifest {
49    /// Create a new manifest from a root CID and config.
50    pub fn new(root: Option<Cid>, config: Config) -> Self {
51        Self {
52            root,
53            config,
54            created_at_millis: None,
55            updated_at_millis: None,
56        }
57    }
58
59    /// Set optional creation and update timestamps in Unix milliseconds.
60    pub fn with_timestamps_millis(
61        mut self,
62        created_at_millis: Option<u64>,
63        updated_at_millis: Option<u64>,
64    ) -> Self {
65        self.created_at_millis = created_at_millis;
66        self.updated_at_millis = updated_at_millis;
67        self
68    }
69
70    /// Set the creation timestamp in Unix milliseconds.
71    pub fn with_created_at_millis(mut self, created_at_millis: u64) -> Self {
72        self.created_at_millis = Some(created_at_millis);
73        self
74    }
75
76    /// Set the update timestamp in Unix milliseconds.
77    pub fn with_updated_at_millis(mut self, updated_at_millis: u64) -> Self {
78        self.updated_at_millis = Some(updated_at_millis);
79        self
80    }
81
82    /// Create a manifest from a tree and timestamp metadata.
83    pub fn from_tree_with_timestamps_millis(
84        tree: &Tree,
85        created_at_millis: Option<u64>,
86        updated_at_millis: Option<u64>,
87    ) -> Self {
88        Self {
89            root: tree.root.clone(),
90            config: tree.config.clone(),
91            created_at_millis,
92            updated_at_millis,
93        }
94    }
95
96    /// Create a manifest from an existing tree handle.
97    pub fn from_tree(tree: &Tree) -> Self {
98        Self {
99            root: tree.root.clone(),
100            config: tree.config.clone(),
101            created_at_millis: None,
102            updated_at_millis: None,
103        }
104    }
105
106    /// Convert this manifest into a tree handle.
107    pub fn into_tree(self) -> Tree {
108        Tree {
109            root: self.root,
110            config: self.config,
111        }
112    }
113
114    /// Convert this manifest into a cloned tree handle.
115    pub fn to_tree(&self) -> Tree {
116        Tree {
117            root: self.root.clone(),
118            config: self.config.clone(),
119        }
120    }
121
122    /// Serialize this manifest to a versioned, deterministic binary payload.
123    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
124        let wire = RootManifestWire {
125            version: ROOT_MANIFEST_VERSION,
126            root: self.root.clone(),
127            config: self.config.clone(),
128            created_at_millis: self.created_at_millis,
129            updated_at_millis: self.updated_at_millis,
130        };
131        serde_cbor::ser::to_vec_packed(&wire).map_err(|err| Error::Deserialize(err.to_string()))
132    }
133
134    /// Decode a manifest from bytes produced by [`RootManifest::to_bytes`].
135    pub fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
136        let wire: RootManifestWire =
137            serde_cbor::from_slice(bytes).map_err(|err| Error::Deserialize(err.to_string()))?;
138        if wire.version != ROOT_MANIFEST_VERSION {
139            return Err(Error::Deserialize(format!(
140                "unsupported root manifest version: {}",
141                wire.version
142            )));
143        }
144        Ok(Self {
145            root: wire.root,
146            config: wire.config,
147            created_at_millis: wire.created_at_millis,
148            updated_at_millis: wire.updated_at_millis,
149        })
150    }
151}
152
153impl From<Tree> for RootManifest {
154    fn from(tree: Tree) -> Self {
155        Self {
156            root: tree.root,
157            config: tree.config,
158            created_at_millis: None,
159            updated_at_millis: None,
160        }
161    }
162}
163
164impl From<&Tree> for RootManifest {
165    fn from(tree: &Tree) -> Self {
166        Self::from_tree(tree)
167    }
168}
169
170impl From<RootManifest> for Tree {
171    fn from(manifest: RootManifest) -> Self {
172        manifest.into_tree()
173    }
174}
175
176/// A named root manifest returned by manifest-store scans.
177#[derive(Clone, Debug, PartialEq)]
178pub struct NamedRootManifest {
179    /// Durable name of the root manifest.
180    pub name: Vec<u8>,
181    /// Manifest stored under `name`.
182    pub manifest: RootManifest,
183}
184
185impl NamedRootManifest {
186    /// Create a named manifest entry.
187    pub fn new(name: Vec<u8>, manifest: RootManifest) -> Self {
188        Self { name, manifest }
189    }
190
191    /// Convert this manifest entry into a tree-oriented named root.
192    pub fn into_named_root(self) -> NamedRoot {
193        NamedRoot {
194            name: self.name,
195            tree: self.manifest.into_tree(),
196        }
197    }
198
199    /// Convert this manifest entry into a cloned tree-oriented named root.
200    pub fn to_named_root(&self) -> NamedRoot {
201        NamedRoot {
202            name: self.name.clone(),
203            tree: self.manifest.to_tree(),
204        }
205    }
206}
207
208/// A named root loaded through a [`crate::Prolly`] manager.
209#[derive(Clone, Debug, PartialEq)]
210pub struct NamedRoot {
211    /// Durable name of the root.
212    pub name: Vec<u8>,
213    /// Tree handle stored under `name`.
214    pub tree: Tree,
215}
216
217impl NamedRoot {
218    /// Create a named tree root.
219    pub fn new(name: Vec<u8>, tree: Tree) -> Self {
220        Self { name, tree }
221    }
222}
223
224/// Result of loading roots for a retention policy.
225#[derive(Clone, Debug, Default, PartialEq)]
226pub struct NamedRootSelection {
227    /// Roots selected by the retention policy.
228    pub roots: Vec<NamedRoot>,
229    /// Exact names requested by the policy that were not present.
230    pub missing_names: Vec<Vec<u8>>,
231}
232
233impl NamedRootSelection {
234    /// Create a new selection from roots and missing exact names.
235    pub fn new(roots: Vec<NamedRoot>, missing_names: Vec<Vec<u8>>) -> Self {
236        Self {
237            roots,
238            missing_names,
239        }
240    }
241
242    /// Whether every exact name requested by the retention policy was present.
243    pub fn is_complete(&self) -> bool {
244        self.missing_names.is_empty()
245    }
246
247    /// Clone selected tree handles for GC APIs.
248    pub fn trees(&self) -> Vec<Tree> {
249        self.roots.iter().map(|root| root.tree.clone()).collect()
250    }
251
252    /// Consume the selection and return tree handles for GC APIs.
253    pub fn into_trees(self) -> Vec<Tree> {
254        self.roots.into_iter().map(|root| root.tree).collect()
255    }
256}
257
258/// Policy for selecting named roots to retain during garbage collection.
259///
260/// `NewestByName` keeps the lexicographically greatest names matching `prefix`.
261/// This works well for names that include sortable sequence numbers or
262/// timestamps, for example `checkpoint/000042` or
263/// `checkpoint/2026-07-01T12:00:00Z`.
264#[derive(Clone, Debug, PartialEq)]
265pub enum NamedRootRetention {
266    /// Retain every named root in the manifest store.
267    All,
268    /// Retain an explicit list of root names.
269    Exact {
270        /// Exact names to load.
271        names: Vec<Vec<u8>>,
272    },
273    /// Retain every named root whose name starts with `prefix`.
274    Prefix {
275        /// Name prefix to retain.
276        prefix: Vec<u8>,
277    },
278    /// Retain the lexicographically newest `count` roots with a prefix.
279    NewestByName {
280        /// Name prefix to retain.
281        prefix: Vec<u8>,
282        /// Maximum number of roots to retain.
283        count: usize,
284    },
285    /// Retain roots with `updated_at_millis >= min_updated_at_millis`.
286    UpdatedSince {
287        /// Optional name prefix to retain.
288        prefix: Vec<u8>,
289        /// Minimum update timestamp in Unix milliseconds.
290        min_updated_at_millis: u64,
291    },
292}
293
294impl NamedRootRetention {
295    /// Retain every named root in the manifest store.
296    pub fn all() -> Self {
297        Self::All
298    }
299
300    /// Retain an explicit list of named roots.
301    pub fn exact<I, N>(names: I) -> Self
302    where
303        I: IntoIterator<Item = N>,
304        N: AsRef<[u8]>,
305    {
306        Self::Exact {
307            names: names
308                .into_iter()
309                .map(|name| name.as_ref().to_vec())
310                .collect(),
311        }
312    }
313
314    /// Retain every named root whose name starts with `prefix`.
315    pub fn prefix(prefix: impl AsRef<[u8]>) -> Self {
316        Self::Prefix {
317            prefix: prefix.as_ref().to_vec(),
318        }
319    }
320
321    /// Retain the lexicographically newest `count` roots with `prefix`.
322    pub fn newest_by_name(prefix: impl AsRef<[u8]>, count: usize) -> Self {
323        Self::NewestByName {
324            prefix: prefix.as_ref().to_vec(),
325            count,
326        }
327    }
328
329    /// Retain roots updated at or after `min_updated_at_millis`.
330    pub fn updated_since(prefix: impl AsRef<[u8]>, min_updated_at_millis: u64) -> Self {
331        Self::UpdatedSince {
332            prefix: prefix.as_ref().to_vec(),
333            min_updated_at_millis,
334        }
335    }
336
337    /// Retain roots updated within `max_age` before `now_millis`.
338    ///
339    /// `now_millis` is explicit so tests, replay, and distributed systems can
340    /// choose their own clock source. The cutoff saturates at zero when
341    /// `max_age` is larger than `now_millis`.
342    pub fn updated_within(prefix: impl AsRef<[u8]>, now_millis: u64, max_age: Duration) -> Self {
343        Self::updated_within_millis(prefix, now_millis, duration_millis_saturating(max_age))
344    }
345
346    /// Retain roots updated within `window_millis` before `now_millis`.
347    pub fn updated_within_millis(
348        prefix: impl AsRef<[u8]>,
349        now_millis: u64,
350        window_millis: u64,
351    ) -> Self {
352        Self::updated_since(prefix, now_millis.saturating_sub(window_millis))
353    }
354
355    /// Retain roots updated within `days` calendar-day windows before `now_millis`.
356    ///
357    /// This is a convenience wrapper for retention rules such as "keep roots
358    /// updated in the last 7 days". A day is treated as exactly 86,400,000
359    /// milliseconds.
360    pub fn updated_within_days(prefix: impl AsRef<[u8]>, now_millis: u64, days: u64) -> Self {
361        Self::updated_within_millis(prefix, now_millis, days.saturating_mul(86_400_000))
362    }
363}
364
365fn duration_millis_saturating(duration: Duration) -> u64 {
366    duration.as_millis().min(u128::from(u64::MAX)) as u64
367}
368
369pub(crate) fn sort_named_root_manifests(roots: &mut [NamedRootManifest]) {
370    roots.sort_by(|left, right| left.name.cmp(&right.name));
371}
372
373/// Result of a named-root compare-and-swap update.
374#[derive(Clone, Debug, PartialEq)]
375#[allow(clippy::large_enum_variant)]
376pub enum ManifestUpdate {
377    /// The expected manifest matched and the update was applied.
378    Applied,
379    /// The expected manifest did not match the current manifest.
380    Conflict {
381        /// Current manifest stored under the requested name.
382        current: Option<RootManifest>,
383    },
384}
385
386impl ManifestUpdate {
387    /// Whether the update was applied.
388    pub fn is_applied(&self) -> bool {
389        matches!(self, Self::Applied)
390    }
391
392    /// Whether the update failed because the current manifest differed.
393    pub fn is_conflict(&self) -> bool {
394        matches!(self, Self::Conflict { .. })
395    }
396
397    /// Current manifest for conflicts, or `None` for applied updates.
398    pub fn current(&self) -> Option<&RootManifest> {
399        match self {
400            Self::Applied => None,
401            Self::Conflict { current } => current.as_ref(),
402        }
403    }
404}
405
406/// Result of a named-root compare-and-swap through a [`crate::Prolly`] manager.
407#[derive(Clone, Debug, PartialEq)]
408#[allow(clippy::large_enum_variant)]
409pub enum NamedRootUpdate {
410    /// The expected tree matched and the update was applied.
411    Applied,
412    /// The expected tree did not match the current named root.
413    Conflict {
414        /// Current tree stored under the requested name.
415        current: Option<Tree>,
416    },
417}
418
419impl NamedRootUpdate {
420    /// Whether the update was applied.
421    pub fn is_applied(&self) -> bool {
422        matches!(self, Self::Applied)
423    }
424
425    /// Whether the update failed because the current tree differed.
426    pub fn is_conflict(&self) -> bool {
427        matches!(self, Self::Conflict { .. })
428    }
429
430    /// Current tree for conflicts, or `None` for applied updates.
431    pub fn current(&self) -> Option<&Tree> {
432        match self {
433            Self::Applied => None,
434            Self::Conflict { current } => current.as_ref(),
435        }
436    }
437}
438
439impl From<ManifestUpdate> for NamedRootUpdate {
440    fn from(update: ManifestUpdate) -> Self {
441        match update {
442            ManifestUpdate::Applied => Self::Applied,
443            ManifestUpdate::Conflict { current } => Self::Conflict {
444                current: current.map(RootManifest::into_tree),
445            },
446        }
447    }
448}
449
450/// Storage for named root manifests.
451///
452/// This trait is separate from [`crate::Store`] so content-addressed node stores
453/// remain simple. Backends that can update a named root atomically should
454/// implement compare-and-swap directly in their native transaction mechanism.
455pub trait ManifestStore: Send + Sync {
456    /// Error type for manifest operations.
457    type Error: std::error::Error + Send + Sync + 'static;
458
459    /// Load a named root manifest.
460    fn get_root(&self, name: &[u8]) -> Result<Option<RootManifest>, Self::Error>;
461
462    /// Unconditionally insert or replace a named root manifest.
463    fn put_root(&self, name: &[u8], manifest: &RootManifest) -> Result<(), Self::Error>;
464
465    /// Delete a named root manifest. Deleting a missing name is not an error.
466    fn delete_root(&self, name: &[u8]) -> Result<(), Self::Error>;
467
468    /// Atomically update a named root if the current manifest matches
469    /// `expected`.
470    ///
471    /// `expected == None` means the name must be absent. `new == None` deletes
472    /// the name when the compare succeeds.
473    fn compare_and_swap_root(
474        &self,
475        name: &[u8],
476        expected: Option<&RootManifest>,
477        new: Option<&RootManifest>,
478    ) -> Result<ManifestUpdate, Self::Error>;
479}
480
481/// Manifest stores that can enumerate durable named roots.
482///
483/// This trait is separate from [`ManifestStore`] because point lookups and CAS
484/// updates are enough for simple applications, while store-wide garbage
485/// collection needs an explicit listing capability. Implementations must return
486/// roots sorted by raw name bytes for deterministic retention planning.
487pub trait ManifestStoreScan: ManifestStore {
488    /// List all durable named root manifests.
489    fn list_roots(&self) -> Result<Vec<NamedRootManifest>, Self::Error>;
490}
491
492/// Async storage for named root manifests.
493///
494/// This mirrors [`ManifestStore`] for remote, browser, object-store, and
495/// cloud-database backends. The mutable root layer stays separate from
496/// immutable node storage so remote implementations can use their native
497/// compare-and-swap or transaction mechanism for branch/head updates.
498#[cfg(feature = "async-store")]
499#[allow(async_fn_in_trait)]
500pub trait AsyncManifestStore {
501    /// Error type for manifest operations.
502    type Error: std::error::Error + 'static;
503
504    /// Load a named root manifest.
505    async fn get_root(&self, name: &[u8]) -> Result<Option<RootManifest>, Self::Error>;
506
507    /// Unconditionally insert or replace a named root manifest.
508    async fn put_root(&self, name: &[u8], manifest: &RootManifest) -> Result<(), Self::Error>;
509
510    /// Delete a named root manifest. Deleting a missing name is not an error.
511    async fn delete_root(&self, name: &[u8]) -> Result<(), Self::Error>;
512
513    /// Atomically update a named root if the current manifest matches
514    /// `expected`.
515    ///
516    /// `expected == None` means the name must be absent. `new == None` deletes
517    /// the name when the compare succeeds.
518    async fn compare_and_swap_root(
519        &self,
520        name: &[u8],
521        expected: Option<&RootManifest>,
522        new: Option<&RootManifest>,
523    ) -> Result<ManifestUpdate, Self::Error>;
524}
525
526/// Async manifest stores that can enumerate durable named roots.
527///
528/// Scanning remains optional because point lookups and compare-and-swap are
529/// enough for normal publish/fetch paths, while retention and GC need listing.
530#[cfg(feature = "async-store")]
531#[allow(async_fn_in_trait)]
532pub trait AsyncManifestStoreScan: AsyncManifestStore {
533    /// List all durable named root manifests.
534    async fn list_roots(&self) -> Result<Vec<NamedRootManifest>, Self::Error>;
535}
536
537impl<T: ManifestStore> ManifestStore for Arc<T> {
538    type Error = T::Error;
539
540    fn get_root(&self, name: &[u8]) -> Result<Option<RootManifest>, Self::Error> {
541        (**self).get_root(name)
542    }
543
544    fn put_root(&self, name: &[u8], manifest: &RootManifest) -> Result<(), Self::Error> {
545        (**self).put_root(name, manifest)
546    }
547
548    fn delete_root(&self, name: &[u8]) -> Result<(), Self::Error> {
549        (**self).delete_root(name)
550    }
551
552    fn compare_and_swap_root(
553        &self,
554        name: &[u8],
555        expected: Option<&RootManifest>,
556        new: Option<&RootManifest>,
557    ) -> Result<ManifestUpdate, Self::Error> {
558        (**self).compare_and_swap_root(name, expected, new)
559    }
560}
561
562impl<T: ManifestStoreScan> ManifestStoreScan for Arc<T> {
563    fn list_roots(&self) -> Result<Vec<NamedRootManifest>, Self::Error> {
564        (**self).list_roots()
565    }
566}
567
568#[cfg(feature = "async-store")]
569impl<T: AsyncManifestStore> AsyncManifestStore for Arc<T> {
570    type Error = T::Error;
571
572    async fn get_root(&self, name: &[u8]) -> Result<Option<RootManifest>, Self::Error> {
573        (**self).get_root(name).await
574    }
575
576    async fn put_root(&self, name: &[u8], manifest: &RootManifest) -> Result<(), Self::Error> {
577        (**self).put_root(name, manifest).await
578    }
579
580    async fn delete_root(&self, name: &[u8]) -> Result<(), Self::Error> {
581        (**self).delete_root(name).await
582    }
583
584    async fn compare_and_swap_root(
585        &self,
586        name: &[u8],
587        expected: Option<&RootManifest>,
588        new: Option<&RootManifest>,
589    ) -> Result<ManifestUpdate, Self::Error> {
590        (**self).compare_and_swap_root(name, expected, new).await
591    }
592}
593
594#[cfg(feature = "async-store")]
595impl<T: AsyncManifestStoreScan> AsyncManifestStoreScan for Arc<T> {
596    async fn list_roots(&self) -> Result<Vec<NamedRootManifest>, Self::Error> {
597        (**self).list_roots().await
598    }
599}
600
601#[cfg(test)]
602mod tests {
603    use super::*;
604    use crate::{Config, Encoding};
605    use std::time::Duration;
606
607    #[derive(Serialize)]
608    struct LegacyRootManifestWire {
609        version: u64,
610        root: Option<Cid>,
611        config: Config,
612    }
613
614    #[test]
615    fn manifest_round_trips_through_bytes() {
616        let cid = Cid::from_bytes(b"root");
617        let manifest = RootManifest::new(
618            Some(cid),
619            Config::builder()
620                .min_chunk_size(2)
621                .max_chunk_size(8)
622                .chunking_factor(4)
623                .hash_seed(99)
624                .encoding(Encoding::Json)
625                .node_cache_max_nodes(128)
626                .build(),
627        )
628        .with_timestamps_millis(Some(100), Some(200));
629
630        let bytes = manifest.to_bytes().unwrap();
631        assert_eq!(RootManifest::from_bytes(&bytes).unwrap(), manifest);
632    }
633
634    #[test]
635    fn manifest_reads_legacy_bytes_without_timestamps() {
636        let legacy = LegacyRootManifestWire {
637            version: ROOT_MANIFEST_VERSION,
638            root: Some(Cid::from_bytes(b"legacy-root")),
639            config: Config::default(),
640        };
641        let bytes = serde_cbor::ser::to_vec_packed(&legacy).unwrap();
642        let manifest = RootManifest::from_bytes(&bytes).unwrap();
643
644        assert_eq!(manifest.root, legacy.root);
645        assert_eq!(manifest.config, legacy.config);
646        assert_eq!(manifest.created_at_millis, None);
647        assert_eq!(manifest.updated_at_millis, None);
648    }
649
650    #[test]
651    fn manifest_converts_to_and_from_tree() {
652        let tree = Tree {
653            root: Some(Cid::from_bytes(b"root")),
654            config: Config::default(),
655        };
656
657        let manifest = RootManifest::from_tree(&tree);
658        assert_eq!(manifest.to_tree(), tree);
659        assert_eq!(Tree::from(manifest), tree);
660    }
661
662    #[test]
663    fn named_root_update_converts_conflict_to_tree() {
664        let tree = Tree {
665            root: Some(Cid::from_bytes(b"root")),
666            config: Config::default(),
667        };
668        let update = ManifestUpdate::Conflict {
669            current: Some(RootManifest::from_tree(&tree)),
670        };
671
672        assert_eq!(
673            NamedRootUpdate::from(update),
674            NamedRootUpdate::Conflict {
675                current: Some(tree)
676            }
677        );
678    }
679
680    #[test]
681    fn retention_duration_helpers_build_cutoffs() {
682        assert_eq!(
683            NamedRootRetention::updated_within(b"checkpoint/", 1_000, Duration::from_millis(250)),
684            NamedRootRetention::updated_since(b"checkpoint/", 750)
685        );
686        assert_eq!(
687            NamedRootRetention::updated_within_millis(b"checkpoint/", 100, 250),
688            NamedRootRetention::updated_since(b"checkpoint/", 0)
689        );
690        assert_eq!(
691            NamedRootRetention::updated_within_days(b"checkpoint/", 172_800_050, 1),
692            NamedRootRetention::updated_since(b"checkpoint/", 86_400_050)
693        );
694        assert_eq!(
695            NamedRootRetention::updated_within_days(b"checkpoint/", 42, u64::MAX),
696            NamedRootRetention::updated_since(b"checkpoint/", 0)
697        );
698    }
699}
prolly/prolly/manifest.rs

prolly/prolly/
manifest.rs
