alice.pds.demo.boltless.dev
This repository has no description
18 kB
529 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::{HashMap, HashSet};
2use std::path::{Path, PathBuf};
3use std::sync::atomic::AtomicBool;
4use std::time::{Duration, SystemTime};
5
6use gix::progress::Discard;
7use knot_types::{Oid, UnixSeconds};
8
9use crate::fsio::{self, PackStem};
10use crate::{FileCount, MaintError, ObjectCount, PruneReport};
11
12const MTIMES_MAGIC: u32 = 0x4d54_4d45;
13const MTIMES_VERSION: u32 = 1;
14
15pub fn run(
16 objects_dir: &Path,
17 kind: gix::hash::Kind,
18 reachable: &HashSet<Oid>,
19 new_reachable_stem: Option<&PackStem>,
20 kept_large: &[PackStem],
21 loose: &[(Oid, PathBuf)],
22 grace: Duration,
23) -> Result<PruneReport, MaintError> {
24 let pack_dir = objects_dir.join("pack");
25 let idxs = fsio::pack_idx_paths(objects_dir);
26 let now = SystemTime::now();
27 let now_mtime = PackedMtime::from_system_time(now);
28
29 let kept_pack =
30 |stem: &PackStem| -> bool { Some(stem) == new_reachable_stem || kept_large.contains(stem) };
31
32 let kept_pack_oids: HashSet<Oid> = idxs
33 .iter()
34 .filter(|idx| PackStem::of(idx).is_some_and(|stem| kept_pack(&stem)))
35 .filter_map(|idx| fsio::pack_oids(idx, kind))
36 .flatten()
37 .collect();
38
39 let recorded = read_recorded_mtimes(&idxs, kind);
40 let mtime_of = mtime_index(&idxs, &pack_dir, kind, &recorded, loose);
41
42 let unreachable =
43 unreachable_candidates(&idxs, kind, loose, reachable, &kept_pack_oids, &kept_pack);
44 let mut keep: Vec<Oid> = unreachable
45 .into_iter()
46 .filter(|oid| !is_expired(mtime_of.get(oid), now, grace))
47 .collect();
48 keep.sort();
49
50 let cruft_stem = if keep.is_empty() {
51 None
52 } else {
53 let stem = write_cruft_pack(objects_dir, keep.clone(), kind)?;
54 if let Some(stem) = &stem {
55 write_mtimes(&pack_dir, stem, kind, |oid| {
56 mtime_of
57 .get(&oid)
58 .map(PackedMtime::from_unix)
59 .unwrap_or(now_mtime)
60 })?;
61 }
62 stem
63 };
64 knot_resource::fsync_path(&pack_dir)?;
65
66 let kept_oids: HashSet<Oid> = kept_pack_oids
67 .iter()
68 .copied()
69 .chain(cruft_stem.as_ref().into_iter().flat_map(|stem| {
70 fsio::pack_oids(&stem.file(&pack_dir, "idx"), kind).unwrap_or_default()
71 }))
72 .collect();
73
74 let keep_covered = keep.iter().all(|oid| kept_oids.contains(oid));
75 if !closure_is_covered(reachable, &kept_oids) || !keep_covered {
76 if let Some(stem) = cruft_stem.as_ref() {
77 fsio::remove_pack_files(&stem.file(&pack_dir, "idx"), &pack_dir);
78 }
79 return Ok(PruneReport::skipped());
80 }
81
82 let removed_packs = idxs
83 .iter()
84 .filter(|idx| {
85 PackStem::of(idx)
86 .is_some_and(|stem| !kept_pack(&stem) && Some(&stem) != cruft_stem.as_ref())
87 })
88 .filter(|idx| fsio::remove_pack_files(idx, &pack_dir))
89 .count();
90 let removed_loose = loose
91 .iter()
92 .filter(|(oid, _)| !reachable.contains(oid))
93 .filter(|(_, path)| std::fs::remove_file(path).is_ok())
94 .count();
95 knot_resource::fsync_path(&pack_dir)?;
96 knot_resource::fsync_path(objects_dir)?;
97
98 Ok(PruneReport {
99 removed: FileCount::new(removed_loose),
100 removed_packs: FileCount::new(removed_packs),
101 crufted: ObjectCount::new(keep.len()),
102 ran: true,
103 })
104}
105
106fn closure_is_covered(reachable: &HashSet<Oid>, kept: &HashSet<Oid>) -> bool {
107 reachable.is_subset(kept)
108}
109
110fn unreachable_candidates<K: Fn(&PackStem) -> bool>(
111 idxs: &[PathBuf],
112 kind: gix::hash::Kind,
113 loose: &[(Oid, PathBuf)],
114 reachable: &HashSet<Oid>,
115 kept_pack_oids: &HashSet<Oid>,
116 kept_pack: &K,
117) -> Vec<Oid> {
118 idxs.iter()
119 .filter(|idx| PackStem::of(idx).is_some_and(|stem| !kept_pack(&stem)))
120 .filter_map(|idx| fsio::pack_oids(idx, kind))
121 .flatten()
122 .chain(loose.iter().map(|(oid, _)| *oid))
123 .filter(|oid| !reachable.contains(oid))
124 .filter(|oid| !kept_pack_oids.contains(oid))
125 .collect::<HashSet<Oid>>()
126 .into_iter()
127 .collect()
128}
129
130fn mtime_index(
131 idxs: &[PathBuf],
132 pack_dir: &Path,
133 kind: gix::hash::Kind,
134 recorded: &HashMap<Oid, UnixSeconds>,
135 loose: &[(Oid, PathBuf)],
136) -> HashMap<Oid, UnixSeconds> {
137 let from_packs = idxs.iter().flat_map(|idx| {
138 let is_cruft = idx.with_extension("mtimes").exists();
139 let pack_secs = fsio::pack_mtime(idx, pack_dir);
140 fsio::pack_oids(idx, kind)
141 .unwrap_or_default()
142 .into_iter()
143 .filter_map(move |oid| {
144 let secs = if is_cruft {
145 recorded.get(&oid).copied()
146 } else {
147 pack_secs
148 };
149 secs.map(|secs| (oid, secs))
150 })
151 });
152 let from_loose = loose
153 .iter()
154 .filter_map(|(oid, path)| loose_mtime(path).map(|secs| (*oid, secs)));
155 newest_by_oid(from_packs.chain(from_loose))
156}
157
158fn newest_by_oid(pairs: impl Iterator<Item = (Oid, UnixSeconds)>) -> HashMap<Oid, UnixSeconds> {
159 pairs.fold(HashMap::new(), |mut acc, (oid, secs)| {
160 acc.entry(oid)
161 .and_modify(|current| {
162 if secs.get() > current.get() {
163 *current = secs;
164 }
165 })
166 .or_insert(secs);
167 acc
168 })
169}
170
171fn read_recorded_mtimes(idxs: &[PathBuf], kind: gix::hash::Kind) -> HashMap<Oid, UnixSeconds> {
172 idxs.iter()
173 .filter(|idx| idx.with_extension("mtimes").exists())
174 .filter_map(|idx| {
175 let bytes = std::fs::read(idx.with_extension("mtimes")).ok()?;
176 let oids = fsio::pack_oids(idx, kind)?;
177 let checksum = gix_pack::data::File::at(idx.with_extension("pack"), kind)
178 .ok()?
179 .checksum();
180 let table = validated_mtimes(&bytes, oids.len(), kind, checksum.as_slice())?;
181 Some(
182 oids.into_iter()
183 .zip(table)
184 .map(|(oid, mtime)| (oid, mtime.to_unix()))
185 .collect::<Vec<_>>(),
186 )
187 })
188 .flatten()
189 .collect()
190}
191
192fn validated_mtimes(
193 bytes: &[u8],
194 count: usize,
195 kind: gix::hash::Kind,
196 pack_checksum: &[u8],
197) -> Option<Vec<PackedMtime>> {
198 let hash_len = kind.len_in_bytes();
199 let header = 12usize;
200 let total = header + count * 4 + hash_len * 2;
201 if bytes.len() != total
202 || bytes[0..4] != MTIMES_MAGIC.to_be_bytes()
203 || bytes[4..8] != MTIMES_VERSION.to_be_bytes()
204 || bytes[8..12] != hash_id(kind).to_be_bytes()
205 {
206 return None;
207 }
208 let table_end = header + count * 4;
209 if bytes[table_end..table_end + hash_len] != *pack_checksum {
210 return None;
211 }
212 let mut hasher = gix_hash::hasher(kind);
213 hasher.update(&bytes[..total - hash_len]);
214 let digest = hasher.try_finalize().ok()?;
215 if digest.as_slice() != &bytes[total - hash_len..] {
216 return None;
217 }
218 Some(
219 (0..count)
220 .map(|index| {
221 let offset = header + index * 4;
222 PackedMtime(u32::from_be_bytes(
223 bytes[offset..offset + 4].try_into().unwrap(),
224 ))
225 })
226 .collect(),
227 )
228}
229
230fn is_expired(mtime: Option<&UnixSeconds>, now: SystemTime, grace: Duration) -> bool {
231 let Some(mtime) = mtime else {
232 return false;
233 };
234 let when = SystemTime::UNIX_EPOCH + Duration::from_secs(mtime.get().max(0) as u64);
235 now.duration_since(when)
236 .map(|age| age >= grace)
237 .unwrap_or(false)
238}
239
240fn loose_mtime(path: &Path) -> Option<UnixSeconds> {
241 let modified = path.metadata().ok()?.modified().ok()?;
242 Some(PackedMtime::from_system_time(modified).to_unix())
243}
244
245#[derive(Debug, Clone, Copy)]
246struct PackedMtime(u32);
247
248impl PackedMtime {
249 fn from_unix(secs: &UnixSeconds) -> Self {
250 Self(secs.get().clamp(0, u32::MAX as i64) as u32)
251 }
252
253 fn from_system_time(time: SystemTime) -> Self {
254 Self(
255 time.duration_since(SystemTime::UNIX_EPOCH)
256 .map(|delta| delta.as_secs().min(u32::MAX as u64) as u32)
257 .unwrap_or(0),
258 )
259 }
260
261 fn to_unix(self) -> UnixSeconds {
262 UnixSeconds::new(self.0 as i64)
263 }
264
265 fn to_be_bytes(self) -> [u8; 4] {
266 self.0.to_be_bytes()
267 }
268}
269
270fn hash_id(kind: gix::hash::Kind) -> u32 {
271 match kind {
272 gix::hash::Kind::Sha256 => 2,
273 _ => 1,
274 }
275}
276
277fn write_cruft_pack(
278 objects_dir: &Path,
279 oids: Vec<Oid>,
280 kind: gix::hash::Kind,
281) -> Result<Option<PackStem>, MaintError> {
282 let pack_dir = objects_dir.join("pack");
283 std::fs::create_dir_all(&pack_dir).map_err(|error| fsio::io_error(&pack_dir, error))?;
284 knot_resource::clear_stale(&pack_dir, ".knot-cruft.");
285 let staging = pack_dir.join(format!(
286 ".knot-cruft.{}.pack",
287 knot_resource::staging_nonce()
288 ));
289 let outcome = stream_pack(objects_dir, oids, kind, &staging)
290 .and_then(|()| install_pack(&pack_dir, &staging, kind));
291 let _ = std::fs::remove_file(&staging);
292 outcome
293}
294
295fn stream_pack(
296 objects_dir: &Path,
297 oids: Vec<Oid>,
298 kind: gix::hash::Kind,
299 staging: &Path,
300) -> Result<(), MaintError> {
301 let file = std::fs::File::create(staging).map_err(|error| fsio::io_error(staging, error))?;
302 let mut writer = std::io::BufWriter::new(file);
303 knot_pack::write_pack(objects_dir, oids, None, &mut writer, kind)
304 .map_err(|error| MaintError::Pack(error.to_string()))?;
305 writer
306 .into_inner()
307 .map(|_| ())
308 .map_err(|error| fsio::io_error(staging, error.into_error()))
309}
310
311fn install_pack(
312 pack_dir: &Path,
313 staging: &Path,
314 kind: gix::hash::Kind,
315) -> Result<Option<PackStem>, MaintError> {
316 let file = std::fs::File::open(staging).map_err(|error| fsio::io_error(staging, error))?;
317 let mut reader = std::io::BufReader::new(file);
318 let outcome = gix_pack::Bundle::write_to_directory(
319 &mut reader,
320 Some(pack_dir),
321 &mut Discard,
322 &AtomicBool::new(false),
323 None::<gix::odb::Handle>,
324 gix_pack::bundle::write::Options {
325 thread_limit: Some(1),
326 iteration_mode: gix_pack::data::input::Mode::Verify,
327 index_version: gix_pack::index::Version::default(),
328 object_hash: kind,
329 },
330 )
331 .map_err(|error| MaintError::Pack(error.to_string()))?;
332
333 if let Some(keep) = &outcome.keep_path {
334 let _ = std::fs::remove_file(keep);
335 }
336 [&outcome.data_path, &outcome.index_path]
337 .into_iter()
338 .flatten()
339 .try_for_each(|path| knot_resource::fsync_path(path))?;
340
341 Ok(outcome
342 .data_path
343 .as_ref()
344 .and_then(|path| PackStem::of(path)))
345}
346
347fn write_mtimes(
348 pack_dir: &Path,
349 stem: &PackStem,
350 kind: gix::hash::Kind,
351 mtime_for: impl Fn(Oid) -> PackedMtime,
352) -> Result<(), MaintError> {
353 let idx = stem.file(pack_dir, "idx");
354 let pack = stem.file(pack_dir, "pack");
355 let index = gix_pack::index::File::at(&idx, kind)
356 .map_err(|error| MaintError::Pack(format!("open cruft index: {error}")))?;
357 let checksum = gix_pack::data::File::at(&pack, kind)
358 .map_err(|error| MaintError::Pack(format!("open cruft pack: {error}")))?
359 .checksum();
360
361 let mut out = Vec::new();
362 out.extend_from_slice(&MTIMES_MAGIC.to_be_bytes());
363 out.extend_from_slice(&MTIMES_VERSION.to_be_bytes());
364 out.extend_from_slice(&hash_id(kind).to_be_bytes());
365 index
366 .iter()
367 .for_each(|entry| out.extend_from_slice(&mtime_for(Oid::from(entry.oid)).to_be_bytes()));
368 out.extend_from_slice(checksum.as_slice());
369 let mut hasher = gix_hash::hasher(kind);
370 hasher.update(&out);
371 let digest = hasher
372 .try_finalize()
373 .map_err(|error| MaintError::Pack(format!("cruft mtimes checksum: {error}")))?;
374 out.extend_from_slice(digest.as_slice());
375
376 knot_resource::atomic_write_bytes(
377 &stem.file(pack_dir, "mtimes"),
378 &out,
379 knot_resource::FileMode::Inherited,
380 )?;
381 Ok(())
382}
383
384#[cfg(test)]
385mod tests {
386 use super::*;
387
388 fn oid(byte: u8) -> Oid {
389 Oid::from_hex(&format!("{byte:02x}").repeat(20)).unwrap()
390 }
391
392 #[test]
393 fn closure_covered_when_every_reachable_oid_survives() {
394 let reachable: HashSet<Oid> = [oid(1), oid(2)].into_iter().collect();
395 let kept: HashSet<Oid> = [oid(1), oid(2), oid(3)].into_iter().collect();
396 assert!(closure_is_covered(&reachable, &kept));
397 }
398
399 #[test]
400 fn closure_uncovered_when_a_reachable_oid_is_missing() {
401 let reachable: HashSet<Oid> = [oid(1), oid(2)].into_iter().collect();
402 let kept: HashSet<Oid> = [oid(1)].into_iter().collect();
403 assert!(!closure_is_covered(&reachable, &kept));
404 }
405
406 #[test]
407 fn unknown_mtime_is_never_expired() {
408 assert!(!is_expired(None, SystemTime::now(), Duration::ZERO));
409 }
410
411 #[test]
412 fn future_mtime_is_never_expired() {
413 let future = PackedMtime::from_system_time(SystemTime::now())
414 .to_unix()
415 .saturating_add_secs(100_000);
416 assert!(!is_expired(
417 Some(&future),
418 SystemTime::now(),
419 Duration::from_secs(1)
420 ));
421 }
422
423 #[test]
424 fn newest_mtime_wins_regardless_of_iteration_order() {
425 let shared = oid(7);
426 let old = UnixSeconds::new(1_000);
427 let fresh = UnixSeconds::new(2_000);
428 let forward = newest_by_oid([(shared, old), (shared, fresh)].into_iter());
429 let reverse = newest_by_oid([(shared, fresh), (shared, old)].into_iter());
430 assert_eq!(forward.get(&shared), Some(&fresh));
431 assert_eq!(reverse.get(&shared), Some(&fresh));
432 }
433
434 #[test]
435 fn old_mtime_past_grace_is_expired() {
436 let old = UnixSeconds::new(1_000);
437 assert!(is_expired(
438 Some(&old),
439 SystemTime::now(),
440 Duration::from_secs(60)
441 ));
442 }
443
444 #[test]
445 fn corrupt_mtimes_are_rejected_rather_than_trusted() {
446 let count = 3usize;
447 let hash_len = gix::hash::Kind::Sha1.len_in_bytes();
448 let total = 12 + count * 4 + hash_len * 2;
449 let zero_checksum = vec![0u8; hash_len];
450 let unsigned = vec![0u8; total];
451 assert!(
452 validated_mtimes(&unsigned, count, gix::hash::Kind::Sha1, &zero_checksum).is_none(),
453 "a correctly-sized but unsigned mtimes table isn't trusted"
454 );
455 let truncated = vec![0u8; total - 1];
456 assert!(
457 validated_mtimes(&truncated, count, gix::hash::Kind::Sha1, &zero_checksum).is_none()
458 );
459 let mut wrong_pack = vec![0u8; total];
460 wrong_pack[0..4].copy_from_slice(&MTIMES_MAGIC.to_be_bytes());
461 wrong_pack[4..8].copy_from_slice(&MTIMES_VERSION.to_be_bytes());
462 wrong_pack[8..12].copy_from_slice(&hash_id(gix::hash::Kind::Sha1).to_be_bytes());
463 let mismatched = vec![0xabu8; hash_len];
464 assert!(
465 validated_mtimes(&wrong_pack, count, gix::hash::Kind::Sha1, &mismatched).is_none(),
466 "an mtimes table whose pack checksum names a different pack isn't trusted"
467 );
468 }
469
470 #[test]
471 fn run_fail_closes_when_survivors_miss_the_closure() {
472 use knot_git::{Layout, RefUpdate};
473 use knot_types::{BranchName, RefName, RepoDid};
474
475 use crate::test_support::{commit_on, empty_tree};
476
477 let scan = tempfile::tempdir().unwrap();
478 let layout = Layout::new(scan.path()).with_default_branch(BranchName::new("main").unwrap());
479 let did = RepoDid::new("did:plc:limpet").unwrap();
480 let repo = layout.create(&did).unwrap();
481 let tip = commit_on(&repo, empty_tree(repo.object_format()), Vec::new(), "a");
482 repo.update_ref(&RefUpdate::Create {
483 name: RefName::new("refs/heads/main").unwrap(),
484 new: tip,
485 })
486 .unwrap();
487
488 let options = crate::Options {
489 repack_max_objects: crate::ObjectCount::new(1_000_000),
490 geometric_factor: crate::GeometricFactor::full_repack(),
491 prune_grace: crate::PruneGrace::from_secs(0),
492 reflog_floor: crate::ReflogRetention::from_secs(i64::MAX as u64 / 4),
493 commit_graph: false,
494 multi_pack_index: false,
495 bitmap: false,
496 };
497 crate::run_repo(&repo, UnixSeconds::new(1_700_000_500), &options).unwrap();
498
499 let objects_dir = repo.objects_dir();
500 let kind = repo.object_format().kind();
501 let reachable: HashSet<Oid> = repo
502 .select_pack_objects(knot_git::Wants::new(&[tip]), knot_git::Haves::new(&[]))
503 .unwrap()
504 .into_iter()
505 .collect();
506 assert!(!reachable.is_empty());
507
508 let absent = PackStem::of(Path::new(
509 "pack-0000000000000000000000000000000000000000.idx",
510 ))
511 .unwrap();
512 let report = run(
513 &objects_dir,
514 kind,
515 &reachable,
516 Some(&absent),
517 &[],
518 &[],
519 Duration::ZERO,
520 )
521 .unwrap();
522 assert!(!report.ran, "an uncovered closure fail-closes the prune");
523 let reopened = knot_git::Repo::open(repo.git().git_dir()).unwrap();
524 assert!(
525 reachable.iter().all(|oid| reopened.contains(*oid)),
526 "no reachable object is deleted when survivors don't cover the closure"
527 );
528 }
529}