wowlab_sentinel/cron/
mod.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use tokio_cron_scheduler::{Job, JobScheduler};
5use tokio_util::sync::CancellationToken;
6
7use crate::{presence, scheduler::reclaim, state::ServerState, telemetry::UPTIME_SECONDS};
8
9pub(crate) mod docs;
10mod integrity;
11mod queue_depth;
12mod ranking;
13mod retention;
14mod rollup;
15
16crate::db_client!(CronDb, "cron", max = 3);
17
18/// Decode a `ResultCoreV1` from bytes that may be wrapped in a `ResultViewV1`.
19pub(crate) fn decode_result_core(
20    bytes: &[u8],
21) -> Result<wowlab_types::proto::ResultCoreV1, prost::DecodeError> {
22    use prost::Message;
23
24    if let Ok(view) = wowlab_types::proto::ResultViewV1::decode(bytes) {
25        if let Some(core) = view.core {
26            return Ok(core);
27        }
28    }
29
30    wowlab_types::proto::ResultCoreV1::decode(bytes)
31}
32
33#[async_trait]
34pub(crate) trait CronJob: Send + Sync + 'static {
35    fn name(&self) -> &'static str;
36    fn schedule(&self) -> &str;
37    async fn run(&self, state: &ServerState);
38}
39
40struct RecordUptimeJob {
41    schedule: String,
42}
43
44#[async_trait]
45impl CronJob for RecordUptimeJob {
46    fn name(&self) -> &'static str {
47        "record_uptime"
48    }
49
50    fn schedule(&self) -> &str {
51        &self.schedule
52    }
53
54    async fn run(&self, state: &ServerState) {
55        #[expect(
56            clippy::cast_precision_loss,
57            reason = "metrics gauges accept f64 and process uptime is operationally below exact-integer limits"
58        )]
59        let uptime = state.health.started_at.elapsed().as_secs() as f64;
60
61        metrics::gauge!(UPTIME_SECONDS).set(uptime);
62    }
63}
64
65struct CleanupJobsJob {
66    schedule: String,
67}
68
69#[async_trait]
70impl CronJob for CleanupJobsJob {
71    fn name(&self) -> &'static str {
72        "cleanup_jobs"
73    }
74
75    fn schedule(&self) -> &str {
76        &self.schedule
77    }
78
79    async fn run(&self, state: &ServerState) {
80        let days = state.config.cron_cleanup_jobs_days;
81        let result = sqlx::query_file!("queries/cleanup_delete_jobs.sql", days)
82            .execute(state.dbs.get::<CronDb>())
83            .await;
84
85        match result {
86            Ok(r) if r.rows_affected() > 0 => {
87                tracing::info!(
88                    count = r.rows_affected(),
89                    days,
90                    "Cleaned up old completed jobs"
91                );
92            }
93            Ok(_) => {}
94            Err(e) => tracing::error!(error = %e, "Failed to cleanup old jobs"),
95        }
96    }
97}
98
99pub(crate) async fn run(
100    state: Arc<ServerState>,
101    shutdown: CancellationToken,
102) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
103    let mut sched = JobScheduler::new().await?;
104
105    let jobs: Vec<Arc<dyn CronJob>> = vec![
106        Arc::new(reclaim::ReclaimChunksJob::new(&state.config.cron_reclaim)),
107        Arc::new(presence::PresenceJob::new(&state.config.cron_presence)),
108        Arc::new(RecordUptimeJob {
109            schedule: state.config.cron_uptime.clone(),
110        }),
111        Arc::new(CleanupJobsJob {
112            schedule: state.config.cron_cleanup.clone(),
113        }),
114        Arc::new(rollup::RollupDailyJob::new(&state.config.cron_rollup)),
115        Arc::new(rollup::RollupRollingJob::new(&state.config.cron_rollup)),
116        Arc::new(ranking::RankingRefreshJob {
117            schedule: state.config.cron_ranking.clone(),
118            top_n: state.config.ranking_top_n,
119        }),
120        Arc::new(integrity::IntegrityCheckJob::new(
121            &state.config.cron_integrity,
122        )),
123        Arc::new(retention::RetentionCleanupJob {
124            schedule: state.config.cron_retention.clone(),
125            snapshots_retention_days: state.config.retention_snapshots_days,
126            rolling_retention_days: state.config.retention_rolling_days,
127        }),
128        Arc::new(docs::DocsCacheJob::new(&state.config.cron_docs)),
129        Arc::new(queue_depth::QueueDepthJob::new(
130            &state.config.cron_queue_depth,
131        )),
132        Arc::new(crate::scheduler::burst::BurstScheduler::new(
133            &state.config.cron_burst_scheduler,
134        )),
135    ];
136
137    let registrations: Vec<_> = jobs
138        .iter()
139        .map(|job| (job.name(), job.schedule()))
140        .collect();
141
142    tracing::info!(?registrations, "Registering cron jobs");
143
144    for job in jobs {
145        let s = Arc::clone(&state);
146        // #t(rust_alloc_in_loop) schedule string built once per job during startup
147        let schedule = job.schedule().to_string();
148
149        sched
150            .add(Job::new_async(&schedule, move |_, _| {
151                let state = Arc::clone(&s);
152                let job = Arc::clone(&job);
153
154                Box::pin(async move {
155                    job.run(&state).await;
156                    state.touch_cron();
157                })
158            })?)
159            .await?;
160    }
161
162    tracing::info!("Cron scheduler started");
163    sched.start().await?;
164
165    shutdown.cancelled().await;
166    tracing::info!("Cron scheduler shutting down");
167    sched.shutdown().await?;
168
169    Ok(())
170}
wowlab_sentinel/cron/mod.rs

wowlab_sentinel/cron/
mod.rs
