af_workflow/
ingress.rs

1//! Pure ingress planning helpers for durable workflow hosts.
2
3use chrono::{DateTime, Utc};
4
5use crate::{CatchUpPolicy, NodeRegistry, ScheduleCadence, SchedulePolicy, Spec};
6
7/// Result of [`schedule_decision`]: whether an overdue schedule runs and where
8/// its durable cursor moves.
9#[derive(Debug, Clone, Copy, PartialEq, Eq)]
10pub struct ScheduleDecision {
11    /// Occurrence to run now; `None` when the policy skips the missed window.
12    pub tick_at: Option<DateTime<Utc>>,
13    /// Next occurrence to persist as the schedule cursor.
14    pub next_at: DateTime<Utc>,
15    /// Catch-up ticks completed so far in the current `CatchUpAll` burst.
16    pub catch_up_count: u32,
17}
18
19/// Next UTC occurrence of a cron expression after `after`.
20pub fn next_cron_at(expression: &str, after: DateTime<Utc>) -> Result<DateTime<Utc>, String> {
21    expression
22        .parse::<cron::Schedule>()
23        .map_err(|error| format!("invalid cron '{expression}': {error}"))?
24        .after(&after)
25        .next()
26        .ok_or_else(|| format!("cron '{expression}' has no future occurrence"))
27}
28
29/// Next occurrence of a schedule after `after`; fixed-delay cadences measure from `last_completed_at`.
30pub fn next_scheduled_at(
31    policy: &SchedulePolicy,
32    after: DateTime<Utc>,
33    last_completed_at: Option<DateTime<Utc>>,
34) -> Result<DateTime<Utc>, String> {
35    policy.validate().map_err(|error| error.to_string())?;
36    match &policy.cadence {
37        ScheduleCadence::Cron { expression } => {
38            let timezone = policy
39                .timezone
40                .parse::<chrono_tz::Tz>()
41                .map_err(|_| format!("unknown IANA timezone '{}'", policy.timezone))?;
42            expression
43                .parse::<cron::Schedule>()
44                .map_err(|error| format!("invalid cron '{expression}': {error}"))?
45                .after(&after.with_timezone(&timezone))
46                .next()
47                .map(|next| next.with_timezone(&Utc))
48                .ok_or_else(|| format!("cron '{expression}' has no future occurrence"))
49        }
50        ScheduleCadence::FixedRate { milliseconds } => after
51            .checked_add_signed(chrono::Duration::milliseconds(
52                i64::try_from(*milliseconds).map_err(|_| "fixed rate exceeds i64")?,
53            ))
54            .ok_or_else(|| "fixed rate overflows time".into()),
55        ScheduleCadence::FixedDelay { milliseconds } => last_completed_at
56            .unwrap_or(after)
57            .checked_add_signed(chrono::Duration::milliseconds(
58                i64::try_from(*milliseconds).map_err(|_| "fixed delay exceeds i64")?,
59            ))
60            .ok_or_else(|| "fixed delay overflows time".into()),
61    }
62}
63
64fn first_after_now(
65    policy: &SchedulePolicy,
66    scheduled_at: DateTime<Utc>,
67    now: DateTime<Utc>,
68) -> Result<DateTime<Utc>, String> {
69    match &policy.cadence {
70        ScheduleCadence::FixedRate { milliseconds } => {
71            let period = i64::try_from(*milliseconds).map_err(|_| "fixed rate exceeds i64")?;
72            let overdue = now
73                .signed_duration_since(scheduled_at)
74                .num_milliseconds()
75                .max(0);
76            let periods = overdue / period + 1;
77            scheduled_at
78                .checked_add_signed(chrono::Duration::milliseconds(
79                    period.saturating_mul(periods),
80                ))
81                .ok_or_else(|| "fixed rate overflows time".into())
82        }
83        ScheduleCadence::FixedDelay { .. } => next_scheduled_at(policy, now, Some(now)),
84        ScheduleCadence::Cron { .. } => next_scheduled_at(policy, now, None),
85    }
86}
87
88/// Decide whether an overdue schedule runs and where its durable cursor moves.
89/// A run is considered missed only when at least one later occurrence is also
90/// due, avoiding false skips from ordinary scheduler latency.
91pub fn schedule_decision(
92    policy: &SchedulePolicy,
93    scheduled_at: DateTime<Utc>,
94    now: DateTime<Utc>,
95    catch_up_count: u32,
96) -> Result<ScheduleDecision, String> {
97    let next_due = next_scheduled_at(policy, scheduled_at, Some(now))?;
98    let has_missed_window =
99        !matches!(&policy.cadence, ScheduleCadence::FixedDelay { .. }) && next_due <= now;
100    if !has_missed_window {
101        return Ok(ScheduleDecision {
102            tick_at: Some(scheduled_at),
103            next_at: next_due,
104            catch_up_count: 0,
105        });
106    }
107    match policy.catch_up {
108        CatchUpPolicy::Skip => Ok(ScheduleDecision {
109            tick_at: None,
110            next_at: first_after_now(policy, scheduled_at, now)?,
111            catch_up_count: 0,
112        }),
113        CatchUpPolicy::CatchUpOnce => Ok(ScheduleDecision {
114            tick_at: Some(scheduled_at),
115            next_at: first_after_now(policy, scheduled_at, now)?,
116            catch_up_count: 0,
117        }),
118        CatchUpPolicy::CatchUpAll { limit } => {
119            let completed = catch_up_count.saturating_add(1);
120            let keep_catching_up = completed < limit && next_due <= now;
121            Ok(ScheduleDecision {
122                tick_at: Some(scheduled_at),
123                next_at: if keep_catching_up {
124                    next_due
125                } else {
126                    first_after_now(policy, scheduled_at, now)?
127                },
128                catch_up_count: u32::from(keep_catching_up) * completed,
129            })
130        }
131    }
132}
133
134/// Ingress node of one branch, for hosts that feed triggers.
135#[derive(Debug, Clone, PartialEq)]
136pub struct BranchIngressPlan {
137    /// Workflow branch this record belongs to.
138    pub branch_id: String,
139    /// Ingress node type.
140    pub ingress_type: String,
141    /// Ingress node config.
142    pub ingress_config: serde_json::Value,
143}
144
145/// Ingress plan per branch.
146pub fn ingress_plans(spec: &Spec, registry: &NodeRegistry) -> Vec<BranchIngressPlan> {
147    spec.branches
148        .iter()
149        .filter_map(|branch| {
150            let ingress = branch
151                .nodes
152                .iter()
153                .find(|node| registry.is_ingress(&node.node_type))?;
154            Some(BranchIngressPlan {
155                branch_id: branch.branch_id.clone(),
156                ingress_type: ingress.node_type.clone(),
157                ingress_config: ingress.config.clone(),
158            })
159        })
160        .collect()
161}
162
163#[cfg(test)]
164mod tests {
165    use super::*;
166    use chrono::TimeZone;
167
168    #[test]
169    fn timezone_and_fixed_delay_are_explicit() {
170        let after = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
171        let cron = SchedulePolicy {
172            cadence: ScheduleCadence::Cron {
173                expression: "0 30 9 * * * *".into(),
174            },
175            timezone: "Asia/Shanghai".into(),
176            catch_up: CatchUpPolicy::CatchUpOnce,
177        };
178        assert_eq!(
179            next_scheduled_at(&cron, after, None).unwrap(),
180            Utc.with_ymd_and_hms(2026, 1, 1, 1, 30, 0).unwrap()
181        );
182
183        let delay = SchedulePolicy {
184            cadence: ScheduleCadence::FixedDelay {
185                milliseconds: 1_000,
186            },
187            timezone: "UTC".into(),
188            catch_up: CatchUpPolicy::CatchUpOnce,
189        };
190        assert_eq!(
191            next_scheduled_at(&delay, after, Some(after + chrono::Duration::seconds(10))).unwrap(),
192            after + chrono::Duration::seconds(11)
193        );
194        assert!(SchedulePolicy {
195            timezone: "not/a-zone".into(),
196            ..delay
197        }
198        .validate()
199        .is_err());
200    }
201
202    #[test]
203    fn catch_up_policies_advance_the_persisted_cursor_without_a_burst() {
204        let scheduled = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
205        let now = scheduled + chrono::Duration::seconds(35);
206        let policy = |catch_up| SchedulePolicy {
207            cadence: ScheduleCadence::FixedRate {
208                milliseconds: 10_000,
209            },
210            timezone: "UTC".into(),
211            catch_up,
212        };
213        let skipped = schedule_decision(&policy(CatchUpPolicy::Skip), scheduled, now, 0).unwrap();
214        assert_eq!(skipped.tick_at, None);
215        assert_eq!(skipped.next_at, scheduled + chrono::Duration::seconds(40));
216
217        let once =
218            schedule_decision(&policy(CatchUpPolicy::CatchUpOnce), scheduled, now, 0).unwrap();
219        assert_eq!(once.tick_at, Some(scheduled));
220        assert_eq!(once.next_at, scheduled + chrono::Duration::seconds(40));
221
222        let all = policy(CatchUpPolicy::CatchUpAll { limit: 2 });
223        let first = schedule_decision(&all, scheduled, now, 0).unwrap();
224        assert_eq!(first.next_at, scheduled + chrono::Duration::seconds(10));
225        assert_eq!(first.catch_up_count, 1);
226        let second = schedule_decision(&all, first.next_at, now, first.catch_up_count).unwrap();
227        assert_eq!(second.next_at, scheduled + chrono::Duration::seconds(40));
228        assert_eq!(second.catch_up_count, 0);
229    }
230
231    #[test]
232    fn cron_uses_timezone_rules_across_dst() {
233        let policy = SchedulePolicy {
234            cadence: ScheduleCadence::Cron {
235                expression: "0 30 2 * * * *".into(),
236            },
237            timezone: "America/New_York".into(),
238            catch_up: CatchUpPolicy::CatchUpOnce,
239        };
240        let before_spring_forward = Utc.with_ymd_and_hms(2026, 3, 8, 6, 59, 0).unwrap();
241        assert_eq!(
242            next_scheduled_at(&policy, before_spring_forward, None).unwrap(),
243            Utc.with_ymd_and_hms(2026, 3, 9, 6, 30, 0).unwrap(),
244            "nonexistent local times are skipped by chrono-tz/cron"
245        );
246    }
247
248    #[test]
249    fn next_cron_at_reports_invalid_expressions() {
250        let after = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
251        assert_eq!(
252            next_cron_at("0 0 12 * * * *", after).unwrap(),
253            Utc.with_ymd_and_hms(2026, 1, 1, 12, 0, 0).unwrap()
254        );
255        assert!(next_cron_at("not a cron", after)
256            .unwrap_err()
257            .contains("invalid cron"));
258    }
259
260    #[test]
261    fn next_scheduled_at_rejects_unparseable_cron_and_fixed_rate_measures_from_after() {
262        let after = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
263        let broken = SchedulePolicy {
264            cadence: ScheduleCadence::Cron {
265                expression: "every day".into(),
266            },
267            timezone: "UTC".into(),
268            catch_up: CatchUpPolicy::Skip,
269        };
270        assert!(next_scheduled_at(&broken, after, None).is_err());
271        let rate = SchedulePolicy {
272            cadence: ScheduleCadence::FixedRate {
273                milliseconds: 1_500,
274            },
275            timezone: "UTC".into(),
276            catch_up: CatchUpPolicy::Skip,
277        };
278        assert_eq!(
279            next_scheduled_at(&rate, after, Some(after + chrono::Duration::hours(1))).unwrap(),
280            after + chrono::Duration::milliseconds(1_500)
281        );
282    }
283}
af_workflow/ingress.rs

af_workflow/
ingress.rs
