aion-rs 0.31.0

//! Behaviour of the run-scoped append guard (aion#213 R2).

use std::sync::Arc;

use aion_core::{Event, EventEnvelope, PackageVersion, Payload, RunId, WorkflowId};
use aion_store::{EventStore, InMemoryStore};

use super::{RunAdmission, run_generation_admission};
use crate::durability::{Recorder, WorkflowStartRecord};

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn payload(label: &str) -> Result<Payload, aion_core::PayloadError> {
    Payload::from_json(&serde_json::json!({ "label": label }))
}

fn start_record(run_id: &RunId) -> Result<WorkflowStartRecord, Box<dyn std::error::Error>> {
    Ok(WorkflowStartRecord {
        workflow_type: String::from("checkout"),
        input: payload("input")?,
        run_id: run_id.clone(),
        parent_run_id: None,
        parent_workflow_id: None,
        package_version: PackageVersion::new("a".repeat(64)),
    })
}

fn started(
    seq: u64,
    workflow_id: &WorkflowId,
    run_id: &RunId,
) -> Result<Event, Box<dyn std::error::Error>> {
    Ok(Event::WorkflowStarted {
        envelope: EventEnvelope {
            seq,
            recorded_at: chrono::Utc::now(),
            workflow_id: workflow_id.clone(),
        },
        workflow_type: String::from("checkout"),
        input: payload("input")?,
        run_id: run_id.clone(),
        parent_run_id: None,
        parent_workflow_id: None,
        package_version: PackageVersion::new("a".repeat(64)),
    })
}

/// The ordinary path: the generation the recorder is on is admitted, and the
/// admission costs no history read (there is nothing in the store to read).
#[tokio::test]
async fn the_recorders_own_generation_is_admitted_without_a_history_read() -> TestResult {
    let store: Arc<dyn EventStore> = Arc::new(InMemoryStore::default());
    let workflow_id = WorkflowId::new_v4();
    let run_id = RunId::new_v4();
    let mut recorder = Recorder::new(workflow_id.clone(), Arc::clone(&store));
    recorder
        .record_workflow_started(
            chrono::Utc::now(),
            WorkflowStartRecord {
                workflow_type: String::from("checkout"),
                input: payload("input")?,
                run_id: run_id.clone(),
                parent_run_id: None,
                parent_workflow_id: None,
                package_version: PackageVersion::new("a".repeat(64)),
            },
        )
        .await?;

    assert_eq!(
        recorder.admit_run_append(&run_id).await?,
        RunAdmission::Open
    );
    Ok(())
}

/// The defect this guard exists for: once the recorder has followed a
/// successor generation, the PREDECESSOR's run is refused. Without the
/// refusal, a predecessor process still executing after continue-as-new
/// appends into the successor's segment.
#[tokio::test]
async fn a_generation_the_recorder_has_left_behind_is_refused() -> TestResult {
    let store: Arc<dyn EventStore> = Arc::new(InMemoryStore::default());
    let workflow_id = WorkflowId::new_v4();
    let predecessor = RunId::new_v4();
    let successor = RunId::new_v4();
    let mut recorder = Recorder::new(workflow_id.clone(), Arc::clone(&store));
    recorder
        .record_workflow_started(
            chrono::Utc::now(),
            WorkflowStartRecord {
                workflow_type: String::from("checkout"),
                input: payload("input")?,
                run_id: predecessor.clone(),
                parent_run_id: None,
                parent_workflow_id: None,
                package_version: PackageVersion::new("a".repeat(64)),
            },
        )
        .await?;
    assert_eq!(
        recorder.admit_run_append(&predecessor).await?,
        RunAdmission::Open,
        "control: the predecessor is admitted while it IS the generation"
    );

    recorder
        .record_continue_as_new_boundary(
            chrono::Utc::now(),
            crate::durability::ContinuedGeneration {
                run_id: predecessor.clone(),
                terminal: Some(crate::durability::ContinuationTerminal {
                    input: payload("carry")?,
                    workflow_type: None,
                }),
                outstanding_deadline: None,
            },
            crate::durability::OpeningGeneration {
                start: WorkflowStartRecord {
                    workflow_type: String::from("checkout"),
                    input: payload("carry")?,
                    run_id: successor.clone(),
                    parent_run_id: Some(predecessor.clone()),
                    parent_workflow_id: None,
                    package_version: PackageVersion::new("a".repeat(64)),
                },
                deadline: None,
            },
        )
        .await?;

    assert_eq!(
        recorder.admit_run_append(&predecessor).await?,
        RunAdmission::RefusedTerminal
    );
    assert_eq!(
        recorder.admit_run_append(&successor).await?,
        RunAdmission::Open
    );
    Ok(())
}

/// A recorder with NO tracked generation settles the question against history
/// rather than admitting by default.
#[tokio::test]
async fn a_recorder_without_a_generation_settles_the_question_against_history() -> TestResult {
    let store: Arc<dyn EventStore> = Arc::new(InMemoryStore::default());
    let workflow_id = WorkflowId::new_v4();
    let predecessor = RunId::new_v4();
    let successor = RunId::new_v4();
    // Seeded through a recorder — the only append authority — which is then
    // dropped, so the recorder under test genuinely has no run context.
    let mut seeding = Recorder::new(workflow_id.clone(), Arc::clone(&store));
    seeding
        .record_workflow_started(chrono::Utc::now(), start_record(&predecessor)?)
        .await?;
    seeding
        .record_continue_as_new_boundary(
            chrono::Utc::now(),
            crate::durability::ContinuedGeneration {
                run_id: predecessor.clone(),
                terminal: Some(crate::durability::ContinuationTerminal {
                    input: payload("carry")?,
                    workflow_type: None,
                }),
                outstanding_deadline: None,
            },
            crate::durability::OpeningGeneration {
                start: WorkflowStartRecord {
                    parent_run_id: Some(predecessor.clone()),
                    ..start_record(&successor)?
                },
                deadline: None,
            },
        )
        .await?;
    let head = seeding.current_head();
    drop(seeding);

    // `resume_at` alone carries no run context; the guard must not shrug.
    let recorder = Recorder::resume_at(workflow_id.clone(), Arc::clone(&store), head);
    assert_eq!(
        recorder.admit_run_append(&predecessor).await?,
        RunAdmission::RefusedTerminal
    );
    assert_eq!(
        recorder.admit_run_append(&successor).await?,
        RunAdmission::Open
    );
    assert_eq!(
        recorder.admit_run_append(&RunId::new_v4()).await?,
        RunAdmission::RefusedTerminal,
        "a run this history never started has no generation to append into"
    );
    Ok(())
}

/// The history predicate itself, at the three shapes that matter.
#[test]
fn the_window_predicate_separates_the_live_generation_from_the_closed_ones() -> TestResult {
    let workflow_id = WorkflowId::new_v4();
    let first = RunId::new_v4();
    let second = RunId::new_v4();
    let history = [
        started(1, &workflow_id, &first)?,
        started(2, &workflow_id, &second)?,
    ];

    assert_eq!(
        run_generation_admission(&history, &second),
        RunAdmission::Open
    );
    assert_eq!(
        run_generation_admission(&history, &first),
        RunAdmission::RefusedTerminal
    );
    assert_eq!(
        run_generation_admission(&history, &RunId::new_v4()),
        RunAdmission::RefusedTerminal
    );
    assert_eq!(
        run_generation_admission(&history[..1], &first),
        RunAdmission::Open,
        "the only generation in a history is the live one"
    );
    Ok(())
}