aion/workloop/

close.rs

//! The iteration-close path, shared by the engine API verb and the
//! `close_iteration/3` NIF.
//!
//! Both callers must produce the SAME durable effect — one atomic
//! `[IterationClosed + WorkflowContinuedAsNew + successor WorkflowStarted]`
//! batch through the loop's one Recorder, no successor process, invariant
//! records installed with retention pruning, samples fed into tolerance
//! accounting — so the behaviour lives here once rather than in each caller.
//! A close reached from compiled workflow code and a close reached from an
//! operator API call are the same close.

use std::sync::Arc;

use aion_core::{Event, RunId, WorkflowId};
use aion_store::EventStore;
use aion_store::visibility::VisibilityStore;
use aion_store::workloop::WorkloopStore;
use chrono::Utc;

use super::iteration::{self, WorkloopIterationClose};
use super::service::{WorkloopService, window_context};
use crate::durability::Recorder;
use crate::error::EngineError;
use crate::registry::{Registry, TerminalOutcome};

/// The engine components one iteration close needs.
///
/// Held by both `Engine` (through its workloop runtime) and the NIF bridge,
/// so compiled workflow code reaches exactly the engine-side close the API
/// verb reaches.
#[derive(Clone)]
pub struct IterationCloseContext {
    /// Registration and invariant-record store.
    pub workloop_store: Arc<dyn WorkloopStore>,
    /// The cadence service, for health accounting.
    pub service: Arc<WorkloopService>,
    /// Event store backing the loop's history.
    pub store: Arc<dyn EventStore>,
    /// Visibility projection store.
    pub visibility_store: Arc<dyn VisibilityStore>,
    /// Registry holding the closing generation's live handle, if resident.
    pub registry: Arc<Registry>,
}

/// Closes the current iteration at the continue-as-new boundary (R3.1).
///
/// Derives one health sample per declared invariant from the taken routes
/// (R3.3), records the boundary batch through the loop's Recorder — spawning
/// NO successor process (R13.3) — installs the produced invariant
/// current-state records with retention pruning (R7/R8), and feeds the
/// samples into tolerance accounting. Returns the successor generation's run
/// id.
///
/// # Errors
///
/// Refuses an unregistered loop, a terminal run, pending work, and undeclared
/// invariants; propagates store/append failures.
pub async fn close_iteration(
    context: &IterationCloseContext,
    loop_id: &WorkflowId,
    close: WorkloopIterationClose,
) -> Result<RunId, EngineError> {
    let record = context
        .workloop_store
        .get_workloop(loop_id)
        .await
        .map_err(EngineError::from)?
        .ok_or_else(|| EngineError::InvalidState {
            reason: format!("workflow {loop_id} is not a registered workloop"),
        })?;
    let window_seq = window_context(&record);
    let spec = record.spec.clone();
    let retention = record.spec.retention();

    let carry_for_notify = close.carry.clone();
    let close_for_recorder = close;
    let outcome = with_loop_recorder(context, loop_id, move |recorder, history| {
        let spec = spec.clone();
        let close = close_for_recorder.clone();
        let loop_id = loop_id.clone();
        Box::pin(async move {
            let run_id = active_run_id(history).ok_or_else(|| {
                crate::durability::DurabilityError::HistoryShape {
                    reason: format!("workloop {loop_id} has no recorded generation"),
                }
            })?;
            iteration::close_iteration(
                recorder,
                iteration::IterationContext {
                    history,
                    run_id: &run_id,
                    loop_id: &loop_id,
                    spec: &spec,
                    window_seq,
                    recorded_at: Utc::now(),
                },
                close,
            )
            .await
            .map_err(|error| crate::durability::DurabilityError::HistoryShape {
                reason: error.to_string(),
            })
        })
    })
    .await?;

    // The closing generation's live handle (if the iteration ran resident) is
    // retired exactly as continue-as-new retires it: notify the continuation
    // and drop the registry entry. The successor stays unregistered —
    // SUSPENDED — until a cadence fire or an armed signal wakes it. That is
    // the whole park: no resident process between fires.
    if let Some(handle) = registry_handle(context, loop_id)? {
        let closed_run = handle.run_id().clone();
        handle.completion().notify(TerminalOutcome::ContinuedAsNew {
            input: carry_for_notify,
            workflow_type: None,
            parent_run_id: closed_run.clone(),
        });
        context.registry.remove(loop_id, &closed_run)?;
    }

    // Invariant current-state records (R7) + declared-window retention (R8):
    // ONE write per invariant installs the record and applies the declared
    // window in the same commit, so retention that is declared is retention
    // that happens — and a park costs `2 + N` durable commits rather than
    // `2 + 2N`.
    let cutoff = retention_cutoff(Utc::now(), retention)?;
    for state_record in outcome.records.clone() {
        context
            .workloop_store
            .put_invariant_record(state_record, cutoff)
            .await
            .map_err(EngineError::from)?;
    }

    context
        .service
        .note_iteration_closed(loop_id, &outcome.samples)
        .await
        .map_err(EngineError::from)?;

    Ok(outcome.next_run_id)
}

