gc.rs 92.6%

150 of 162 lines run, 7 of 14 functions · run #383 on main · run missed

//! Garbage collection (spec §21): delete objects
//! nothing reachable refers to.
//!
//! Roots: the current view, the views of every operation newer than
//! `core.op-retention` (their working copies only within the shorter
//! `core.snapshot-retention`), reference attestations, and the policy; receipts
//! live as long as the changes of their state. Operation
//! objects are always kept so the log stays walkable, but an operation older
//! than the retention window no longer keeps its states alive, so undoing
//! that far back may fail once its objects are gone.

use super::*;
use crate::store::{PackReport, PruneReport, RetiredReport};
use std::collections::HashSet;

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct GcReport {
    pub packed: PackReport,
    pub pruned: PruneReport,
    pub reachable: usize,
    /// Packs folded into the one a full repack writes.
    pub merged_packs: usize,
    /// Packs and loose objects earlier repacks retired, deleted now that
    /// [`crate::store::RETIRED_FOR`] has passed (oxygenhub#249).
    pub retired: RetiredReport,
}

impl Repo {
    /// Delete what earlier repacks retired more than
    /// [`crate::store::RETIRED_FOR`] ago, prune unreachable objects older
    /// than the grace period (`now` ignores it, but not the retired
    /// files' hour), then repack everything into one pack (spec §13.4).
    /// The packs and loose objects this repack replaces are retired in
    /// turn, for the next collection. A store that is not this disk's
    /// directory then folds its signatures and index entries (§13.7).
    pub fn gc(&self, now: bool) -> Result<GcReport> {
        self.gc_with(now, &mut |_| {})
    }

    /// [`Repo::gc`], reporting each phase and the packing's progress
    /// (oxygen#66), for a spinner or a log.
    pub fn gc_with(&self, now: bool, progress: crate::store::Progress) -> Result<GcReport> {
        self.gc_keeping(now, &[], progress)
    }

    /// [`Repo::gc_with`], keeping `extra` blobs and their chunks as well:
    /// objects a host stored for itself and names from outside the
    /// repository's history, such as a forge's job artifacts
    /// (oxygenhub#126).
    pub fn gc_keeping(
        &self,
        now: bool,
        extra: &[ObjectRef],
        progress: crate::store::Progress,
    ) -> Result<GcReport> {
        // Retired files first: the readers that used them have had their
        // hour, and what replaced them is in the packs read from here on.
        let retired = self
            .store
            .expire_retired(crate::store::RETIRED_FOR, progress)?;
        progress("finding what is reachable");
        let mut keep = self.reachable()?;
        for r in extra {
            if !self.store.has(r) || !keep.insert(*r) {
                continue;
            }
            if let Ok(body) = self.store.get_typed(r, ObjectType::Blob) {
                if let Ok(blob) = Blob::decode(&body) {
                    keep.extend(blob.chunks);
                }
            }
        }
        let grace = if now {
            std::time::Duration::ZERO
        } else {
            std::time::Duration::from_secs(self.config.gc_grace_days * 86_400)
        };
        let pruned = self.store.prune_with(&keep, grace, progress)?;
        // Then everything left in one pack, compressed harder than a
        // commit's (spec §13.4), as `git gc` repacks.
        let (packed, merged_packs) = self
            .store
            .repack_all_with(crate::store::REPACK_ZSTD_LEVEL, progress)?;
        // Last, on a store that is not this disk's directory, every
        // signature and index entry into one list (spec §13.7, #116).
        self.store.fold_entries(progress)?;
        Ok(GcReport {
            packed,
            pruned,
            reachable: keep.len(),
            merged_packs,
            retired,
        })
    }

