aion/workloop/

iteration.rs

//! Iteration close and retirement: the continue-as-new machinery, surfaced
//! for workloops (R3.1) — never a second loop mechanism.
//!
//! Each cadence tick opens a bounded history generation; `route start`
//! compiles to THIS boundary: the iteration's routes land as health samples
//! against the loop's invariants (R3.3), `IterationClosed` and the existing
//! `WorkflowContinuedAsNew` terminal are recorded through the loop's single
//! Recorder, and the successor generation's `WorkflowStarted` is recorded
//! WITHOUT spawning a process — between fires a workloop is store bytes plus
//! its sweep-set row (R13.3); the cadence service wakes the generation when
//! its window fires.

use aion_core::{
    Event, HealthSample, HealthStatus, Payload, RunId, WorkloopSpec, current_lease_terminal,
};
use aion_store::workloop::InvariantStateRecord;
use chrono::{DateTime, Utc};

use super::error::WorkloopError;
use crate::durability::{Recorder, WorkflowStartRecord};
use crate::lifecycle::continue_as_new::guard_no_pending_work;
use crate::time::retire_run_deadline;

/// One iteration close, as the compiled `route start(carry)` terminal hands
/// it to the engine.
#[derive(Clone, Debug)]
pub struct WorkloopIterationClose {
    /// Routes the iteration took, in order; the last is its terminal.
    pub routes: Vec<String>,
    /// Carry payload threaded into the successor generation (the route's
    /// payload — `route start(seen: ...)`).
    pub carry: Payload,
    /// Typed invariant current-state values the iteration produced, keyed by
    /// invariant name (type-checked at the surface, type-erased here).
    pub invariant_states: Vec<(String, Payload)>,
}

/// What a close produced: the derived samples, the successor generation, and
/// the invariant records to persist.
#[derive(Clone, Debug)]
pub struct IterationOutcome {
    /// The health samples derived from the routes (R3.3) and recorded on the
    /// `IterationClosed` event.
    pub samples: Vec<HealthSample>,
    /// The successor generation's run id.
    pub next_run_id: RunId,
    /// Invariant current-state records to install (R7), one per produced
    /// value.
    pub records: Vec<InvariantStateRecord>,
}

/// Derives one health sample per declared invariant from the iteration's
/// taken routes: `Confirmed` when any taken route is declared as confirming
/// the invariant, `Unconfirmed` otherwise — every invariant is sampled on the
/// same tick (R2.2), so a closing iteration never leaves an invariant
/// unsampled.
#[must_use]
pub fn derive_health_samples(
    spec: &WorkloopSpec,
    routes: &[String],
    window_seq: Option<u64>,
) -> Vec<HealthSample> {
    spec.invariants()
        .iter()
        .map(|invariant| {
            let confirmed = routes
                .iter()
                .any(|route| invariant.confirms.contains(route));
            HealthSample {
                invariant: invariant.name.clone(),
                status: if confirmed {
                    HealthStatus::Confirmed
                } else {
                    HealthStatus::Unconfirmed
                },
                window_seq,
            }
        })
        .collect()
}

/// Health samples for an iteration that FAILED: a failed iteration is a red
/// health sample against every invariant, not a failed loop (R3.3) — the next
/// iteration IS the retry.
#[must_use]
pub fn failed_iteration_samples(spec: &WorkloopSpec, window_seq: Option<u64>) -> Vec<HealthSample> {
    spec.invariants()
        .iter()
        .map(|invariant| HealthSample {
            invariant: invariant.name.clone(),
            status: HealthStatus::Unconfirmed,
            window_seq,
        })
        .collect()
}

/// Refuses invariant-state values naming undeclared invariants and builds
/// their store records.
fn invariant_records(
    spec: &WorkloopSpec,
    close: &WorkloopIterationClose,
    loop_id: &aion_core::WorkflowId,
    window_seq: Option<u64>,
    recorded_at: DateTime<Utc>,
) -> Result<Vec<InvariantStateRecord>, WorkloopError> {
    close
        .invariant_states
        .iter()
        .map(|(name, payload)| {
            let invariant = spec
                .invariants()
                .iter()
                .find(|invariant| &invariant.name == name)
                .ok_or_else(|| WorkloopError::UndeclaredInvariant {
                    loop_id: loop_id.clone(),
                    invariant: name.clone(),
                })?;
            Ok(InvariantStateRecord {
                loop_id: loop_id.clone(),
                invariant: name.clone(),
                payload: payload.clone(),
                record_type: invariant.record_type.clone(),
                window_seq,
                recorded_at,
            })
        })
        .collect()
}

/// The current generation's recorded workflow type and package version — the
/// successor is pinned to the same recorded version; version upgrade for
/// long-lived loops rides the deploy surface, not the iteration boundary.
fn current_generation_identity(
    history: &[Event],
) -> Result<(String, aion_core::PackageVersion), WorkloopError> {
    history
        .iter()
        .rev()
        .find_map(|event| {
            if let Event::WorkflowStarted {
                workflow_type,
                package_version,
                ..
            } = event
            {
                Some((workflow_type.clone(), package_version.clone()))
            } else {
                None
            }
        })
        .ok_or_else(|| WorkloopError::Engine {
            reason: "workloop history has no WorkflowStarted".to_owned(),
        })
}

/// Everything the iteration boundary needs to know about the loop and its
/// current generation, gathered by the caller under the recorder lock.
#[derive(Clone, Copy, Debug)]
pub struct IterationContext<'a> {
    /// The loop's full history, read under the same recorder acquisition.
    pub history: &'a [Event],
    /// The closing generation's run id.
    pub run_id: &'a RunId,
    /// The loop's workflow identity.
    pub loop_id: &'a aion_core::WorkflowId,
    /// The loop's declared spec.
    pub spec: &'a WorkloopSpec,
    /// The current fired window, when the loop is cadenced and has fired.
    pub window_seq: Option<u64>,
    /// Deterministic recording timestamp for the boundary's events.
    pub recorded_at: DateTime<Utc>,
}

