aion-rs 0.27.1

//! The engine-side cadence service: dead-man switch, tolerance sweep, wake.
//!
//! ONE service task sweeps every registered loop off the store's
//! `due_workloops` set — a sleeping loop is store bytes plus this row's
//! `next_check_at`, with no resident process, no timer-wheel entry, and no
//! per-loop task (R13.3). Missed-window evaluation happens HERE, engine-side,
//! never inside the loop's own fault domain (R4.3): a loop crash is the
//! simultaneous non-confirmation of every invariant it carried, which is what
//! makes the single alarm path total (R4.2).

use std::sync::Arc;
use std::time::Duration;

use aion_core::{HealthSample, HealthStatus, InvariantAlarm, WorkflowId, WorkloopSpec};
use aion_store::workloop::{WorkloopRecord, WorkloopStore};
use async_trait::async_trait;
use chrono::{DateTime, Utc};

use super::error::WorkloopError;
use super::health::{
    UnconfirmedEvidence, latch_alarm, observe_confirmed, observe_unconfirmed, peek_alarm,
};
use super::windows::{advance_window, initial_window, next_check_at};
use crate::engine_seam::RecordOutcome;

/// Durable append seam for engine-raised workloop events. Both paths MUST go
/// through the target loop's single Recorder (the one-writer law).
#[async_trait]
pub trait LoopEventSink: Send + Sync {
    /// Record `CadenceFired { window_seq }` in the loop's history.
    ///
    /// Returns [`RecordOutcome::RefusedTerminal`] when the loop's active run
    /// already holds a terminal — a dead loop has no windows, and the refusal
    /// is the engine's positive evidence of death.
    ///
    /// # Errors
    ///
    /// Returns [`WorkloopError`] when the append fails outright.
    async fn record_cadence_fired(
        &self,
        loop_id: &WorkflowId,
        window_seq: u64,
    ) -> Result<RecordOutcome, WorkloopError>;

    /// Record `InvariantUnconfirmed` (the one alarm path) in the loop's
    /// history. UNLIKE cadence fires, this append is honoured even after the
    /// run's terminal: the alarm that reports a loop's death must not be
    /// silenced by the very death it reports (R5.3's premise). The event is
    /// status-invisible, so the terminal projection is untouched.
    ///
    /// # Errors
    ///
    /// Returns [`WorkloopError`] when the append fails outright.
    async fn record_invariant_unconfirmed(
        &self,
        loop_id: &WorkflowId,
        alarm: InvariantAlarm,
    ) -> Result<(), WorkloopError>;
}

/// Wake seam: bring a suspended loop's current generation to a live process so
/// the fired iteration can run (wake = load carry, run iteration, suspend —
/// R13.3). Called only after the fire is durably recorded.
#[async_trait]
pub trait WorkloopWaker: Send + Sync {
    /// Wake the loop's current generation.
    ///
    /// # Errors
    ///
    /// Returns [`WorkloopError`] when the generation cannot be woken; the
    /// sweep reports the fault and the recorded fire stands (the next sweep's
    /// dead-man evaluation counts the unrun window — a failed wake degrades to
    /// a missed window, never to silence).
    async fn wake(&self, loop_id: &WorkflowId) -> Result<(), WorkloopError>;
}

/// What one sweep pass did — the service's own observability surface.
#[derive(Debug, Default)]
pub struct SweepReport {
    /// Loops evaluated this pass.
    pub swept: usize,
    /// Cadence fires recorded: (loop, window sequence).
    pub fired: Vec<(WorkflowId, u64)>,
    /// Alarms raised on the one path: (loop, alarm).
    pub alarms: Vec<(WorkflowId, InvariantAlarm)>,
    /// Loops found dead (cadence fire refused on a terminal run) and
    /// deregistered after their loop-dead alarms were raised.
    pub dead: Vec<WorkflowId>,
    /// Loops found RETIRED (cadence fire refused on a run carrying
    /// `LoopRetired`) and withdrawn from the sweep set with no alarms — a
    /// declared stop, reported separately from a death so an operator reading
    /// this report is never told a decommission was an incident.
    pub retired: Vec<WorkflowId>,
    /// Per-loop faults that did not stop the sweep, with their loop.
    pub faults: Vec<(WorkflowId, String)>,
}

