aion-store 0.31.0

//! The stream-head handshake between a workflow's event stream and its
//! visibility row: how a store decides, per workflow, whether the row can
//! answer "is this one in flight?" on its own or the history must be read.
//!
//! Both backends run the same fold ([`verdict`]) over the same two facts —
//! the stream's head `seq` ([`StreamHead`]) and the row it may or may not
//! have — so `list_active` and `list_paused` give the same answer whichever
//! store served them, and a finished workflow is never opened again at boot.

use aion_core::{WorkflowId, WorkflowStatus};

use super::VisibilityRecord;

/// One workflow's event stream and the `seq` of its last appended event.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamHead {
    /// The workflow whose stream this is.
    pub workflow_id: WorkflowId,
    /// The `seq` of the stream's last event (Aion sequences are 1-based, so a
    /// stream with `n` events has head `n`). Never 0: a stream with no events
    /// is not listed.
    pub head_seq: u64,
}

/// What a workflow's row, checked against its stream head, settles.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RowVerdict {
    /// The row is at the head and its status is terminal: the workflow is
    /// finished, it is in neither list, and its history is NOT read.
    Finished,
    /// The row is at the head and its status is `Paused`: it belongs to the
    /// paused list, and its history is NOT read.
    Paused,
    /// The row is at the head and its status is `Running`. The workflow is in
    /// flight, and the fold must still decide the exact status: a generation
    /// that continued as new projects `Running` on its row (the identity is
    /// alive) but folds `ContinuedAsNew`, which the active list excludes. The
    /// history is read — the same read replay pays for the in-flight set.
    InFlight,
    /// No row, or a row behind (or ahead of) the head: nothing is settled and
    /// the history decides, exactly as it did before rows carried a head.
    Unsettled,
}

/// Decides, from the row (if any) and the stream's head `seq`, whether the
/// row answers on its own.
///
/// A row is written AFTER its durable append and never ahead of it, so
/// `row.head_seq == head_seq` means no event has landed since the row's
/// fold. Terminal and `Paused` are fold statuses the projection never
/// rewrites, so such a row is authoritative for them by construction. The
/// projection's one rewrite (a continued generation reads `Running`) lands in
/// [`RowVerdict::InFlight`], which the fold re-derives. An unstamped row
/// (`head_seq == 0`) is never at the head, because no event has `seq` 0.
#[must_use]
pub fn verdict(row: Option<&VisibilityRecord>, head_seq: u64) -> RowVerdict {
    let Some(row) = row else {
        return RowVerdict::Unsettled;
    };
    if row.head_seq == 0 || row.head_seq != head_seq {
        return RowVerdict::Unsettled;
    }
    match row.status {
        WorkflowStatus::Completed
        | WorkflowStatus::Failed
        | WorkflowStatus::Cancelled
        | WorkflowStatus::TimedOut
        | WorkflowStatus::ContinuedAsNew => RowVerdict::Finished,
        WorkflowStatus::Paused => RowVerdict::Paused,
        WorkflowStatus::Running => RowVerdict::InFlight,
    }
}

/// Which list, if either, a folded status belongs to. Shared by both
/// backends so the two lists partition histories identically.
#[must_use]
pub const fn list_for(status: WorkflowStatus) -> Option<ListMembership> {
    match status {
        WorkflowStatus::Running => Some(ListMembership::Active),
        WorkflowStatus::Paused => Some(ListMembership::Paused),
        WorkflowStatus::Completed
        | WorkflowStatus::Failed
        | WorkflowStatus::Cancelled
        | WorkflowStatus::TimedOut
        | WorkflowStatus::ContinuedAsNew => None,
    }
}

/// The list a workflow belongs to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ListMembership {
    /// `list_active`: projected status exactly `Running`.
    Active,
    /// `list_paused`: projected status exactly `Paused`.
    Paused,
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use aion_core::{RunId, WorkflowId, WorkflowStatus};
    use chrono::Utc;

    use super::{ListMembership, RowVerdict, list_for, verdict};
    use crate::visibility::VisibilityRecord;

    fn row(status: WorkflowStatus, head_seq: u64) -> VisibilityRecord {
        VisibilityRecord {
            namespace: String::from("default"),
            workflow_id: WorkflowId::new(uuid::Uuid::from_u128(7)),
            run_id: RunId::new(uuid::Uuid::from_u128(8)),
            workflow_type: String::from("checkout"),
            status,
            started_at: Utc::now(),
            updated_at: Utc::now(),
            ended_at: None,
            parent: None,
            display_name: None,
            kind: None,
            failed_step: None,
            failure_reason: None,
            search_attributes: HashMap::new(),
            outstanding_leases: Vec::new(),
            package_version: None,
            head_seq,
        }
    }

    #[test]
    fn a_terminal_row_at_the_head_is_finished_and_unread() {
        for status in [
            WorkflowStatus::Completed,
            WorkflowStatus::Failed,
            WorkflowStatus::Cancelled,
            WorkflowStatus::TimedOut,
        ] {
            assert_eq!(verdict(Some(&row(status, 5)), 5), RowVerdict::Finished);
        }
    }

    #[test]
    fn a_paused_row_at_the_head_is_paused_and_unread() {
        assert_eq!(
            verdict(Some(&row(WorkflowStatus::Paused, 3)), 3),
            RowVerdict::Paused
        );
    }

    #[test]
    fn a_running_row_at_the_head_is_in_flight_and_the_fold_decides() {
        assert_eq!(
            verdict(Some(&row(WorkflowStatus::Running, 3)), 3),
            RowVerdict::InFlight
        );
    }

    #[test]
    fn a_row_behind_or_ahead_of_the_head_settles_nothing() {
        assert_eq!(
            verdict(Some(&row(WorkflowStatus::Completed, 4)), 5),
            RowVerdict::Unsettled
        );
        assert_eq!(
            verdict(Some(&row(WorkflowStatus::Completed, 6)), 5),
            RowVerdict::Unsettled
        );
    }

    #[test]
    fn an_unstamped_row_is_never_at_the_head() {
        // A pre-upgrade row decodes with head_seq 0; the stream head is never
        // 0, and the compare alone would already refuse it — the explicit
        // check keeps the contract honest if a caller ever passes head 0.
        assert_eq!(
            verdict(Some(&row(WorkflowStatus::Completed, 0)), 0),
            RowVerdict::Unsettled
        );
    }

    #[test]
    fn no_row_settles_nothing() {
        assert_eq!(verdict(None, 5), RowVerdict::Unsettled);
    }

    #[test]
    fn the_two_lists_partition_the_fold_exactly() {
        assert_eq!(
            list_for(WorkflowStatus::Running),
            Some(ListMembership::Active)
        );
        assert_eq!(
            list_for(WorkflowStatus::Paused),
            Some(ListMembership::Paused)
        );
        for status in [
            WorkflowStatus::Completed,
            WorkflowStatus::Failed,
            WorkflowStatus::Cancelled,
            WorkflowStatus::TimedOut,
            WorkflowStatus::ContinuedAsNew,
        ] {
            assert_eq!(list_for(status), None);
        }
    }
}