OXYGENHUB
# a2387d547373 oxygen / crates / oxy-core / src / repo / gc.rs
7.8 KiB · Rust · last changed The issuer countersigns when it accepted a certified update (keyless step 4c) 19 days ago source blame history raw
1 
//! Garbage collection (spec §21): delete objects
2 
//! nothing reachable refers to.
3 
//!
4 
//! Roots: the current view, the views of every operation newer than
5 
//! `core.op-retention`, reference attestations, and the policy; receipts
6 
//! live as long as the changes of their state. Operation
7 
//! objects are always kept so the log stays walkable, but an operation older
8 
//! than the retention window no longer keeps its states alive, so undoing
9 
//! that far back may fail once its objects are gone.
10
11 
use super::*;
12 
use crate::store::{PackReport, PruneReport};
13 
use std::collections::HashSet;
14
15 
#[derive(Debug, Clone, Default, PartialEq, Eq)]
16 
pub struct GcReport {
17 
    pub packed: PackReport,
18 
    pub pruned: PruneReport,
19 
    pub reachable: usize,
20 
    /// Packs folded into the one a full repack writes.
21 
    pub merged_packs: usize,
22 
}
23
24 
impl Repo {
25 
    /// Prune unreachable objects older than the grace period (`now`
26 
    /// ignores it), then repack everything into one pack (spec §13.4).
27 
    pub fn gc(&self, now: bool) -> Result<GcReport> {
28 
        self.gc_with(now, &mut |_| {})
29 
    }
30
31 
    /// [`Repo::gc`], reporting each phase and the packing's progress
32 
    /// (oxygen#66), for a spinner or a log.
33 
    pub fn gc_with(&self, now: bool, progress: crate::store::Progress) -> Result<GcReport> {
34 
        progress("finding what is reachable");
35 
        let keep = self.reachable()?;
36 
        let grace = if now {
37 
            std::time::Duration::ZERO
38 
        } else {
39 
            std::time::Duration::from_secs(self.config.gc_grace_days * 86_400)
40 
        };
41 
        let pruned = self.store.prune_with(&keep, grace, progress)?;
42 
        // Then everything left in one pack, compressed harder than a
43 
        // commit's (spec §13.4), as `git gc` repacks.
44 
        let (packed, merged_packs) = self
45 
            .store
46 
            .repack_all_with(crate::store::REPACK_ZSTD_LEVEL, progress)?;
47 
        Ok(GcReport {
48 
            packed,
49 
            pruned,
50 
            reachable: keep.len(),
51 
            merged_packs,
52 
        })
53 
    }
54
55 
    /// Every object that must survive garbage collection.
56 
    pub fn reachable(&self) -> Result<HashSet<ObjectRef>> {
57 
        let mut keep: HashSet<ObjectRef> = HashSet::new();
58 
        let mut states: Vec<ObjectRef> = Vec::new();
59 
        let mut changes: Vec<ChangeHash> = Vec::new();
60 
        let mut attests: Vec<ObjectRef> = Vec::new();
61 
        let mut policies: Vec<ObjectRef> = Vec::new();
62
63 
        // Operations: always kept; views within retention are roots.
64 
        let cutoff = Hlc::now()
65 
            .wall_ms
66 
            .saturating_sub(self.config.op_retention_days * 86_400_000);
67 
        for (i, (r, op)) in self.op_log()?.into_iter().enumerate() {
68 
            keep.insert(r);
69 
            if i == 0 || op.hlc.wall_ms >= cutoff {
70 
                states.extend(op.view.bookmarks.values().copied());
71 
                states.extend(op.view.remotes.values().flat_map(|b| b.values().copied()));
72 
                changes.extend(op.view.workspaces.values().copied());
73 
                policies.extend(op.view.policy);
74 
            }
75 
        }
76 
        // Checkpoints: the object, its payload blob, and the blob's chunks.
77 
        for r in self.checkpoint_refs()? {
78 
            if let Ok(bytes) = self.store.get_typed(&r, ObjectType::Checkpoint) {
79 
                keep.insert(r);
80 
                let cp = crate::objects::Checkpoint::decode(&bytes)?;
81 
                keep.insert(cp.payload);
82 
                if let Ok(b) = self.store.get_typed(&cp.payload, ObjectType::Blob) {
83 
                    keep.extend(Blob::decode(&b)?.chunks);
84 
                }
85 
                // A version 2 payload's graph blobs and their chunks.
86 
                for g in self.checkpoint_graph_blobs(&cp).unwrap_or_default() {
87 
                    keep.insert(g);
88 
                    if let Ok(b) = self.store.get_typed(&g, ObjectType::Blob) {
89 
                        keep.extend(Blob::decode(&b)?.chunks);
90 
                    }
91 
                }
92 
            }
93 
        }
94 
        // Tags: their states and attestations.
95 
        let view_now = self.view()?;
96 
        states.extend(view_now.tags.values().copied());
97 
        for name in view_now.tags.keys() {
98 
            if let Some((r, _)) = self.tag_attest(name)? {
99 
                attests.push(r);
100 
            }
101 
        }
102 
        // Reference attestations.
103 
        for name in self.view()?.bookmarks.keys() {
104 
            if let Some((r, _)) = self.latest_ref_attest(name)? {
105 
                // With a host's acceptance of it (spec §14.7).
106 
                if let Some(acc) = self.acceptance_of(&r)? {
107 
                    attests.push(acc.attest.hash());
108 
                }
109 
                attests.push(r);
110 
            }
111 
        }
112 
        // A historical view may name objects pruned under an earlier, shorter
113 
        // retention; there is nothing to keep for those.
114 
        for st in states {
115 
            if keep.insert(st) {
116 
                match self.store.get_state(&st) {
117 
                    Ok(s) => changes.extend(s.heads),
118 
                    Err(StoreError::NotFound(_)) => continue,
119 
                    Err(e) => return Err(e.into()),
120 
                }
121 
            }
122 
        }
123 
        for p in policies {
124 
            if keep.insert(p) {
125 
                let bytes = match self.store.get_typed(&p, ObjectType::Policy) {
126 
                    Ok(b) => b,
127 
                    Err(StoreError::NotFound(_)) => continue,
128 
                    Err(e) => return Err(e.into()),
129 
                };
130 
                let policy = crate::policy::Policy::decode(&bytes)?;
131 
                for id in &policy.identities {
132 
                    keep.insert(id.hash());
133 
                }
134 
                for sig in self.store.signatures(&p)? {
135 
                    keep.insert(sig.key);
136 
                }
137 
            }
138 
        }
139 
        // Changes: their closure, identities, blobs, chunks, seals.
140 
        let mut stack = changes;
141 
        while let Some(h) = stack.pop() {
142 
            if !keep.insert(h) {
143 
                continue;
144 
            }
145 
            // Beyond the shallow boundary, or pruned under an earlier
146 
            // retention, there is nothing to keep.
147 
            let c = match self.store.get_change_parts(&h) {
148 
                Ok((c, texts)) => {
149 
                    keep.extend(texts);
150 
                    c
151 
                }
152 
                Err(StoreError::NotFound(_)) => continue,
153 
                Err(e) => return Err(e.into()),
154 
            };
155 
            stack.extend(c.deps.iter().copied());
156 
            if !c.author.is_zero() {
157 
                keep.insert(c.author);
158 
            }
159 
            for sig in self.store.signatures(&h)? {
160 
                self.keep_signer(&mut keep, sig.key)?;
161 
            }
162 
            for e in &c.edits {
163 
                if let Edit::Replace { old, new, .. } = e {
164 
                    for b in [old, new].into_iter().flatten() {
165 
                        if keep.insert(*b) {
166 
                            if let Ok(bytes) = self.store.get_typed(b, ObjectType::Blob) {
167 
                                keep.extend(Blob::decode(&bytes)?.chunks);
168 
                            }
169 
                        }
170 
                    }
171 
                }
172 
            }
173 
            attests.extend(self.store.seals_for(&h)?);
174 
        }
175 
        self.reachable_receipts(&mut keep)?;
176 
        for a in attests {
177 
            if keep.insert(a) {
178 
                for sig in self.store.signatures(&a)? {
179 
                    self.keep_signer(&mut keep, sig.key)?;
180 
                }
181 
            }
182 
        }
183 
        Ok(keep)
184 
    }
185
186 
    /// Keep a signer's identity and, for a short-lived key, its
187 
    /// certificate and the certificate's issuer (spec §14.7).
188 
    pub(crate) fn keep_signer(&self, keep: &mut HashSet<ObjectRef>, key: ObjectRef) -> Result<()> {
189 
        keep.insert(key);
190 
        if let Some(c) = self.store.certificate_of(&key)? {
191 
            keep.insert(c);
192 
            if let Ok(cert) = self.store.get_certificate(&c) {
193 
                keep.insert(cert.cert.issuer);
194 
            }
195 
        }
196 
        Ok(())
197 
    }
198 
}