/// The engine-side cadence service. One instance, one sweep task, N loops.
pub struct WorkloopService {
    store: Arc<dyn WorkloopStore>,
    sink: Arc<dyn LoopEventSink>,
    waker: Arc<dyn WorkloopWaker>,
    sweep_interval: Duration,
    now: Arc<dyn Fn() -> DateTime<Utc> + Send + Sync>,
}

impl WorkloopService {
    /// Creates the service. `sweep_interval` is the operator-declared sweep
    /// cadence — REQUIRED, never defaulted, and it bounds dead-man detection
    /// latency.
    ///
    /// # Errors
    ///
    /// Refuses a zero interval ([`WorkloopError::ZeroSweepInterval`]).
    pub fn new(
        store: Arc<dyn WorkloopStore>,
        sink: Arc<dyn LoopEventSink>,
        waker: Arc<dyn WorkloopWaker>,
        sweep_interval: Duration,
    ) -> Result<Self, WorkloopError> {
        Self::with_clock(store, sink, waker, sweep_interval, Utc::now)
    }

    /// [`WorkloopService::new`] with an injected clock, for deterministic
    /// tests.
    ///
    /// # Errors
    ///
    /// Refuses a zero interval ([`WorkloopError::ZeroSweepInterval`]).
    pub fn with_clock(
        store: Arc<dyn WorkloopStore>,
        sink: Arc<dyn LoopEventSink>,
        waker: Arc<dyn WorkloopWaker>,
        sweep_interval: Duration,
        now: impl Fn() -> DateTime<Utc> + Send + Sync + 'static,
    ) -> Result<Self, WorkloopError> {
        if sweep_interval.is_zero() {
            return Err(WorkloopError::ZeroSweepInterval);
        }
        Ok(Self {
            store,
            sink,
            waker,
            sweep_interval,
            now: Arc::new(now),
        })
    }

    /// The declared sweep interval.
    #[must_use]
    pub const fn sweep_interval(&self) -> Duration {
        self.sweep_interval
    }

    /// Registers a loop: persists its declared spec and arms the first cadence
    /// window (and/or duration deadlines) on the sweep set.
    ///
    /// # Errors
    ///
    /// Refuses a duplicate registration and propagates store failures.
    pub async fn register(
        &self,
        loop_id: WorkflowId,
        namespace: String,
        spec: WorkloopSpec,
    ) -> Result<WorkloopRecord, WorkloopError> {
        if self.store.get_workloop(&loop_id).await?.is_some() {
            return Err(WorkloopError::AlreadyRegistered { loop_id });
        }
        let now = (self.now)();
        let next_window_at = match spec.arming().cadence_period() {
            Some(period) => Some(initial_window(now, period)?),
            None => None,
        };
        let mut record = WorkloopRecord {
            loop_id,
            namespace,
            spec,
            window_seq: 0,
            next_window_at,
            next_check_at: None,
            last_iteration_closed_window: None,
            invariant_health: std::collections::BTreeMap::new(),
            registered_at: now,
            updated_at: now,
        };
        record.next_check_at = next_check_at(&record);
        self.store.put_workloop(record.clone()).await?;
        Ok(record)
    }

    /// Wakes a loop's current generation immediately — the signal-armed fire
    /// (R2.4): a declared signal arrived, so the iteration runs now, no
    /// window involved.
    ///
    /// # Errors
    ///
    /// Propagates the waker's failure.
    pub async fn wake_now(&self, loop_id: &WorkflowId) -> Result<(), WorkloopError> {
        self.waker.wake(loop_id).await
    }

    /// Deregisters a loop (retirement or death). Invariant current-state
    /// records are untouched — the current record survives indefinitely
    /// (R8.1).
    ///
    /// # Errors
    ///
    /// Propagates store failures.
    pub async fn deregister(&self, loop_id: &WorkflowId) -> Result<bool, WorkloopError> {
        Ok(self.store.remove_workloop(loop_id).await?)
    }

