| 1 | //! Deleting old job logs (`LOG_RETENTION_DAYS`, limits.rs): once an hour, |
| 2 | //! the logs of jobs, and of earlier attempts, that finished longer ago than |
| 3 | //! that. A job keeps its steps, conclusion, annotations and summary; only |
| 4 | //! its log lines go, and the job is marked (`logs_purged_at`) so it is not |
| 5 | //! looked at again. A few hundred jobs a sweep, so a backlog drains over |
| 6 | //! hours instead of in one heavy query. |
| 7 | |
| 8 | use g1t_contracts::time::rfc3339; |
| 9 | use serde::Deserialize; |
| 10 | use worker::Result; |
| 11 | |
| 12 | use crate::Actions; |
| 13 | |
| 14 | /// Jobs (and earlier attempts) whose logs one sweep deletes, at most. |
| 15 | const PER_SWEEP: u32 = 200; |
| 16 | const DAY_MS: u64 = 24 * 60 * 60 * 1000; |
| 17 | |
| 18 | /// The moment before which a finished job's log is deleted; None keeps |
| 19 | /// every log. |
| 20 | pub fn cutoff(now_ms: u64, days: u64) -> Option<String> { |
| 21 | (days > 0).then(|| rfc3339(now_ms.saturating_sub(days.saturating_mul(DAY_MS)))) |
| 22 | } |
| 23 | |
| 24 | #[derive(Deserialize)] |
| 25 | struct Due { |
| 26 | id: String, |
| 27 | log_id: String, |
| 28 | } |
| 29 | |
| 30 | impl Actions { |
| 31 | pub(crate) async fn sweep_logs(&self, now_ms: u64) -> Result<()> { |
| 32 | let Some(before) = cutoff(now_ms, self.limits.log_retention_days) else { |
| 33 | return Ok(()); |
| 34 | }; |
| 35 | for (table, log_id) in [("jobs", "id"), ("job_attempts", "log_id")] { |
| 36 | let due = self |
| 37 | .db |
| 38 | .prepare(format!( |
| 39 | "SELECT id, {log_id} AS log_id FROM {table} |
| 40 | WHERE logs_purged_at IS NULL AND finished_at IS NOT NULL AND finished_at < ? |
| 41 | LIMIT {PER_SWEEP}" |
| 42 | )) |
| 43 | .bind(&[before.as_str().into()])? |
| 44 | .all() |
| 45 | .await? |
| 46 | .results::<Due>()?; |
| 47 | if due.is_empty() { |
| 48 | continue; |
| 49 | } |
| 50 | let at = rfc3339(now_ms); |
| 51 | let mut statements = Vec::with_capacity(due.len() * 2); |
| 52 | for job in &due { |
| 53 | statements.push(self.db.prepare("DELETE FROM logs WHERE job_id = ?").bind(&[job.log_id.as_str().into()])?); |
| 54 | statements.push( |
| 55 | self.db |
| 56 | .prepare(format!("UPDATE {table} SET logs_purged_at = ? WHERE id = ?")) |
| 57 | .bind(&[at.as_str().into(), job.id.as_str().into()])?, |
| 58 | ); |
| 59 | } |
| 60 | self.db.batch(statements).await?; |
| 61 | } |
| 62 | Ok(()) |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | #[cfg(test)] |
| 67 | mod tests { |
| 68 | use super::*; |
| 69 | |
| 70 | #[test] |
| 71 | fn logs_older_than_the_retention_go_and_zero_keeps_them() { |
| 72 | let now = 100 * DAY_MS; |
| 73 | assert_eq!(cutoff(now, 90), Some(rfc3339(10 * DAY_MS))); |
| 74 | assert_eq!(cutoff(now, 0), None); |
| 75 | // A retention longer than time itself deletes nothing it should keep. |
| 76 | assert_eq!(cutoff(now, 1000), Some(rfc3339(0))); |
| 77 | } |
| 78 | } |