aion-rs 0.31.0

//! Pure tolerance accounting for workloop invariants (R2.3/R4.2).
//!
//! Health hangs off invariants, not the loop (R4.1). Every function here is
//! pure over [`InvariantHealthState`] — the durable accounting the store
//! carries between sweeps — so the whole tolerance surface is testable without
//! an engine, a store, or a clock.

use aion_core::{AlarmCause, InvariantAlarm, ToleranceSpec};
use aion_store::InvariantHealthState;
use chrono::{DateTime, Utc};

/// The two kinds of unconfirmed evidence an invariant can accrue. A closed
/// input type, so a caller cannot feed `loop-dead` or `unconfirmed-unknown`
/// (those are conclusions, never observations).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum UnconfirmedEvidence {
    /// The loop ran and did not confirm the invariant (a red sample, R3.3).
    SampleRed,
    /// The loop missed a cadence window — including the hung iteration, which
    /// produces no terminal by its next window (R3.3a) and is therefore a
    /// missed window, evaluated engine-side, never waited on.
    WindowMissed,
}

impl UnconfirmedEvidence {
    /// The alarm cause this evidence carries onto the one alarm path.
    #[must_use]
    pub const fn cause(self) -> AlarmCause {
        match self {
            Self::SampleRed => AlarmCause::SampleRed,
            Self::WindowMissed => AlarmCause::WindowMissed,
        }
    }
}

/// Record a confirmation: the invariant was confirmed held at `at`. Resets the
/// consecutive-unconfirmed count, the accrued evidence, and the alarm latch —
/// so a recovered invariant that breaches again alarms again.
pub fn observe_confirmed(state: &mut InvariantHealthState, at: DateTime<Utc>) {
    state.last_confirmed_at = Some(at);
    state.consecutive_unconfirmed = 0;
    state.last_evidence = None;
    state.alarmed = false;
}

/// Record one unit of unconfirmed evidence (a red sample or a missed window —
/// missed windows are themselves countable samples on the engine's clock,
/// R2.4a).
pub fn observe_unconfirmed(state: &mut InvariantHealthState, evidence: UnconfirmedEvidence) {
    state.consecutive_unconfirmed = state.consecutive_unconfirmed.saturating_add(1);
    state.last_evidence = Some(evidence.cause());
}

/// Whether the declared tolerance is exceeded NOW, and with what cause.
///
/// - **Count form** (*tolerates N consecutive unhealthy samples*): exceeded
///   when the consecutive count is strictly greater than N — tolerance zero
///   alarms on the first unhealthy sample, exactly the babysitters' undeclared
///   behaviour made declarable. The cause is the accrued evidence.
/// - **Duration form** (*unconfirmed for D*): exceeded when `now` is strictly
///   past the last confirmation (or `anchor`, for a never-confirmed invariant
///   — registration starts the clock) plus D. The cause is the accrued
///   evidence, or `unconfirmed-unknown` when there is none — total silence,
///   the R2.4a family.
///
/// When both forms are declared, either exceeding alarms (each is a declared
/// bound). The engine alarms when tolerance is exceeded — never before,
/// always then (R2.3).
#[must_use]
pub fn tolerance_breach(
    tolerance: &ToleranceSpec,
    state: &InvariantHealthState,
    anchor: DateTime<Utc>,
    now: DateTime<Utc>,
) -> Option<AlarmCause> {
    if let Some(windows) = tolerance.consecutive_windows()
        && state.consecutive_unconfirmed > windows
    {
        // Count breaches always carry accrued evidence: the count only moves
        // through observe_unconfirmed, which records what moved it.
        return Some(
            state
                .last_evidence
                .unwrap_or(AlarmCause::UnconfirmedUnknown),
        );
    }
    if let Some(unconfirmed_for) = tolerance.unconfirmed_for() {
        let since = state.last_confirmed_at.unwrap_or(anchor);
        // A declared duration beyond the representable clock range has a
        // deadline no real clock can reach — it simply never expires, which
        // is the only reading that invents no value.
        if let Ok(delta) = chrono::Duration::from_std(unconfirmed_for)
            && let Some(deadline) = since.checked_add_signed(delta)
            && now > deadline
        {
            return Some(
                state
                    .last_evidence
                    .unwrap_or(AlarmCause::UnconfirmedUnknown),
            );
        }
    }
    None
}

