alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::HashSet;
2use std::time::{Duration, SystemTime};
3
4use knot_git::{GitError, Haves, Repo, Wants};
5use knot_types::{Oid, RepoDid, UnixSeconds};
6
7use crate::store::{DiskStore, Reclaimed, expired};
8use crate::{LfsError, LfsOid, LfsSize, scan_pointers};
9
10#[derive(Debug, thiserror::Error)]
11pub enum GcError {
12 #[error("git: {0}")]
13 Git(#[from] GitError),
14 #[error("store: {0}")]
15 Store(#[from] LfsError),
16}
17
18#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
19pub struct GcReport {
20 pub scanned: usize,
21 pub marked: usize,
22 pub swept: usize,
23 pub bytes: LfsSize,
24}
25
26fn unix_seconds(now: SystemTime) -> UnixSeconds {
27 let secs = now
28 .duration_since(SystemTime::UNIX_EPOCH)
29 .map(|delta| delta.as_secs() as i64)
30 .unwrap_or(0);
31 UnixSeconds::new(secs)
32}
33
34fn reachable_roots(repo: &Repo, floor: UnixSeconds) -> Result<Vec<Oid>, GitError> {
35 let mut roots: HashSet<Oid> = repo
36 .references()?
37 .into_iter()
38 .filter(|record| !knot_git::is_reserved(&record.name))
39 .map(|record| record.target)
40 .collect();
41 repo.reflog_updates_since(floor)
42 .into_iter()
43 .for_each(|update| {
44 roots.insert(update.new);
45 if let Some(old) = update.old {
46 roots.insert(old);
47 }
48 });
49 Ok(roots
50 .into_iter()
51 .filter(|oid| repo.contains(*oid))
52 .collect())
53}
54
55fn reachable_pointers(repo: &Repo, floor: UnixSeconds) -> Result<HashSet<LfsOid>, GitError> {
56 let roots = reachable_roots(repo, floor)?;
57 if roots.is_empty() {
58 return Ok(HashSet::new());
59 }
60 Ok(scan_pointers(repo, Wants::new(&roots), Haves::new(&[]))?
61 .into_keys()
62 .collect())
63}
64
65pub fn collect_repo(
66 store: &DiskStore,
67 repo: &Repo,
68 did: &RepoDid,
69 grace: Duration,
70 now: SystemTime,
71) -> Result<GcReport, GcError> {
72 let stored = store.enumerate(did)?;
73 if stored.is_empty() {
74 return Ok(GcReport::default());
75 }
76 let floor = unix_seconds(now).saturating_sub_secs(grace.as_secs().min(i64::MAX as u64) as i64);
77 let reachable = reachable_pointers(repo, floor)?;
78 stored
79 .iter()
80 .filter(|object| !reachable.contains(&object.oid))
81 .filter(|object| expired(now, object.mtime, grace))
82 .map(|object| store.collect_expired(did, &object.oid, grace, now))
83 .try_fold(
84 GcReport {
85 scanned: stored.len(),
86 ..GcReport::default()
87 },
88 |mut report, outcome| {
89 report.marked += 1;
90 if let Reclaimed::Swept(size) = outcome? {
91 report.swept += 1;
92 report.bytes = report.bytes.saturating_add(size);
93 }
94 Ok::<_, GcError>(report)
95 },
96 )
97}
98
99#[cfg(test)]
100mod tests {
101 use std::time::Duration;
102
103 use knot_git::{
104 EntryKind, Identity, Layout, NewCommit, RefUpdate, Repo, StagedAction, StagedChange,
105 };
106 use knot_types::{AuthorName, BranchName, Email, Oid, RefName, RepoDid};
107 use sha2::{Digest, Sha256};
108
109 use super::*;
110 use crate::store::DiskStore;
111 use crate::{ClaimedSize, LfsOid, LfsSize, LfsStore, LfsStorePath};
112
113 const EMPTY_TREE: &str = "4b825dc642cb6eb9a060e54bf8d69288fbee4904";
114 const DAY: Duration = Duration::from_secs(86_400);
115 const MONTHS: Duration = Duration::from_secs(60 * 86_400);
116
117 struct Fixture {
118 _scan: tempfile::TempDir,
119 _lfs: tempfile::TempDir,
120 store: DiskStore,
121 did: RepoDid,
122 repo: Repo,
123 }
124
125 fn fixture() -> Fixture {
126 let scan = tempfile::tempdir().unwrap();
127 let lfs = tempfile::tempdir().unwrap();
128 let layout = Layout::new(scan.path()).with_default_branch(BranchName::new("main").unwrap());
129 let did = RepoDid::new("did:plc:squid").unwrap();
130 let repo = layout.create(&did).unwrap();
131 let store = DiskStore::open(LfsStorePath::new(lfs.path())).unwrap();
132 Fixture {
133 _scan: scan,
134 _lfs: lfs,
135 store,
136 did,
137 repo,
138 }
139 }
140
141 fn who(secs: i64) -> Identity {
142 Identity {
143 name: AuthorName::new("nel"),
144 email: Email::new("nel@oyster.cafe"),
145 time: knot_types::UnixSeconds::new(secs),
146 offset_seconds: 0,
147 }
148 }
149
150 fn put_media(f: &Fixture, bytes: &[u8]) -> (LfsOid, LfsSize) {
151 let oid = LfsOid::from_digest(Sha256::digest(bytes).into());
152 let bytes_len = bytes.len() as u64;
153 f.store
154 .put(&f.did, &oid, ClaimedSize::new(bytes_len), &mut &bytes[..])
155 .unwrap();
156 (oid, LfsSize::new(bytes_len))
157 }
158
159 fn empty_commit(f: &Fixture, message: &str, secs: i64) -> Oid {
160 f.repo
161 .write_commit(&NewCommit {
162 tree: Oid::from_hex(EMPTY_TREE).unwrap(),
163 parents: Vec::new(),
164 author: who(secs),
165 committer: who(secs),
166 message: message.to_string(),
167 extra_headers: Vec::new(),
168 })
169 .unwrap()
170 }
171
172 fn commit_pointer(f: &Fixture, name: &str, oid: &LfsOid, size: LfsSize) -> Oid {
173 let pointer =
174 format!("version https://git-lfs.github.com/spec/v1\noid sha256:{oid}\nsize {size}\n")
175 .into_bytes();
176 let tree = f
177 .repo
178 .write_staged_tree(
179 Oid::from_hex(EMPTY_TREE).unwrap(),
180 &[StagedChange {
181 path: knot_types::RepoPath::new("clip.bin").unwrap(),
182 action: StagedAction::Put {
183 content: pointer,
184 kind: EntryKind::Blob,
185 },
186 }],
187 )
188 .unwrap();
189 let tip = f
190 .repo
191 .write_commit(&NewCommit {
192 tree,
193 parents: Vec::new(),
194 author: who(1_700_000_000),
195 committer: who(1_700_000_000),
196 message: "add media".to_string(),
197 extra_headers: Vec::new(),
198 })
199 .unwrap();
200 f.repo
201 .update_ref(&RefUpdate::Create {
202 name: RefName::new(name).unwrap(),
203 new: tip,
204 })
205 .unwrap();
206 tip
207 }
208
209 fn age(f: &Fixture, oid: &LfsOid, past: Duration) {
210 let path = f.store.object_file(&f.did, oid).unwrap().unwrap().1;
211 std::fs::OpenOptions::new()
212 .write(true)
213 .open(&path)
214 .unwrap()
215 .set_modified(SystemTime::now() - past)
216 .unwrap();
217 }
218
219 fn collect(f: &Fixture, grace: Duration) -> GcReport {
220 collect_repo(&f.store, &f.repo, &f.did, grace, SystemTime::now()).unwrap()
221 }
222
223 #[test]
224 fn a_reachable_object_is_never_swept_regardless_of_retention_source() {
225 let f = fixture();
226
227 let (live, live_size) = put_media(&f, b"referenced media");
228 commit_pointer(&f, "refs/heads/keep", &live, live_size);
229 age(&f, &live, MONTHS);
230
231 let (fresh, _) = put_media(&f, b"just uploaded, pointer still in flight");
232
233 let (forced, forced_size) = put_media(&f, b"orphaned by a force push");
234 let old = commit_pointer(&f, "refs/heads/main", &forced, forced_size);
235 let replacement = empty_commit(&f, "drop media", 1_700_000_100);
236 f.repo
237 .update_ref(&RefUpdate::Update {
238 name: RefName::new("refs/heads/main").unwrap(),
239 old,
240 new: replacement,
241 })
242 .unwrap();
243 age(&f, &forced, MONTHS);
244
245 let report = collect(&f, Duration::from_secs(14 * 86_400));
246 assert_eq!(report.marked, 0, "no reachable object is ever marked");
247 assert_eq!(report.swept, 0);
248 [&live, &fresh, &forced].iter().for_each(|oid| {
249 assert!(
250 f.store.probe(&f.did, oid).unwrap().is_some(),
251 "a live ref, the grace window, and the reflog old tip each keep their object"
252 );
253 });
254 }
255
256 #[test]
257 fn an_unreferenced_expired_object_is_swept_even_after_its_branch_is_gone() {
258 let f = fixture();
259
260 let (orphan, orphan_size) = put_media(&f, b"orphaned media");
261 age(&f, &orphan, MONTHS);
262
263 let (dropped, dropped_size) = put_media(&f, b"lived on a branch that was deleted");
264 let tip = commit_pointer(&f, "refs/heads/topic", &dropped, dropped_size);
265 f.repo
266 .update_ref(&RefUpdate::Delete {
267 name: RefName::new("refs/heads/topic").unwrap(),
268 old: tip,
269 })
270 .unwrap();
271 age(&f, &dropped, MONTHS);
272
273 let report = collect(&f, DAY);
274 assert_eq!(report.marked, 2);
275 assert_eq!(
276 report.swept, 2,
277 "a plain orphan and a deleted-branch orphan are both swept once the mtime grace expires"
278 );
279 assert_eq!(report.bytes, orphan_size.saturating_add(dropped_size));
280 assert_eq!(f.store.probe(&f.did, &orphan).unwrap(), None);
281 assert_eq!(f.store.probe(&f.did, &dropped).unwrap(), None);
282 }
283
284 #[test]
285 fn a_repo_with_no_stored_objects_never_walks_git() {
286 let f = fixture();
287 let orphan_pointer = LfsOid::from_digest(Sha256::digest(b"pointer without bytes").into());
288 commit_pointer(&f, "refs/heads/main", &orphan_pointer, LfsSize::new(21));
289 assert_eq!(collect(&f, DAY), GcReport::default());
290 }
291}