cja/cron/
worker.rs

1use std::{collections::HashMap, time::Duration};
2
3use chrono::{DateTime, Utc};
4use chrono_tz::Tz;
5use tokio_util::sync::CancellationToken;
6
7use crate::app_state::AppState as AS;
8
9use super::registry::{CronRegistry, TickError};
10
11/// Worker that executes cron jobs on a schedule
12///
13/// The worker runs cron jobs based on their configured schedules. For cron expressions,
14/// the timezone parameter determines when the cron expression is evaluated. For example,
15/// a cron expression "0 9 * * *" (9 AM daily) will run at 9 AM in the configured timezone.
16///
17/// Interval-based jobs ignore the timezone and run based on elapsed time since last run.
18pub struct Worker<AppState: AS> {
19    id: uuid::Uuid,
20    state: AppState,
21    registry: CronRegistry<AppState>,
22    pub(crate) started_at: DateTime<Utc>,
23    timezone: Tz,
24    sleep_duration: Duration,
25}
26
27impl<AppState: AS> Worker<AppState> {
28    /// Create a new Worker with UTC as the default timezone
29    pub fn new(state: AppState, registry: CronRegistry<AppState>) -> Self {
30        Self::new_with_timezone(state, registry, chrono_tz::UTC, Duration::from_mins(1))
31    }
32
33    /// Create a new Worker with a specific timezone
34    pub fn new_with_timezone(
35        state: AppState,
36        registry: CronRegistry<AppState>,
37        timezone: Tz,
38        sleep_duration: Duration,
39    ) -> Self {
40        Self {
41            id: uuid::Uuid::new_v4(),
42            state,
43            registry,
44            started_at: Utc::now(),
45            timezone,
46            sleep_duration,
47        }
48    }
49
50    /// Run the cron worker until the shutdown token is cancelled.
51    ///
52    /// The worker will execute cron jobs on their configured schedules until shutdown
53    /// is requested. When the `shutdown_token` is cancelled:
54    /// - The worker stops scheduling new cron job executions
55    /// - Any currently running tick completes before shutdown
56    /// - The sleep between ticks can be interrupted for faster shutdown
57    ///
58    /// # Arguments
59    ///
60    /// * `shutdown_token` - Cancellation token for graceful shutdown
61    ///
62    /// # Example
63    ///
64    /// ```ignore
65    /// use tokio_util::sync::CancellationToken;
66    ///
67    /// let shutdown_token = CancellationToken::new();
68    /// let worker_token = shutdown_token.clone();
69    ///
70    /// tokio::spawn(async move {
71    ///     worker.run(worker_token).await.unwrap();
72    /// });
73    ///
74    /// // Later, trigger shutdown
75    /// shutdown_token.cancel();
76    /// ```
77    pub async fn run(self, shutdown_token: CancellationToken) -> Result<(), TickError> {
78        tracing::debug!(cron_worker_id = %self.id, "Starting Cron loop");
79        while !shutdown_token.is_cancelled() {
80            // Dropping tick on cancellation can interrupt a direct callback
81            // after its side effect but before last_run_at is persisted.
82            // Finish the current tick, including its error or completion write.
83            self.tick().await?;
84
85            tokio::select! {
86                () = tokio::time::sleep(self.sleep_duration) => {}
87                () = shutdown_token.cancelled() => {
88                    tracing::info!(cron_worker_id = %self.id, "Cron worker shutdown during sleep");
89                    break;
90                }
91            }
92        }
93
94        tracing::info!(cron_worker_id = %self.id, "Cron worker shutdown complete");
95        Ok(())
96    }
97
98    async fn last_enqueue_map(&self) -> Result<HashMap<String, DateTime<Utc>>, TickError> {
99        let last_runs = sqlx::query!("SELECT name, last_run_at FROM Crons")
100            .fetch_all(self.state.db())
101            .await
102            .map_err(TickError::SqlxError)?;
103
104        let last_run_map: HashMap<String, DateTime<Utc>> = last_runs
105            .iter()
106            .map(|row| (row.name.clone(), row.last_run_at))
107            .collect();
108
109        Ok(last_run_map)
110    }
111
112    #[tracing::instrument(name = "cron.tick", level = "trace", skip_all, fields(cron_worker.id = %self.id))]
113    pub(crate) async fn tick(&self) -> Result<(), TickError> {
114        let last_enqueue_map = self.last_enqueue_map().await?;
115        for job in self.registry.jobs.values() {
116            job.tick(
117                self.state.clone(),
118                &last_enqueue_map,
119                self.started_at,
120                self.timezone,
121            )
122            .await?;
123        }
124
125        Ok(())
126    }
127}
cja/cron/worker.rs

cja/cron/
worker.rs