/// The alarm that WOULD be raised now, without latching: `None` when within
/// tolerance or when an alarm is already latched for the current unconfirmed
/// run. Callers that durably record the alarm latch only after the record
/// succeeds ([`latch_alarm`]), so a failed append cannot silence the breach
/// forever.
#[must_use]
pub fn peek_alarm(
    invariant: &str,
    tolerance: &ToleranceSpec,
    state: &InvariantHealthState,
    anchor: DateTime<Utc>,
    now: DateTime<Utc>,
    window_seq: Option<u64>,
) -> Option<InvariantAlarm> {
    if state.alarmed {
        return None;
    }
    let cause = tolerance_breach(tolerance, state, anchor, now)?;
    Some(InvariantAlarm {
        invariant: invariant.to_owned(),
        cause,
        window_seq,
        last_confirmed_at: state.last_confirmed_at,
        consecutive_unconfirmed: state.consecutive_unconfirmed,
    })
}

/// Latch the alarm for the current unconfirmed run, once its record durably
/// landed. A sustained breach then alarms once; [`observe_confirmed`] resets
/// the latch, so recovery followed by a new breach alarms again.
pub fn latch_alarm(state: &mut InvariantHealthState) {
    state.alarmed = true;
}

/// Edge-triggered alarm draw: [`peek_alarm`] + [`latch_alarm`] in one step,
/// for callers with no durable append between the two.
pub fn take_alarm(
    invariant: &str,
    tolerance: &ToleranceSpec,
    state: &mut InvariantHealthState,
    anchor: DateTime<Utc>,
    now: DateTime<Utc>,
    window_seq: Option<u64>,
) -> Option<InvariantAlarm> {
    let alarm = peek_alarm(invariant, tolerance, state, anchor, now, window_seq)?;
    latch_alarm(state);
    Some(alarm)
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    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)
    }

    fn count_tolerance(windows: u64) -> ToleranceSpec {
        ToleranceSpec::count(windows)
    }

    fn duration_tolerance(seconds: u64) -> Result<ToleranceSpec, Box<dyn std::error::Error>> {
        Ok(ToleranceSpec::duration(Duration::from_secs(seconds))?)
    }

    #[test]
    fn count_form_tolerates_exactly_n_and_alarms_on_the_next()
    -> Result<(), Box<dyn std::error::Error>> {
        let tolerance = count_tolerance(3);
        let mut state = InvariantHealthState::default();

        for miss in 0..3 {
            observe_unconfirmed(&mut state, UnconfirmedEvidence::WindowMissed);
            assert!(
                take_alarm("serving", &tolerance, &mut state, at(0), at(miss), Some(1)).is_none(),
                "within tolerance ({miss}) must not alarm — never before"
            );
        }
        observe_unconfirmed(&mut state, UnconfirmedEvidence::WindowMissed);
        let alarm = take_alarm("serving", &tolerance, &mut state, at(0), at(4), Some(4))
            .ok_or("exceeding tolerance must alarm — always then")?;
        assert_eq!(alarm.cause, AlarmCause::WindowMissed);
        assert_eq!(alarm.consecutive_unconfirmed, 4);
        assert_eq!(alarm.window_seq, Some(4));
        assert_eq!(alarm.last_confirmed_at, None);
        Ok(())
    }

    #[test]
    fn tolerance_zero_alarms_on_the_first_unhealthy_sample()
    -> Result<(), Box<dyn std::error::Error>> {
        let tolerance = count_tolerance(0);
        let mut state = InvariantHealthState::default();
        observe_unconfirmed(&mut state, UnconfirmedEvidence::SampleRed);
        let alarm = take_alarm("serving", &tolerance, &mut state, at(0), at(1), Some(1))
            .ok_or("declared tolerance zero must alarm on one bad sample")?;
        assert_eq!(alarm.cause, AlarmCause::SampleRed);
        Ok(())
    }

    #[test]
    fn a_sustained_breach_alarms_once_and_recovery_rearms() {
        let tolerance = count_tolerance(0);
        let mut state = InvariantHealthState::default();
        observe_unconfirmed(&mut state, UnconfirmedEvidence::SampleRed);
        assert!(take_alarm("serving", &tolerance, &mut state, at(0), at(1), None).is_some());
        // Still breached, already latched: no second alarm.
        observe_unconfirmed(&mut state, UnconfirmedEvidence::SampleRed);
        assert!(take_alarm("serving", &tolerance, &mut state, at(0), at(2), None).is_none());
        // Confirmation resets; a new breach alarms again.
        observe_confirmed(&mut state, at(3));
        assert!(!state.alarmed);
        assert_eq!(state.consecutive_unconfirmed, 0);
        observe_unconfirmed(&mut state, UnconfirmedEvidence::SampleRed);
        assert!(take_alarm("serving", &tolerance, &mut state, at(0), at(4), None).is_some());
    }

    #[test]
    fn duration_form_alarms_on_total_silence_with_unknown_cause()
    -> Result<(), Box<dyn std::error::Error>> {
        let tolerance = duration_tolerance(100)?;
        let mut state = InvariantHealthState::default();

        // Before the deadline (anchored at registration): quiet is tolerated.
        assert!(take_alarm("serving", &tolerance, &mut state, at(0), at(100), None).is_none());
        // Past the deadline with zero samples ever: the silent-death family —
        // the cause names the absence of evidence, not a fabricated failure.
        let alarm = take_alarm("serving", &tolerance, &mut state, at(0), at(101), None)
            .ok_or("duration expiry with zero samples must alarm (R2.4a)")?;
        assert_eq!(alarm.cause, AlarmCause::UnconfirmedUnknown);
        assert_eq!(alarm.consecutive_unconfirmed, 0);
        assert_eq!(alarm.last_confirmed_at, None);
        assert_eq!(alarm.window_seq, None);
        Ok(())
    }

    #[test]
    fn duration_form_measures_from_the_last_confirmation() -> Result<(), Box<dyn std::error::Error>>
    {
        let tolerance = duration_tolerance(100)?;
        let mut state = InvariantHealthState::default();
        observe_confirmed(&mut state, at(50));

        assert!(take_alarm("serving", &tolerance, &mut state, at(0), at(150), None).is_none());
        let alarm = take_alarm("serving", &tolerance, &mut state, at(0), at(151), None)
            .ok_or("unconfirmed past D after the last confirmation must alarm")?;
        assert_eq!(alarm.last_confirmed_at, Some(at(50)));
        Ok(())
    }

    #[test]
    fn duration_expiry_names_the_accrued_evidence_when_there_is_some()
    -> Result<(), Box<dyn std::error::Error>> {
        let tolerance = ToleranceSpec::both(10, Duration::from_secs(100))?;
        let mut state = InvariantHealthState::default();
        // Two red samples — under the count bound of 10, but the duration
        // bound expires first. The cause is the evidence, not "unknown".
        observe_unconfirmed(&mut state, UnconfirmedEvidence::SampleRed);
        observe_unconfirmed(&mut state, UnconfirmedEvidence::SampleRed);
        let alarm = take_alarm("serving", &tolerance, &mut state, at(0), at(101), Some(2))
            .ok_or("duration expiry must alarm")?;
        assert_eq!(alarm.cause, AlarmCause::SampleRed);
        assert_eq!(alarm.consecutive_unconfirmed, 2);
        Ok(())
    }
}