aion/workloop/

windows.rs

//! Pure cadence-window arithmetic (R4.3/R13.3).
//!
//! One clock per loop (R2.2): windows sit on a fixed grid anchored at
//! registration, so a late sweep never drifts the cadence — it fires the
//! elapsed window once and re-arms on the original grid past `now`
//! (lateness-not-death: a missed window is COUNTED by the dead-man switch,
//! never silently replayed as a burst of catch-up fires).

use std::time::Duration;

use aion_store::WorkloopRecord;
use chrono::{DateTime, Utc};

use super::error::WorkloopError;

fn chrono_period(period: Duration) -> Result<chrono::Duration, WorkloopError> {
    chrono::Duration::from_std(period).map_err(|error| WorkloopError::UnrepresentableDuration {
        reason: format!("cadence period {period:?}: {error}"),
    })
}

/// The first cadence window after registration: `registered_at + period`.
///
/// # Errors
///
/// Refuses a period beyond the representable clock range.
pub fn initial_window(
    registered_at: DateTime<Utc>,
    period: Duration,
) -> Result<DateTime<Utc>, WorkloopError> {
    let delta = chrono_period(period)?;
    registered_at
        .checked_add_signed(delta)
        .ok_or_else(|| WorkloopError::UnrepresentableDuration {
            reason: format!("first window past {registered_at} overflows the clock"),
        })
}

/// The next window boundary on the loop's own grid strictly after `now`,
/// starting from the boundary that just fired.
///
/// A sweep that arrives late (or after engine downtime) advances to the first
/// future boundary rather than scheduling an immediate burst; the windows
/// skipped in between are visible to the dead-man switch as missed windows,
/// so lateness is counted, never papered over.
///
/// # Errors
///
/// Refuses a period beyond the representable clock range.
pub fn advance_window(
    fired_at: DateTime<Utc>,
    period: Duration,
    now: DateTime<Utc>,
) -> Result<DateTime<Utc>, WorkloopError> {
    let delta = chrono_period(period)?;
    let mut next = fired_at.checked_add_signed(delta).ok_or_else(|| {
        WorkloopError::UnrepresentableDuration {
            reason: format!("window past {fired_at} overflows the clock"),
        }
    })?;
    while next <= now {
        next = next.checked_add_signed(delta).ok_or_else(|| {
            WorkloopError::UnrepresentableDuration {
                reason: format!("window past {next} overflows the clock"),
            }
        })?;
    }
    Ok(next)
}