/// The instant prior invariant generations are pruned against: `now` less the
/// declared retention window.
///
/// # 🔴 A FALLBACK HERE POINTS THE WRONG WAY, SO THERE IS NONE
///
/// This was `chrono::Duration::from_std(retention).unwrap_or_else(|_|
/// chrono::Duration::zero())`. A zero fallback makes the cutoff `now`, which
/// prunes EVERY prior generation — the precise inverse of what an out-of-range
/// (that is, enormous) retention declares. A swallowed conversion whose
/// fallback destroys the data the declaration asked to keep is strictly worse
/// than a refused close: the close is retryable, the deleted generations are
/// not.
///
/// `WorkloopSpec` already refuses a retention this cannot convert, so a spec
/// declared through the engine cannot reach the first branch. It is still
/// propagated rather than asserted away: unreachable-by-construction is a
/// property of today's declaration path, and a spec is durable bytes that
/// outlive it.
fn retention_cutoff(
    now: chrono::DateTime<Utc>,
    retention: std::time::Duration,
) -> Result<chrono::DateTime<Utc>, EngineError> {
    let window =
        chrono::Duration::from_std(retention).map_err(|error| EngineError::InvalidState {
            reason: format!(
                "workloop retention window of {seconds}s cannot be expressed as a calendar \
                 duration ({error}), so no retention cutoff exists; refusing rather than \
                 pruning against a fallback that would delete every prior generation",
                seconds = retention.as_secs()
            ),
        })?;
    now.checked_sub_signed(window)
        .ok_or_else(|| EngineError::InvalidState {
            reason: format!(
                "subtracting the declared workloop retention window of {seconds}s from \
                 {now} left the representable calendar range, so no retention cutoff exists",
                seconds = retention.as_secs()
            ),
        })
}

fn registry_handle(
    context: &IterationCloseContext,
    loop_id: &WorkflowId,
) -> Result<Option<crate::registry::WorkflowHandle>, EngineError> {
    // aion#213: `with_loop_recorder` APPENDS DURABLY through whatever this
    // returns, so a first-match scan here is the coin toss `Registry::sole_handle`
    // exists to refuse — it would pick a generation's recorder at random and
    // write the loop's history through it.
    context.registry.sole_handle(loop_id)
}

