aion-rs 0.31.0

//! Exact-version structural admission before workflow start mutation.

use std::sync::Arc;

use aion_core::{Event, Payload, WorkflowId};
use aion_package::ContentHash;
use aion_store::EventStore;

use crate::EngineError;
use crate::loader::{ActivityServing, LoadedWorkflow, PinnedWorkflow, WorkflowCatalog};
use crate::registry::Registry;

pub(in crate::lifecycle) fn resolve_contract(
    catalog: &WorkflowCatalog,
    workflow_type: &str,
    loaded_version: Option<&ContentHash>,
) -> Result<PinnedWorkflow, EngineError> {
    let pinned = match loaded_version {
        Some(version) => catalog.resolve_exact(workflow_type, version)?,
        None => catalog.resolve_routed(workflow_type)?,
    }
    .ok_or_else(|| EngineError::WorkflowNotFound {
        workflow_type: workflow_type.to_owned(),
    })?;
    let loaded = pinned.workflow();
    // The `.v4` identity floor holds in every serving posture: a pre-`.v4`
    // package is never grandfathered.
    let contract = loaded
        .contract()
        .map_err(|source| EngineError::ContractIdentity {
            workflow_type: workflow_type.to_owned(),
            source,
        })?;
    // The queue-service check protects only queue-routed fulfilment (an
    // unserved queue waits silently forever). Under declared in-process
    // serving no queue exists: dispatch runs on the configured dispatcher
    // immediately, and a missing dispatcher fails loudly at dispatch.
    if catalog.activity_serving() == ActivityServing::QueueRouted
        && !contract.unscoped_activities.is_empty()
    {
        let mut activities = contract.unscoped_activities.clone();
        activities.sort();
        return Err(EngineError::NoQueueDeclaration {
            workflow_type: workflow_type.to_owned(),
            version: loaded.version().clone(),
            activities: activities.join(","),
        });
    }
    Ok(pinned)
}

/// Admit a caller-supplied start input against the input type declared by the
/// exact package identity the start resolved to.
///
/// Runs BEFORE `workflow_identity`, before the recorder, and before any
/// process is spawned, so a refused start records nothing and leaves nothing to
/// clean up: the caller sees their own mistake at the moment they made it,
/// rather than the run dying from the inside some time later.
///
/// A contract whose entry declares no input shape — a legacy or defaulted
/// manifest, whose `input_schema` is `null` — admits everything, because that
/// is precisely what an absent declaration says.
///
/// A schema that IS declared but cannot be compiled is a defect in the
/// PACKAGE, not in the caller's value. It is logged and admitted: refusing
/// would blame the caller for something they cannot fix and would make every
/// start of that package impossible, which is a worse failure than the one
/// this admission exists to prevent.
///
/// # Errors
///
/// Returns [`EngineError::ContractIdentity`] for a pre-`.v4` identity (already
/// refused by [`resolve_contract`], propagated rather than assumed) and
/// [`EngineError::StartInputRefused`] when the input does not satisfy the
/// declared schema.
pub(super) fn admit_declared_input(
    loaded: &LoadedWorkflow,
    input: &Payload,
) -> Result<(), EngineError> {
    let workflow_type = loaded.workflow_type();
    let contract = loaded
        .contract()
        .map_err(|source| EngineError::ContractIdentity {
            workflow_type: workflow_type.to_owned(),
            source,
        })?;
    let schema = contract.entry_input_schema(workflow_type);
    if aion_package::declares_nothing(schema) {
        return Ok(());
    }
    let refuse = |reason: String| EngineError::StartInputRefused {
        workflow_type: workflow_type.to_owned(),
        version: loaded.version().clone(),
        reason,
    };
    let value = input
        .to_json()
        .map_err(|error| refuse(format!("the input is not decodable JSON: {error}")))?;
    match aion_package::admit_value(schema, &value) {
        Ok(()) => Ok(()),
        Err(aion_package::AdmissionError::UnusableSchema { reason }) => {
            tracing::warn!(
                workflow_type,
                version = %loaded.version(),
                %reason,
                "package declares an input schema that is not valid JSON Schema; the start input could not be admitted against it and was allowed through"
            );
            Ok(())
        }
        Err(error @ aion_package::AdmissionError::Mismatch { .. }) => {
            Err(refuse(error.to_string()))
        }
    }
}

/// The identity a start records under, and the sequence head its FRESH
/// recorder resumes from.
///
/// # 🔴 THE HEAD READ IS FOR AN ID NOTHING IS WRITING, AND ONLY FOR ONE
///
/// This read is unlocked, and it is the seed of a brand-new `Recorder` that
/// will own the workflow's tracked sequence head from here on. That is sound
/// for a fresh execution — a start under a minted id, a recorded-but-unstarted
/// child, a hatch identity, a steered start — because nothing else holds a
/// writer for that id, so nothing can move the head between this read and the
/// first append.
///
/// It is NOT sound for an id that already has a live handle: that handle's
/// recorder owns the head, its process is still executing, and any append it
/// makes in the window advances the durable head under the new recorder's
/// feet. Seeding a second recorder there is aion#213 exactly — continue-as-new
/// used to do it, the predecessor's in-flight `sleep` arm landed in the
/// window, and the successor's first append died on `SequenceConflict
/// { expected: 4, found: 5 }`. So a start for an id the registry already holds
/// is REFUSED here, before any append and before any process is spawned. A
/// continuation of a live workflow goes through
/// [`crate::lifecycle::continuation`], which appends through the incumbent
/// recorder instead of minting a rival.
///
/// # Errors
///
/// Returns [`EngineError::WorkflowIdAlreadyLive`] when a run of `requested`
/// already holds a live handle, [`EngineError::WorkflowWritersAmbiguous`] when
/// more than one does, and store errors from the history read.
pub(in crate::lifecycle) async fn workflow_identity(
    store: &Arc<dyn EventStore>,
    registry: &Registry,
    requested: Option<WorkflowId>,
) -> Result<(WorkflowId, u64), EngineError> {
    let Some(workflow_id) = requested else {
        return Ok((WorkflowId::new_v4(), 0));
    };
    if let Some(incumbent) = registry.sole_handle(&workflow_id)? {
        return Err(EngineError::WorkflowIdAlreadyLive {
            workflow_id: workflow_id.to_string(),
            holder_run_id: incumbent.run_id().to_string(),
            holder_pid: incumbent.pid(),
        });
    }
    let initial_head = store
        .read_history(&workflow_id)
        .await?
        .iter()
        .map(Event::seq)
        .max()
        .unwrap_or_default();
    Ok((workflow_id, initial_head))
}