/// When the sweeper must next look at this loop: the earliest of the next
/// cadence window and every invariant's duration-form deadline (skipping
/// invariants whose alarm is already latched — a latched alarm re-arms only
/// through a confirmation, so there is nothing to poll for).
///
/// `None` means nothing is left to check engine-side: a signal-only loop with
/// every duration alarm latched. Cadenced loops always have a next window.
#[must_use]
pub fn next_check_at(record: &WorkloopRecord) -> Option<DateTime<Utc>> {
    let mut earliest: Option<DateTime<Utc>> = record.next_window_at;

    for invariant in record.spec.invariants() {
        let Some(unconfirmed_for) = invariant.tolerance.unconfirmed_for() else {
            continue;
        };
        let state = record.invariant_health.get(&invariant.name);
        if state.is_some_and(|state| state.alarmed) {
            continue;
        }
        let since = state
            .and_then(|state| state.last_confirmed_at)
            .unwrap_or(record.registered_at);
        let Ok(delta) = chrono::Duration::from_std(unconfirmed_for) else {
            // Beyond the representable range: the deadline never arrives.
            continue;
        };
        let Some(deadline) = since.checked_add_signed(delta) else {
            continue;
        };
        earliest = Some(match earliest {
            Some(current) if current <= deadline => current,
            _ => deadline,
        });
    }

    earliest
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use aion_core::{InvariantSpec, ToleranceSpec, WorkloopArming, WorkloopSpec};
    use aion_store::InvariantHealthState;
    use chrono::TimeZone;

    use super::*;

    fn at(offset: i64) -> DateTime<Utc> {
        Utc.with_ymd_and_hms(2026, 8, 25, 6, 0, 0)
            .single()
            .unwrap_or_default()
            + chrono::Duration::seconds(offset)
    }

    #[test]
    fn windows_stay_on_the_registration_grid() -> Result<(), Box<dyn std::error::Error>> {
        let period = Duration::from_secs(100);
        let first = initial_window(at(0), period)?;
        assert_eq!(first, at(100));

        // On-time advance: the very next boundary.
        assert_eq!(advance_window(first, period, at(100))?, at(200));
        // A late sweep (two windows of downtime) lands on the grid past now —
        // one boundary, not a burst.
        assert_eq!(advance_window(first, period, at(350))?, at(400));
        Ok(())
    }

    fn record_with(
        arming: WorkloopArming,
        tolerance: ToleranceSpec,
        health: Option<InvariantHealthState>,
        next_window_at: Option<DateTime<Utc>>,
    ) -> Result<WorkloopRecord, Box<dyn std::error::Error>> {
        let spec = WorkloopSpec::new(
            arming,
            vec![InvariantSpec {
                name: String::from("serving"),
                record_type: String::from("ServeState"),
                tolerance,
                confirms: vec![String::from("sweep")],
            }],
            Duration::from_secs(86_400),
        )?;
        Ok(WorkloopRecord {
            loop_id: aion_core::WorkflowId::new_v4(),
            namespace: String::from("default"),
            spec,
            window_seq: 0,
            next_window_at,
            next_check_at: None,
            last_iteration_closed_window: None,
            invariant_health: health
                .map(|state| BTreeMap::from([(String::from("serving"), state)]))
                .unwrap_or_default(),
            registered_at: at(0),
            updated_at: at(0),
        })
    }

    #[test]
    fn next_check_is_the_earliest_of_window_and_duration_deadline()
    -> Result<(), Box<dyn std::error::Error>> {
        // Cadence at +100, duration deadline at +300 (anchor = registration).
        let record = record_with(
            WorkloopArming::every(Duration::from_secs(100))?,
            ToleranceSpec::both(3, Duration::from_secs(300))?,
            None,
            Some(at(100)),
        )?;
        assert_eq!(next_check_at(&record), Some(at(100)));

        // Duration deadline earlier than the window: it wins.
        let record = record_with(
            WorkloopArming::every(Duration::from_secs(1000))?,
            ToleranceSpec::both(3, Duration::from_secs(300))?,
            None,
            Some(at(1000)),
        )?;
        assert_eq!(next_check_at(&record), Some(at(300)));
        Ok(())
    }

    #[test]
    fn signal_only_loop_checks_at_the_duration_deadline_and_rests_when_latched()
    -> Result<(), Box<dyn std::error::Error>> {
        let arming = WorkloopArming::signal_only(vec![String::from("task_ready")])?;
        let tolerance = ToleranceSpec::duration(Duration::from_secs(300))?;

        let unlatched = record_with(arming.clone(), tolerance.clone(), None, None)?;
        assert_eq!(next_check_at(&unlatched), Some(at(300)));

        // Confirmation moves the deadline out from the confirmation instant.
        let confirmed = record_with(
            arming.clone(),
            tolerance.clone(),
            Some(InvariantHealthState {
                last_confirmed_at: Some(at(200)),
                consecutive_unconfirmed: 0,
                last_evidence: None,
                alarmed: false,
            }),
            None,
        )?;
        assert_eq!(next_check_at(&confirmed), Some(at(500)));

        // A latched alarm has nothing left to poll: the loop rests entirely.
        let latched = record_with(
            arming,
            tolerance,
            Some(InvariantHealthState {
                last_confirmed_at: None,
                consecutive_unconfirmed: 0,
                last_evidence: None,
                alarmed: true,
            }),
            None,
        )?;
        assert_eq!(next_check_at(&latched), None);
        Ok(())
    }
}