zuka
zuka/src/jobs/mod.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/mod.rs
RSmod.rs5.3 KBDownload
1// Scheduled maintenance.
2//
3// Runs in its own process (`<binary> jobs`) behind its own service unit, not in the
4// web entrypoint — the house rule, and here it also means a long repack cannot stall
5// request handling.
6//
7// Three jobs, each of which fixes something the service otherwise leaks forever:
8// garbage collection (loose objects accumulate at roughly three per commit),
9// the deleted-repository sweep (a soft delete reclaims nothing without it), and
10// fsck (metadata and disk can disagree, and nothing else reports it).
11
12pub mod fsck;
13pub mod gc;
14pub mod sweep;
15
16use crate::config::Config;
17use std::sync::Arc;
18use std::time::Instant;
19
20/// Run every maintenance job once.
21pub fn run_once(config: &Config) -> Report {
22 let started = Instant::now();
23 let gc = gc::run(config);
24 let swept = sweep::run(config);
25 let checked = fsck::run(config, false);
26
27 Report {
28 repos_packed: gc.packed,
29 bytes_reclaimed: gc.reclaimed,
30 repos_swept: swept.removed,
31 bytes_swept: swept.bytes,
32 orphans: checked.orphans.len(),
33 elapsed_ms: started.elapsed().as_millis() as u64,
34 }
35}
36
37#[derive(Debug, Default)]
38pub struct Report {
39 pub repos_packed: usize,
40 pub bytes_reclaimed: u64,
41 pub repos_swept: usize,
42 pub bytes_swept: u64,
43 pub orphans: usize,
44 pub elapsed_ms: u64,
45}
46
47impl std::fmt::Display for Report {
48 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
49 write!(
50 f,
51 "packed={} reclaimed={}B swept={} swept_bytes={}B orphans={} in {}ms",
52 self.repos_packed,
53 self.bytes_reclaimed,
54 self.repos_swept,
55 self.bytes_swept,
56 self.orphans,
57 self.elapsed_ms
58 )
59 }
60}
61
62/// Run maintenance on an interval until the process is stopped.
63pub async fn serve(config: Arc<Config>) {
64 let interval = config.gc_interval;
65 eprintln!("[jobs] maintenance every {interval:?}");
66
67 loop {
68 let config = Arc::clone(&config);
69 let outcome = tokio::task::spawn_blocking(move || run_once(&config)).await;
70
71 match outcome {
72 Ok(report) => eprintln!("[jobs] {report}"),
73 // A panic in one cycle must not stop the scheduler; the next cycle is
74 // the retry.
75 Err(e) => eprintln!("[jobs] cycle failed: {e}"),
76 }
77 tokio::time::sleep(interval).await;
78 }
79}
80
81/// Every bare repository under the git root, as `(account, repo, path)`.
82pub fn walk_repos(git_root: &std::path::Path) -> Vec<(String, String, std::path::PathBuf)> {
83 let mut found = Vec::new();
84 let Ok(accounts) = std::fs::read_dir(git_root) else {
85 return found;
86 };
87
88 for account in accounts.flatten() {
89 if !account.path().is_dir() {
90 continue;
91 }
92 let account_name = account.file_name().to_string_lossy().to_string();
93
94 let Ok(repos) = std::fs::read_dir(account.path()) else {
95 continue;
96 };
97 for repo in repos.flatten() {
98 let path = repo.path();
99 let name = repo.file_name().to_string_lossy().to_string();
100 let Some(stripped) = name.strip_suffix(".git") else {
101 continue;
102 };
103 if path.join("HEAD").is_file() {
104 found.push((account_name.clone(), stripped.to_string(), path));
105 }
106 }
107 }
108 found.sort();
109 found
110}
111
112/// Total bytes under a directory, following no symlinks.
113pub fn dir_bytes(path: &std::path::Path) -> u64 {
114 let mut total = 0u64;
115 let mut stack = vec![path.to_path_buf()];
116
117 while let Some(dir) = stack.pop() {
118 let Ok(entries) = std::fs::read_dir(&dir) else {
119 continue;
120 };
121 for entry in entries.flatten() {
122 let Ok(meta) = entry.metadata() else { continue };
123 if meta.is_dir() {
124 stack.push(entry.path());
125 } else if meta.is_file() {
126 total = total.saturating_add(meta.len());
127 }
128 }
129 }
130 total
131}
132
133#[cfg(test)]
134mod tests {
135 use super::*;
136 use std::process::Command;
137
138 #[test]
139 fn walks_only_bare_repositories() {
140 let dir = tempfile::tempdir().unwrap();
141 let root = dir.path();
142
143 std::fs::create_dir_all(root.join("alice")).unwrap();
144 Command::new("git")
145 .args(["init", "--bare", "--quiet"])
146 .arg(root.join("alice").join("site.git"))
147 .status()
148 .unwrap();
149
150 // A directory that is not a repository must be ignored, not walked into.
151 std::fs::create_dir_all(root.join("alice").join("notarepo.git")).unwrap();
152 std::fs::create_dir_all(root.join("alice").join("plain")).unwrap();
153
154 let found = walk_repos(root);
155 assert_eq!(found.len(), 1);
156 assert_eq!(found[0].0, "alice");
157 assert_eq!(found[0].1, "site");
158 }
159
160 #[test]
161 fn an_absent_root_walks_to_nothing() {
162 assert!(walk_repos(std::path::Path::new("/nonexistent")).is_empty());
163 }
164
165 #[test]
166 fn dir_bytes_sums_files_recursively() {
167 let dir = tempfile::tempdir().unwrap();
168 std::fs::write(dir.path().join("a"), b"12345").unwrap();
169 std::fs::create_dir(dir.path().join("sub")).unwrap();
170 std::fs::write(dir.path().join("sub").join("b"), b"123").unwrap();
171
172 assert_eq!(dir_bytes(dir.path()), 8);
173 }
174}