alice.pds.demo.boltless.dev
This repository has no description
8.1 kB
262 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::HashSet;
2use std::path::{Path, PathBuf};
3use std::sync::atomic::AtomicBool;
4
5use gix::progress::Discard;
6use knot_git::{Haves, Repo, Wants};
7use knot_types::Oid;
8
9use crate::fsio::{self, PackStem};
10use crate::{FileCount, GeometricFactor, MaintError, ObjectCount, RepackReport, RepackStatus};
11
12type RepackOutcome = (
13 RepackReport,
14 Option<HashSet<Oid>>,
15 Option<PackStem>,
16 Vec<PackStem>,
17);
18
19pub fn run(
20 repo: &Repo,
21 objects_dir: &Path,
22 kind: gix::hash::Kind,
23 roots: Vec<Oid>,
24 max_objects: ObjectCount,
25 factor: GeometricFactor,
26 loose: &[(Oid, PathBuf)],
27) -> Result<RepackOutcome, MaintError> {
28 let closure = match repo.select_pack_objects(Wants::new(&roots), Haves::new(&[])) {
29 Ok(closure) => closure,
30 Err(_) => {
31 return Ok((
32 RepackReport::skipped(RepackStatus::ClosureFailed),
33 None,
34 None,
35 Vec::new(),
36 ));
37 }
38 };
39 if closure.is_empty() {
40 return Ok((
41 RepackReport::skipped(RepackStatus::NothingReachable),
42 Some(HashSet::new()),
43 None,
44 Vec::new(),
45 ));
46 }
47 if closure.len() > max_objects.get() {
48 return Ok((
49 RepackReport::skipped(RepackStatus::SkippedTooLarge),
50 None,
51 None,
52 Vec::new(),
53 ));
54 }
55 let reachable: HashSet<Oid> = closure.iter().copied().collect();
56
57 let kept_large = if factor == GeometricFactor::full_repack() {
58 Vec::new()
59 } else {
60 kept_large_packs(objects_dir, kind, factor)
61 };
62 let kept_large_oids: HashSet<Oid> = kept_large
63 .iter()
64 .filter_map(|idx| fsio::pack_oids(idx, kind))
65 .flatten()
66 .collect();
67 let new_objects: Vec<Oid> = closure
68 .into_iter()
69 .filter(|oid| !kept_large_oids.contains(oid))
70 .collect();
71
72 let new_stem = if new_objects.is_empty() {
73 None
74 } else {
75 build_and_install_pack(objects_dir, new_objects, kind)?
76 };
77
78 let removed_loose = loose
79 .iter()
80 .filter(|(oid, _)| reachable.contains(oid))
81 .filter(|(_, path)| std::fs::remove_file(path).is_ok())
82 .count();
83
84 let kept_large_stems: Vec<PackStem> = kept_large
85 .iter()
86 .filter_map(|idx| PackStem::of(idx))
87 .collect();
88
89 Ok((
90 RepackReport {
91 status: RepackStatus::Repacked,
92 packed_objects: ObjectCount::new(reachable.len()),
93 removed_loose: FileCount::new(removed_loose),
94 removed_packs: FileCount::new(0),
95 },
96 Some(reachable),
97 new_stem,
98 kept_large_stems,
99 ))
100}
101
102fn kept_large_packs(
103 objects_dir: &Path,
104 kind: gix::hash::Kind,
105 factor: GeometricFactor,
106) -> Vec<PathBuf> {
107 let pack_dir = objects_dir.join("pack");
108 let mut eligible: Vec<(PathBuf, usize)> = fsio::pack_idx_paths(objects_dir)
109 .into_iter()
110 .filter(|idx| !is_excluded(idx, &pack_dir))
111 .filter_map(|idx| fsio::pack_oids(&idx, kind).map(|oids| (idx, oids.len())))
112 .collect();
113 eligible.sort_by_key(|(_, count)| *count);
114 let weights: Vec<usize> = eligible.iter().map(|(_, count)| *count).collect();
115 let split = compute_split(&weights, factor);
116 eligible
117 .into_iter()
118 .skip(split)
119 .map(|(idx, _)| idx)
120 .collect()
121}
122
123fn is_excluded(idx: &Path, pack_dir: &Path) -> bool {
124 let Some(stem) = idx.file_stem().and_then(|stem| stem.to_str()) else {
125 return true;
126 };
127 pack_dir.join(format!("{stem}.mtimes")).exists()
128}
129
130fn compute_split(weights: &[usize], factor: GeometricFactor) -> usize {
131 let n = weights.len();
132 if n == 0 {
133 return 0;
134 }
135 let geometric =
136 |big: usize, small: usize| (small as u64).saturating_mul(factor.get()) <= big as u64;
137 let split1 = (1..n)
138 .rev()
139 .find(|&i| !geometric(weights[i], weights[i - 1]))
140 .map(|i| i + 1)
141 .unwrap_or(0);
142 let total: u64 = weights[..split1].iter().map(|weight| *weight as u64).sum();
143 let extended = (split1..n).try_fold((split1, total), |(split, total), j| {
144 match total.checked_mul(factor.get()) {
145 Some(threshold) if (weights[j] as u64) < threshold => std::ops::ControlFlow::Continue(
146 (split + 1, total.saturating_add(weights[j] as u64)),
147 ),
148 _ => std::ops::ControlFlow::Break((split, total)),
149 }
150 });
151 match extended {
152 std::ops::ControlFlow::Continue((split, _)) => split,
153 std::ops::ControlFlow::Break((split, _)) => split,
154 }
155}
156
157fn build_and_install_pack(
158 objects_dir: &Path,
159 closure: Vec<Oid>,
160 kind: gix::hash::Kind,
161) -> Result<Option<PackStem>, MaintError> {
162 let pack_dir = objects_dir.join("pack");
163 std::fs::create_dir_all(&pack_dir).map_err(|error| fsio::io_error(&pack_dir, error))?;
164 knot_resource::clear_stale(&pack_dir, ".knot-repack.");
165 let staging = pack_dir.join(format!(
166 ".knot-repack.{}.pack",
167 knot_resource::staging_nonce()
168 ));
169 let outcome = write_streaming_pack(objects_dir, closure, kind, &staging)
170 .and_then(|()| install_streamed_pack(&pack_dir, &staging, kind));
171 let _ = std::fs::remove_file(&staging);
172 outcome
173}
174
175fn write_streaming_pack(
176 objects_dir: &Path,
177 closure: Vec<Oid>,
178 kind: gix::hash::Kind,
179 staging: &Path,
180) -> Result<(), MaintError> {
181 let file = std::fs::File::create(staging).map_err(|error| fsio::io_error(staging, error))?;
182 let mut writer = std::io::BufWriter::new(file);
183 knot_pack::write_pack(objects_dir, closure, None, &mut writer, kind)
184 .map_err(|error| MaintError::Pack(error.to_string()))?;
185 writer
186 .into_inner()
187 .map(|_| ())
188 .map_err(|error| fsio::io_error(staging, error.into_error()))
189}
190
191fn install_streamed_pack(
192 pack_dir: &Path,
193 staging: &Path,
194 kind: gix::hash::Kind,
195) -> Result<Option<PackStem>, MaintError> {
196 let file = std::fs::File::open(staging).map_err(|error| fsio::io_error(staging, error))?;
197 let mut reader = std::io::BufReader::new(file);
198 let outcome = gix_pack::Bundle::write_to_directory(
199 &mut reader,
200 Some(pack_dir),
201 &mut Discard,
202 &AtomicBool::new(false),
203 None::<gix::odb::Handle>,
204 gix_pack::bundle::write::Options {
205 thread_limit: Some(1),
206 iteration_mode: gix_pack::data::input::Mode::Verify,
207 index_version: gix_pack::index::Version::default(),
208 object_hash: kind,
209 },
210 )
211 .map_err(|error| MaintError::Pack(error.to_string()))?;
212
213 if let Some(keep) = &outcome.keep_path {
214 let _ = std::fs::remove_file(keep);
215 }
216 [&outcome.data_path, &outcome.index_path]
217 .into_iter()
218 .flatten()
219 .try_for_each(|path| knot_resource::fsync_path(path))?;
220 knot_resource::fsync_path(pack_dir)?;
221
222 Ok(outcome
223 .data_path
224 .as_ref()
225 .and_then(|path| PackStem::of(path)))
226}
227
228#[cfg(test)]
229mod tests {
230 use super::{GeometricFactor, compute_split};
231
232 #[test]
233 fn uniform_small_packs_all_roll_up() {
234 assert_eq!(compute_split(&[1, 1, 1, 1], GeometricFactor::new(2)), 4);
235 }
236
237 #[test]
238 fn a_clean_geometric_progression_rolls_up_nothing() {
239 assert_eq!(compute_split(&[1, 2, 4, 8], GeometricFactor::new(2)), 0);
240 }
241
242 #[test]
243 fn one_large_pack_with_tiny_additions_keeps_the_large_one() {
244 let split = compute_split(&[1, 1, 1, 1000], GeometricFactor::new(2));
245 assert_eq!(split, 3);
246 }
247
248 #[test]
249 fn a_single_pack_is_never_rolled_up() {
250 assert_eq!(compute_split(&[42], GeometricFactor::new(2)), 0);
251 }
252
253 #[test]
254 fn empty_input_rolls_up_nothing() {
255 assert_eq!(compute_split(&[], GeometricFactor::new(2)), 0);
256 }
257
258 #[test]
259 fn a_huge_factor_saturates_without_overflow() {
260 assert_eq!(compute_split(&[1, 1, 1], GeometricFactor::full_repack()), 3);
261 }
262}