aion-store 0.30.0

//! WA-010 R4: the visibility row's `current_worker`, pinned for every
//! backend through the same rows.
//!
//! The row stores `outstanding_leases` and projects `current_worker` from
//! the last entry. These scenarios write rows the way the engine's Recorder
//! writes them — one event at a time through
//! [`aion_core::apply_lease_transition`] — and read the summary back
//! through the store, so a backend that mangles the list (drops it, reorders
//! it, fails to decode a pre-field row) fails here rather than in a console.

use std::collections::HashMap;
use std::sync::Arc;

use aion_core::{
    ActivityId, Event, EventEnvelope, OutstandingLease, Payload, RunId, WorkerAttribution,
    WorkerTransport, WorkflowId, WorkflowStatus, apply_lease_transition,
};
use chrono::{DateTime, Utc};

use super::expect_eq;
use crate::StoreError;
use crate::visibility::{VisibilityRecord, VisibilityStore};

const NAMESPACE: &str = "conformance-leases";

fn workflow_id() -> WorkflowId {
    WorkflowId::new(uuid::Uuid::from_u128(0x1ea5e))
}

fn run_id() -> RunId {
    RunId::new(uuid::Uuid::from_u128(0x0001_ea5e_0001))
}

fn instant(seq: u64) -> DateTime<Utc> {
    DateTime::<Utc>::default() + chrono::Duration::seconds(i64::try_from(seq).unwrap_or(0))
}

fn envelope(seq: u64) -> EventEnvelope {
    EventEnvelope {
        seq,
        recorded_at: instant(seq),
        workflow_id: workflow_id(),
    }
}

fn worker(identity: &str, transport: WorkerTransport) -> WorkerAttribution {
    WorkerAttribution {
        identity: identity.to_owned(),
        task_queue: String::from("billing"),
        node: Some(String::from("n1")),
        deployment: Some(String::from("billing-workers")),
        instance_id: None,
        transport,
    }
}

fn leased(seq: u64, activity: u64, attempt: u32, identity: &str) -> Event {
    Event::ActivityLeased {
        envelope: envelope(seq),
        activity_id: ActivityId::from_sequence_position(activity),
        attempt,
        worker: worker(
            identity,
            if seq.is_multiple_of(2) {
                WorkerTransport::Grpc
            } else {
                WorkerTransport::Liminal
            },
        ),
    }
}

fn completed(seq: u64, activity: u64, attempt: u32) -> Result<Event, StoreError> {
    Ok(Event::ActivityCompleted {
        envelope: envelope(seq),
        activity_id: ActivityId::from_sequence_position(activity),
        attempt,
        result: Payload::from_json(&serde_json::json!({ "ok": true }))
            .map_err(|error| StoreError::Backend(error.to_string()))?,
    })
}

/// A running row with no leases, as the start-time upsert writes it.
fn fresh_row() -> VisibilityRecord {
    VisibilityRecord {
        namespace: NAMESPACE.to_owned(),
        workflow_id: workflow_id(),
        run_id: run_id(),
        workflow_type: String::from("checkout"),
        status: WorkflowStatus::Running,
        started_at: instant(1),
        updated_at: instant(1),
        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,
    }
}

/// Apply `events` to the stored row one at a time, the way the Recorder's
/// per-append touch does, reading the row back between events so every
/// intermediate write round-trips through the backend.
async fn touch_through(
    store: &Arc<dyn VisibilityStore>,
    events: &[Event],
) -> Result<VisibilityRecord, StoreError> {
    store.record_visibility(fresh_row()).await?;
    for event in events {
        let mut row = store
            .get_visibility(&workflow_id())
            .await?
            .ok_or_else(|| StoreError::Backend(String::from("the row written above is gone")))?;
        row.updated_at = *event.recorded_at();
        apply_lease_transition(&mut row.outstanding_leases, event);
        store.record_visibility(row).await?;
    }
    store
        .get_visibility(&workflow_id())
        .await?
        .ok_or_else(|| StoreError::Backend(String::from("the row written above is gone")))
}

