forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
4.5 KiB
1//! Background upkeep for CI. Every 30 seconds it marks running jobs whose runner stopped
2//! heartbeating as Error and syncs the statuses of their runs. Once a day it deletes job logs and
3//! artifacts past their retention, from `[ci]` in the config.
4
5use std::collections::HashSet;
6use std::path::PathBuf;
7use std::time::Duration;
8
9use forge_core::time::stamp_ago;
10use tokio::time::{Instant, MissedTickBehavior};
11
12use super::{artifact_key, log_path, sync_run_statuses};
13use crate::state::AppState;
14
15/// How often lost jobs are reaped.
16const REAP_EVERY: Duration = Duration::from_secs(30);
17/// A running job with no heartbeat for this long is lost.
18const LOST_AFTER: Duration = Duration::from_secs(120);
19/// How often retention runs.
20const PURGE_EVERY: Duration = Duration::from_secs(24 * 60 * 60);
21
22/// Starts the reaper task. It runs until the process exits.
23pub fn spawn(state: AppState) {
24 tokio::spawn(run(state));
25}
26
27async fn run(state: AppState) {
28 let mut tick = tokio::time::interval(REAP_EVERY);
29 tick.set_missed_tick_behavior(MissedTickBehavior::Delay);
30 // Retention runs on the first pass after boot, then once a day.
31 let mut next_purge = Instant::now();
32 loop {
33 tick.tick().await;
34 reap_once(&state).await;
35 if Instant::now() >= next_purge {
36 purge_once(&state).await;
37 next_purge = Instant::now() + PURGE_EVERY;
38 }
39 }
40}
41
42/// Marks lost jobs as Error and syncs the statuses of every run they belong to. Returns how many
43/// jobs were reaped.
44pub async fn reap_once(state: &AppState) -> usize {
45 reap_before(state, &stamp_ago(LOST_AFTER)).await
46}
47
48/// [`reap_once`] with an explicit cutoff: running jobs whose heartbeat is older than `older_than`
49/// are lost.
50pub async fn reap_before(state: &AppState, older_than: &str) -> usize {
51 let jobs = match state.store.reap_lost_jobs(older_than).await {
52 Ok(jobs) => jobs,
53 Err(e) => {
54 tracing::error!(error = %e, "ci: reaping lost jobs failed");
55 return 0;
56 }
57 };
58 let mut seen = HashSet::new();
59 let mut wake = false;
60 for job in &jobs {
61 state.ci_live.close_job(&job.id);
62 state.ci_live.bump_run(&job.run_id);
63 if !seen.insert(job.run_id.clone()) {
64 continue;
65 }
66 match state.store.run_by_id(&job.run_id).await {
67 Ok(Some(run)) => match sync_run_statuses(state, &run).await {
68 Ok(queued) => wake |= queued,
69 Err(e) => {
70 tracing::error!(error = %e, run_id = %run.id, "ci: syncing statuses failed")
71 }
72 },
73 Ok(None) => {}
74 Err(e) => tracing::error!(error = %e, run_id = %job.run_id, "ci: loading run failed"),
75 }
76 }
77 if wake {
78 state.ci_waker.notify_waiters();
79 }
80 if !jobs.is_empty() {
81 tracing::warn!(count = jobs.len(), "ci: reaped lost jobs");
82 }
83 jobs.len()
84}
85
86/// Deletes the logs and artifacts past retention. Rows go first (in the store); files follow.
87/// Missing files are fine. Returns `(logs deleted, artifacts deleted)` as attempted.
88pub async fn purge_once(state: &AppState) -> (usize, usize) {
89 let ci = &state.config.ci;
90 let day = |days: u64| Duration::from_secs(days.saturating_mul(24 * 60 * 60));
91 let logs_before = stamp_ago(day(ci.log_retention_days));
92 let artifacts_before = stamp_ago(day(ci.artifact_retention_days));
93 let (job_ids, artifact_ids) = match state.store.purge_old(&logs_before, &artifacts_before).await
94 {
95 Ok(ids) => ids,
96 Err(e) => {
97 tracing::error!(error = %e, "ci: retention purge failed");
98 return (0, 0);
99 }
100 };
101 let logs: Vec<PathBuf> = job_ids
102 .iter()
103 .map(|id| log_path(&state.config.data_dir, id))
104 .collect();
105 let log_count = logs.len();
106 let removed = tokio::task::spawn_blocking(move || {
107 for path in &logs {
108 if let Err(e) = std::fs::remove_file(path)
109 && e.kind() != std::io::ErrorKind::NotFound
110 {
111 tracing::error!(error = %e, path = %path.display(), "ci: deleting a log failed");
112 }
113 }
114 })
115 .await;
116 if let Err(e) = removed {
117 tracing::error!(error = %e, "ci: log deletion task failed");
118 }
119 let artifact_count = artifact_ids.len();
120 for id in &artifact_ids {
121 if let Err(e) = state.blobs.delete(&artifact_key(id)).await {
122 tracing::error!(error = %e, artifact_id = %id, "ci: deleting an artifact failed");
123 }
124 }
125 (log_count, artifact_count)
126}