forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
4.4 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 if !seen.insert(job.run_id.clone()) {
62 continue;
63 }
64 match state.store.run_by_id(&job.run_id).await {
65 Ok(Some(run)) => match sync_run_statuses(state, &run).await {
66 Ok(queued) => wake |= queued,
67 Err(e) => {
68 tracing::error!(error = %e, run_id = %run.id, "ci: syncing statuses failed")
69 }
70 },
71 Ok(None) => {}
72 Err(e) => tracing::error!(error = %e, run_id = %job.run_id, "ci: loading run failed"),
73 }
74 }
75 if wake {
76 state.ci_waker.notify_one();
77 }
78 if !jobs.is_empty() {
79 tracing::warn!(count = jobs.len(), "ci: reaped lost jobs");
80 }
81 jobs.len()
82}
83
84/// Deletes the logs and artifacts past retention. Rows go first (in the store); files follow.
85/// Missing files are fine. Returns `(logs deleted, artifacts deleted)` as attempted.
86pub async fn purge_once(state: &AppState) -> (usize, usize) {
87 let ci = &state.config.ci;
88 let day = |days: u64| Duration::from_secs(days.saturating_mul(24 * 60 * 60));
89 let logs_before = stamp_ago(day(ci.log_retention_days));
90 let artifacts_before = stamp_ago(day(ci.artifact_retention_days));
91 let (job_ids, artifact_ids) = match state.store.purge_old(&logs_before, &artifacts_before).await
92 {
93 Ok(ids) => ids,
94 Err(e) => {
95 tracing::error!(error = %e, "ci: retention purge failed");
96 return (0, 0);
97 }
98 };
99 let logs: Vec<PathBuf> = job_ids
100 .iter()
101 .map(|id| log_path(&state.config.data_dir, id))
102 .collect();
103 let log_count = logs.len();
104 let removed = tokio::task::spawn_blocking(move || {
105 for path in &logs {
106 if let Err(e) = std::fs::remove_file(path)
107 && e.kind() != std::io::ErrorKind::NotFound
108 {
109 tracing::error!(error = %e, path = %path.display(), "ci: deleting a log failed");
110 }
111 }
112 })
113 .await;
114 if let Err(e) = removed {
115 tracing::error!(error = %e, "ci: log deletion task failed");
116 }
117 let artifact_count = artifact_ids.len();
118 for id in &artifact_ids {
119 if let Err(e) = state.blobs.delete(&artifact_key(id)).await {
120 tracing::error!(error = %e, artifact_id = %id, "ci: deleting an artifact failed");
121 }
122 }
123 (log_count, artifact_count)
124}