fn current_identity(row: &VisibilityRecord) -> Option<String> {
    row.summary().current_worker.map(|worker| worker.identity)
}

pub(super) async fn a_lease_sets_current_worker_and_its_terminal_clears_it(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    let row = touch_through(&store, &[leased(2, 2, 1, "worker-a@host-1")]).await?;
    expect_eq(
        current_identity(&row),
        Some(String::from("worker-a@host-1")),
        "a recorded lease names the worker on the row",
    )?;
    expect_eq(
        row.summary().current_worker.map(|worker| worker.transport),
        Some(WorkerTransport::Grpc),
        "the attribution round-trips whole, transport included",
    )?;

    let row = touch_through(
        &store,
        &[leased(2, 2, 1, "worker-a@host-1"), completed(3, 2, 1)?],
    )
    .await?;
    expect_eq(
        current_identity(&row),
        None,
        "the attempt's completion clears its lease",
    )?;
    expect_eq(
        row.outstanding_leases,
        Vec::<OutstandingLease>::new(),
        "nothing is outstanding after the only attempt completed",
    )
}

pub(super) async fn the_last_lease_of_an_attempt_wins(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    let row = touch_through(
        &store,
        &[
            leased(2, 2, 1, "worker-a@host-1"),
            leased(3, 2, 1, "worker-b@host-2"),
        ],
    )
    .await?;
    expect_eq(
        current_identity(&row),
        Some(String::from("worker-b@host-2")),
        "an at-least-once redelivery's later lease supersedes the earlier one",
    )?;
    expect_eq(
        row.outstanding_leases.len(),
        1,
        "the same attempt is never listed twice",
    )
}

pub(super) async fn an_earlier_outstanding_attempt_is_shown_when_the_later_one_terminates(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    let row = touch_through(
        &store,
        &[
            leased(2, 2, 1, "worker-a@host-1"),
            leased(3, 3, 1, "worker-b@host-2"),
            completed(4, 3, 1)?,
        ],
    )
    .await?;
    expect_eq(
        current_identity(&row),
        Some(String::from("worker-a@host-1")),
        "with two attempts outstanding, the later one's terminal falls back to the earlier",
    )
}

pub(super) async fn a_row_with_no_lease_is_unattributed(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    let row = touch_through(&store, &[]).await?;
    expect_eq(
        row.summary().current_worker,
        None,
        "a run nothing has leased carries no worker",
    )?;
    // A row written by a server that predates the field: the backend must
    // decode it as UNATTRIBUTED, not refuse it. Only the in-memory and JSON
    // backends can be exercised through the trait here; the serde default on
    // the field is what both rely on.
    let mut legacy = serde_json::to_value(fresh_row())
        .map_err(|error| StoreError::Backend(error.to_string()))?;
    if let serde_json::Value::Object(map) = &mut legacy {
        drop(map.remove("outstanding_leases"));
    }
    let decoded: VisibilityRecord = serde_json::from_value(legacy)
        .map_err(|error| StoreError::Backend(format!("a pre-field row must decode: {error}")))?;
    expect_eq(
        decoded.outstanding_leases,
        Vec::new(),
        "a row recorded before lease events existed decodes as unattributed",
    )
}

/// The stored row after a sequence of touches equals the row the whole-
/// history fold would build: the two writers share one transition and this
/// pins that they also share one RESULT through the backend.
pub(super) async fn a_touched_row_matches_the_fold_from_history(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    let events = [
        leased(2, 2, 1, "worker-a@host-1"),
        leased(3, 3, 1, "worker-b@host-2"),
        leased(4, 2, 1, "worker-c@host-3"),
        completed(5, 3, 1)?,
    ];
    let row = touch_through(&store, &events).await?;
    expect_eq(
        row.outstanding_leases,
        aion_core::outstanding_leases(&events),
        "the row's leases after per-event touches equal the whole-history fold",
    )
}