aion-rs 0.31.0

//! Production [`LoopEventSink`]: every workloop event through the loop's ONE
//! Recorder.
//!
//! Two acquisition paths, one writer: a REGISTERED loop's events append under
//! its live handle's recorder lock; an unregistered (suspended, store-bytes-
//! only) loop's events append through a one-shot `Recorder::resume_at` — the
//! sanctioned pattern for a run with no live recorder (see
//! `lifecycle::pause`'s discipline note). The two can never race: `resume_at`
//! is used only when no registry entry exists.

use aion_core::{Event, InvariantAlarm, RunId, WorkflowId};
use aion_store::EventStore;
use aion_store::visibility::VisibilityStore;
use async_trait::async_trait;
use chrono::Utc;
use std::sync::Arc;

use super::error::WorkloopError;
use super::service::LoopEventSink;
use crate::durability::Recorder;
use crate::engine_seam::RecordOutcome;
use crate::registry::Registry;

/// Recorder-backed sink used by the engine's cadence service.
pub struct EngineLoopEventSink {
    registry: Arc<Registry>,
    store: Arc<dyn EventStore>,
    visibility_store: Arc<dyn VisibilityStore>,
}

impl EngineLoopEventSink {
    /// Builds the sink over the engine's registry and stores.
    #[must_use]
    pub fn new(
        registry: Arc<Registry>,
        store: Arc<dyn EventStore>,
        visibility_store: Arc<dyn VisibilityStore>,
    ) -> Self {
        Self {
            registry,
            store,
            visibility_store,
        }
    }

    fn registered_recorder(
        &self,
        loop_id: &WorkflowId,
    ) -> Result<Option<crate::registry::WorkflowHandle>, WorkloopError> {
        // aion#213: the loop's ONE handle. What comes back is APPENDED THROUGH
        // (see the method below), so a first-match scan would pick a
        // generation's recorder at random and write the loop's history with it.
        self.registry
            .sole_handle(loop_id)
            .map_err(|error| WorkloopError::Engine {
                reason: error.to_string(),
            })
    }

    /// Append through the right recorder for the loop's residency, holding
    /// the one-writer law. `refuse_terminal` decides whether an active-run
    /// terminal refuses the append (cadence fires) or not (alarms).
    async fn record_with_recorder<F>(
        &self,
        loop_id: &WorkflowId,
        refuse_terminal: bool,
        record: F,
    ) -> Result<RecordOutcome, WorkloopError>
    where
        F: for<'a> AsyncFnOnce(&'a mut Recorder) -> Result<(), crate::durability::DurabilityError>,
    {
        if let Some(handle) = self.registered_recorder(loop_id)? {
            let recorder = handle.recorder();
            let mut recorder = recorder.lock().await;
            let history = self.store.read_history(loop_id).await?;
            if refuse_terminal && let Some(refusal) = terminal_refusal(&history) {
                return Ok(refusal);
            }
            record(&mut recorder).await?;
            return Ok(RecordOutcome::Recorded);
        }

        let history = self.store.read_history(loop_id).await?;
        // 🔴 A LOOP WITH NO RECORDED START IS NOT A LOOP TO FIRE AT.
        //
        // `start_workloop` writes the sweep-set row BEFORE the workflow
        // exists, so the loop can be registered for the width of one start.
        // A crash in that window leaves the row behind permanently. Either
        // way, appending a `CadenceFired` here would open the workflow's
        // history with an event that is not a `WorkflowStarted` — a history
        // no replay, projection or recovery can read. The fault is REPORTED
        // (the sweep carries it per-loop and keeps going) rather than
        // absorbed, and boot reconciliation withdraws the genuinely orphaned
        // rows.
        if history.is_empty() {
            return Err(WorkloopError::Engine {
                reason: format!(
                    "workloop {loop_id} is on the sweep set but its workflow has no recorded \
                     history, so there is nothing to fire at. This is either the width of an \
                     in-flight `start_workloop` — the registration precedes the start by \
                     design — or a registration whose start never landed, which engine boot \
                     reconciliation withdraws"
                ),
            });
        }
        if refuse_terminal && let Some(refusal) = terminal_refusal(&history) {
            return Ok(refusal);
        }
        let head = history.iter().map(Event::seq).max().unwrap_or_default();
        let mut recorder = Recorder::resume_at(loop_id.clone(), Arc::clone(&self.store), head);
        if let Some(run_id) = active_run_id(&history) {
            recorder = recorder.with_visibility(run_id, Arc::clone(&self.visibility_store));
        }
        record(&mut recorder).await?;
        Ok(RecordOutcome::Recorded)
    }
}

#[async_trait]
impl LoopEventSink for EngineLoopEventSink {
    async fn record_cadence_fired(
        &self,
        loop_id: &WorkflowId,
        window_seq: u64,
    ) -> Result<RecordOutcome, WorkloopError> {
        self.record_with_recorder(loop_id, true, async move |recorder: &mut Recorder| {
            recorder.record_cadence_fired(Utc::now(), window_seq).await
        })
        .await
    }

    async fn record_invariant_unconfirmed(
        &self,
        loop_id: &WorkflowId,
        alarm: InvariantAlarm,
    ) -> Result<(), WorkloopError> {
        // The alarm 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. InvariantUnconfirmed is status-invisible, so the
        // terminal projection is untouched.
        self.record_with_recorder(loop_id, false, async move |recorder: &mut Recorder| {
            recorder
                .record_invariant_unconfirmed(Utc::now(), alarm)
                .await
        })
        .await
        .map(|_| ())
    }
}

