OXYGENHUB
# 771661aa50f4 oxygen / crates / oxy-core / src / repo / gc.rs
10.1 KiB · Rust · last changed Signatures and index entries from a journal and entry lists, not a LIST per lookup 17 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` (their working copies only within the shorter
6 
//! `core.snapshot-retention`), reference attestations, and the policy; receipts
7 
//! live as long as the changes of their state. Operation
8 
//! objects are always kept so the log stays walkable, but an operation older
9 
//! than the retention window no longer keeps its states alive, so undoing
10 
//! that far back may fail once its objects are gone.
11
12 
use super::*;
13 
use crate::store::{PackReport, PruneReport, RetiredReport};
14 
use std::collections::HashSet;
15
16 
#[derive(Debug, Clone, Default, PartialEq, Eq)]
17 
pub struct GcReport {
18 
    pub packed: PackReport,
19 
    pub pruned: PruneReport,
20 
    pub reachable: usize,
21 
    /// Packs folded into the one a full repack writes.
22 
    pub merged_packs: usize,
23 
    /// Packs and loose objects earlier repacks retired, deleted now that
24 
    /// [`crate::store::RETIRED_FOR`] has passed (oxygenhub#249).
25 
    pub retired: RetiredReport,
26 
}
27
28 
impl Repo {
29 
    /// Delete what earlier repacks retired more than
30 
    /// [`crate::store::RETIRED_FOR`] ago, prune unreachable objects older
31 
    /// than the grace period (`now` ignores it, but not the retired
32 
    /// files' hour), then repack everything into one pack (spec §13.4).
33 
    /// The packs and loose objects this repack replaces are retired in
34 
    /// turn, for the next collection. A store that is not this disk's
35 
    /// directory then folds its signatures and index entries (§13.7).
36 
    pub fn gc(&self, now: bool) -> Result<GcReport> {
37 
        self.gc_with(now, &mut |_| {})
38 
    }
39
40 
    /// [`Repo::gc`], reporting each phase and the packing's progress
41 
    /// (oxygen#66), for a spinner or a log.
42 
    pub fn gc_with(&self, now: bool, progress: crate::store::Progress) -> Result<GcReport> {
43 
        self.gc_keeping(now, &[], progress)
44 
    }
45
46 
    /// [`Repo::gc_with`], keeping `extra` blobs and their chunks as well:
47 
    /// objects a host stored for itself and names from outside the
48 
    /// repository's history, such as a forge's job artifacts
49 
    /// (oxygenhub#126).
50 
    pub fn gc_keeping(
51 
        &self,
52 
        now: bool,
53 
        extra: &[ObjectRef],
54 
        progress: crate::store::Progress,
55 
    ) -> Result<GcReport> {
56 
        // Retired files first: the readers that used them have had their
57 
        // hour, and what replaced them is in the packs read from here on.
58 
        let retired = self
59 
            .store
60 
            .expire_retired(crate::store::RETIRED_FOR, progress)?;
61 
        progress("finding what is reachable");
62 
        let mut keep = self.reachable()?;
63 
        for r in extra {
64 
            if !self.store.has(r) || !keep.insert(*r) {
65 
                continue;
66 
            }
67 
            if let Ok(body) = self.store.get_typed(r, ObjectType::Blob) {
68 
                if let Ok(blob) = Blob::decode(&body) {
69 
                    keep.extend(blob.chunks);
70 
                }
71 
            }
72 
        }
73 
        let grace = if now {
74 
            std::time::Duration::ZERO
75 
        } else {
76 
            std::time::Duration::from_secs(self.config.gc_grace_days * 86_400)
77 
        };
78 
        let pruned = self.store.prune_with(&keep, grace, progress)?;
79 
        // Then everything left in one pack, compressed harder than a
80 
        // commit's (spec §13.4), as `git gc` repacks.
81 
        let (packed, merged_packs) = self
82 
            .store
83 
            .repack_all_with(crate::store::REPACK_ZSTD_LEVEL, progress)?;
84 
        // Last, on a store that is not this disk's directory, every
85 
        // signature and index entry into one list (spec §13.7, #116).
86 
        self.store.fold_entries(progress)?;
87 
        Ok(GcReport {
88 
            packed,
89 
            pruned,
90 
            reachable: keep.len(),
91 
            merged_packs,
92 
            retired,
93 
        })
94 
    }
95
96 
    /// Every object that must survive garbage collection.
97 
    pub fn reachable(&self) -> Result<HashSet<ObjectRef>> {
98 
        let mut keep: HashSet<ObjectRef> = HashSet::new();
99 
        let mut states: Vec<ObjectRef> = Vec::new();
100 
        let mut changes: Vec<ChangeHash> = Vec::new();
101 
        let mut attests: Vec<ObjectRef> = Vec::new();
102 
        let mut policies: Vec<ObjectRef> = Vec::new();
103
104 
        // Operations: always kept; views within retention are roots. Past
105 
        // working copies only within the shorter `core.snapshot-retention`
106 
        // (oxygen#98): a file only ever snapshotted, such as a build output
107 
        // a status took in, goes once they expire.
108 
        let now_ms = Hlc::now().wall_ms;
109 
        let cutoff = now_ms.saturating_sub(self.config.op_retention_days * 86_400_000);
110 
        let wc_cutoff = now_ms
111 
            .saturating_sub(self.config.snapshot_retention_days * 86_400_000)
112 
            .max(cutoff);
113 
        for (i, (r, op)) in self.op_log()?.into_iter().enumerate() {
114 
            keep.insert(r);
115 
            if i == 0 || op.hlc.wall_ms >= cutoff {
116 
                states.extend(op.view.bookmarks.values().copied());
117 
                states.extend(op.view.remotes.values().flat_map(|b| b.values().copied()));
118 
                policies.extend(op.view.policy);
119 
            }
120 
            if i == 0 || (self.config.snapshot_retention_days > 0 && op.hlc.wall_ms >= wc_cutoff) {
121 
                changes.extend(op.view.workspaces.values().copied());
122 
            }
123 
        }
124 
        // Checkpoints: the object, its payload blob, and the blob's chunks.
125 
        for r in self.checkpoint_refs()? {
126 
            if let Ok(bytes) = self.store.get_typed(&r, ObjectType::Checkpoint) {
127 
                keep.insert(r);
128 
                let cp = crate::objects::Checkpoint::decode(&bytes)?;
129 
                keep.insert(cp.payload);
130 
                if let Ok(b) = self.store.get_typed(&cp.payload, ObjectType::Blob) {
131 
                    keep.extend(Blob::decode(&b)?.chunks);
132 
                }
133 
                // A version 2 payload's graph blobs and their chunks.
134 
                for g in self.checkpoint_graph_blobs(&cp).unwrap_or_default() {
135 
                    keep.insert(g);
136 
                    if let Ok(b) = self.store.get_typed(&g, ObjectType::Blob) {
137 
                        keep.extend(Blob::decode(&b)?.chunks);
138 
                    }
139 
                }
140 
            }
141 
        }
142 
        // Tags: their states and attestations.
143 
        let view_now = self.view()?;
144 
        states.extend(view_now.tags.values().copied());
145 
        for name in view_now.tags.keys() {
146 
            if let Some((r, _)) = self.tag_attest(name)? {
147 
                attests.push(r);
148 
            }
149 
        }
150 
        // Reference attestations.
151 
        for name in self.view()?.bookmarks.keys() {
152 
            if let Some((r, _)) = self.latest_ref_attest(name)? {
153 
                // With a host's acceptance of it (spec §14.7).
154 
                if let Some(acc) = self.acceptance_of(&r)? {
155 
                    attests.push(acc.attest.hash());
156 
                }
157 
                attests.push(r);
158 
            }
159 
        }
160 
        // A historical view may name objects pruned under an earlier, shorter
161 
        // retention; there is nothing to keep for those.
162 
        for st in states {
163 
            if keep.insert(st) {
164 
                match self.store.get_state(&st) {
165 
                    Ok(s) => changes.extend(s.heads),
166 
                    Err(StoreError::NotFound(_)) => continue,
167 
                    Err(e) => return Err(e.into()),
168 
                }
169 
            }
170 
        }
171 
        for p in policies {
172 
            if keep.insert(p) {
173 
                let bytes = match self.store.get_typed(&p, ObjectType::Policy) {
174 
                    Ok(b) => b,
175 
                    Err(StoreError::NotFound(_)) => continue,
176 
                    Err(e) => return Err(e.into()),
177 
                };
178 
                let policy = crate::policy::Policy::decode(&bytes)?;
179 
                for id in &policy.identities {
180 
                    keep.insert(id.hash());
181 
                }
182 
                for sig in self.store.signatures(&p)? {
183 
                    keep.insert(sig.key);
184 
                }
185 
            }
186 
        }
187 
        // Changes: their closure, identities, blobs, chunks, seals.
188 
        let mut stack = changes;
189 
        while let Some(h) = stack.pop() {
190 
            if !keep.insert(h) {
191 
                continue;
192 
            }
193 
            // Beyond the shallow boundary, or pruned under an earlier
194 
            // retention, there is nothing to keep.
195 
            let c = match self.store.get_change_parts(&h) {
196 
                Ok((c, texts)) => {
197 
                    keep.extend(texts);
198 
                    c
199 
                }
200 
                Err(StoreError::NotFound(_)) => continue,
201 
                Err(e) => return Err(e.into()),
202 
            };
203 
            stack.extend(c.deps.iter().copied());
204 
            if !c.author.is_zero() {
205 
                keep.insert(c.author);
206 
            }
207 
            for sig in self.store.signatures(&h)? {
208 
                self.keep_signer(&mut keep, sig.key)?;
209 
            }
210 
            for e in &c.edits {
211 
                if let Edit::Replace { old, new, .. } = e {
212 
                    for b in [old, new].into_iter().flatten() {
213 
                        if keep.insert(*b) {
214 
                            if let Ok(bytes) = self.store.get_typed(b, ObjectType::Blob) {
215 
                                keep.extend(Blob::decode(&bytes)?.chunks);
216 
                            }
217 
                        }
218 
                    }
219 
                }
220 
            }
221 
            attests.extend(self.store.seals_for(&h)?);
222 
        }
223 
        self.reachable_receipts(&mut keep)?;
224 
        self.reachable_approvals(&mut keep)?;
225 
        for a in attests {
226 
            if keep.insert(a) {
227 
                for sig in self.store.signatures(&a)? {
228 
                    self.keep_signer(&mut keep, sig.key)?;
229 
                }
230 
            }
231 
        }
232 
        Ok(keep)
233 
    }
234
235 
    /// Keep a signer's identity and, for a short-lived key, its
236 
    /// certificate and the certificate's issuer (spec §14.7).
237 
    pub(crate) fn keep_signer(&self, keep: &mut HashSet<ObjectRef>, key: ObjectRef) -> Result<()> {
238 
        keep.insert(key);
239 
        if let Some(c) = self.store.certificate_of(&key)? {
240 
            keep.insert(c);
241 
            if let Ok(cert) = self.store.get_certificate(&c) {
242 
                keep.insert(cert.cert.issuer);
243 
            }
244 
        }
245 
        Ok(())
246 
    }
247 
}