/// Closes the current iteration through the loop's single Recorder: records
/// `IterationClosed { routes, health_samples }`, the existing
/// `WorkflowContinuedAsNew` terminal carrying the carry payload, retires the
/// run's deadline, and records the successor generation's `WorkflowStarted` —
/// all sequential appends under ONE recorder, no process spawned.
///
/// The caller holds the recorder lock for the whole call and owns feeding the
/// returned samples into the cadence service's health accounting plus
/// persisting the returned invariant records with retention pruning.
///
/// # Errors
///
/// Refuses a close on a terminal run, with pending work (the continue-as-new
/// guard, unchanged), or naming undeclared invariants; propagates append
/// failures.
pub async fn close_iteration(
    recorder: &mut Recorder,
    context: IterationContext<'_>,
    close: WorkloopIterationClose,
) -> Result<IterationOutcome, WorkloopError> {
    let IterationContext {
        history,
        run_id,
        loop_id,
        spec,
        window_seq,
        recorded_at,
    } = context;
    if current_lease_terminal(history).is_some() {
        return Err(WorkloopError::Engine {
            reason: format!("workloop {loop_id} run {run_id} already recorded a terminal"),
        });
    }
    // 🔴 THE PENDING-WORK GUARD IS SCOPED TO THIS GENERATION'S SEGMENT.
    //
    // `guard_no_pending_work` accumulates unsettled activities and children by
    // forward-scanning whatever slice it is handed, and `WorkflowStarted` is in
    // its NO-OP arm — a generation boundary does not clear the pending sets. On
    // an ordinary workflow that is invisible, because the history it is handed
    // is one run or a few. On a workloop it is neither invisible nor harmless:
    // the history is EVERY generation the loop has ever had, so one unsettled
    // activity left behind by generation 7 would refuse the close of generation
    // 4,000, permanently, and the loop would stop parking for a reason four
    // thousand generations in its past.
    //
    // Pending work is run-scoped by nature — a generation can only settle what
    // it started — so the segment is the correct unit, and it is also the
    // bounded one: this scan no longer grows with the loop's age.
    guard_no_pending_work(aion_core::run_segment(history, run_id)).map_err(|error| {
        WorkloopError::Engine {
            reason: error.to_string(),
        }
    })?;

    let samples = derive_health_samples(spec, &close.routes, window_seq);
    let records = invariant_records(spec, &close, loop_id, window_seq, recorded_at)?;
    let (workflow_type, package_version) = current_generation_identity(history)?;

    // ONE atomic batch: IterationClosed + WorkflowContinuedAsNew + the
    // successor generation's WorkflowStarted — no crash window between the
    // boundary's halves, and NO process spawned for the successor (R13.3);
    // the cadence service wakes it when its window fires. The successor is
    // pinned to the predecessor's recorded package version.
    let next_run_id = RunId::new_v4();
    recorder
        .record_workloop_iteration_boundary(
            recorded_at,
            close.routes.clone(),
            samples.clone(),
            close.carry.clone(),
            run_id.clone(),
            WorkflowStartRecord {
                workflow_type,
                input: close.carry,
                run_id: next_run_id.clone(),
                parent_run_id: Some(run_id.clone()),
                parent_workflow_id: None,
                package_version,
            },
        )
        .await?;
    retire_run_deadline(recorder, history, run_id).await?;

    Ok(IterationOutcome {
        samples,
        next_run_id,
        records,
    })
}

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

    use aion_core::{InvariantSpec, ToleranceSpec, WorkloopArming};

    use super::*;

    fn spec() -> Result<WorkloopSpec, Box<dyn std::error::Error>> {
        Ok(WorkloopSpec::new(
            WorkloopArming::every(Duration::from_secs(1500))?,
            vec![
                InvariantSpec {
                    name: String::from("serving"),
                    record_type: String::from("ServeState"),
                    tolerance: ToleranceSpec::count(3),
                    confirms: vec![String::from("sweep"), String::from("dispatch")],
                },
                InvariantSpec {
                    name: String::from("drained"),
                    record_type: String::from("DrainState"),
                    tolerance: ToleranceSpec::count(0),
                    confirms: vec![String::from("drain")],
                },
            ],
            Duration::from_secs(86_400),
        )?)
    }

    #[test]
    fn every_invariant_is_sampled_on_the_same_tick() -> Result<(), Box<dyn std::error::Error>> {
        let samples = derive_health_samples(
            &spec()?,
            &[String::from("sweep"), String::from("start")],
            Some(4),
        );
        assert_eq!(samples.len(), 2);
        assert_eq!(samples[0].invariant, "serving");
        assert_eq!(samples[0].status, HealthStatus::Confirmed);
        assert_eq!(samples[0].window_seq, Some(4));
        // The other invariant's confirming route was not taken: an unhealthy
        // sample, not an unsampled invariant.
        assert_eq!(samples[1].invariant, "drained");
        assert_eq!(samples[1].status, HealthStatus::Unconfirmed);
        Ok(())
    }

    #[test]
    fn a_failed_iteration_reds_every_invariant() -> Result<(), Box<dyn std::error::Error>> {
        let samples = failed_iteration_samples(&spec()?, None);
        assert!(
            samples
                .iter()
                .all(|sample| sample.status == HealthStatus::Unconfirmed)
        );
        assert_eq!(samples.len(), 2);
        Ok(())
    }
}