    /// Feed an iteration close into health accounting (R3.3): every declared
    /// invariant is sampled on the same tick — `Confirmed` resets its
    /// accounting, `Unconfirmed` accrues red-sample evidence and may exceed
    /// the count-form tolerance immediately. Returns the alarms raised.
    ///
    /// # Errors
    ///
    /// Refuses samples naming undeclared invariants; propagates store and
    /// append failures.
    pub async fn note_iteration_closed(
        &self,
        loop_id: &WorkflowId,
        samples: &[HealthSample],
    ) -> Result<Vec<InvariantAlarm>, WorkloopError> {
        let mut record = self.store.get_workloop(loop_id).await?.ok_or_else(|| {
            WorkloopError::NotRegistered {
                loop_id: loop_id.clone(),
            }
        })?;
        for sample in samples {
            if !record
                .spec
                .invariants()
                .iter()
                .any(|invariant| invariant.name == sample.invariant)
            {
                return Err(WorkloopError::UndeclaredInvariant {
                    loop_id: loop_id.clone(),
                    invariant: sample.invariant.clone(),
                });
            }
        }
        let now = (self.now)();
        for sample in samples {
            let state = record.invariant_health.entry(sample.invariant.clone());
            let state = state.or_default();
            match sample.status {
                HealthStatus::Confirmed => observe_confirmed(state, now),
                HealthStatus::Unconfirmed => {
                    observe_unconfirmed(state, UnconfirmedEvidence::SampleRed);
                }
            }
        }
        record.last_iteration_closed_window = Some(record.window_seq);

        let mut alarms = Vec::new();
        let window_ctx = window_context(&record);
        let anchor = record.registered_at;
        for invariant in record.spec.invariants().to_vec() {
            let Some(state) = record.invariant_health.get_mut(&invariant.name) else {
                continue;
            };
            if let Some(alarm) = peek_alarm(
                &invariant.name,
                &invariant.tolerance,
                state,
                anchor,
                now,
                window_ctx,
            ) {
                self.sink
                    .record_invariant_unconfirmed(loop_id, alarm.clone())
                    .await?;
                latch_alarm(state);
                alarms.push(alarm);
            }
        }

        record.next_check_at = next_check_at(&record);
        record.updated_at = now;
        self.store.put_workloop(record).await?;
        Ok(alarms)
    }

    /// One sweep pass over every due loop: fire elapsed cadence windows
    /// (recording `CadenceFired` through the one Recorder, then waking the
    /// generation), evaluate the dead-man switch (an iteration with no
    /// terminal by its next window is a missed window — R3.3a — counted
    /// against every invariant), and evaluate every declared tolerance,
    /// raising `InvariantUnconfirmed` with its cause on the one alarm path.
    ///
    /// # Errors
    ///
    /// Returns an error only when the due-set itself cannot be read; per-loop
    /// faults are carried in the report so one sick loop cannot silence the
    /// sweep for the rest.
    pub async fn tick(&self) -> Result<SweepReport, WorkloopError> {
        let now = (self.now)();
        let mut report = SweepReport::default();
        for record in self.store.due_workloops(now).await? {
            let loop_id = record.loop_id.clone();
            if let Err(error) = self.sweep_loop(record, now, &mut report).await {
                report.faults.push((loop_id, error.to_string()));
            }
            report.swept += 1;
        }
        Ok(report)
    }