    /// Every object that must survive garbage collection.
    pub fn reachable(&self) -> Result<HashSet<ObjectRef>> {
        let mut keep: HashSet<ObjectRef> = HashSet::new();
        let mut states: Vec<ObjectRef> = Vec::new();
        let mut changes: Vec<ChangeHash> = Vec::new();
        let mut attests: Vec<ObjectRef> = Vec::new();
        let mut policies: Vec<ObjectRef> = Vec::new();

        // Operations: always kept; views within retention are roots. Past
        // working copies only within the shorter `core.snapshot-retention`
        // (oxygen#98): a file only ever snapshotted, such as a build output
        // a status took in, goes once they expire.
        let now_ms = Hlc::now().wall_ms;
        let cutoff = now_ms.saturating_sub(self.config.op_retention_days * 86_400_000);
        let wc_cutoff = now_ms
            .saturating_sub(self.config.snapshot_retention_days * 86_400_000)
            .max(cutoff);
        for (i, (r, op)) in self.op_log()?.into_iter().enumerate() {
            keep.insert(r);
            if i == 0 || op.hlc.wall_ms >= cutoff {
                states.extend(op.view.bookmarks.values().copied());
                states.extend(op.view.remotes.values().flat_map(|b| b.values().copied()));
                policies.extend(op.view.policy);
            }
            if i == 0 || (self.config.snapshot_retention_days > 0 && op.hlc.wall_ms >= wc_cutoff) {
                changes.extend(op.view.workspaces.values().copied());
            }
        }
        // Checkpoints: the object, its payload blob, and the blob's chunks.
        for r in self.checkpoint_refs()? {
            if let Ok(bytes) = self.store.get_typed(&r, ObjectType::Checkpoint) {
                keep.insert(r);
                let cp = crate::objects::Checkpoint::decode(&bytes)?;
                keep.insert(cp.payload);
                if let Ok(b) = self.store.get_typed(&cp.payload, ObjectType::Blob) {
                    keep.extend(Blob::decode(&b)?.chunks);
                }
                // A version 2 payload's graph blobs and their chunks.
                for g in self.checkpoint_graph_blobs(&cp).unwrap_or_default() {
                    keep.insert(g);
                    if let Ok(b) = self.store.get_typed(&g, ObjectType::Blob) {
                        keep.extend(Blob::decode(&b)?.chunks);
                    }
                }
            }
        }
        // Tags: their states and attestations.
        let view_now = self.view()?;
        states.extend(view_now.tags.values().copied());
        for name in view_now.tags.keys() {
            if let Some((r, _)) = self.tag_attest(name)? {
                // With its issuer's acceptance (spec §14.7).
                if let Some(acc) = self.acceptance_of(&r)? {
                    attests.push(acc.attest.hash());
                }
                attests.push(r);
            }
        }
        // Reference attestations.
        for name in self.view()?.bookmarks.keys() {
            if let Some((r, _)) = self.latest_ref_attest(name)? {
                // With a host's acceptance of it (spec §14.7).
                if let Some(acc) = self.acceptance_of(&r)? {
                    attests.push(acc.attest.hash());
                }
                attests.push(r);
            }
        }
        // A historical view may name objects pruned under an earlier, shorter
        // retention; there is nothing to keep for those.
        for st in states {
            if keep.insert(st) {
                match self.store.get_state(&st) {
                    Ok(s) => changes.extend(s.heads),
                    Err(StoreError::NotFound(_)) => continue,
                    Err(e) => return Err(e.into()),
                }
            }
        }
        for p in policies {
            if keep.insert(p) {
                let bytes = match self.store.get_typed(&p, ObjectType::Policy) {
                    Ok(b) => b,
                    Err(StoreError::NotFound(_)) => continue,
                    Err(e) => return Err(e.into()),
                };
                let policy = crate::policy::Policy::decode(&bytes)?;
                for id in &policy.identities {
                    keep.insert(id.hash());
                }
                for sig in self.store.signatures(&p)? {
                    keep.insert(sig.key);
                }
            }
        }
        // Changes: their closure, identities, blobs, chunks, seals.
        let mut stack = changes;
        while let Some(h) = stack.pop() {
            if !keep.insert(h) {
                continue;
            }
            // Beyond the shallow boundary, or pruned under an earlier
            // retention, there is nothing to keep.
            let c = match self.store.get_change_parts(&h) {
                Ok((c, texts)) => {
                    keep.extend(texts);
                    c
                }
                Err(StoreError::NotFound(_)) => continue,
                Err(e) => return Err(e.into()),
            };
            stack.extend(c.deps.iter().copied());
            if !c.author.is_zero() {
                keep.insert(c.author);
            }
            for sig in self.store.signatures(&h)? {
                self.keep_signer(&mut keep, sig.key)?;
            }
            for e in &c.edits {
                if let Edit::Replace { old, new, .. } = e {
                    for b in [old, new].into_iter().flatten() {
                        if keep.insert(*b) {
                            if let Ok(bytes) = self.store.get_typed(b, ObjectType::Blob) {
                                keep.extend(Blob::decode(&bytes)?.chunks);
                            }
                        }
                    }
                }
            }
            attests.extend(self.store.seals_for(&h)?);
        }
        self.reachable_receipts(&mut keep)?;
        self.reachable_approvals(&mut keep)?;
        for a in attests {
            if keep.insert(a) {
                for sig in self.store.signatures(&a)? {
                    self.keep_signer(&mut keep, sig.key)?;
                }
            }
        }
        Ok(keep)
    }

