1use anyhow::{Result, anyhow};
4use serde::{Deserialize, Serialize};
5use std::path::{Path, PathBuf};
6use tracing::{debug, info, warn};
7
8use crate::jobstore::resolve_root;
9use crate::schema::{GcData, JobState, JobStatus, Response};
10
11const DEFAULT_OLDER_THAN: &str = "30d";
12const DEFAULT_AUTO_SCAN_LIMIT: usize = 200;
13const DEFAULT_AUTO_DELETE_LIMIT: usize = 20;
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum GcMode {
17 Manual,
18 Automatic,
19}
20
21#[derive(Debug, Clone)]
22pub struct GcPolicy {
23 pub older_than: String,
24 pub max_jobs: Option<usize>,
25 pub max_bytes: Option<u64>,
26 pub dry_run: bool,
27 pub mode: GcMode,
28 pub scan_limit: Option<usize>,
29 pub delete_limit: Option<usize>,
30}
31
32#[derive(Debug)]
33pub struct GcOpts<'a> {
34 pub root: Option<&'a str>,
35 pub older_than: Option<&'a str>,
36 pub max_jobs: Option<u64>,
37 pub max_bytes: Option<u64>,
38 pub dry_run: bool,
39}
40
41#[derive(Debug, Clone)]
42struct Candidate {
43 job_id: String,
44 path: PathBuf,
45 gc_ts: String,
46 bytes: u64,
47 reasons: Vec<&'static str>,
48}
49
50#[derive(Debug, Clone, Serialize, Deserialize)]
51pub struct AutoGcConfig {
52 pub enabled: bool,
53 pub older_than: String,
54 pub max_jobs: Option<usize>,
55 pub max_bytes: Option<u64>,
56 pub scan_limit: usize,
57 pub delete_limit: usize,
58}
59
60impl Default for AutoGcConfig {
61 fn default() -> Self {
62 Self {
63 enabled: true,
64 older_than: DEFAULT_OLDER_THAN.to_string(),
65 max_jobs: None,
66 max_bytes: None,
67 scan_limit: DEFAULT_AUTO_SCAN_LIMIT,
68 delete_limit: DEFAULT_AUTO_DELETE_LIMIT,
69 }
70 }
71}
72
73pub fn execute(opts: GcOpts) -> Result<()> {
74 let root = resolve_root(opts.root);
75 let root_str = root.display().to_string();
76
77 let (older_than_str, older_than_source) = match opts.older_than {
78 Some(s) => (s.to_string(), "flag".to_string()),
79 None => (DEFAULT_OLDER_THAN.to_string(), "default".to_string()),
80 };
81
82 let max_jobs = opts
83 .max_jobs
84 .map(|v| usize::try_from(v).map_err(|_| anyhow!("invalid --max-jobs: {v}")))
85 .transpose()?;
86
87 let policy = GcPolicy {
88 older_than: older_than_str.clone(),
89 max_jobs,
90 max_bytes: opts.max_bytes,
91 dry_run: opts.dry_run,
92 mode: GcMode::Manual,
93 scan_limit: None,
94 delete_limit: None,
95 };
96
97 let outcome = run_gc(&root, &policy)?;
98
99 Response::new(
100 "gc",
101 GcData {
102 root: root_str,
103 dry_run: opts.dry_run,
104 older_than: older_than_str,
105 older_than_source,
106 deleted: outcome.deleted,
107 skipped: outcome.skipped,
108 out_of_scope: outcome.out_of_scope,
109 failed: outcome.failed,
110 freed_bytes: outcome.freed_bytes,
111 scanned_dirs: outcome.scanned_dirs,
112 candidate_count: outcome.candidate_count,
113 },
114 )
115 .print();
116
117 Ok(())
118}
119
120pub fn maybe_run_auto_gc(root: &Path, cfg: &AutoGcConfig) {
121 if !cfg.enabled {
122 debug!("auto-gc disabled");
123 return;
124 }
125
126 let policy = GcPolicy {
127 older_than: cfg.older_than.clone(),
128 max_jobs: cfg.max_jobs,
129 max_bytes: cfg.max_bytes,
130 dry_run: false,
131 mode: GcMode::Automatic,
132 scan_limit: Some(cfg.scan_limit),
133 delete_limit: Some(cfg.delete_limit),
134 };
135
136 if let Err(e) = run_gc_with_lock(root, &policy) {
137 warn!(error = %e, "auto-gc failed (best-effort)");
138 }
139}
140
141#[derive(Debug)]
142struct GcOutcome {
143 deleted: u64,
144 skipped: u64,
145 out_of_scope: u64,
146 failed: u64,
147 freed_bytes: u64,
148 scanned_dirs: u64,
149 candidate_count: u64,
150}
151
152fn run_gc_with_lock(root: &Path, policy: &GcPolicy) -> Result<GcOutcome> {
153 if policy.mode == GcMode::Manual {
154 return run_gc(root, policy);
155 }
156
157 let lock_path = root.join(".gc.lock");
158 let lock = std::fs::OpenOptions::new()
159 .write(true)
160 .create_new(true)
161 .open(&lock_path);
162
163 let lock_file = match lock {
164 Ok(f) => f,
165 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
166 debug!(path = %lock_path.display(), "auto-gc lock already held; skipping");
167 return Ok(empty_outcome());
168 }
169 Err(e) => return Err(anyhow!("create auto-gc lock {}: {e}", lock_path.display())),
170 };
171
172 let result = run_gc(root, policy);
173 drop(lock_file);
174 let _ = std::fs::remove_file(&lock_path);
175 result
176}
177
178fn empty_outcome() -> GcOutcome {
179 GcOutcome {
180 deleted: 0,
181 skipped: 0,
182 out_of_scope: 0,
183 failed: 0,
184 freed_bytes: 0,
185 scanned_dirs: 0,
186 candidate_count: 0,
187 }
188}
189
190fn run_gc(root: &Path, policy: &GcPolicy) -> Result<GcOutcome> {
191 if !root.exists() {
192 return Ok(empty_outcome());
193 }
194
195 let retention_secs = parse_duration(&policy.older_than).ok_or_else(|| {
196 anyhow!(
197 "invalid duration: {}; expected formats: 30d, 24h, 60m, 3600s",
198 policy.older_than
199 )
200 })?;
201
202 let now_secs = std::time::SystemTime::now()
203 .duration_since(std::time::UNIX_EPOCH)
204 .unwrap_or_default()
205 .as_secs();
206 let cutoff_secs = now_secs.saturating_sub(retention_secs);
207 let cutoff = format_rfc3339(cutoff_secs);
208
209 let mut scanned_dirs = 0u64;
210 let mut out_of_scope = 0u64;
211 let mut skipped = 0u64;
212 let mut failed = 0u64;
213
214 let mut candidates = Vec::<Candidate>::new();
215
216 let read_dir = std::fs::read_dir(root)
217 .map_err(|e| anyhow!("failed to read root directory {}: {}", root.display(), e))?;
218
219 for entry in read_dir {
220 let entry = match entry {
221 Ok(v) => v,
222 Err(e) => {
223 skipped += 1;
224 failed += 1;
225 warn!(error = %e, "gc: failed to read directory entry");
226 continue;
227 }
228 };
229
230 let path = entry.path();
231 if !path.is_dir() {
232 continue;
233 }
234
235 scanned_dirs += 1;
236 if let Some(limit) = policy.scan_limit
237 && scanned_dirs as usize > limit
238 {
239 break;
240 }
241
242 let job_id = match path.file_name().and_then(|n| n.to_str()) {
243 Some(s) => s.to_string(),
244 None => {
245 skipped += 1;
246 out_of_scope += 1;
247 continue;
248 }
249 };
250
251 let state_path = path.join("state.json");
252 let state = match std::fs::read(&state_path)
253 .ok()
254 .and_then(|b| serde_json::from_slice::<JobState>(&b).ok())
255 {
256 Some(s) => s,
257 None => {
258 skipped += 1;
259 out_of_scope += 1;
260 continue;
261 }
262 };
263
264 let status = state.status().clone();
265 if matches!(status, JobStatus::Running | JobStatus::Created) {
266 skipped += 1;
267 out_of_scope += 1;
268 continue;
269 }
270
271 if !matches!(
272 status,
273 JobStatus::Exited | JobStatus::Killed | JobStatus::Failed
274 ) {
275 skipped += 1;
276 out_of_scope += 1;
277 continue;
278 }
279
280 let gc_ts = state
281 .finished_at
282 .as_deref()
283 .or(Some(state.updated_at.as_str()))
284 .unwrap_or_default()
285 .to_string();
286
287 if gc_ts.is_empty() {
288 skipped += 1;
289 out_of_scope += 1;
290 continue;
291 }
292
293 if !is_older_than(&gc_ts, &cutoff) {
294 skipped += 1;
295 out_of_scope += 1;
296 continue;
297 }
298
299 let bytes = dir_size_bytes(&path);
300 candidates.push(Candidate {
301 job_id,
302 path,
303 gc_ts,
304 bytes,
305 reasons: vec!["older_than"],
306 });
307 }
308
309 candidates.sort_by(|a, b| a.gc_ts.cmp(&b.gc_ts)); if let Some(max_jobs) = policy.max_jobs
312 && candidates.len() > max_jobs
313 {
314 let cut = candidates.len() - max_jobs;
316 for c in &mut candidates[..cut] {
317 c.reasons.push("max_jobs");
318 }
319 for c in &mut candidates[cut..] {
320 c.reasons.retain(|r| *r != "older_than");
321 }
322 }
323
324 if let Some(max_bytes) = policy.max_bytes {
325 let mut all_terminal_total = candidates.iter().map(|c| c.bytes).sum::<u64>();
326 if all_terminal_total > max_bytes {
327 for c in &mut candidates {
328 if all_terminal_total <= max_bytes {
329 break;
330 }
331 if !c.reasons.contains(&"max_bytes") {
332 c.reasons.push("max_bytes");
333 }
334 all_terminal_total = all_terminal_total.saturating_sub(c.bytes);
335 }
336 }
337 }
338
339 let mut selected = Vec::new();
340 for c in candidates {
341 if c.reasons.is_empty() {
342 skipped += 1;
343 out_of_scope += 1;
344 continue;
345 }
346 selected.push(c);
347 }
348
349 let candidate_count = selected.len() as u64;
350 let mut deleted = 0u64;
351 let mut freed_bytes = 0u64;
352 let mut deletions = 0usize;
353
354 for c in selected {
355 if let Some(limit) = policy.delete_limit
356 && deletions >= limit
357 {
358 skipped += 1;
359 out_of_scope += 1;
360 continue;
361 }
362
363 if policy.dry_run {
364 freed_bytes = freed_bytes.saturating_add(c.bytes);
365 continue;
366 }
367
368 match std::fs::remove_dir_all(&c.path) {
369 Ok(()) => {
370 if c.path.exists() {
371 skipped += 1;
372 failed += 1;
373 } else {
374 deletions += 1;
375 deleted += 1;
376 freed_bytes = freed_bytes.saturating_add(c.bytes);
377 }
378 }
379 Err(e) => {
380 skipped += 1;
381 failed += 1;
382 warn!(job_id = %c.job_id, error = %e, "gc: failed to delete job directory");
383 }
384 }
385 }
386
387 info!(
388 mode = ?policy.mode,
389 deleted,
390 skipped,
391 out_of_scope,
392 failed,
393 freed_bytes,
394 scanned_dirs,
395 candidate_count,
396 "gc complete"
397 );
398
399 Ok(GcOutcome {
400 deleted,
401 skipped,
402 out_of_scope,
403 failed,
404 freed_bytes,
405 scanned_dirs,
406 candidate_count,
407 })
408}
409
410pub fn parse_duration(s: &str) -> Option<u64> {
411 let s = s.trim();
412 if let Some(n) = s.strip_suffix('d') {
413 n.parse::<u64>().ok().map(|v| v * 86_400)
414 } else if let Some(n) = s.strip_suffix('h') {
415 n.parse::<u64>().ok().map(|v| v * 3_600)
416 } else if let Some(n) = s.strip_suffix('m') {
417 n.parse::<u64>().ok().map(|v| v * 60)
418 } else if let Some(n) = s.strip_suffix('s') {
419 n.parse::<u64>().ok()
420 } else {
421 s.parse::<u64>().ok()
422 }
423}
424
425fn is_older_than(ts: &str, cutoff: &str) -> bool {
426 let ts_prefix = &ts[..ts.len().min(19)];
427 let cutoff_prefix = &cutoff[..cutoff.len().min(19)];
428 ts_prefix < cutoff_prefix
429}
430
431pub fn dir_size_bytes(path: &Path) -> u64 {
432 let mut total = 0u64;
433 let Ok(entries) = std::fs::read_dir(path) else {
434 return 0;
435 };
436 for entry in entries.flatten() {
437 let p = entry.path();
438 if let Ok(meta) = p.metadata() {
439 if meta.is_file() {
440 total += meta.len();
441 } else if meta.is_dir() {
442 total += dir_size_bytes(&p);
443 }
444 }
445 }
446 total
447}
448
449fn format_rfc3339(secs: u64) -> String {
450 let mut s = secs;
451 let seconds = s % 60;
452 s /= 60;
453 let minutes = s % 60;
454 s /= 60;
455 let hours = s % 24;
456 s /= 24;
457
458 let mut days = s;
459 let mut year = 1970u64;
460 loop {
461 let days_in_year = if is_leap(year) { 366 } else { 365 };
462 if days < days_in_year {
463 break;
464 }
465 days -= days_in_year;
466 year += 1;
467 }
468
469 let leap = is_leap(year);
470 let month_days: [u64; 12] = [
471 31,
472 if leap { 29 } else { 28 },
473 31,
474 30,
475 31,
476 30,
477 31,
478 31,
479 30,
480 31,
481 30,
482 31,
483 ];
484 let mut month = 0usize;
485 for (i, &d) in month_days.iter().enumerate() {
486 if days < d {
487 month = i;
488 break;
489 }
490 days -= d;
491 }
492 let day = days + 1;
493
494 format!(
495 "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
496 year,
497 month + 1,
498 day,
499 hours,
500 minutes,
501 seconds
502 )
503}
504
505fn is_leap(year: u64) -> bool {
506 (year.is_multiple_of(4) && !year.is_multiple_of(100)) || year.is_multiple_of(400)
507}
508
509#[cfg(test)]
510mod tests {
511 use super::*;
512
513 #[test]
514 fn parse_duration_days() {
515 assert_eq!(parse_duration("30d"), Some(30 * 86_400));
516 }
517
518 #[test]
519 fn parse_duration_invalid() {
520 assert!(parse_duration("abc").is_none());
521 }
522
523 #[test]
524 fn older_than_logic() {
525 assert!(is_older_than(
526 "2020-01-01T00:00:00Z",
527 "2024-01-01T00:00:00Z"
528 ));
529 assert!(!is_older_than(
530 "2024-01-01T00:00:00Z",
531 "2024-01-01T00:00:00Z"
532 ));
533 }
534}