/// The refusal a cadence fire earns when the loop's active run (latest
/// `WorkflowStarted`) already recorded a terminal — and WHICH refusal.
///
/// `None` means the run is live and the fire may proceed.
///
/// # 🔴 THE TWO TERMINALS ARE TOLD APART HERE
///
/// A terminal alone is the engine's positive evidence that the loop cannot
/// run: [`RecordOutcome::RefusedTerminal`], which the sweep answers by
/// declaring the loop dead and alarming every invariant. A terminal preceded
/// by `LoopRetired` in the SAME run segment is a declared stop:
/// [`RecordOutcome::RefusedRetired`], which the sweep answers by withdrawing
/// the row quietly. Reading only "is there a terminal" made a clean retirement
/// indistinguishable from a death, and the retirement path passes through that
/// exact state on its way out.
///
/// The scan is segment-scoped, not whole-history: a `LoopRetired` can only
/// speak for the run it was recorded in, and a workloop's history holds every
/// generation it ever had.
fn terminal_refusal(history: &[Event]) -> Option<RecordOutcome> {
    let run_id = active_run_id(history)?;
    crate::lifecycle::completion::terminal_outcome_from_history(history, &run_id)?;
    let retired = aion_core::run_segment(history, &run_id)
        .iter()
        .any(|event| matches!(event, Event::LoopRetired { .. }));
    Some(if retired {
        RecordOutcome::RefusedRetired
    } else {
        RecordOutcome::RefusedTerminal
    })
}

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 aion_core::{ContentType, EventEnvelope, PackageVersion, Payload, WorkflowId};

    use super::{Event, RecordOutcome, RunId, terminal_refusal};

    fn envelope(workflow_id: &WorkflowId, seq: u64) -> EventEnvelope {
        EventEnvelope {
            seq,
            recorded_at: chrono::Utc::now(),
            workflow_id: workflow_id.clone(),
        }
    }

    fn started(workflow_id: &WorkflowId, run_id: &RunId, seq: u64) -> Event {
        Event::WorkflowStarted {
            envelope: envelope(workflow_id, seq),
            workflow_type: String::from("queue_watch"),
            input: Payload::new(ContentType::Json, b"{}".to_vec()),
            run_id: run_id.clone(),
            parent_run_id: None,
            parent_workflow_id: None,
            package_version: PackageVersion::new("a".repeat(64)),
        }
    }

    fn retired(workflow_id: &WorkflowId, seq: u64) -> Event {
        Event::LoopRetired {
            envelope: envelope(workflow_id, seq),
            reason: String::from("decommissioned"),
        }
    }

    fn completed(workflow_id: &WorkflowId, seq: u64) -> Event {
        Event::WorkflowCompleted {
            envelope: envelope(workflow_id, seq),
            result: Payload::new(ContentType::Json, b"{}".to_vec()),
        }
    }

    /// A live run is no refusal at all — the control for both cases below.
    #[test]
    fn a_live_run_refuses_no_cadence_fire() {
        let workflow_id = WorkflowId::new_v4();
        let run = RunId::new_v4();
        assert!(terminal_refusal(&[started(&workflow_id, &run, 1)]).is_none());
    }

    /// 🔴 THE TWO TERMINALS ARE TOLD APART, AND THE DIFFERENCE DECIDES WHETHER
    /// A LOOP IS DECLARED DEAD.
    ///
    /// An undeclared terminal is positive evidence the loop cannot run, and the
    /// sweep answers it with `AlarmCause::LoopDead` against every invariant. A
    /// terminal preceded by `LoopRetired` is a declared stop and must raise
    /// nothing. Retirement passes through exactly this state on its way out —
    /// terminal recorded, sweep-set row not yet withdrawn — so collapsing the
    /// two wrote a permanent death record into the history of a loop that was
    /// decommissioned on purpose.
    #[test]
    fn a_terminal_without_a_retirement_is_a_death_and_with_one_is_not() {
        let workflow_id = WorkflowId::new_v4();
        let run = RunId::new_v4();

        let died = vec![started(&workflow_id, &run, 1), completed(&workflow_id, 2)];
        assert!(matches!(
            terminal_refusal(&died),
            Some(RecordOutcome::RefusedTerminal)
        ));

        let was_retired = vec![
            started(&workflow_id, &run, 1),
            retired(&workflow_id, 2),
            completed(&workflow_id, 3),
        ];
        assert!(matches!(
            terminal_refusal(&was_retired),
            Some(RecordOutcome::RefusedRetired)
        ));
    }

    /// 🔴 A `LoopRetired` SPEAKS ONLY FOR THE RUN IT WAS RECORDED IN.
    ///
    /// A workloop's history holds every generation it ever had. A whole-history
    /// scan would let a retirement recorded generations ago make a LATER
    /// generation's undeclared death read as a declared stop — the loop would
    /// die silently and the sweep would withdraw it without an alarm. This is
    /// the case a non-segment-scoped check waves through, and it is reachable:
    /// a retirement that failed after recording its marker leaves exactly this
    /// shape.
    #[test]
    fn a_prior_generations_retirement_does_not_excuse_this_generations_death() {
        let workflow_id = WorkflowId::new_v4();
        let old_run = RunId::new_v4();
        let current_run = RunId::new_v4();
        let history = vec![
            started(&workflow_id, &old_run, 1),
            retired(&workflow_id, 2),
            started(&workflow_id, &current_run, 3),
            completed(&workflow_id, 4),
        ];

        // Fixture control: the marker really is in the history being scanned,
        // so a green here is scoping and not an absent event.
        assert!(
            history
                .iter()
                .any(|event| matches!(event, Event::LoopRetired { .. })),
            "fixture control: the history must carry a LoopRetired for this to prove scoping"
        );
        assert!(
            matches!(
                terminal_refusal(&history),
                Some(RecordOutcome::RefusedTerminal)
            ),
            "a retirement in an earlier generation must not excuse this one's death"
        );
    }
}