/// Append through the loop's ONE Recorder: the live handle's recorder when
/// registered, a one-shot `Recorder::resume_at` when suspended (the
/// sanctioned non-resident pattern). The closure receives the history read
/// under the same acquisition, so check-then-append is not interleaved.
async fn with_loop_recorder<T>(
    context: &IterationCloseContext,
    loop_id: &WorkflowId,
    record: impl for<'a> FnOnce(
        &'a mut Recorder,
        &'a [Event],
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<Output = Result<T, crate::durability::DurabilityError>>
                + Send
                + 'a,
        >,
    >,
) -> Result<T, EngineError> {
    if let Some(handle) = registry_handle(context, loop_id)? {
        let recorder = handle.recorder();
        let mut recorder = recorder.lock().await;
        let history = context.store.read_history(loop_id).await?;
        let value = record(&mut recorder, &history).await?;
        return Ok(value);
    }
    let history = context.store.read_history(loop_id).await?;
    let head = history.iter().map(Event::seq).max().unwrap_or_default();
    let mut recorder = Recorder::resume_at(loop_id.clone(), Arc::clone(&context.store), head);
    if let Some(run_id) = active_run_id(&history) {
        recorder = recorder.with_visibility(run_id, Arc::clone(&context.visibility_store));
    }
    let value = record(&mut recorder, &history).await?;
    Ok(value)
}

/// The loop's currently active generation — the latest recorded start.
pub(crate) fn active_run_id(history: &[Event]) -> Option<RunId> {
    history.iter().rev().find_map(|event| match event {
        Event::WorkflowStarted { run_id, .. } => Some(run_id.clone()),
        _ => None,
    })
}

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

    use chrono::TimeZone;

    use super::retention_cutoff;
    use crate::error::EngineError;

    fn now() -> Result<chrono::DateTime<chrono::Utc>, Box<dyn std::error::Error>> {
        chrono::Utc
            .with_ymd_and_hms(2026, 8, 26, 12, 0, 0)
            .single()
            .ok_or_else(|| "test instant must be valid".into())
    }

    /// The ordinary case, and the control for the refusal below: a declared
    /// window subtracts to an instant that far predates `now`.
    #[test]
    fn a_declared_window_subtracts_to_its_own_past() -> Result<(), Box<dyn std::error::Error>> {
        let now = now()?;
        let cutoff = retention_cutoff(now, Duration::from_secs(14 * 86_400))?;
        assert_eq!(cutoff, now - chrono::Duration::days(14));
        Ok(())
    }

    /// 🔴 THE SWALLOWED FALLBACK POINTED THE WRONG WAY.
    ///
    /// With `unwrap_or_else(|_| Duration::zero())` this input produced a
    /// cutoff of exactly `now` — which prunes EVERY prior generation, the
    /// inverse of a retention window so long it could not be represented. The
    /// assertion is therefore not merely "an error is returned": it is that
    /// the function does not answer `now`, because `now` is the specific wrong
    /// answer the old code gave.
    #[test]
    fn an_unrepresentable_window_refuses_instead_of_pruning_everything()
    -> Result<(), Box<dyn std::error::Error>> {
        let now = now()?;
        let outcome = retention_cutoff(now, Duration::from_secs(u64::MAX / 1_000));
        assert!(
            !matches!(&outcome, Ok(cutoff) if *cutoff == now),
            "an unrepresentable retention must never yield a cutoff of `now`: that prunes \
             every prior generation, which is the opposite of what it declares"
        );
        let refusal = outcome
            .err()
            .ok_or("an unrepresentable retention window must be refused")?;
        assert!(
            matches!(&refusal, EngineError::InvalidState { reason }
                if reason.contains("retention")),
            "the refusal must name the retention window: {refusal}"
        );
        Ok(())
    }

    /// The other end of the same conversion: a window that converts but whose
    /// subtraction leaves the calendar range still has no cutoff, and must not
    /// saturate into one.
    #[test]
    fn a_window_that_underflows_the_calendar_refuses() {
        let early = chrono::DateTime::<chrono::Utc>::MIN_UTC + chrono::Duration::days(1);
        let outcome = retention_cutoff(early, Duration::from_secs(1_000 * 365 * 86_400));
        assert!(
            outcome.is_err(),
            "subtracting past the representable range must refuse, not saturate: {outcome:?}"
        );
    }
}