    /// Keep a signer's identity and, for a short-lived key, its
    /// certificate and the certificate's issuer (spec §14.7).
    pub(crate) fn keep_signer(&self, keep: &mut HashSet<ObjectRef>, key: ObjectRef) -> Result<()> {
        keep.insert(key);
        if let Some(c) = self.store.certificate_of(&key)? {
            keep.insert(c);
            if let Ok(cert) = self.store.get_certificate(&c) {
                keep.insert(cert.cert.issuer);
            }
        }
        Ok(())
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

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	22×	pub fn gc(&self, now: bool) -> Result<GcReport> {
37	121×	self.gc_with(now, &mut \|_\| {})
38	22×	}
39
40		/// [`Repo::gc`], reporting each phase and the packing's progress
41		/// (oxygen#66), for a spinner or a log.
42	23×	pub fn gc_with(&self, now: bool, progress: crate::store::Progress) -> Result<GcReport> {
43	23×	self.gc_keeping(now, &[], progress)
44	23×	}
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	24×	pub fn gc_keeping(
51	24×	&self,
52	24×	now: bool,
53	24×	extra: &[ObjectRef],
54	24×	progress: crate::store::Progress,
55	24×	) -> 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	24×	let retired = self
59	24×	.store
60	24×	.expire_retired(crate::store::RETIRED_FOR, progress)?;
61	24×	progress("finding what is reachable");
62	24×	let mut keep = self.reachable()?;
63	24×	for r in extra {
64	1×	if !self.store.has(r) \|\| !keep.insert(*r) {
65	0×	continue;
66	1×	}
67	1×	if let Ok(body) = self.store.get_typed(r, ObjectType::Blob) {
68	1×	if let Ok(blob) = Blob::decode(&body) {
69	1×	keep.extend(blob.chunks);
70	1×	}
71	0×	}
72		}
73	24×	let grace = if now {
74	22×	std::time::Duration::ZERO
75		} else {
76	2×	std::time::Duration::from_secs(self.config.gc_grace_days * 86_400)
77		};
78	24×	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	24×	let (packed, merged_packs) = self
82	24×	.store
83	24×	.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	24×	self.store.fold_entries(progress)?;
87	24×	Ok(GcReport {
88	24×	packed,
89	24×	pruned,
90	24×	reachable: keep.len(),
91	24×	merged_packs,
92	24×	retired,
93	24×	})
94	24×	}
95
96		/// Every object that must survive garbage collection.
97	24×	pub fn reachable(&self) -> Result<HashSet<ObjectRef>> {
98	24×	let mut keep: HashSet<ObjectRef> = HashSet::new();
99	24×	let mut states: Vec<ObjectRef> = Vec::new();
100	24×	let mut changes: Vec<ChangeHash> = Vec::new();
101	24×	let mut attests: Vec<ObjectRef> = Vec::new();
102	24×	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	24×	let now_ms = Hlc::now().wall_ms;
109	24×	let cutoff = now_ms.saturating_sub(self.config.op_retention_days * 86_400_000);
110	24×	let wc_cutoff = now_ms
111	24×	.saturating_sub(self.config.snapshot_retention_days * 86_400_000)
112	24×	.max(cutoff);
113	154×	for (i, (r, op)) in self.op_log()?.into_iter().enumerate() {
114	154×	keep.insert(r);
115	154×	if i == 0 \|\| op.hlc.wall_ms >= cutoff {
116	116×	states.extend(op.view.bookmarks.values().copied());
117	116×	states.extend(op.view.remotes.values().flat_map(\|b\| b.values().copied()));
118	116×	policies.extend(op.view.policy);
119	38×	}
120	154×	if i == 0 \|\| (self.config.snapshot_retention_days > 0 && op.hlc.wall_ms >= wc_cutoff) {
121	113×	changes.extend(op.view.workspaces.values().copied());
122	113×	}
123		}
124		// Checkpoints: the object, its payload blob, and the blob's chunks.
125	24×	for r in self.checkpoint_refs()? {
126	2×	if let Ok(bytes) = self.store.get_typed(&r, ObjectType::Checkpoint) {
127	2×	keep.insert(r);
128	2×	let cp = crate::objects::Checkpoint::decode(&bytes)?;
129	2×	keep.insert(cp.payload);
130	2×	if let Ok(b) = self.store.get_typed(&cp.payload, ObjectType::Blob) {
131	2×	keep.extend(Blob::decode(&b)?.chunks);
132	0×	}
133		// A version 2 payload's graph blobs and their chunks.
134	2×	for g in self.checkpoint_graph_blobs(&cp).unwrap_or_default() {
135	1×	keep.insert(g);
136	1×	if let Ok(b) = self.store.get_typed(&g, ObjectType::Blob) {
137	1×	keep.extend(Blob::decode(&b)?.chunks);
138	0×	}
139		}
140	0×	}
141		}
142		// Tags: their states and attestations.
143	24×	let view_now = self.view()?;
144	24×	states.extend(view_now.tags.values().copied());
145	24×	for name in view_now.tags.keys() {
146	4×	if let Some((r, _)) = self.tag_attest(name)? {
147		// With its issuer's acceptance (spec §14.7).
148	4×	if let Some(acc) = self.acceptance_of(&r)? {
149	1×	attests.push(acc.attest.hash());
150	3×	}
151	4×	attests.push(r);
152	0×	}
153		}
154		// Reference attestations.
155	24×	for name in self.view()?.bookmarks.keys() {
156	24×	if let Some((r, _)) = self.latest_ref_attest(name)? {
157		// With a host's acceptance of it (spec §14.7).
158	2×	if let Some(acc) = self.acceptance_of(&r)? {
159	1×	attests.push(acc.attest.hash());
160	1×	}
161	2×	attests.push(r);
162	22×	}
163		}
164		// A historical view may name objects pruned under an earlier, shorter
165		// retention; there is nothing to keep for those.
166	136×	for st in states {
167	136×	if keep.insert(st) {
168	78×	match self.store.get_state(&st) {
169	75×	Ok(s) => changes.extend(s.heads),
170	3×	Err(StoreError::NotFound(_)) => continue,
171	0×	Err(e) => return Err(e.into()),
172		}
173	58×	}
174		}
175	24×	for p in policies {
176	14×	if keep.insert(p) {
177	5×	let bytes = match self.store.get_typed(&p, ObjectType::Policy) {
178	5×	Ok(b) => b,
179	0×	Err(StoreError::NotFound(_)) => continue,
180	0×	Err(e) => return Err(e.into()),
181		};
182	5×	let policy = crate::policy::Policy::decode(&bytes)?;
183	7×	for id in &policy.identities {
184	7×	keep.insert(id.hash());
185	7×	}
186	5×	for sig in self.store.signatures(&p)? {
187	5×	keep.insert(sig.key);
188	5×	}
189	9×	}
190		}
191		// Changes: their closure, identities, blobs, chunks, seals.
192	24×	let mut stack = changes;
193	394×	while let Some(h) = stack.pop() {
194	370×	if !keep.insert(h) {
195	208×	continue;
196	162×	}
197		// Beyond the shallow boundary, or pruned under an earlier
198		// retention, there is nothing to keep.
199	162×	let c = match self.store.get_change_parts(&h) {
200	152×	Ok((c, texts)) => {
201	152×	keep.extend(texts);
202	152×	c
203		}
204	10×	Err(StoreError::NotFound(_)) => continue,
205	0×	Err(e) => return Err(e.into()),
206		};
207	152×	stack.extend(c.deps.iter().copied());
208	152×	if !c.author.is_zero() {
209	33×	keep.insert(c.author);
210	119×	}
211	152×	for sig in self.store.signatures(&h)? {
212	13×	self.keep_signer(&mut keep, sig.key)?;
213		}
214	152×	for e in &c.edits {
215	134×	if let Edit::Replace { old, new, .. } = e {
216	18×	for b in [old, new].into_iter().flatten() {
217	18×	if keep.insert(*b) {
218	18×	if let Ok(bytes) = self.store.get_typed(b, ObjectType::Blob) {
219	18×	keep.extend(Blob::decode(&bytes)?.chunks);
220	0×	}
221	0×	}
222		}
223	116×	}
224		}
225	152×	attests.extend(self.store.seals_for(&h)?);
226		}
227	24×	self.reachable_receipts(&mut keep)?;
228	24×	self.reachable_approvals(&mut keep)?;
229	24×	for a in attests {
230	13×	if keep.insert(a) {
231	12×	for sig in self.store.signatures(&a)? {
232	12×	self.keep_signer(&mut keep, sig.key)?;
233		}
234	1×	}
235		}
236	24×	Ok(keep)
237	24×	}
238
239		/// Keep a signer's identity and, for a short-lived key, its
240		/// certificate and the certificate's issuer (spec §14.7).
241	27×	pub(crate) fn keep_signer(&self, keep: &mut HashSet<ObjectRef>, key: ObjectRef) -> Result<()> {
242	27×	keep.insert(key);
243	27×	if let Some(c) = self.store.certificate_of(&key)? {
244	5×	keep.insert(c);
245	5×	if let Ok(cert) = self.store.get_certificate(&c) {
246	5×	keep.insert(cert.cert.issuer);
247	5×	}
248	22×	}
249	27×	Ok(())
250	27×	}
251		}

Coverage / crates/oxy-core/src/repo/gc.rs 92.6%

Keyboard