zuka
zuka/src/jobs/sweep.rs

worklyn / zukapublic

Agent-first git hosting. One Rust binary: git over HTTP and SSH, a REST API, MCP, CI, and multi-tenant isolation.

Get a copy: git clone https://zuka.worklyn.com/worklyn/zuka.git
branchbc985d6bbd321be340e3750b3da4734ee1553cbd ▾
main All branches →
zuka/src/jobs/sweep.rs
RSsweep.rs9.2 KBDownload
1// The deleted-repository sweep.
2//
3// `soft_delete` moves a repository into `tmp/deleted/` so an agent that deletes the
4// wrong thing can get it back. Without this job that directory only grows: a delete
5// reclaims nothing, forever, on the same disk the live repositories use.
6
7use crate::config::Config;
8use crate::jobs::dir_bytes;
9use std::time::{Duration, SystemTime};
10
11#[derive(Debug, Default)]
12pub struct Outcome {
13 pub removed: usize,
14 pub bytes: u64,
15}
16
17impl Outcome {
18 fn merge(&mut self, other: Outcome) {
19 self.removed += other.removed;
20 self.bytes = self.bytes.saturating_add(other.bytes);
21 }
22}
23
24pub fn run(config: &Config) -> Outcome {
25 let mut outcome = remove_older_than(&config.deleted_dir(), config.deleted_retention);
26
27 let runs = prune_runs(&config.runs_dir(), config.ci_keep_runs);
28 outcome.removed += runs.removed;
29 outcome.bytes = outcome.bytes.saturating_add(runs.bytes);
30
31 // Stale CI workspaces: a run killed mid-flight cannot clean up after itself.
32 let work = remove_older_than(
33 &config.data_dir.join("tmp").join("work"),
34 Duration::from_secs(3600),
35 );
36 outcome.removed += work.removed;
37 outcome.bytes = outcome.bytes.saturating_add(work.bytes);
38
39 outcome
40}
41
42/// Keep the newest `keep` runs per repository, with their logs.
43///
44/// Without this, run records and their captured output accumulate for the life of
45/// the repository — one more thing that only grows.
46pub fn prune_runs(runs_dir: &std::path::Path, keep: usize) -> Outcome {
47 let mut outcome = Outcome::default();
48 if keep == 0 {
49 return outcome;
50 }
51
52 let Ok(accounts) = std::fs::read_dir(runs_dir) else {
53 return outcome;
54 };
55 for account in accounts.flatten() {
56 let Ok(repos) = std::fs::read_dir(account.path()) else {
57 continue;
58 };
59 for repo in repos.flatten() {
60 outcome.merge(prune_repo_runs(&repo.path(), keep));
61 }
62 }
63 outcome
64}
65
66fn prune_repo_runs(dir: &std::path::Path, keep: usize) -> Outcome {
67 let mut outcome = Outcome::default();
68 let Ok(entries) = std::fs::read_dir(dir) else {
69 return outcome;
70 };
71
72 // Order by mtime rather than by parsing every record: the sweep should not
73 // depend on the run schema.
74 let mut records: Vec<(SystemTime, std::path::PathBuf)> = entries
75 .flatten()
76 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
77 .filter_map(|e| {
78 let modified = e.metadata().ok()?.modified().ok()?;
79 Some((modified, e.path()))
80 })
81 .collect();
82
83 if records.len() <= keep {
84 return outcome;
85 }
86 records.sort_by_key(|(modified, _)| std::cmp::Reverse(*modified));
87
88 for (_, path) in records.into_iter().skip(keep) {
89 let log = path.with_extension("log");
90 let bytes = file_bytes(&path) + file_bytes(&log);
91 if std::fs::remove_file(&path).is_ok() {
92 let _ = std::fs::remove_file(&log);
93 outcome.removed += 1;
94 outcome.bytes = outcome.bytes.saturating_add(bytes);
95 }
96 }
97 outcome
98}
99
100fn file_bytes(path: &std::path::Path) -> u64 {
101 std::fs::metadata(path).map(|m| m.len()).unwrap_or(0)
102}
103
104/// Remove soft-deleted repositories older than `retention`.
105pub fn remove_older_than(deleted_dir: &std::path::Path, retention: Duration) -> Outcome {
106 let mut outcome = Outcome::default();
107 let Ok(entries) = std::fs::read_dir(deleted_dir) else {
108 return outcome;
109 };
110
111 let now = SystemTime::now();
112 for entry in entries.flatten() {
113 let path = entry.path();
114 let Ok(meta) = entry.metadata() else { continue };
115 if !meta.is_dir() {
116 continue;
117 }
118
119 // Age is taken from mtime, which `rename` preserves — so the clock starts
120 // when the repository was deleted, not when the directory was created.
121 let age = meta
122 .modified()
123 .ok()
124 .and_then(|t| now.duration_since(t).ok())
125 .unwrap_or_default();
126 if age < retention {
127 continue;
128 }
129
130 let bytes = dir_bytes(&path);
131 match std::fs::remove_dir_all(&path) {
132 Ok(()) => {
133 outcome.removed += 1;
134 outcome.bytes = outcome.bytes.saturating_add(bytes);
135 }
136 Err(e) => eprintln!("[jobs] sweep {} failed: {e}", path.display()),
137 }
138 }
139 outcome
140}
141
142#[cfg(test)]
143mod tests {
144 use super::*;
145 use std::time::UNIX_EPOCH;
146
147 #[test]
148 fn run_records_are_pruned_to_the_newest_few() {
149 let dir = tempfile::tempdir().unwrap();
150 let repo = dir.path().join("alice").join("site");
151 std::fs::create_dir_all(&repo).unwrap();
152
153 for i in 0..10 {
154 std::fs::write(repo.join(format!("run{i}.json")), b"{}").unwrap();
155 std::fs::write(repo.join(format!("run{i}.log")), b"output").unwrap();
156 // Distinct mtimes so "newest" is well defined.
157 filetime_set(
158 &repo.join(format!("run{i}.json")),
159 UNIX_EPOCH + Duration::from_secs(1_000_000 + i),
160 );
161 }
162
163 let outcome = prune_runs(dir.path(), 3);
164 assert_eq!(outcome.removed, 7);
165 assert!(outcome.bytes > 0);
166
167 let left: Vec<_> = std::fs::read_dir(&repo)
168 .unwrap()
169 .flatten()
170 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
171 .collect();
172 assert_eq!(left.len(), 3, "the newest three must survive");
173 assert!(
174 repo.join("run9.json").exists() && !repo.join("run0.json").exists(),
175 "pruning must keep the newest, not an arbitrary three"
176 );
177 assert!(
178 !repo.join("run0.log").exists(),
179 "a pruned run must take its log with it"
180 );
181 }
182
183 #[test]
184 fn pruning_is_disabled_by_a_zero_retention() {
185 let dir = tempfile::tempdir().unwrap();
186 let repo = dir.path().join("alice").join("site");
187 std::fs::create_dir_all(&repo).unwrap();
188 std::fs::write(repo.join("a.json"), b"{}").unwrap();
189
190 assert_eq!(prune_runs(dir.path(), 0).removed, 0);
191 assert!(repo.join("a.json").exists());
192 }
193
194 #[test]
195 fn a_recent_deletion_is_retained() {
196 let dir = tempfile::tempdir().unwrap();
197 let grave = dir.path().join("alice__site__1.git");
198 std::fs::create_dir_all(&grave).unwrap();
199 std::fs::write(grave.join("HEAD"), b"ref: refs/heads/main").unwrap();
200
201 let outcome = remove_older_than(dir.path(), Duration::from_secs(3600));
202 assert_eq!(outcome.removed, 0);
203 assert!(grave.exists(), "a fresh delete must stay recoverable");
204 }
205
206 #[test]
207 fn an_expired_deletion_is_reclaimed() {
208 let dir = tempfile::tempdir().unwrap();
209 let grave = dir.path().join("alice__site__1.git");
210 std::fs::create_dir_all(&grave).unwrap();
211 std::fs::write(grave.join("HEAD"), b"ref: refs/heads/main").unwrap();
212
213 // Zero retention makes everything already expired.
214 let outcome = remove_older_than(dir.path(), Duration::ZERO);
215 assert_eq!(outcome.removed, 1);
216 assert!(outcome.bytes > 0, "reclaimed bytes must be reported");
217 assert!(!grave.exists());
218 }
219
220 #[test]
221 fn an_absent_directory_sweeps_to_nothing() {
222 let outcome = remove_older_than(std::path::Path::new("/nonexistent"), Duration::ZERO);
223 assert_eq!(outcome.removed, 0);
224 }
225
226 #[test]
227 fn a_stray_file_is_left_alone() {
228 let dir = tempfile::tempdir().unwrap();
229 std::fs::write(dir.path().join("note.txt"), b"not a repository").unwrap();
230
231 let outcome = remove_older_than(dir.path(), Duration::ZERO);
232 assert_eq!(outcome.removed, 0);
233 assert!(dir.path().join("note.txt").exists());
234 }
235
236 #[test]
237 fn age_is_measured_from_when_the_repository_was_deleted() {
238 let dir = tempfile::tempdir().unwrap();
239 let grave = dir.path().join("alice__old__1.git");
240 std::fs::create_dir_all(&grave).unwrap();
241
242 // Backdate it the way a real delete a week ago would look.
243 let week_ago = UNIX_EPOCH
244 + Duration::from_secs(
245 SystemTime::now()
246 .duration_since(UNIX_EPOCH)
247 .unwrap()
248 .as_secs()
249 - 7 * 86_400,
250 );
251 filetime_set(&grave, week_ago);
252
253 assert_eq!(
254 remove_older_than(dir.path(), Duration::from_secs(86_400)).removed,
255 1
256 );
257 }
258
259 /// Set mtime without pulling in a crate for it.
260 fn filetime_set(path: &std::path::Path, when: SystemTime) {
261 let secs = when.duration_since(UNIX_EPOCH).unwrap().as_secs();
262 let times = [
263 libc::timeval {
264 tv_sec: secs as libc::time_t,
265 tv_usec: 0,
266 },
267 libc::timeval {
268 tv_sec: secs as libc::time_t,
269 tv_usec: 0,
270 },
271 ];
272 let c_path = std::ffi::CString::new(path.to_string_lossy().as_bytes()).unwrap();
273 // SAFETY: `c_path` is a valid NUL-terminated path and `times` is two entries.
274 unsafe {
275 libc::utimes(c_path.as_ptr(), times.as_ptr());
276 }
277 }
278}