pgtask_core/
schedule.rs

1use std::{num::NonZeroU16, str::FromStr, time::Duration};
2
3use chrono::{DateTime, TimeDelta, Utc};
4use cron::Schedule as CronSchedule;
5use thiserror::Error;
6
7use crate::{EnqueueRequest, ScheduleId, ScheduleName};
8
9#[derive(Clone, Debug, Eq, PartialEq)]
10pub enum ScheduleDefinition {
11    Interval { every: Duration },
12    Cron { expression: String },
13}
14
15impl ScheduleDefinition {
16    pub fn interval(every: Duration) -> Result<Self, ScheduleError> {
17        if every.is_zero() {
18            return Err(ScheduleError::ZeroInterval);
19        }
20        TimeDelta::from_std(every).map_err(|_| ScheduleError::IntervalOutOfRange)?;
21        Ok(Self::Interval { every })
22    }
23
24    pub fn cron(expression: impl Into<String>) -> Result<Self, ScheduleError> {
25        let expression = expression.into();
26        parse_cron(&expression)?;
27        Ok(Self::Cron { expression })
28    }
29
30    pub fn next_after(&self, after: DateTime<Utc>) -> Result<DateTime<Utc>, ScheduleError> {
31        match self {
32            Self::Interval { every } => after
33                .checked_add_signed(TimeDelta::from_std(*every).map_err(|_| ScheduleError::IntervalOutOfRange)?)
34                .ok_or(ScheduleError::DateOutOfRange),
35            Self::Cron { expression } => parse_cron(expression)?
36                .after(&after)
37                .next()
38                .ok_or(ScheduleError::NoFutureOccurrence),
39        }
40    }
41
42    /// Counts occurrences due in `first_due..=now`.
43    ///
44    /// An interval is arithmetic. A cron expression has to be walked, so the walk is capped: a
45    /// schedule that missed more than `DUE_COUNT_LIMIT` occurrences reports the cap rather than
46    /// spending unbounded time inside the materialization transaction.
47    fn due_count(&self, first_due: DateTime<Utc>, now: DateTime<Utc>) -> Result<u64, ScheduleError> {
48        match self {
49            Self::Interval { every } => {
50                let every_milliseconds =
51                    i64::try_from(every.as_millis()).map_err(|_| ScheduleError::IntervalOutOfRange)?;
52                if every_milliseconds == 0 {
53                    return Err(ScheduleError::ZeroInterval);
54                }
55                let elapsed_milliseconds = (now - first_due).num_milliseconds().max(0);
56                Ok(u64::try_from(elapsed_milliseconds / every_milliseconds).unwrap_or(0) + 1)
57            }
58            Self::Cron { .. } => {
59                let mut count = 0_u64;
60                let mut occurrence = first_due;
61                while occurrence <= now && count < DUE_COUNT_LIMIT {
62                    count += 1;
63                    occurrence = self.next_after(occurrence)?;
64                }
65                Ok(count)
66            }
67        }
68    }
69
70    fn latest_due(&self, first_due: DateTime<Utc>, now: DateTime<Utc>) -> Result<DateTime<Utc>, ScheduleError> {
71        match self {
72            Self::Interval { every } => {
73                let every_milliseconds =
74                    i64::try_from(every.as_millis()).map_err(|_| ScheduleError::IntervalOutOfRange)?;
75                let elapsed_milliseconds = (now - first_due).num_milliseconds();
76                let intervals = elapsed_milliseconds / every_milliseconds;
77                first_due
78                    .checked_add_signed(TimeDelta::milliseconds(intervals * every_milliseconds))
79                    .ok_or(ScheduleError::DateOutOfRange)
80            }
81            Self::Cron { expression } => {
82                let schedule = parse_cron(expression)?;
83                let mut low = first_due.timestamp().saturating_sub(1);
84                let mut high = now.timestamp().saturating_add(1);
85                while low + 1 < high {
86                    let middle = low + (high - low) / 2;
87                    let middle = DateTime::from_timestamp(middle, 0).ok_or(ScheduleError::DateOutOfRange)?;
88                    let next = schedule
89                        .after(&middle)
90                        .next()
91                        .ok_or(ScheduleError::NoFutureOccurrence)?;
92                    if next <= now {
93                        low = middle.timestamp();
94                    } else {
95                        high = middle.timestamp();
96                    }
97                }
98                let low = DateTime::from_timestamp(low, 0).ok_or(ScheduleError::DateOutOfRange)?;
99                schedule
100                    .after(&low)
101                    .next()
102                    .filter(|occurrence| *occurrence <= now)
103                    .ok_or(ScheduleError::NoFutureOccurrence)
104            }
105        }
106    }
107
108    pub fn materialize(
109        &self,
110        next_run_at: DateTime<Utc>,
111        now: DateTime<Utc>,
112        policy: MisfirePolicy,
113    ) -> Result<Materialization, ScheduleError> {
114        if next_run_at > now {
115            return Ok(Materialization {
116                occurrences: Vec::new(),
117                next_run_at,
118                skipped: 0,
119            });
120        }
121
122        let occurrences = match policy {
123            MisfirePolicy::Skip => vec![next_run_at],
124            MisfirePolicy::Latest => vec![self.latest_due(next_run_at, now)?],
125            MisfirePolicy::CatchUp { limit } => {
126                let mut occurrences = Vec::with_capacity(usize::from(limit.get()));
127                let mut occurrence = next_run_at;
128                while occurrence <= now && occurrences.len() < usize::from(limit.get()) {
129                    occurrences.push(occurrence);
130                    occurrence = self.next_after(occurrence)?;
131                }
132                occurrences
133            }
134        };
135        let due = self.due_count(next_run_at, now)?;
136        Ok(Materialization {
137            skipped: due.saturating_sub(occurrences.len() as u64),
138            occurrences,
139            next_run_at: self.next_after(now)?,
140        })
141    }
142}
143
144#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
145pub enum MisfirePolicy {
146    Skip,
147    #[default]
148    Latest,
149    CatchUp {
150        limit: NonZeroU16,
151    },
152}
153
154#[derive(Clone, Debug)]
155pub struct ScheduleConfig {
156    pub id: ScheduleId,
157    pub name: ScheduleName,
158    pub definition: ScheduleDefinition,
159    pub misfire_policy: MisfirePolicy,
160    pub task: EnqueueRequest,
161    pub start_at: Option<DateTime<Utc>>,
162}
163
164impl ScheduleConfig {
165    pub fn new(name: ScheduleName, definition: ScheduleDefinition, task: EnqueueRequest) -> Self {
166        Self {
167            id: ScheduleId::new(),
168            name,
169            definition,
170            misfire_policy: MisfirePolicy::default(),
171            task,
172            start_at: None,
173        }
174    }
175}
176
177#[derive(Clone, Debug)]
178pub struct Schedule {
179    pub config: ScheduleConfig,
180    pub next_run_at: DateTime<Utc>,
181    pub paused_at: Option<DateTime<Utc>>,
182    pub created_at: DateTime<Utc>,
183    pub updated_at: DateTime<Utc>,
184}
185
186/// Upper bound on the cron occurrences counted when reporting a missed window.
187const DUE_COUNT_LIMIT: u64 = 10_000;
188
189#[derive(Clone, Debug, Eq, PartialEq)]
190pub struct Materialization {
191    pub occurrences: Vec<DateTime<Utc>>,
192    pub next_run_at: DateTime<Utc>,
193    /// Due occurrences the misfire policy discarded, so a silent gap stays observable.
194    pub skipped: u64,
195}
196
197#[derive(Debug, Error)]
198pub enum ScheduleError {
199    #[error("interval must be greater than zero")]
200    ZeroInterval,
201    #[error("interval exceeds the supported date range")]
202    IntervalOutOfRange,
203    #[error("cron expression must contain exactly six fields: second minute hour day-of-month month day-of-week")]
204    InvalidCronFieldCount,
205    #[error("invalid cron expression: {0}")]
206    InvalidCron(String),
207    #[error("schedule has no future occurrence")]
208    NoFutureOccurrence,
209    #[error("schedule date exceeds the supported range")]
210    DateOutOfRange,
211}
212
213fn parse_cron(expression: &str) -> Result<CronSchedule, ScheduleError> {
214    if expression.split_whitespace().count() != 6 {
215        return Err(ScheduleError::InvalidCronFieldCount);
216    }
217    CronSchedule::from_str(&format!("{expression} *")).map_err(|error| ScheduleError::InvalidCron(error.to_string()))
218}
219
220#[cfg(test)]
221mod tests {
222    use std::{num::NonZeroU16, time::Duration};
223
224    use super::{MisfirePolicy, ScheduleDefinition, ScheduleError};
225    use chrono::{TimeZone, Utc};
226
227    #[test]
228    fn validates_interval_and_six_field_cron_definitions() {
229        assert!(matches!(
230            ScheduleDefinition::interval(Duration::ZERO),
231            Err(ScheduleError::ZeroInterval)
232        ));
233        assert!(matches!(
234            ScheduleDefinition::cron("0 * * * *"),
235            Err(ScheduleError::InvalidCronFieldCount)
236        ));
237        assert!(matches!(
238            ScheduleDefinition::cron("invalid * * * * *"),
239            Err(ScheduleError::InvalidCron(_))
240        ));
241        assert!(matches!(
242            ScheduleDefinition::cron("TZ=Europe/Madrid 0 */5 * * * *"),
243            Err(ScheduleError::InvalidCronFieldCount)
244        ));
245        assert!(ScheduleDefinition::cron("0 */5 * * * *").is_ok());
246    }
247
248    #[test]
249    fn interval_misfire_policies_are_bounded() {
250        let definition = ScheduleDefinition::interval(Duration::from_secs(10)).unwrap();
251        let first = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
252        let now = first + chrono::TimeDelta::seconds(35);
253
254        let skipped = definition.materialize(first, now, MisfirePolicy::Skip).unwrap();
255        assert_eq!(skipped.occurrences, vec![first]);
256        assert_eq!(skipped.next_run_at, first + chrono::TimeDelta::seconds(45));
257
258        let latest = definition.materialize(first, now, MisfirePolicy::Latest).unwrap();
259        assert_eq!(latest.occurrences, vec![first + chrono::TimeDelta::seconds(30)]);
260
261        let caught_up = definition
262            .materialize(
263                first,
264                now,
265                MisfirePolicy::CatchUp {
266                    limit: NonZeroU16::new(2).unwrap(),
267                },
268            )
269            .unwrap();
270        assert_eq!(
271            caught_up.occurrences,
272            vec![first, first + chrono::TimeDelta::seconds(10)]
273        );
274        assert_eq!(caught_up.next_run_at, first + chrono::TimeDelta::seconds(45));
275    }
276
277    #[test]
278    fn cron_latest_finds_the_last_due_occurrence_without_scanning_backlog() {
279        let definition = ScheduleDefinition::cron("0 */5 * * * *").unwrap();
280        let first = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
281        let now = Utc.with_ymd_and_hms(2026, 1, 2, 12, 3, 0).unwrap();
282        let materialized = definition.materialize(first, now, MisfirePolicy::Latest).unwrap();
283        assert_eq!(
284            materialized.occurrences,
285            vec![Utc.with_ymd_and_hms(2026, 1, 2, 12, 0, 0).unwrap()]
286        );
287        assert_eq!(
288            materialized.next_run_at,
289            Utc.with_ymd_and_hms(2026, 1, 2, 12, 5, 0).unwrap()
290        );
291    }
292
293    #[test]
294    fn future_schedule_is_not_materialized() {
295        let definition = ScheduleDefinition::interval(Duration::from_secs(10)).unwrap();
296        let now = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
297        let future = now + chrono::TimeDelta::seconds(10);
298        let materialized = definition.materialize(future, now, MisfirePolicy::Latest).unwrap();
299        assert!(materialized.occurrences.is_empty());
300        assert_eq!(materialized.next_run_at, future);
301    }
302}
pgtask_core/schedule.rs

pgtask_core/
schedule.rs
