aion/durability/

error.rs

//! `NonDeterminismError` and `DurabilityError` taxonomy.

use aion_core::{SearchAttributeError, WorkflowId};
use aion_store::StoreError;

/// A deterministic replay mismatch between the workflow command stream and recorded history.
#[derive(thiserror::Error, Clone, Debug, PartialEq, Eq)]
#[error(
    "non-determinism in workflow {workflow_id} at sequence {seq}: expected {expected}, found {found}"
)]
pub struct NonDeterminismError {
    /// Workflow whose recorded history diverged from the replayed command stream.
    pub workflow_id: WorkflowId,
    /// Sequence position of the recorded event at the cursor mismatch.
    pub seq: u64,
    /// Shape of the command the workflow issued, including family and correlation key.
    pub expected: String,
    /// Shape of the recorded event found at the cursor position, including family and key.
    pub found: String,
}

/// Errors returned by durability recording, replay, and recovery operations.
#[derive(thiserror::Error, Debug)]
pub enum DurabilityError {
    /// The backing event store rejected or failed a durability operation.
    #[error("store error: {0}")]
    Store(#[from] StoreError),

    /// Replay detected that workflow code no longer matches recorded history.
    #[error("non-determinism violation: {0}")]
    NonDeterminism(#[from] NonDeterminismError),

    /// Recorded history is malformed or internally inconsistent.
    #[error("history shape error: {reason}")]
    HistoryShape {
        /// Human-readable description of the malformed recorded history.
        reason: String,
    },

    /// A search attribute update did not satisfy the registered schema.
    #[error("search attribute validation error: {0}")]
    SearchAttribute(#[from] SearchAttributeError),

    /// A durable write was refused because this engine's task epoch has closed.
    ///
    /// 🔴 DELIBERATELY NOT [`crate::EngineError::EngineTaskEpochClosed`]. That
    /// variant carries a documented single-construction-site invariant —
    /// `aion-client`'s `map_engine_error` gives it no arm on the stated ground
    /// that it is unreachable through any transport-exposed operation — and a
    /// second construction site would silently take that transport's catch-all.
    /// This is a different layer with a different audience: it surfaces as
    /// `{error, _}` inside running workflow code, never over the wire.
    #[error("engine task epoch closed: {reason}")]
    EngineTaskEpochClosed {
        /// What was refused, and what it means for the run.
        reason: String,
    },

    /// A durable write was refused because it was offered on behalf of a run
    /// whose generation the history has already moved past (aion#213).
    ///
    /// Continue-as-new does not stop the predecessor's process at the instant
    /// its terminal lands — the API path never touches the pid, and the NIF
    /// path's `cancel_pid` follows the append — so a run that is durably over
    /// can still reach its recorder with work it had already begun. Sequencing
    /// that append would put a `TimerStarted` (or an activity record, or a
    /// signal) for a dead run inside the SUCCESSOR's segment, where replay
    /// reads it as the successor's own. Nothing is appended, and the run this
    /// names is over: the refusal is the correct outcome, not a fault to
    /// retry.
    #[error(
        "run `{run_id}` of workflow `{workflow_id}` cannot append: its generation has been \
         superseded, so the append belongs to no live run"
    )]
    RunSuperseded {
        /// Workflow whose history the append was offered into.
        workflow_id: aion_core::WorkflowId,
        /// The superseded run the append was offered on behalf of.
        run_id: aion_core::RunId,
    },
}

#[cfg(test)]
mod tests {
    use super::{DurabilityError, NonDeterminismError};
    use aion_core::WorkflowId;
    use aion_store::StoreError;

    fn non_determinism_error() -> NonDeterminismError {
        NonDeterminismError {
            workflow_id: WorkflowId::new(uuid::Uuid::nil()),
            seq: 42,
            expected: "activity schedule ordinal 7".to_owned(),
            found: "timer fired timer:named:deadline".to_owned(),
        }
    }

    #[test]
    fn non_determinism_display_includes_context() {
        let error = non_determinism_error();

        let message = error.to_string();

        assert!(message.contains("00000000-0000-0000-0000-000000000000"));
        assert!(message.contains("42"));
        assert!(message.contains("activity schedule ordinal 7"));
        assert!(message.contains("timer fired timer:named:deadline"));
    }

    /// The operator-facing `RunSuperseded` sentence: it names both identities,
    /// and it is ONE sentence rather than two halves with a hole between them.
    ///
    /// # 🔴 A WRAPPED LITERAL THAT LOSES ITS `\` LEAVES A HOLE NO TOOL CATCHES
    ///
    /// This message shipped once as `"…has been          superseded, …"` — the
    /// line-continuation backslash was dropped when the literal was re-wrapped.
    /// `cargo fmt` does not touch the inside of string literals and clippy has
    /// nothing to say about them, so nothing failed. The string is not internal:
    /// `aion-server`'s wire mapping sends `source.to_string()` to API callers
    /// verbatim, so an operator reads it in an incident. The run of spaces is
    /// what the assertion below looks for, because that is the shape the defect
    /// takes every time.
    #[test]
    fn run_superseded_reads_as_one_sentence_naming_both_identities() {
        // Two DISTINCT identifiers: with identical fixtures the two
        // `contains` assertions below collapse into one assertion twice, and
        // dropping either `{workflow_id}` or `{run_id}` from the message
        // would go unnoticed.
        let workflow_id = WorkflowId::new(uuid::Uuid::from_u128(0xA1));
        let run_id = aion_core::RunId::new(uuid::Uuid::from_u128(0xB2));
        let error = DurabilityError::RunSuperseded {
            workflow_id: workflow_id.clone(),
            run_id: run_id.clone(),
        };

        let message = error.to_string();

        assert!(message.contains(&workflow_id.to_string()));
        assert!(message.contains(&run_id.to_string()));
        assert!(
            message.contains("its generation has been superseded"),
            "the wrapped literal must read as one sentence: {message}"
        );
        assert!(
            !message.contains("  "),
            "a run of spaces means a dropped `\\` line continuation: {message}"
        );
    }

    #[test]
    fn durability_error_display_mentions_underlying_cause() {
        let store = DurabilityError::Store(StoreError::SequenceConflict {
            expected: 10,
            found: 11,
        });
        let non_determinism = DurabilityError::NonDeterminism(non_determinism_error());
        let history_shape = DurabilityError::HistoryShape {
            reason: "activity result without preceding schedule".to_owned(),
        };

        let store_message = store.to_string();
        let non_determinism_message = non_determinism.to_string();
        let history_shape_message = history_shape.to_string();

        assert!(!store_message.is_empty());
        assert!(store_message.contains("sequence conflict"));
        assert!(!non_determinism_message.is_empty());
        assert!(non_determinism_message.contains("activity schedule ordinal 7"));
        assert!(non_determinism_message.contains("timer fired timer:named:deadline"));
        assert!(!history_shape_message.is_empty());
        assert!(history_shape_message.contains("activity result without preceding schedule"));
    }
}