cufflink_types/manifest/
schedule.rs

1use super::JobConfig;
2use crate::features;
3use serde::{Deserialize, Serialize};
4
5/// A handler the platform runs on a timetable, once per fire across every
6/// replica.
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
8pub struct ScheduleDefinition {
9    pub name: String,
10    pub handler: String,
11    pub spec: ScheduleSpec,
12    /// The IANA time zone a cron expression is read in; UTC when unset.
13    #[serde(default, skip_serializing_if = "Option::is_none")]
14    pub timezone: Option<String>,
15    #[serde(default, skip_serializing_if = "Overlap::is_skip")]
16    pub overlap: Overlap,
17    #[serde(default, skip_serializing_if = "CatchUp::is_latest")]
18    pub catch_up: CatchUp,
19    /// Run each fire as a job on a worker instead of inline.
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub job: Option<JobConfig>,
22    /// The inline handler's execution budget; 60 s when unset.
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub timeout_secs: Option<u32>,
25}
26
27impl ScheduleDefinition {
28    pub const NAME_MAX: usize = 64;
29    pub const DEFAULT_TIMEOUT_SECS: u32 = 60;
30    /// The range `timeout_secs` must fall in.
31    pub const TIMEOUT_SECS: std::ops::RangeInclusive<u32> = 1..=300;
32    /// The shortest interval between two fires.
33    pub const MIN_INTERVAL_SECS: u32 = 10;
34
35    /// A schedule with the platform's defaults: a handler run inline, in
36    /// UTC, that skips a fire while the last one runs.
37    pub fn new(name: impl Into<String>, handler: impl Into<String>, spec: ScheduleSpec) -> Self {
38        Self {
39            name: name.into(),
40            handler: handler.into(),
41            spec,
42            timezone: None,
43            overlap: Overlap::default(),
44            catch_up: CatchUp::default(),
45            job: None,
46            timeout_secs: None,
47        }
48    }
49
50    /// `^[a-z0-9_]{1,64}$`
51    pub fn is_valid_name(name: &str) -> bool {
52        (1..=Self::NAME_MAX).contains(&name.len())
53            && name
54                .bytes()
55                .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
56    }
57
58    /// Platform features this schedule's declaration requires beyond
59    /// `schedules` itself.
60    pub fn derived_features(&self) -> impl Iterator<Item = &'static str> {
61        self.job
62            .as_ref()
63            .is_some_and(JobConfig::uses_v2)
64            .then_some(features::JOBS_V2)
65            .into_iter()
66    }
67}
68
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
70#[serde(tag = "kind", rename_all = "snake_case")]
71pub enum ScheduleSpec {
72    /// A cron expression of five fields, or six with seconds first.
73    Cron { expr: String },
74    /// Every `secs` seconds, at the Unix-epoch multiples of `secs`.
75    Every { secs: u32 },
76}
77
78/// What a fire does while the previous run is still going.
79#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
80#[serde(rename_all = "snake_case")]
81pub enum Overlap {
82    #[default]
83    Skip,
84    Allow,
85}
86
87impl Overlap {
88    pub fn is_skip(&self) -> bool {
89        *self == Overlap::Skip
90    }
91}
92
93/// Which of the fires missed during an outage still run.
94#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
95#[serde(rename_all = "snake_case")]
96pub enum CatchUp {
97    /// The newest one.
98    #[default]
99    Latest,
100    None,
101    /// The oldest ones, up to the platform's cap.
102    All,
103}
104
105impl CatchUp {
106    pub fn is_latest(&self) -> bool {
107        *self == CatchUp::Latest
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114    use crate::{BackoffSpec, RetryOn};
115    use serde_json::json;
116
117    fn every(secs: u32) -> ScheduleDefinition {
118        ScheduleDefinition {
119            name: "order_retry".into(),
120            handler: "retry_orders".into(),
121            spec: ScheduleSpec::Every { secs },
122            timezone: None,
123            overlap: Overlap::Skip,
124            catch_up: CatchUp::Latest,
125            job: None,
126            timeout_secs: None,
127        }
128    }
129
130    #[test]
131    fn a_minimal_schedule_carries_only_what_it_declares() {
132        let wire = json!({
133            "name": "order_retry",
134            "handler": "retry_orders",
135            "spec": {"kind": "every", "secs": 30}
136        });
137        assert_eq!(serde_json::to_value(every(30)).unwrap(), wire);
138        assert_eq!(
139            serde_json::from_value::<ScheduleDefinition>(wire).unwrap(),
140            every(30)
141        );
142        assert_eq!(every(30).derived_features().count(), 0);
143    }
144
145    #[test]
146    fn every_field_round_trips() {
147        let schedule = ScheduleDefinition {
148            name: "payment_reminders".into(),
149            handler: "run_payment_reminders".into(),
150            spec: ScheduleSpec::Cron {
151                expr: "0 * * * *".into(),
152            },
153            timezone: Some("Europe/London".into()),
154            overlap: Overlap::Allow,
155            catch_up: CatchUp::None,
156            job: Some(JobConfig {
157                timeout_secs: 900,
158                ..Default::default()
159            }),
160            timeout_secs: Some(120),
161        };
162        let wire = json!({
163            "name": "payment_reminders",
164            "handler": "run_payment_reminders",
165            "spec": {"kind": "cron", "expr": "0 * * * *"},
166            "timezone": "Europe/London",
167            "overlap": "allow",
168            "catch_up": "none",
169            "job": {"timeout_secs": 900, "max_attempts": 3},
170            "timeout_secs": 120
171        });
172        assert_eq!(serde_json::to_value(&schedule).unwrap(), wire);
173        assert_eq!(
174            serde_json::from_value::<ScheduleDefinition>(wire).unwrap(),
175            schedule
176        );
177        assert_eq!(serde_json::to_value(CatchUp::All).unwrap(), json!("all"));
178    }
179
180    #[test]
181    fn a_retrying_job_demands_jobs_v2() {
182        let plain = ScheduleDefinition {
183            job: Some(JobConfig::default()),
184            ..every(60)
185        };
186        assert_eq!(plain.derived_features().count(), 0);
187        for job in [
188            JobConfig {
189                retry_on: vec![RetryOn::Error],
190                ..Default::default()
191            },
192            JobConfig {
193                backoff: Some(BackoffSpec::default()),
194                ..Default::default()
195            },
196        ] {
197            let schedule = ScheduleDefinition {
198                job: Some(job),
199                ..every(60)
200            };
201            assert_eq!(
202                schedule.derived_features().collect::<Vec<_>>(),
203                [features::JOBS_V2]
204            );
205        }
206    }
207
208    #[test]
209    fn names_are_lowercase_identifiers() {
210        for valid in ["order_retry", "a", "sync_2", &"s".repeat(64)] {
211            assert!(ScheduleDefinition::is_valid_name(valid), "{valid}");
212        }
213        for invalid in [
214            "",
215            "Order",
216            "order-retry",
217            "order.retry",
218            "a b",
219            &"s".repeat(65),
220        ] {
221            assert!(!ScheduleDefinition::is_valid_name(invalid), "{invalid}");
222        }
223    }
224}
cufflink_types/manifest/schedule.rs

cufflink_types/manifest/
schedule.rs