    async fn sweep_loop(
        &self,
        mut record: WorkloopRecord,
        now: DateTime<Utc>,
        report: &mut SweepReport,
    ) -> Result<(), WorkloopError> {
        let loop_id = record.loop_id.clone();
        let mut wake_pending = false;

        // --- cadence half: fire the elapsed window (R4.3). ---
        if let (Some(period), Some(window_at)) =
            (record.spec.arming().cadence_period(), record.next_window_at)
            && window_at <= now
        {
            // Dead-man first: did the PREVIOUS fired window's iteration close?
            // An iteration that produced no terminal by its next window is a
            // missed window, evaluated engine-side, never waited on (R3.3a).
            if record.window_seq >= 1
                && record.last_iteration_closed_window != Some(record.window_seq)
            {
                for invariant in record.spec.invariants().to_vec() {
                    let state = record
                        .invariant_health
                        .entry(invariant.name.clone())
                        .or_default();
                    observe_unconfirmed(state, UnconfirmedEvidence::WindowMissed);
                }
            }

            let window_seq = record.window_seq.saturating_add(1);
            match self.sink.record_cadence_fired(&loop_id, window_seq).await? {
                // AlreadyRecorded is an acknowledgement-lost duplicate of OUR
                // own append (single writer): the fire stands and the owed
                // wake must still follow, exactly as for Recorded.
                RecordOutcome::Recorded | RecordOutcome::AlreadyRecorded => {
                    record.window_seq = window_seq;
                    record.next_window_at = Some(advance_window(window_at, period, now)?);
                    report.fired.push((loop_id.clone(), window_seq));
                    wake_pending = true;
                }
                RecordOutcome::RefusedTerminal => {
                    // The loop's run is terminal without a declared retirement:
                    // the engine positively knows the loop cannot run. Loop
                    // death is the simultaneous non-confirmation of every
                    // invariant it carried — the missed-window event fanned out
                    // with cause loop-dead (R4.3), recorded even though the run
                    // is terminal.
                    return self.declare_loop_dead(record, report).await;
                }
                RecordOutcome::RefusedRetired => {
                    // 🔴 A RETIREMENT IS NOT A DEATH.
                    //
                    // Retirement records `LoopRetired` + its terminal and THEN
                    // withdraws the sweep-set row, so a sweep landing between
                    // those two steps finds a registered loop whose run is
                    // terminal — the same observation a dead loop produces.
                    // Answering it with `declare_loop_dead` wrote an
                    // `AlarmCause::LoopDead` against every invariant of a loop
                    // that was decommissioned on purpose, and those alarms are
                    // durable and permanent.
                    //
                    // The row is withdrawn here instead. Doing so is not a
                    // race with the retirement's own `deregister`: removal is
                    // idempotent and reports whether a row existed, and the
                    // retirement logs when it finds the row already gone.
                    let removed = self.store.remove_workloop(&loop_id).await?;
                    tracing::info!(
                        %loop_id,
                        row_existed = removed,
                        "sweep found a retired workloop still on the sweep set and withdrew it; \
                         a declared retirement raises no alarms"
                    );
                    report.retired.push(loop_id);
                    return Ok(());
                }
            }
        }

        // --- evaluation half: every declared tolerance, one alarm path. ---
        let window_ctx = window_context(&record);
        let anchor = record.registered_at;
        for invariant in record.spec.invariants().to_vec() {
            let state = record
                .invariant_health
                .entry(invariant.name.clone())
                .or_default();
            if let Some(alarm) = peek_alarm(
                &invariant.name,
                &invariant.tolerance,
                state,
                anchor,
                now,
                window_ctx,
            ) {
                self.sink
                    .record_invariant_unconfirmed(&loop_id, alarm.clone())
                    .await?;
                latch_alarm(state);
                report.alarms.push((loop_id.clone(), alarm));
            }
        }

        record.next_check_at = next_check_at(&record);
        record.updated_at = now;
        self.store.put_workloop(record).await?;

        if wake_pending && let Err(error) = self.waker.wake(&loop_id).await {
            // The fire is durably recorded; a failed wake degrades to a
            // missed window at the next sweep, never to silence.
            report
                .faults
                .push((loop_id, format!("wake failed: {error}")));
        }
        Ok(())
    }

    async fn declare_loop_dead(
        &self,
        record: WorkloopRecord,
        report: &mut SweepReport,
    ) -> Result<(), WorkloopError> {
        let loop_id = record.loop_id.clone();
        let window_ctx = window_context(&record);
        for invariant in record.spec.invariants() {
            let state = record
                .invariant_health
                .get(&invariant.name)
                .cloned()
                .unwrap_or_default();
            let alarm = InvariantAlarm {
                invariant: invariant.name.clone(),
                cause: aion_core::AlarmCause::LoopDead,
                window_seq: window_ctx,
                last_confirmed_at: state.last_confirmed_at,
                consecutive_unconfirmed: state.consecutive_unconfirmed,
            };
            self.sink
                .record_invariant_unconfirmed(&loop_id, alarm.clone())
                .await?;
            report.alarms.push((loop_id.clone(), alarm));
        }
        // Deregister: a dead loop has no windows to sweep. Its alarms are
        // durably recorded, its history keeps the un-retired terminal, and
        // its invariant current-state records survive indefinitely (R8.1).
        self.store.remove_workloop(&loop_id).await?;
        report.dead.push(loop_id);
        Ok(())
    }

