use areev_core::types::{Catchup, Trigger, TriggerKind};
use croner::Cron;
use crate::error::{Result, TriggerError};
const SUPPORTED_TIMEZONE: &str = "UTC";
#[cfg(feature = "tz")]
fn resolve_zone(tz: &str) -> Option<chrono_tz::Tz> {
tz.parse::<chrono_tz::Tz>().ok()
}
pub fn validate(trigger: &Trigger) -> Result<()> {
if let Some(why) = trigger.incoherence() {
return Err(TriggerError::Malformed { what: why });
}
if let Err(e) = areev_core::error::Hash::from_hex(trigger.workflow_hash()) {
return Err(TriggerError::UnresolvedWorkflow {
hash: trigger.workflow.clone(),
why: format!(
"{e} — a workflow is named by its 64-hex content address, \
optionally prefixed `sha256:`"
),
});
}
if let Some(spec) = trigger.context_query.as_deref() {
if let Err(why) = crate::context::parse_context_query(spec) {
return Err(TriggerError::Malformed {
what: format!("context_query: {why}"),
});
}
}
if trigger.kind == TriggerKind::Schedule {
let expr = trigger.cron.as_deref().unwrap_or_default();
parse_cron(expr)?;
if let Some(tz) = trigger.config_str(areev_core::types::config_keys::TIMEZONE) {
check_timezone(expr, tz)?;
}
}
if trigger.kind == TriggerKind::Composite {
if let Some(raw) = trigger.predicate.as_ref() {
let predicate = crate::predicate::from_value(raw)?;
for referenced in crate::predicate::referenced_members(&predicate) {
if !trigger.members.contains_key(&referenced) {
return Err(TriggerError::UnknownMember {
trigger: "being declared".into(),
member: referenced,
});
}
}
}
}
Ok(())
}
fn check_timezone(expr: &str, tz: &str) -> Result<()> {
if tz.eq_ignore_ascii_case(SUPPORTED_TIMEZONE) {
return Ok(());
}
#[cfg(feature = "tz")]
{
if resolve_zone(tz).is_some() {
return Ok(());
}
Err(TriggerError::BadSchedule {
expr: expr.to_string(),
why: format!(
"timezone {tz:?} is not an IANA zone name this build's zone data knows \
(e.g. \"America/New_York\", \"Asia/Kolkata\", \"UTC\")"
),
})
}
#[cfg(not(feature = "tz"))]
{
Err(TriggerError::BadSchedule {
expr: expr.to_string(),
why: format!(
"timezone {tz:?} needs zone data this build does not carry — rebuild \
areev-trigger with the `tz` feature, or declare the expression in UTC"
),
})
}
}
pub fn declared_zone(trigger: &Trigger) -> Option<String> {
if trigger.kind != TriggerKind::Schedule {
return None;
}
trigger_zone(trigger)
}
#[cfg(feature = "tz")]
pub fn render_local(trigger: &Trigger, at_ms: i64) -> Option<String> {
let tz = declared_zone(trigger)?;
let zone = resolve_zone(&tz)?;
let utc = chrono::DateTime::from_timestamp_millis(at_ms)?;
Some(format!(
"{} {}",
utc.with_timezone(&zone).format("%Y-%m-%dT%H:%M:%S%:z"),
tz
))
}
#[cfg(not(feature = "tz"))]
pub fn render_local(_trigger: &Trigger, _at_ms: i64) -> Option<String> {
None
}
fn trigger_zone(trigger: &Trigger) -> Option<String> {
trigger
.config_str(areev_core::types::config_keys::TIMEZONE)
.filter(|tz| !tz.eq_ignore_ascii_case(SUPPORTED_TIMEZONE))
.map(str::to_string)
}
fn parse_cron(expr: &str) -> Result<Cron> {
Cron::new(expr)
.with_seconds_optional()
.parse()
.map_err(|e| TriggerError::BadSchedule { expr: expr.to_string(), why: e.to_string() })
}
pub fn next_due_after(trigger: &Trigger, after_ms: i64) -> Result<Option<i64>> {
match trigger.kind {
TriggerKind::Once => {
let at = trigger.at_ms.ok_or_else(|| TriggerError::Malformed {
what: "a once trigger needs at_ms".into(),
})?;
Ok(if at > after_ms { Some(at) } else { None })
}
TriggerKind::Interval | TriggerKind::Polling => {
let secs = trigger.interval_secs.ok_or_else(|| TriggerError::Malformed {
what: format!("a {} trigger needs interval_secs", trigger.kind.as_str()),
})?;
Ok(Some(after_ms.saturating_add(i64::from(secs) * 1000)))
}
TriggerKind::Schedule => {
let expr = trigger.cron.as_deref().unwrap_or_default();
let cron = parse_cron(expr)?;
let after = chrono::DateTime::from_timestamp_millis(after_ms).ok_or_else(|| {
TriggerError::BadSchedule {
expr: expr.to_string(),
why: format!("{after_ms} is not a representable instant"),
}
})?;
match trigger_zone(trigger) {
None => {
let next = cron.find_next_occurrence(&after, false).map_err(|e| {
TriggerError::BadSchedule { expr: expr.to_string(), why: e.to_string() }
})?;
Ok(Some(next.timestamp_millis()))
}
Some(tz) => next_local_occurrence(expr, &cron, &tz, after_ms, after),
}
}
TriggerKind::Memory | TriggerKind::Webhook | TriggerKind::Manual | TriggerKind::Composite => {
Ok(None)
}
}
}
#[cfg(feature = "tz")]
fn next_local_occurrence(
expr: &str,
cron: &Cron,
tz: &str,
after_ms: i64,
after: chrono::DateTime<chrono::Utc>,
) -> Result<Option<i64>> {
use chrono::TimeZone;
let zone = resolve_zone(tz).ok_or_else(|| TriggerError::BadSchedule {
expr: expr.to_string(),
why: format!("timezone {tz:?} is not a zone name this build knows"),
})?;
let mut cursor = after.with_timezone(&zone);
for _ in 0..8 {
let next = cron.find_next_occurrence(&cursor, false).map_err(|e| {
TriggerError::BadSchedule { expr: expr.to_string(), why: e.to_string() }
})?;
let ms = next.timestamp_millis();
if ms > after_ms {
return Ok(Some(ms));
}
cursor = zone.from_utc_datetime(
&chrono::DateTime::from_timestamp_millis(ms + 1_000)
.ok_or_else(|| TriggerError::BadSchedule {
expr: expr.to_string(),
why: "instant out of range while resolving a DST fold".into(),
})?
.naive_utc(),
);
}
Err(TriggerError::BadSchedule {
expr: expr.to_string(),
why: format!(
"no occurrence strictly after {after_ms} in {tz} within the DST search bound"
),
})
}
#[cfg(not(feature = "tz"))]
fn next_local_occurrence(
expr: &str,
_cron: &Cron,
tz: &str,
_after_ms: i64,
_after: chrono::DateTime<chrono::Utc>,
) -> Result<Option<i64>> {
Err(TriggerError::BadSchedule {
expr: expr.to_string(),
why: format!(
"timezone {tz:?} needs zone data this build does not carry — rebuild \
areev-trigger with the `tz` feature, or declare the expression in UTC"
),
})
}
pub fn advance_after_firing(trigger: &Trigger, fired_at_ms: i64, now_ms: i64) -> Result<Option<i64>> {
let Some(mut next) = next_due_after(trigger, fired_at_ms)? else {
return Ok(None);
};
if next > now_ms {
return Ok(Some(next));
}
match trigger.catchup {
Catchup::All => Ok(Some(next)),
Catchup::Last | Catchup::None => {
const MAX_STEPS: u32 = 10_000;
let mut steps = 0;
while next <= now_ms && steps < MAX_STEPS {
match next_due_after(trigger, next)? {
Some(n) if n > next => next = n,
_ => break,
}
steps += 1;
}
if next <= now_ms {
next = next_due_after(trigger, now_ms)?.unwrap_or(next);
}
Ok(Some(next))
}
}
}
pub fn initial_due(trigger: &Trigger, now_ms: i64) -> Result<Option<i64>> {
Ok(match trigger.kind {
TriggerKind::Schedule => next_due_after(trigger, now_ms)?,
TriggerKind::Once => trigger.at_ms.filter(|at| *at > now_ms),
TriggerKind::Interval
| TriggerKind::Polling
| TriggerKind::Memory
| TriggerKind::Webhook
| TriggerKind::Manual
| TriggerKind::Composite => None,
})
}
pub fn is_due(trigger: &Trigger, state: &areev_store::TriggerState, now_ms: i64) -> Result<bool> {
if !state.permits_firing(now_ms) {
return Ok(false);
}
if state.next_due_at.is_some() {
return Ok(true); }
Ok(initial_due(trigger, now_ms)?.is_none())
}
pub fn exhausts_after(trigger: &Trigger, fired_at_ms: i64) -> bool {
trigger.kind == TriggerKind::Once
&& next_due_after(trigger, fired_at_ms).ok().flatten().is_none()
}
pub fn backoff_until(trigger: &Trigger, consecutive_failures: u32, now_ms: i64) -> i64 {
const BASE_MS: i64 = 30_000;
const CAP_MS: i64 = 3_600_000;
let shift = consecutive_failures.saturating_sub(1).min(7);
let penalty = (BASE_MS << shift).min(CAP_MS);
let floor = trigger.interval_secs.map(|s| i64::from(s) * 1000).unwrap_or(0);
now_ms.saturating_add(penalty.max(floor))
}
#[cfg(test)]
mod tests {
use super::*;
const WF: &str = "a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f90";
const T0: i64 = 1_767_225_600_000;
fn interval(secs: u32) -> Trigger {
Trigger::new(TriggerKind::Interval, WF).interval_secs(secs)
}
fn schedule(expr: &str) -> Trigger {
Trigger::new(TriggerKind::Schedule, WF).cron(expr)
}
#[test]
fn interval_anchors_to_the_last_firing() {
let t = interval(120);
assert_eq!(next_due_after(&t, T0).unwrap(), Some(T0 + 120_000));
assert_eq!(next_due_after(&t, T0 + 5_000).unwrap(), Some(T0 + 125_000));
}
#[test]
fn cron_anchors_to_wall_clock_boundaries() {
let t = schedule("*/5 * * * *");
let first = next_due_after(&t, T0).unwrap().unwrap();
assert_eq!(first, T0 + 300_000, "next boundary from midnight");
let after_late = next_due_after(&t, T0 + 301_000).unwrap().unwrap();
assert_eq!(after_late, T0 + 600_000);
}
#[test]
fn a_five_field_weekday_expression_parses_with_posix_numbering() {
let t = schedule("0 9 * * 1-5");
let next = next_due_after(&t, T0).unwrap().unwrap();
let dt = chrono::DateTime::from_timestamp_millis(next).unwrap();
use chrono::{Datelike, Timelike};
assert_eq!(dt.hour(), 9);
assert_eq!(dt.minute(), 0);
let wd = dt.weekday().num_days_from_monday();
assert!(wd < 5, "must land Mon-Fri, got {:?}", dt.weekday());
}
#[test]
fn a_one_shot_fires_once_and_then_never() {
let t = Trigger::new(TriggerKind::Once, WF).at_ms(T0 + 1_000);
assert_eq!(next_due_after(&t, T0).unwrap(), Some(T0 + 1_000));
assert_eq!(next_due_after(&t, T0 + 1_000).unwrap(), None, "not due at its own instant again");
assert_eq!(next_due_after(&t, T0 + 9_999).unwrap(), None);
}
#[test]
fn push_and_gate_kinds_are_not_clocked() {
for kind in [TriggerKind::Webhook, TriggerKind::Manual, TriggerKind::Memory] {
let t = Trigger::new(kind, WF);
assert_eq!(next_due_after(&t, T0).unwrap(), None, "{kind:?} should not be clocked");
}
}
#[test]
fn catchup_last_collapses_a_weekend_of_missed_occurrences() {
let t = schedule("*/5 * * * *").catchup(Catchup::Last);
let three_days = 3 * 86_400_000;
let next = advance_after_firing(&t, T0, T0 + three_days).unwrap().unwrap();
assert!(next > T0 + three_days, "must be scheduled ahead of now, not in the backlog");
assert!(next <= T0 + three_days + 300_000, "and at the very next boundary");
}
#[test]
fn catchup_all_leaves_the_backlog_in_place() {
let t = schedule("*/5 * * * *").catchup(Catchup::All);
let three_days = 3 * 86_400_000;
let next = advance_after_firing(&t, T0, T0 + three_days).unwrap().unwrap();
assert_eq!(next, T0 + 300_000, "the stale occurrence stays due, to be replayed");
}
#[test]
fn collapsing_is_bounded_for_a_fast_interval_over_a_long_gap() {
let t = interval(1).catchup(Catchup::Last);
let a_year = 365 * 86_400_000i64;
let started = std::time::Instant::now();
let next = advance_after_firing(&t, T0, T0 + a_year).unwrap().unwrap();
assert!(next > T0 + a_year);
assert!(started.elapsed().as_millis() < 500, "the collapse walk must be bounded");
}
#[test]
fn a_firing_that_is_still_ahead_of_now_is_left_alone() {
let t = interval(300);
let next = advance_after_firing(&t, T0, T0 + 1_000).unwrap().unwrap();
assert_eq!(next, T0 + 300_000);
}
#[test]
fn backoff_grows_then_caps_and_never_polls_faster_than_the_interval() {
let t = interval(600); let first = backoff_until(&t, 1, T0) - T0;
let second = backoff_until(&t, 2, T0) - T0;
assert!(second >= first, "backoff must not shrink as failures accumulate");
assert!(first >= 600_000, "floor is the declared interval");
let far = backoff_until(&t, 99, T0) - T0;
assert!(far <= 3_600_000, "and it caps rather than growing without bound");
}
#[test]
fn an_unparseable_expression_is_refused_with_its_code() {
let t = schedule("not a cron expression");
let e = validate(&t).unwrap_err();
assert_eq!(e.code(), "TRG-E006");
assert!(e.to_string().starts_with("TRG-E006"));
}
#[test]
fn utc_is_accepted_however_it_is_spelled() {
let ok = schedule("0 9 * * *").config(serde_json::json!({ "int:timezone": "utc" }));
assert!(validate(&ok).is_ok());
let ok = schedule("0 9 * * *").config(serde_json::json!({ "int:timezone": "UTC" }));
assert!(validate(&ok).is_ok());
}
#[test]
fn an_unknown_zone_name_is_refused_whatever_the_build() {
let t = schedule("0 9 * * *")
.config(serde_json::json!({ "int:timezone": "Mars/Olympus" }));
let e = validate(&t).unwrap_err();
assert_eq!(e.code(), "TRG-E006");
}
#[cfg(not(feature = "tz"))]
#[test]
fn a_build_without_zone_data_refuses_and_names_the_feature() {
let t = schedule("0 9 * * *")
.config(serde_json::json!({ "int:timezone": "America/New_York" }));
let e = validate(&t).unwrap_err();
assert_eq!(e.code(), "TRG-E006");
assert!(e.to_string().contains("`tz` feature"), "{e}");
}
#[cfg(feature = "tz")]
mod tz_tests {
use super::*;
fn zoned(expr: &str, tz: &str) -> Trigger {
schedule(expr).config(serde_json::json!({ "int:timezone": tz }))
}
fn ms(iso: &str) -> i64 {
areev_core::time::iso8601_to_ms(iso).unwrap()
}
#[test]
fn a_declared_zone_validates() {
assert!(validate(&zoned("0 8 * * 1", "America/New_York")).is_ok());
assert!(validate(&zoned("0 9 * * *", "Asia/Kolkata")).is_ok());
}
#[test]
fn a_weekly_local_time_tracks_the_offset_across_the_dst_boundary() {
let t = zoned("0 8 * * 1", "America/New_York");
let first = next_due_after(&t, ms("2026-10-19T12:00:01Z")).unwrap().unwrap();
assert_eq!(first, ms("2026-10-26T12:00:00Z"), "still EDT");
let second = next_due_after(&t, first).unwrap().unwrap();
assert_eq!(second, ms("2026-11-02T13:00:00Z"), "EST after the change");
}
#[test]
fn a_time_skipped_by_spring_forward_fires_once_at_the_next_valid_minute() {
let t = zoned("30 2 * * *", "America/New_York");
let gap_day = next_due_after(&t, ms("2026-03-07T12:00:00Z")).unwrap().unwrap();
assert_eq!(
gap_day,
ms("2026-03-08T07:00:00Z"),
"03:00 EDT — the first valid instant after the gap"
);
let next = next_due_after(&t, gap_day).unwrap().unwrap();
assert_eq!(next, ms("2026-03-09T06:30:00Z"));
}
#[test]
fn an_ordinary_day_fires_at_the_declared_local_time() {
let t = zoned("30 2 * * *", "America/New_York");
let fired = next_due_after(&t, ms("2026-03-06T12:00:00Z")).unwrap().unwrap();
assert_eq!(fired, ms("2026-03-07T07:30:00Z"));
}
#[test]
fn a_repeated_time_fires_once_at_the_earlier_occurrence() {
let t = zoned("30 1 * * *", "America/New_York");
let first = next_due_after(&t, ms("2026-11-01T00:00:00Z")).unwrap().unwrap();
assert_eq!(first, ms("2026-11-01T05:30:00Z"), "the earlier one");
let next = next_due_after(&t, first).unwrap().unwrap();
assert_eq!(
next,
ms("2026-11-02T06:30:00Z"),
"the next DAY, not the second pass of the repeated hour"
);
}
#[test]
fn every_answer_is_strictly_after_the_question_across_the_fold() {
let t = zoned("30 1 * * *", "America/New_York");
let start = ms("2026-11-01T05:00:00Z");
let end = ms("2026-11-01T07:00:00Z");
let mut at = start;
while at < end {
let got = next_due_after(&t, at).unwrap().unwrap();
assert!(got > at, "answer {got} must be strictly after {at}");
at += 60_000;
}
}
#[test]
fn catchup_last_across_a_dst_boundary_collapses_to_one_firing() {
let mut t = zoned("30 1 * * *", "America/New_York");
t.catchup = Catchup::Last;
let fired = ms("2026-10-30T05:30:00Z");
let now = ms("2026-11-03T12:00:00Z");
let next = advance_after_firing(&t, fired, now).unwrap().unwrap();
assert!(next > now, "one firing ahead, not a backlog: {next}");
}
#[test]
fn an_absent_or_utc_zone_is_unchanged() {
let utc = schedule("0 9 * * *");
let zoned_utc = zoned("0 9 * * *", "UTC");
let at = ms("2026-03-08T00:00:00Z");
assert_eq!(
next_due_after(&utc, at).unwrap(),
next_due_after(&zoned_utc, at).unwrap()
);
}
}
#[test]
fn validate_rejects_an_incoherent_declaration_before_the_schedule() {
let t = Trigger::new(TriggerKind::Interval, WF); let e = validate(&t).unwrap_err();
assert_eq!(e.code(), "TRG-E001");
}
}