laterite_core/schedule/
mod.rs

1//! The scheduler: entries that run on a cron expression, evaluated every
2//! minute in their own timezone, each on one process at a time.
3//!
4//! An entry runs a closure or pushes a job. A module registers its entries;
5//! the ticker in the application process (or `schedule run` alone) checks
6//! them each minute, takes a lock per entry and minute so one deployment
7//! runs an entry once however many processes it has, skips an entry still
8//! running from last time, and records each run for the screen.
9//!
10//! ```ignore
11//! use laterite_core::schedule::{ScheduleEntry, ScheduleRegistry};
12//!
13//! fn register(&self, registry: &mut Registry) {
14//!     registry.add_schedule(
15//!         ScheduleEntry::new("acme.blog::digest", "0 7 * * *")
16//!             .timezone("Europe/Berlin")
17//!             .job(SendDigest { user_id: 0 }),
18//!     );
19//!     registry.add_schedule(
20//!         ScheduleEntry::new("acme.blog::sweep", "*/5 * * * *")
21//!             .run(|cx| async move { sweep(cx.db()).await }),
22//!     );
23//! }
24//! ```
25
26use std::collections::HashMap;
27use std::future::Future;
28use std::pin::Pin;
29use std::sync::Arc;
30use std::time::Duration;
31
32use crate::cache::Cache;
33use crate::config::ScheduleConfig;
34use crate::error::{CoreError, CoreResult};
35use crate::events::Events;
36use crate::parameters::Parameters;
37use crate::queue::{Job, Queue};
38use crate::registry::Registry;
39use crate::{Db, Module, ModuleId};
40use chrono::{DateTime, Timelike, Utc};
41use chrono_tz::Tz;
42use croner::Cron;
43
44pub mod migrations;
45mod store;
46
47pub use store::{EntryState, RunStatus};
48
49/// The module that owns the `schedule_runs` table, listed among an
50/// application's built-ins.
51pub struct ScheduleModule;
52
53impl Module for ScheduleModule {
54    fn id(&self) -> ModuleId {
55        ModuleId::new(migrations::MODULE_ID)
56    }
57    fn migrations(&self) -> crate::MigrationSet {
58        migrations::migrations()
59    }
60}
61
62/// What a running entry sees.
63#[derive(Clone)]
64pub struct ScheduleCx {
65    db: Db,
66    events: Events,
67    cache: Cache,
68    parameters: Parameters,
69    queue: Queue,
70}
71
72impl ScheduleCx {
73    pub fn db(&self) -> &Db {
74        &self.db
75    }
76    pub fn events(&self) -> &Events {
77        &self.events
78    }
79    pub fn cache(&self) -> &Cache {
80        &self.cache
81    }
82    pub fn parameters(&self) -> &Parameters {
83        &self.parameters
84    }
85    pub fn queue(&self) -> &Queue {
86        &self.queue
87    }
88}
89
90/// What happens when an entry is due while its last run is still going.
91#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
92pub enum Overlap {
93    /// Let this minute's run go: the entry runs once at a time.
94    #[default]
95    Skip,
96    /// Run it beside the one still going.
97    Allow,
98}
99
100type RunFn =
101    dyn Fn(ScheduleCx) -> Pin<Box<dyn Future<Output = CoreResult<()>> + Send>> + Send + Sync;
102
103#[derive(Clone)]
104enum Action {
105    Run(Arc<RunFn>),
106    Job {
107        name: &'static str,
108        payload: serde_json::Value,
109        queue: String,
110    },
111    /// Declared with neither `.run` nor `.job`; refused at boot.
112    None,
113}
114
115/// One scheduled entry: a name, a cron expression, a timezone and what to do.
116#[derive(Clone)]
117pub struct ScheduleEntry {
118    name: String,
119    expression: String,
120    timezone: String,
121    description: String,
122    action: Action,
123    overlap: Overlap,
124    one_server: bool,
125}
126
127impl std::fmt::Debug for ScheduleEntry {
128    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
129        f.debug_struct("ScheduleEntry")
130            .field("name", &self.name)
131            .field("expression", &self.expression)
132            .field("timezone", &self.timezone)
133            .finish_non_exhaustive()
134    }
135}
136
137impl ScheduleEntry {
138    /// An entry named `name` (`acme.blog::digest`) on `expression`: five
139    /// fields (`0 7 * * *`), six with seconds, or a preset (`@hourly`,
140    /// `@daily`, `@weekly`, `@monthly`, `@yearly`). A bad expression is
141    /// refused when the application boots.
142    pub fn new(name: impl Into<String>, expression: impl Into<String>) -> Self {
143        Self {
144            name: name.into(),
145            expression: expression.into(),
146            timezone: "UTC".to_string(),
147            description: String::new(),
148            action: Action::None,
149            overlap: Overlap::Skip,
150            one_server: true,
151        }
152    }
153
154    /// Every `n` minutes.
155    pub fn every_minutes(name: impl Into<String>, n: u32) -> Self {
156        Self::new(name, format!("*/{} * * * *", n.max(1)))
157    }
158
159    /// Every hour, on the hour.
160    pub fn hourly(name: impl Into<String>) -> Self {
161        Self::new(name, "0 * * * *")
162    }
163
164    /// Every day at `HH:MM` in the entry's timezone.
165    pub fn daily_at(name: impl Into<String>, time: &str) -> Self {
166        let (hour, minute) = time.split_once(':').unwrap_or((time, "0"));
167        let hour = hour.trim().parse::<u32>().unwrap_or(0);
168        let minute = minute.trim().parse::<u32>().unwrap_or(0);
169        Self::new(name, format!("{minute} {hour} * * *"))
170    }
171
172    /// The timezone the expression is read in; `UTC` unless set.
173    pub fn timezone(mut self, tz: impl Into<String>) -> Self {
174        self.timezone = tz.into();
175        self
176    }
177
178    /// A line for the screen.
179    pub fn description(mut self, description: impl Into<String>) -> Self {
180        self.description = description.into();
181        self
182    }
183
184    /// Runs `f` when due.
185    pub fn run<F, Fut>(mut self, f: F) -> Self
186    where
187        F: Fn(ScheduleCx) -> Fut + Send + Sync + 'static,
188        Fut: Future<Output = CoreResult<()>> + Send + 'static,
189    {
190        self.action = Action::Run(Arc::new(move |cx| Box::pin(f(cx))));
191        self
192    }
193
194    /// Pushes `job` when due.
195    pub fn job<T: Job>(mut self, job: T) -> Self {
196        let payload = serde_json::to_value(&job).unwrap_or(serde_json::Value::Null);
197        self.action = Action::Job {
198            name: T::NAME,
199            payload,
200            queue: job.queue(),
201        };
202        self
203    }
204
205    pub fn overlap(mut self, overlap: Overlap) -> Self {
206        self.overlap = overlap;
207        self
208    }
209
210    /// Whether one process of the deployment runs it per minute (`true`
211    /// unless set) or every process does.
212    pub fn one_server(mut self, one: bool) -> Self {
213        self.one_server = one;
214        self
215    }
216
217    pub fn name(&self) -> &str {
218        &self.name
219    }
220
221    pub fn expression(&self) -> &str {
222        &self.expression
223    }
224}
225
226/// `registry.add_schedule(entry)`: the typed way to contribute an entry.
227pub trait ScheduleRegistry {
228    fn add_schedule(&mut self, entry: ScheduleEntry);
229}
230
231impl ScheduleRegistry for Registry {
232    fn add_schedule(&mut self, entry: ScheduleEntry) {
233        self.add(entry);
234    }
235}
236
237/// An entry parsed and ready.
238struct Prepared {
239    entry: ScheduleEntry,
240    cron: Cron,
241    tz: Tz,
242}
243
244impl Prepared {
245    fn new(entry: ScheduleEntry) -> CoreResult<Self> {
246        let cron: Cron = entry.expression.parse().map_err(|e| {
247            CoreError::Config(format!(
248                "schedule entry `{}`: expression `{}` does not parse: {e}",
249                entry.name, entry.expression
250            ))
251        })?;
252        let tz: Tz = entry.timezone.parse().map_err(|_| {
253            CoreError::Config(format!(
254                "schedule entry `{}`: timezone `{}` is unknown",
255                entry.name, entry.timezone
256            ))
257        })?;
258        if matches!(entry.action, Action::None) {
259            return Err(CoreError::Config(format!(
260                "schedule entry `{}` has nothing to do: give it `.run(..)` or `.job(..)`",
261                entry.name
262            )));
263        }
264        Ok(Self { entry, cron, tz })
265    }
266
267    fn is_due(&self, minute: DateTime<Utc>) -> bool {
268        let local = minute.with_timezone(&self.tz);
269        self.cron.is_time_matching(&local).unwrap_or(false)
270    }
271
272    fn next_after(&self, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
273        let local = now.with_timezone(&self.tz);
274        self.cron
275            .find_next_occurrence(&local, false)
276            .ok()
277            .map(|t| t.with_timezone(&Utc))
278    }
279}
280
281struct Inner {
282    entries: Vec<Prepared>,
283    by_name: HashMap<String, usize>,
284    db: Db,
285    cx: ScheduleCx,
286}
287
288/// The schedule handle. Cheap to clone: every clone is the same schedule.
289#[derive(Clone)]
290pub struct Schedule {
291    inner: Arc<Inner>,
292}
293
294impl std::fmt::Debug for Schedule {
295    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
296        f.debug_struct("Schedule")
297            .field("entries", &self.inner.entries.len())
298            .finish_non_exhaustive()
299    }
300}
301
302/// The services an entry runs against.
303#[derive(Clone)]
304pub struct Services {
305    pub db: Db,
306    pub events: Events,
307    pub cache: Cache,
308    pub parameters: Parameters,
309    pub queue: Queue,
310}
311
312/// How long the one-server lock on a minute is kept.
313const MINUTE_LOCK: Duration = Duration::from_secs(3600);
314/// How long an entry's overlap lock may be held by a run that never came back.
315const OVERLAP_LOCK: Duration = Duration::from_secs(24 * 3600);
316
317/// How one tick went: what ran, what was skipped.
318#[derive(Debug, Clone, Default, PartialEq, Eq)]
319pub struct Tick {
320    pub ran: Vec<String>,
321    pub skipped: Vec<String>,
322}
323
324impl Schedule {
325    /// A schedule over `entries`; a bad expression, timezone or duplicate
326    /// name is an error.
327    pub fn new(
328        entries: impl IntoIterator<Item = ScheduleEntry>,
329        services: Services,
330    ) -> CoreResult<Self> {
331        let mut prepared = Vec::new();
332        let mut by_name = HashMap::new();
333        for entry in entries {
334            if by_name.contains_key(&entry.name) {
335                return Err(CoreError::Config(format!(
336                    "schedule entry `{}` is registered twice",
337                    entry.name
338                )));
339            }
340            by_name.insert(entry.name.clone(), prepared.len());
341            prepared.push(Prepared::new(entry)?);
342        }
343        Ok(Self {
344            inner: Arc::new(Inner {
345                entries: prepared,
346                by_name,
347                db: services.db.clone(),
348                cx: ScheduleCx {
349                    db: services.db,
350                    events: services.events,
351                    cache: services.cache,
352                    parameters: services.parameters,
353                    queue: services.queue,
354                },
355            }),
356        })
357    }
358
359    /// The entries' names, in registration order.
360    pub fn names(&self) -> Vec<&str> {
361        self.inner
362            .entries
363            .iter()
364            .map(|p| p.entry.name.as_str())
365            .collect()
366    }
367
368    pub fn len(&self) -> usize {
369        self.inner.entries.len()
370    }
371
372    pub fn is_empty(&self) -> bool {
373        self.inner.entries.is_empty()
374    }
375
376    /// Every entry with its recorded state, in registration order.
377    pub async fn list(&self) -> CoreResult<Vec<EntryState>> {
378        let recorded = store::all(&self.inner.db).await?;
379        let now = Utc::now();
380        Ok(self
381            .inner
382            .entries
383            .iter()
384            .map(|p| {
385                let mut state = recorded
386                    .get(&p.entry.name)
387                    .cloned()
388                    .unwrap_or_else(|| EntryState::fresh(&p.entry.name));
389                state.expression = p.entry.expression.clone();
390                state.timezone = p.entry.timezone.clone();
391                state.description = p.entry.description.clone();
392                state.next_run_at = p.next_after(now).map(crate::queue::time_text);
393                state
394            })
395            .collect())
396    }
397
398    /// Runs every entry due at `now`'s minute. Each runs on its own task;
399    /// the tick returns once they are started, naming what ran and what was
400    /// skipped (held by another process, or still running).
401    pub async fn tick(&self, now: DateTime<Utc>) -> CoreResult<Tick> {
402        let minute = now
403            .with_second(0)
404            .and_then(|t| t.with_nanosecond(0))
405            .unwrap_or(now);
406        let mut tick = Tick::default();
407        for (at, prepared) in self.inner.entries.iter().enumerate() {
408            if !prepared.is_due(minute) {
409                continue;
410            }
411            match self.start(at, minute).await? {
412                true => tick.ran.push(prepared.entry.name.clone()),
413                false => tick.skipped.push(prepared.entry.name.clone()),
414            }
415        }
416        Ok(tick)
417    }
418
419    /// Runs one entry now, due or not, under the same locks as a tick, and
420    /// waits for it. `false` when it was skipped (still running elsewhere).
421    pub async fn run_now(&self, name: &str) -> CoreResult<bool> {
422        let Some(&at) = self.inner.by_name.get(name) else {
423            return Err(CoreError::NotFound(format!("schedule entry `{name}`")));
424        };
425        match self.spawn(at, Utc::now()).await? {
426            Some(handle) => {
427                let _ = handle.await;
428                Ok(true)
429            }
430            None => Ok(false),
431        }
432    }
433
434    async fn start(&self, at: usize, minute: DateTime<Utc>) -> CoreResult<bool> {
435        Ok(self.spawn(at, minute).await?.is_some())
436    }
437
438    /// Takes the locks and starts the entry on its own task; `None` when a
439    /// lock was held elsewhere.
440    async fn spawn(
441        &self,
442        at: usize,
443        minute: DateTime<Utc>,
444    ) -> CoreResult<Option<tokio::task::JoinHandle<()>>> {
445        let prepared = &self.inner.entries[at];
446        let name = prepared.entry.name.clone();
447        let cache = &self.inner.cx.cache;
448        if prepared.entry.one_server {
449            let key = format!("laterite.schedule::{name}@{}", minute.format("%Y%m%d%H%M"));
450            if cache.lock(&key, MINUTE_LOCK).await?.is_none() {
451                tracing::debug!(entry = %name, "scheduled entry taken by another process");
452                return Ok(None);
453            }
454        }
455        let overlap = if prepared.entry.overlap == Overlap::Skip {
456            match cache
457                .lock(&format!("laterite.schedule::{name}"), OVERLAP_LOCK)
458                .await?
459            {
460                Some(lock) => Some(lock),
461                None => {
462                    tracing::warn!(entry = %name, "scheduled entry still running; skipped");
463                    store::skipped(&self.inner.db, &name).await?;
464                    return Ok(None);
465                }
466            }
467        } else {
468            None
469        };
470        store::started(&self.inner.db, &name).await?;
471        let schedule = self.clone();
472        Ok(Some(tokio::spawn(async move {
473            let outcome = schedule.execute(at).await;
474            let db = &schedule.inner.db;
475            match &outcome {
476                Ok(()) => {
477                    tracing::info!(entry = %name, "scheduled entry ran");
478                    if let Err(e) = store::finished(db, &name, None).await {
479                        tracing::error!(entry = %name, error = %e, "run could not be recorded");
480                    }
481                }
482                Err(e) => {
483                    tracing::error!(entry = %name, error = %e, "scheduled entry failed");
484                    if let Err(e2) = store::finished(db, &name, Some(&e.to_string())).await {
485                        tracing::error!(entry = %name, error = %e2, "failure could not be recorded");
486                    }
487                }
488            }
489            if let Some(lock) = overlap {
490                let _ = lock.release().await;
491            }
492        })))
493    }
494
495    async fn execute(&self, at: usize) -> CoreResult<()> {
496        let prepared = &self.inner.entries[at];
497        match &prepared.entry.action {
498            Action::Run(f) => f(self.inner.cx.clone()).await,
499            Action::Job {
500                name,
501                payload,
502                queue,
503            } => self
504                .inner
505                .cx
506                .queue
507                .push_named(name, payload, queue)
508                .await
509                .map(|_| ()),
510            Action::None => Ok(()),
511        }
512    }
513
514    /// Ticks at every minute until `stop` says so.
515    pub async fn run_ticker(&self, mut stop: tokio::sync::watch::Receiver<bool>) {
516        loop {
517            let now = Utc::now();
518            let next_minute = (now + chrono::Duration::seconds(60))
519                .with_second(0)
520                .and_then(|t| t.with_nanosecond(0))
521                .unwrap_or(now);
522            let wait = (next_minute - now)
523                .to_std()
524                .unwrap_or(Duration::from_secs(1));
525            tokio::select! {
526                _ = tokio::time::sleep(wait) => {}
527                _ = stop.changed() => return,
528            }
529            if *stop.borrow() {
530                return;
531            }
532            match self.tick(Utc::now()).await {
533                Ok(tick) if !tick.ran.is_empty() || !tick.skipped.is_empty() => {
534                    tracing::debug!(ran = ?tick.ran, skipped = ?tick.skipped, "schedule tick");
535                }
536                Ok(_) => {}
537                Err(e) => tracing::error!(error = %e, "schedule tick failed"),
538            }
539        }
540    }
541
542    /// Starts the ticker on its own task.
543    pub fn spawn_ticker(
544        &self,
545        stop: tokio::sync::watch::Receiver<bool>,
546    ) -> tokio::task::JoinHandle<()> {
547        let schedule = self.clone();
548        tokio::spawn(async move { schedule.run_ticker(stop).await })
549    }
550}
551
552/// Every entry the table records, sorted by name: what a tool outside the
553/// application can see of the schedule.
554pub async fn recorded(db: &Db) -> CoreResult<Vec<EntryState>> {
555    let mut rows: Vec<EntryState> = store::all(db).await?.into_values().collect();
556    rows.sort_by(|a, b| a.name.cmp(&b.name));
557    Ok(rows)
558}
559
560/// `ticker`, `external` or `off`, as `[schedule] mode` names them.
561#[derive(Debug, Clone, Copy, PartialEq, Eq)]
562pub enum Mode {
563    Ticker,
564    External,
565    Off,
566}
567
568impl Mode {
569    pub fn parse(s: &str) -> CoreResult<Self> {
570        match s {
571            "ticker" => Ok(Self::Ticker),
572            "external" => Ok(Self::External),
573            "off" => Ok(Self::Off),
574            other => Err(CoreError::Config(format!(
575                "schedule.mode `{other}` is not one of ticker, external, off"
576            ))),
577        }
578    }
579}
580
581/// The mode `[schedule]` configures.
582pub fn mode_from_config(config: &ScheduleConfig) -> CoreResult<Mode> {
583    Mode::parse(&config.mode)
584}
585
586#[cfg(test)]
587mod tests {
588    use super::*;
589
590    #[test]
591    fn sugar_builds_the_expected_expressions() {
592        assert_eq!(
593            ScheduleEntry::every_minutes("a", 5).expression(),
594            "*/5 * * * *"
595        );
596        assert_eq!(ScheduleEntry::hourly("a").expression(), "0 * * * *");
597        assert_eq!(
598            ScheduleEntry::daily_at("a", "07:30").expression(),
599            "30 7 * * *"
600        );
601        assert_eq!(ScheduleEntry::daily_at("a", "7").expression(), "0 7 * * *");
602    }
603
604    #[test]
605    fn an_entry_is_due_in_its_own_timezone() {
606        let entry = ScheduleEntry::new("a", "0 7 * * *")
607            .timezone("Asia/Colombo")
608            .run(|_| async { Ok(()) });
609        let prepared = Prepared::new(entry).unwrap();
610        // 07:00 in Colombo is 01:30 UTC.
611        let due = "2026-10-24T01:30:00Z".parse::<DateTime<Utc>>().unwrap();
612        assert!(prepared.is_due(due));
613        let not = "2026-10-24T07:00:00Z".parse::<DateTime<Utc>>().unwrap();
614        assert!(!prepared.is_due(not));
615        let next = prepared.next_after(not).unwrap();
616        assert_eq!(next.to_rfc3339(), "2026-10-25T01:30:00+00:00");
617    }
618
619    #[test]
620    fn a_bad_expression_timezone_or_empty_entry_is_refused() {
621        assert!(Prepared::new(ScheduleEntry::new("a", "nope").run(|_| async { Ok(()) })).is_err());
622        assert!(Prepared::new(
623            ScheduleEntry::hourly("a")
624                .timezone("Mars/Olympus")
625                .run(|_| async { Ok(()) })
626        )
627        .is_err());
628        assert!(Prepared::new(ScheduleEntry::hourly("a")).is_err());
629        assert_eq!(Mode::parse("off").unwrap(), Mode::Off);
630    }
631}
laterite_core/schedule/mod.rs

laterite_core/schedule/
mod.rs