    /// Runs the sweep loop until `shutdown` flips true. One task for every
    /// loop in the store — never a task per loop.
    pub async fn run(self: Arc<Self>, mut shutdown: tokio::sync::watch::Receiver<bool>) {
        let mut interval = tokio::time::interval(self.sweep_interval);
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        loop {
            tokio::select! {
                _ = interval.tick() => {
                    match self.tick().await {
                        Ok(report) => {
                            for (loop_id, fault) in &report.faults {
                                tracing::warn!(%loop_id, fault, "workloop sweep fault");
                            }
                        }
                        Err(error) => {
                            tracing::error!(%error, "workloop sweep pass failed");
                        }
                    }
                }
                changed = shutdown.changed() => {
                    if changed.is_err() || *shutdown.borrow() {
                        break;
                    }
                }
            }
        }
    }
}

/// Withdraws every registration whose workflow has NO recorded history — the
/// boot half of `start_workloop`'s non-atomic register-then-start. Returns the
/// loops withdrawn.
///
/// # 🔴 WHY THIS RUNS AT BOOT, BEFORE THE SWEEP TASK EXISTS, AND NOWHERE ELSE
///
/// `Engine::start_workloop` writes the sweep-set row BEFORE the workflow
/// exists, and must: the reverse order lets generation 1 reach
/// `close_iteration` before its own registration lands and fail the run. The
/// price is a window in which a crash strands a row for a workflow that has no
/// history and never will.
///
/// Nothing in a RUNNING process can tell that stranded row from the width of
/// an in-flight start — they are the same observation — so a sweep that
/// withdrew empty-history rows would race the very birth window the ordering
/// exists to protect, and a loop would lose its registration between its start
/// and its first close. A timeout would only make the race longer, and this
/// codebase does not assume durations.
///
/// Boot has the fact the sweep lacks: no start is in flight across a process
/// boundary. A registration whose workflow has no history at the instant this
/// engine comes up is therefore provably orphaned, with no clock involved.
/// It is a FREE function, called before the service and its sweep task are
/// assembled, so the ordering is a property of the call site rather than a
/// hope about which of two tasks ticks first.
///
/// # Errors
///
/// Propagates store failures; a row that cannot be decoded is left in place
/// and reported, never silently withdrawn — an unreadable row is not evidence
/// that its loop never started.
pub async fn withdraw_unstarted_registrations(
    workloop_store: &Arc<dyn WorkloopStore>,
    store: &Arc<dyn aion_store::EventStore>,
) -> Result<Vec<WorkflowId>, WorkloopError> {
    let listing = workloop_store.list_workloops().await?;
    let mut withdrawn = Vec::new();
    for record in listing.workloops {
        let loop_id = record.loop_id;
        if !store.read_history(&loop_id).await?.is_empty() {
            continue;
        }
        let existed = workloop_store.remove_workloop(&loop_id).await?;
        tracing::warn!(
            %loop_id,
            row_existed = existed,
            "withdrawing a workloop registration whose workflow has no recorded history: its \
             start never landed, so the sweep set carried a row for a loop that never ran and \
             the dead-man switch would have alarmed on it forever"
        );
        withdrawn.push(loop_id);
    }
    for row in listing.undecodable {
        tracing::error!(
            loop_id = %row.loop_id,
            "workloop registration row does not decode; it is left in place rather than \
             withdrawn, because an unreadable row is not evidence that its loop never started"
        );
    }
    Ok(withdrawn)
}

/// The window context alarms carry: the current fired window on a cadenced
/// loop that has fired at least once, `None` otherwise (signal-only loops
/// have no windows; a loop that never fired has none yet).
pub(crate) fn window_context(record: &WorkloopRecord) -> Option<u64> {
    (record.spec.arming().cadence_period().is_some() && record.window_seq >= 1)
        .then_some(record.window_seq)
}