scheduler/model/
schedule.rs

1use super::{CronSchedule, JobTimeWindow};
2use chrono::{DateTime, Utc};
3use chrono_tz::{Asia::Shanghai, Tz};
4use std::time::Duration;
5
6/// Interval schedule with a stable phase derived from a seed.
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct StaggeredIntervalSchedule {
9    pub every: Duration,
10    pub seed: Option<String>,
11}
12
13impl StaggeredIntervalSchedule {
14    pub fn new(every: Duration) -> Self {
15        Self { every, seed: None }
16    }
17
18    pub fn with_seed(mut self, seed: impl Into<String>) -> Self {
19        self.seed = Some(seed.into());
20        self
21    }
22}
23
24/// Interval schedule that spreads members evenly across the interval.
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub struct GroupedIntervalSchedule {
27    pub every: Duration,
28    pub group_size: u32,
29    pub member_index: u32,
30    pub group_seed: Option<String>,
31}
32
33impl GroupedIntervalSchedule {
34    pub fn new(every: Duration, group_size: u32, member_index: u32) -> Self {
35        Self {
36            every,
37            group_size,
38            member_index,
39            group_seed: None,
40        }
41    }
42
43    pub fn with_group_seed(mut self, group_seed: impl Into<String>) -> Self {
44        self.group_seed = Some(group_seed.into());
45        self
46    }
47}
48
49/// Cron schedule that spreads members evenly inside a post-anchor window.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct GroupedCronSchedule {
52    pub cron: CronSchedule,
53    pub spread: Duration,
54    pub group_size: u32,
55    pub member_index: u32,
56    pub group_seed: Option<String>,
57}
58
59impl GroupedCronSchedule {
60    pub fn new(
61        cron: CronSchedule,
62        spread: Duration,
63        group_size: u32,
64        member_index: u32,
65    ) -> Self {
66        Self {
67            cron,
68            spread,
69            group_size,
70            member_index,
71            group_seed: None,
72        }
73    }
74
75    pub fn with_group_seed(mut self, group_seed: impl Into<String>) -> Self {
76        self.group_seed = Some(group_seed.into());
77        self
78    }
79}
80
81/// Interval override for a matching time window.
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct IntervalWindow {
84    pub window: JobTimeWindow,
85    pub every: Option<Duration>,
86}
87
88impl IntervalWindow {
89    pub fn new(window: JobTimeWindow, every: Option<Duration>) -> Self {
90        Self { window, every }
91    }
92}
93
94/// Interval schedule with per-window frequencies.
95///
96/// `None` means no trigger is produced while that frequency is active.
97/// Windows are evaluated in order; the first matching window wins.
98#[derive(Debug, Clone, PartialEq, Eq)]
99pub struct WindowedIntervalSchedule {
100    pub default_every: Option<Duration>,
101    pub windows: Vec<IntervalWindow>,
102}
103
104impl WindowedIntervalSchedule {
105    pub fn new(default_every: Option<Duration>) -> Self {
106        Self {
107            default_every,
108            windows: Vec::new(),
109        }
110    }
111
112    pub fn with_window(mut self, window: JobTimeWindow, every: Option<Duration>) -> Self {
113        self.windows.push(IntervalWindow::new(window, every));
114        self
115    }
116}
117
118#[derive(Debug, Clone, PartialEq, Eq)]
119pub enum Schedule {
120    /// Trigger repeatedly after the given interval.
121    Interval(Duration),
122    /// Trigger repeatedly after the given interval with a stable phase.
123    ///
124    /// The first run is aligned to a deterministic phase derived from the
125    /// seed. If no seed is provided, the job id is used.
126    StaggeredInterval(StaggeredIntervalSchedule),
127    /// Trigger repeatedly after the given interval with evenly spaced group
128    /// members.
129    ///
130    /// The first run is aligned to a stable slot determined by the member
131    /// index. An optional group seed can rotate the entire group without
132    /// changing the spacing between members.
133    GroupedInterval(GroupedIntervalSchedule),
134    /// Trigger using a cron anchor and spread members inside each anchor
135    /// window.
136    ///
137    /// Each 5-field cron match becomes an anchor. The scheduler assigns each
138    /// member a stable slot inside `spread`, then repeats that slot on every
139    /// subsequent anchor.
140    GroupedCron(GroupedCronSchedule),
141    /// Trigger repeatedly with frequencies selected by local time windows.
142    WindowedInterval(WindowedIntervalSchedule),
143    /// Trigger at the listed wall-clock times.
144    ///
145    /// The list is sorted before execution starts. An empty list is treated as
146    /// a no-op schedule and exits without running.
147    AtTimes(Vec<DateTime<Tz>>),
148    /// Trigger using a standard 5-field cron expression.
149    ///
150    /// The scheduler timezone is used to calculate the next matching wall-clock
151    /// time. The expression itself does not carry timezone information.
152    Cron(CronSchedule),
153}
154
155#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
156pub enum MissedRunPolicy {
157    Skip,
158    #[default]
159    CatchUpOnce,
160    ReplayAll,
161}
162
163#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
164pub enum OverlapPolicy {
165    #[default]
166    Forbid,
167    QueueOne,
168    AllowParallel,
169}
170
171#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
172pub enum TimeWindowAlignment {
173    #[default]
174    Disabled,
175    AlignToNextWindow,
176}
177
178#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
179pub enum TerminalStatePolicy {
180    #[default]
181    Retain,
182    Delete,
183}
184
185#[derive(Debug, Clone)]
186pub struct SchedulerConfig {
187    /// The timezone forwarded to each [`crate::RunContext`] and used to
188    /// evaluate [`Schedule::Cron`] expressions.
189    ///
190    /// This does not rewrite [`Schedule::AtTimes`] values, which already carry
191    /// their own timezone-aware timestamps.
192    pub timezone: Tz,
193    /// The maximum number of [`crate::RunRecord`] items kept in memory.
194    pub history_limit: usize,
195    /// Controls how persisted state is handled once a job reaches a terminal
196    /// state (`next_run_at == None` and no further work is pending).
197    pub terminal_state_policy: TerminalStatePolicy,
198}
199
200impl Default for SchedulerConfig {
201    fn default() -> Self {
202        Self {
203            timezone: Shanghai,
204            history_limit: 32,
205            terminal_state_policy: TerminalStatePolicy::Retain,
206        }
207    }
208}
209
210impl Schedule {
211    /// Create a staggered interval schedule that defaults to the job id as
212    /// the phase seed.
213    pub fn staggered_interval(every: Duration) -> Self {
214        Self::StaggeredInterval(StaggeredIntervalSchedule::new(every))
215    }
216
217    /// Create a staggered interval schedule with an explicit phase seed.
218    pub fn staggered_interval_with_seed(every: Duration, seed: impl Into<String>) -> Self {
219        Self::StaggeredInterval(StaggeredIntervalSchedule::new(every).with_seed(seed))
220    }
221
222    /// Create a grouped interval schedule with evenly spaced members.
223    pub fn grouped_interval(every: Duration, group_size: u32, member_index: u32) -> Self {
224        Self::GroupedInterval(GroupedIntervalSchedule::new(
225            every,
226            group_size,
227            member_index,
228        ))
229    }
230
231    /// Create a grouped interval schedule and rotate the whole group with an
232    /// explicit seed.
233    pub fn grouped_interval_with_seed(
234        every: Duration,
235        group_size: u32,
236        member_index: u32,
237        group_seed: impl Into<String>,
238    ) -> Self {
239        Self::GroupedInterval(
240            GroupedIntervalSchedule::new(every, group_size, member_index)
241                .with_group_seed(group_seed),
242        )
243    }
244
245    /// Create a grouped cron schedule with evenly spaced members.
246    pub fn grouped_cron(
247        cron: CronSchedule,
248        spread: Duration,
249        group_size: u32,
250        member_index: u32,
251    ) -> Self {
252        Self::GroupedCron(GroupedCronSchedule::new(
253            cron,
254            spread,
255            group_size,
256            member_index,
257        ))
258    }
259
260    /// Create a grouped cron schedule and rotate the whole group with an
261    /// explicit seed.
262    pub fn grouped_cron_with_seed(
263        cron: CronSchedule,
264        spread: Duration,
265        group_size: u32,
266        member_index: u32,
267        group_seed: impl Into<String>,
268    ) -> Self {
269        Self::GroupedCron(
270            GroupedCronSchedule::new(cron, spread, group_size, member_index)
271                .with_group_seed(group_seed),
272        )
273    }
274
275    /// Create a windowed interval schedule.
276    pub fn windowed_interval(default_every: Option<Duration>) -> Self {
277        Self::WindowedInterval(WindowedIntervalSchedule::new(default_every))
278    }
279}
280
281pub(crate) fn utc_time(value: DateTime<Tz>) -> DateTime<Utc> {
282    value.with_timezone(&Utc)
283}
scheduler/model/schedule.rs

scheduler/model/
schedule.rs
