use crate::error::SchedulerError;
use crate::model::{
GroupedIntervalSchedule, Job, JobState, Schedule, StaggeredIntervalSchedule, utc_time,
};
use chrono::{DateTime, TimeDelta, Utc};
use chrono_tz::Tz;
use std::time::Duration;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct DueSummary {
pub(crate) first_due: DateTime<Utc>,
pub(crate) last_due: DateTime<Utc>,
pub(crate) count: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct DueWindow {
pub(crate) first_due: DateTime<Utc>,
pub(crate) has_multiple: bool,
}
pub(crate) fn initial_next_run_at<D>(
job: &Job<D>,
timezone: Tz,
) -> Result<Option<DateTime<Utc>>, SchedulerError> {
if matches!(job.max_runs, Some(0)) {
return Ok(None);
}
match &job.schedule {
Schedule::Interval(every) => duration_to_delta(*every)
.ok_or_else(SchedulerError::invalid_interval_out_of_range)
.and_then(|delta| {
Utc::now()
.checked_add_signed(delta)
.ok_or_else(SchedulerError::invalid_interval_out_of_range)
})
.map(Some),
Schedule::StaggeredInterval(staggered) => {
staggered_initial_next_run_at(Utc::now(), staggered, &job.job_id).map(Some)
}
Schedule::GroupedInterval(grouped) => {
grouped_initial_next_run_at(Utc::now(), grouped).map(Some)
}
Schedule::AtTimes(times) => Ok(times.first().copied().map(utc_time)),
Schedule::Cron(schedule) => Ok(schedule.next_after(Utc::now(), timezone)),
}
}
pub(crate) fn next_run_is_in_future(next_run_at: Option<DateTime<Utc>>) -> bool {
next_run_at.map(|value| value > Utc::now()).unwrap_or(false)
}
pub(crate) fn inspect_due_window<D>(
job: &Job<D>,
state: &JobState,
now: DateTime<Utc>,
timezone: Tz,
) -> Result<Option<DueWindow>, SchedulerError>
where
D: Send + Sync + 'static,
{
let Some(first_due) = state.next_run_at else {
return Ok(None);
};
if first_due > now {
return Ok(None);
}
let next_trigger_count = state.trigger_count + 1;
let has_multiple = compute_next_after(job, first_due, next_trigger_count, timezone)?
.map(|next_due| next_due <= now)
.unwrap_or(false);
Ok(Some(DueWindow {
first_due,
has_multiple,
}))
}
pub(crate) fn consume_due_summary<D>(
job: &Job<D>,
state: &mut JobState,
now: DateTime<Utc>,
timezone: Tz,
) -> Result<Option<DueSummary>, SchedulerError>
where
D: Send + Sync + 'static,
{
let mut first_due = None;
let mut last_due = None;
let mut count = 0u32;
while let Some(value) = state.next_run_at {
if value > now {
break;
}
if first_due.is_none() {
first_due = Some(value);
}
last_due = Some(value);
count += 1;
advance_state_to(job, state, value, timezone)?;
}
Ok(first_due.map(|first_due| DueSummary {
first_due,
last_due: last_due.expect("last_due exists when first_due exists"),
count,
}))
}
pub(crate) fn advance_state_to<D>(
job: &Job<D>,
state: &mut JobState,
scheduled_at: DateTime<Utc>,
timezone: Tz,
) -> Result<(), SchedulerError>
where
D: Send + Sync + 'static,
{
state.trigger_count += 1;
state.next_run_at = compute_next_after(job, scheduled_at, state.trigger_count, timezone)?;
Ok(())
}
pub(crate) fn compute_next_after<D>(
job: &Job<D>,
scheduled_at: DateTime<Utc>,
trigger_count: u32,
timezone: Tz,
) -> Result<Option<DateTime<Utc>>, SchedulerError>
where
D: Send + Sync + 'static,
{
if let Some(max_runs) = job.max_runs
&& trigger_count >= max_runs
{
return Ok(None);
}
match &job.schedule {
Schedule::Interval(every) => {
let delta = duration_to_delta(*every)
.ok_or_else(SchedulerError::invalid_interval_out_of_range)?;
Ok(scheduled_at.checked_add_signed(delta))
}
Schedule::StaggeredInterval(staggered) => {
let delta = duration_to_delta(staggered.every)
.ok_or_else(SchedulerError::invalid_interval_out_of_range)?;
Ok(scheduled_at.checked_add_signed(delta))
}
Schedule::GroupedInterval(grouped) => {
let delta = duration_to_delta(grouped.every)
.ok_or_else(SchedulerError::invalid_interval_out_of_range)?;
Ok(scheduled_at.checked_add_signed(delta))
}
Schedule::AtTimes(times) => Ok(times.get(trigger_count as usize).copied().map(utc_time)),
Schedule::Cron(schedule) => Ok(schedule.next_after(scheduled_at, timezone)),
}
}
pub(crate) fn is_missed(scheduled_at: DateTime<Utc>, now: DateTime<Utc>) -> bool {
let tolerance = TimeDelta::milliseconds(25);
scheduled_at
.checked_add_signed(tolerance)
.map(|adjusted| adjusted < now)
.unwrap_or(false)
}
fn duration_to_delta(duration: Duration) -> Option<TimeDelta> {
TimeDelta::from_std(duration).ok()
}
fn duration_to_nanos(duration: Duration) -> Option<u128> {
let nanos = u128::from(duration.as_secs())
.checked_mul(1_000_000_000)?
.checked_add(u128::from(duration.subsec_nanos()))?;
Some(nanos)
}
fn utc_to_nanos(value: DateTime<Utc>) -> i128 {
i128::from(value.timestamp())
.saturating_mul(1_000_000_000)
.saturating_add(i128::from(value.timestamp_subsec_nanos()))
}
fn nanos_to_utc(value: i128) -> Option<DateTime<Utc>> {
let seconds = value.div_euclid(1_000_000_000);
let nanos = value.rem_euclid(1_000_000_000) as u32;
let seconds = i64::try_from(seconds).ok()?;
chrono::TimeZone::timestamp_opt(&Utc, seconds, nanos).single()
}
fn stable_seed_hash(seed: &str) -> u64 {
let mut hash = 0xcbf29ce484222325u64;
for byte in seed.as_bytes() {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x100000001b3);
}
hash
}
fn interval_nanos(every: Duration) -> Result<u128, SchedulerError> {
let interval_nanos =
duration_to_nanos(every).ok_or_else(SchedulerError::invalid_interval_out_of_range)?;
if interval_nanos == 0 {
return Err(SchedulerError::invalid_zero_interval());
}
Ok(interval_nanos)
}
fn aligned_initial_next_run_at(
now: DateTime<Utc>,
interval_nanos: u128,
phase_nanos: u128,
) -> Result<DateTime<Utc>, SchedulerError> {
let interval_nanos = i128::try_from(interval_nanos)
.map_err(|_| SchedulerError::invalid_interval_out_of_range())?;
let phase_nanos =
i128::try_from(phase_nanos).map_err(|_| SchedulerError::invalid_interval_out_of_range())?;
let now_nanos = utc_to_nanos(now);
let cycle_start = now_nanos.div_euclid(interval_nanos) * interval_nanos;
let mut candidate = cycle_start + phase_nanos;
if candidate < now_nanos {
candidate += interval_nanos;
}
nanos_to_utc(candidate).ok_or_else(SchedulerError::invalid_interval_out_of_range)
}
fn staggered_initial_next_run_at(
now: DateTime<Utc>,
staggered: &StaggeredIntervalSchedule,
seed: &str,
) -> Result<DateTime<Utc>, SchedulerError> {
let interval_nanos = interval_nanos(staggered.every)?;
let phase_nanos =
u128::from(stable_seed_hash(staggered.seed.as_deref().unwrap_or(seed))) % interval_nanos;
aligned_initial_next_run_at(now, interval_nanos, phase_nanos)
}
fn grouped_initial_next_run_at(
now: DateTime<Utc>,
grouped: &GroupedIntervalSchedule,
) -> Result<DateTime<Utc>, SchedulerError> {
validate_grouped_interval(grouped)?;
let interval_nanos = interval_nanos(grouped.every)?;
let base_phase_nanos = interval_nanos
.checked_mul(u128::from(grouped.member_index))
.ok_or_else(SchedulerError::invalid_interval_out_of_range)?
/ u128::from(grouped.group_size);
let group_offset_nanos = grouped
.group_seed
.as_deref()
.map(|seed| u128::from(stable_seed_hash(seed)) % interval_nanos)
.unwrap_or(0);
let phase_nanos = base_phase_nanos
.checked_add(group_offset_nanos)
.ok_or_else(SchedulerError::invalid_interval_out_of_range)?
% interval_nanos;
aligned_initial_next_run_at(now, interval_nanos, phase_nanos)
}
pub(crate) fn validate_grouped_interval(
grouped: &GroupedIntervalSchedule,
) -> Result<(), SchedulerError> {
if grouped.group_size == 0 {
return Err(SchedulerError::invalid_job_with_kind(
crate::InvalidJobKind::Other,
"grouped interval group_size must be greater than zero",
));
}
if grouped.member_index >= grouped.group_size {
return Err(SchedulerError::invalid_job_with_kind(
crate::InvalidJobKind::Other,
"grouped interval member_index must be less than group_size",
));
}
Ok(())
}
#[cfg(test)]
pub(crate) fn collect_due_times<D>(
job: &Job<D>,
state: &JobState,
now: DateTime<Utc>,
timezone: Tz,
) -> Result<Vec<DateTime<Utc>>, SchedulerError>
where
D: Send + Sync + 'static,
{
let mut due_times = Vec::new();
let mut trigger_count = state.trigger_count;
let mut next_run_at = state.next_run_at;
while let Some(value) = next_run_at {
if value > now {
break;
}
due_times.push(value);
trigger_count += 1;
next_run_at = compute_next_after(job, value, trigger_count, timezone)?;
}
Ok(due_times)
}
#[cfg(test)]
pub(crate) fn advance_state_for<D>(
job: &Job<D>,
state: &mut JobState,
due_times: &[DateTime<Utc>],
timezone: Tz,
) -> Result<(), SchedulerError>
where
D: Send + Sync + 'static,
{
for scheduled_at in due_times {
advance_state_to(job, state, *scheduled_at, timezone)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::{
advance_state_for, collect_due_times, compute_next_after, nanos_to_utc, stable_seed_hash,
staggered_initial_next_run_at, utc_to_nanos, validate_grouped_interval,
};
use crate::{
GroupedIntervalSchedule, InvalidJobKind, Job, JobState, MissedRunPolicy, Schedule,
StaggeredIntervalSchedule, Task,
};
use chrono::{TimeDelta, TimeZone, Utc};
use chrono_tz::Asia::Shanghai;
use std::time::Duration;
fn noop_job(schedule: Schedule) -> Job<()> {
Job::without_deps("job", schedule, Task::from_async(|_| async { Ok(()) }))
}
#[test]
fn compute_next_after_staggered_interval_advances_by_the_base_duration() {
let scheduled_at = Utc.with_ymd_and_hms(2026, 4, 3, 1, 2, 0).unwrap();
let job = noop_job(Schedule::StaggeredInterval(
StaggeredIntervalSchedule::new(Duration::from_secs(60)).with_seed("site-a"),
));
let next = compute_next_after(&job, scheduled_at, 1, chrono_tz::UTC).unwrap();
assert_eq!(
next,
Some(Utc.with_ymd_and_hms(2026, 4, 3, 1, 3, 0).unwrap())
);
}
#[test]
fn staggered_initial_next_run_is_deterministic_and_bounded() {
let now = Utc.with_ymd_and_hms(2026, 4, 3, 1, 2, 3).unwrap();
let schedule = StaggeredIntervalSchedule::new(Duration::from_secs(86_400));
let seeded = schedule.clone().with_seed("site-a");
let next = staggered_initial_next_run_at(now, &seeded, "job-a").unwrap();
let interval_nanos = 86_400_u128 * 1_000_000_000;
let phase_nanos = i128::from(stable_seed_hash("site-a")).rem_euclid(interval_nanos as i128);
let now_nanos = utc_to_nanos(now);
let cycle_start = now_nanos.div_euclid(interval_nanos as i128) * interval_nanos as i128;
let mut expected = cycle_start + phase_nanos;
if expected < now_nanos {
expected += interval_nanos as i128;
}
assert_eq!(next, nanos_to_utc(expected).unwrap());
assert_eq!(
staggered_initial_next_run_at(now, &schedule.clone().with_seed("site-a"), "job-a")
.unwrap(),
next
);
assert!(next >= now);
assert!(next < now + TimeDelta::seconds(86_400));
}
#[test]
fn staggered_initial_next_run_falls_back_to_job_id_when_seed_missing() {
let now = Utc.with_ymd_and_hms(2026, 4, 3, 1, 2, 3).unwrap();
let schedule = StaggeredIntervalSchedule::new(Duration::from_secs(3_600));
let next = staggered_initial_next_run_at(now, &schedule, "job-a").unwrap();
let repeated = staggered_initial_next_run_at(now, &schedule, "job-a").unwrap();
let different = staggered_initial_next_run_at(now, &schedule, "job-b").unwrap();
assert_eq!(next, repeated);
assert!(next >= now);
assert!(next < now + TimeDelta::seconds(3_600));
assert_ne!(next, different);
}
#[test]
fn collect_due_times_replays_staggered_interval_in_fixed_steps() {
let now = Utc.with_ymd_and_hms(2026, 4, 3, 1, 2, 3).unwrap();
let first = (now - TimeDelta::seconds(3)).with_timezone(&Shanghai);
let mut state = JobState::new("job", Some(first.with_timezone(&Utc)));
let job = noop_job(Schedule::StaggeredInterval(
StaggeredIntervalSchedule::new(Duration::from_secs(5)).with_seed("site-a"),
))
.with_missed_run_policy(MissedRunPolicy::ReplayAll);
let due = collect_due_times(&job, &state, now, Shanghai).unwrap();
assert_eq!(due, vec![first.with_timezone(&Utc)]);
advance_state_for(&job, &mut state, &due, Shanghai).unwrap();
assert_eq!(
state.next_run_at,
Some((first + TimeDelta::seconds(5)).with_timezone(&Utc))
);
}
#[test]
fn grouped_initial_next_run_is_evenly_spaced() {
let now = Utc.with_ymd_and_hms(2026, 4, 3, 0, 0, 0).unwrap();
let interval = Duration::from_secs(120 * 60);
let member0 = GroupedIntervalSchedule::new(interval, 3, 0);
let member1 = GroupedIntervalSchedule::new(interval, 3, 1);
let member2 = GroupedIntervalSchedule::new(interval, 3, 2);
let next0 = super::grouped_initial_next_run_at(now, &member0).unwrap();
let next1 = super::grouped_initial_next_run_at(now, &member1).unwrap();
let next2 = super::grouped_initial_next_run_at(now, &member2).unwrap();
assert_eq!(next0, now);
assert_eq!(next1, now + TimeDelta::minutes(40));
assert_eq!(next2, now + TimeDelta::minutes(80));
}
#[test]
fn grouped_initial_next_run_uses_group_seed_as_a_shared_rotation() {
let now = Utc.with_ymd_and_hms(2026, 4, 3, 0, 0, 0).unwrap();
let interval = Duration::from_secs(120 * 60);
let a0 = GroupedIntervalSchedule::new(interval, 3, 0).with_group_seed("site-a");
let a1 = GroupedIntervalSchedule::new(interval, 3, 1).with_group_seed("site-a");
let b0 = GroupedIntervalSchedule::new(interval, 3, 0).with_group_seed("site-b");
let next_a0 = super::grouped_initial_next_run_at(now, &a0).unwrap();
let next_a1 = super::grouped_initial_next_run_at(now, &a1).unwrap();
let next_b0 = super::grouped_initial_next_run_at(now, &b0).unwrap();
assert_eq!(
(next_a1 - next_a0).num_seconds().rem_euclid(120 * 60),
40 * 60
);
assert_eq!(
next_a0,
super::grouped_initial_next_run_at(now, &a0).unwrap()
);
assert_ne!(next_a0, next_b0);
}
#[test]
fn grouped_interval_validation_rejects_invalid_parameters() {
let zero_group = GroupedIntervalSchedule::new(Duration::from_secs(60), 0, 0);
let out_of_range = GroupedIntervalSchedule::new(Duration::from_secs(60), 3, 3);
let zero_group_error = validate_grouped_interval(&zero_group).unwrap_err();
let out_of_range_error = validate_grouped_interval(&out_of_range).unwrap_err();
assert!(matches!(
zero_group_error,
crate::SchedulerError::InvalidJob(ref invalid)
if invalid.kind() == InvalidJobKind::Other
));
assert!(matches!(
out_of_range_error,
crate::SchedulerError::InvalidJob(ref invalid)
if invalid.kind() == InvalidJobKind::Other
));
}
}