cratefield_runtime_native/
cron.rs

1//! In-process cron (issue #19): the native counterpart of a Worker's
2//! `[triggers] crons`. Each expression in `CRONS` gets one tokio task
3//! that sleeps to the next occurrence (UTC) and fans out to every
4//! module's `scheduled(ctx, cron)` — exactly what
5//! `cratefield_runtime_cloudflare::serve_scheduled` does per trigger.
6//!
7//! Expressions are strictly **five-field** (minute hour day month
8//! weekday), the form wrangler accepts; a six-field expression with
9//! leading seconds is rejected rather than silently reinterpreted.
10//! "Now" always comes through the clock (`tokio`/`time`), never
11//! `chrono::Utc::now`: croner's chrono dependency arrives without its
12//! `clock` feature, so that call could not compile here even by
13//! accident (PR #50's lesson, applied structurally).
14
15use std::sync::Arc;
16use std::time::Duration;
17
18use cratefield_core::{Clock, Harness, Ports};
19use tokio::task::JoinHandle;
20
21/// A `CRONS` entry that could not be parsed.
22#[derive(Debug, thiserror::Error)]
23#[error("invalid cron expression {expr:?}: {reason}")]
24pub struct CronError {
25    expr: String,
26    reason: String,
27}
28
29fn parse(expr: &str) -> Result<croner::Cron, CronError> {
30    croner::parser::CronParser::builder()
31        .seconds(croner::parser::Seconds::Disallowed)
32        .build()
33        .parse(expr)
34        .map_err(|err| CronError {
35            expr: expr.to_owned(),
36            reason: err.to_string(),
37        })
38}
39
40/// `time` → the datetime library croner computes with (chrono without
41/// its `clock` feature — types only, never `now`).
42fn to_cron_time(t: time::OffsetDateTime) -> Option<chrono::DateTime<chrono::Utc>> {
43    chrono::DateTime::from_timestamp(t.unix_timestamp(), t.nanosecond().rem_euclid(1_000_000_000))
44}
45
46/// One task per expression: sleep to the next occurrence, fan out,
47/// repeat. Ticks are missed, not queued, while the fan-out runs — the
48/// same posture a Worker has when a scheduled handler overruns its
49/// schedule.
50async fn run_expression(harness: Arc<Harness>, ports: Ports, expr: String, cron: croner::Cron) {
51    loop {
52        let now = to_cron_time(crate::ports::TokioClock.now());
53        let Some(now) = now else {
54            tracing::error!(cron = %expr, "clock out of range; cron task stopping");
55            return;
56        };
57        let next = match cron.find_next_occurrence(&now, false) {
58            Ok(next) => next,
59            Err(err) => {
60                tracing::error!(cron = %expr, error = %err, "cannot compute next tick; stopping");
61                return;
62            }
63        };
64        let until_next = (next - now)
65            .to_std()
66            .unwrap_or_else(|_| Duration::from_millis(1));
67        tokio::time::sleep(until_next).await;
68        fan_out(&harness, &ports, &expr).await;
69    }
70}
71
72/// Fans one trigger out to every module's `scheduled(ctx, cron)` with a
73/// per-tick context. Handler errors are logged and never fail the tick
74/// (or the process) — the semantics of `serve_scheduled` on Workers.
75/// Public so a venture that owns its scheduler (or its tests) can drive
76/// the same fan-out itself.
77pub async fn fan_out(harness: &Harness, ports: &Ports, cron: &str) {
78    for module in harness.modules() {
79        let module_ctx = harness.module_context(module.as_ref(), ports);
80        if let Err(err) = module.scheduled(&module_ctx, cron).await {
81            tracing::error!(
82                module = module.name(),
83                cron = %cron,
84                error = %err,
85                "scheduled module work failed",
86            );
87        }
88    }
89}
90
91/// Validates every expression, then spawns one task per expression.
92/// Validation happens before any task starts so an invalid `CRONS`
93/// entry fails the process at startup (wrangler's deploy-time check),
94/// never as a silently dead schedule.
95///
96/// # Errors
97///
98/// [`CronError`] naming the first expression that does not parse as a
99/// five-field cron.
100pub fn spawn_cron_scheduler(
101    harness: &Arc<Harness>,
102    ports: &Ports,
103    expressions: &[String],
104) -> Result<Vec<JoinHandle<()>>, CronError> {
105    let parsed: Vec<(String, croner::Cron)> = expressions
106        .iter()
107        .map(|expr| parse(expr).map(|cron| (expr.clone(), cron)))
108        .collect::<Result<_, _>>()?;
109    Ok(parsed
110        .into_iter()
111        .map(|(expr, cron)| {
112            tokio::spawn(run_expression(
113                Arc::clone(harness),
114                crate::runtime::clone_ports(ports),
115                expr,
116                cron,
117            ))
118        })
119        .collect())
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125
126    #[test]
127    fn five_field_expressions_parse() {
128        for expr in ["*/5 * * * *", "0 3 * * *", "30 4 1,15 8 *"] {
129            parse(expr).unwrap_or_else(|err| panic!("{expr} should parse: {err}"));
130        }
131    }
132
133    #[test]
134    fn six_field_expressions_are_rejected_not_reinterpreted() {
135        let err = parse("0 */5 * * * *").expect_err("leading seconds must be rejected");
136        assert!(err.to_string().contains("0 */5 * * * *"));
137    }
138
139    #[test]
140    fn plain_garbage_is_rejected() {
141        assert!(parse("not a cron").is_err());
142        assert!(parse("").is_err());
143    }
144}
cratefield_runtime_native/cron.rs

cratefield_runtime_native/
cron.rs
