aion-server 0.31.0

Aion workflow server library: HTTP, gRPC, WebSocket, and worker endpoints. Run it with the `aion` binary from the aion-cli crate.
Documentation
//! Focused wire mappings for nested engine error families.

use std::borrow::Cow;

use aion::EngineError;
use aion::durability::DurabilityError;
use aion_proto::WireError;
use aion_store::StoreError;

use super::ErrorTraceFields;

pub(super) fn invalid_state_wire(reason: &str) -> WireError {
    WireError::invalid_state_with_type("InvalidState", reason.to_owned())
}

pub(super) fn backend_wire(error_type: &'static str, source: &EngineError) -> WireError {
    WireError::backend_with_type(error_type, source.to_string())
}

pub(super) fn contract_refusal_wire(error_type: &'static str, source: &EngineError) -> WireError {
    WireError::invalid_state(source.to_string()).with_error_type(error_type)
}

/// Wire mapping for a declared-contract boundary refusal — a start input or a
/// signal payload that did not satisfy the type the package declared.
///
/// These are the CALLER's to fix and cost the target run nothing: the refusal
/// is returned before anything is recorded and before any arrival is consumed.
/// So they are invalid input (HTTP 400), never a backend fault and never a
/// state conflict.
pub(super) fn declared_contract_wire(error_type: &'static str, source: &EngineError) -> WireError {
    WireError::invalid_input(source.to_string()).with_error_type(error_type)
}

pub(super) const fn store_error_type(source: &StoreError) -> &'static str {
    match source {
        StoreError::SequenceConflict { .. } => "SequenceConflict",
        StoreError::NotFound { .. } => "NotFound",
        StoreError::AssistantSessionNotFound { .. } => "AssistantSessionNotFound",
        StoreError::NotOwner { .. } => "NotOwner",
        StoreError::Backend(_) => "Backend",
        StoreError::Serialization(_) => "Serialization",
        StoreError::InvalidQuery(_) => "InvalidQuery",
    }
}

pub(super) fn durability_wire(durability: &DurabilityError, source: &EngineError) -> WireError {
    match durability {
        DurabilityError::Store(store) => super::wire_from_store(store),
        DurabilityError::NonDeterminism(_)
        | DurabilityError::HistoryShape { .. }
        | DurabilityError::SearchAttribute(_) => {
            WireError::backend_with_type("Durability", source.to_string())
        }
        // aion#213: an append refused because the run's generation is closed.
        // A PRECONDITION failure, not a backend fault — the store is fine, the
        // history is readable, and what failed is a condition on the RUN the
        // caller named: it is over, and a later generation has taken its place.
        // A caller reading `backend` would retry against a condition that never
        // clears; reading a precondition failure they look at the run chain,
        // which is where the answer is.
        DurabilityError::RunSuperseded { .. } => {
            WireError::invalid_state_with_type("RunSuperseded", source.to_string())
        }
        // Same wire name as [`aion::EngineError::EngineTaskEpochClosed`] (see
        // `error.rs`'s engine mapping), because a caller is being told about one
        // condition — this engine has begun closing — and which internal seam
        // raised it is not theirs to act on. The two Rust variants stay separate
        // for a reason that lives on the Rust side, not on the wire: the engine
        // variant carries a documented single-construction-site invariant.
        //
        // 🔴 WHY THIS DISCRIMINANT IS DROPPED WHEN A SIBLING CHANGE INSISTS ON
        // KEEPING ONE. `runtime/nif_timer_fire.rs` introduces `RefusedAppend`
        // expressly so a refusal's cause CANNOT be dropped — its own comment
        // says collapsing one "let one sentence be raised for several cases".
        // Landing both in one changeset looks like arguing both sides, so the
        // rule that separates them is stated here rather than left to be
        // inferred.
        //
        // **A discriminant travels when it changes what the recipient should
        // DO.** `RefusedAppend` decides which SENTENCE an operator reads, and
        // the three cases it separates carry three different remedies: wait for
        // workflow code to reissue, check the run's status, or nothing at all.
        // Collapse them and the operator is sent to the wrong one. Here the two
        // variants share a single condition and a single remedy — this engine
        // is closing; the run is untouched; a successor picks it up — so the
        // seam of origin changes nothing a caller can act on, and splitting the
        // label would only fragment the trace an operator searches. Keeping a
        // distinction that cannot alter a decision is not caution; it is a
        // second name for one fact, which is the drift this codebase treats as
        // a defect in its own right.
        //
        // Reached only if a durable write refused by the epoch is ever surfaced
        // through a transport operation. Today the one construction site is the
        // continue-as-new NIF, whose error returns into workflow code rather
        // than into a response — but this mapping is total over the type, so it
        // decides rather than assuming that stays true.
        DurabilityError::EngineTaskEpochClosed { .. } => {
            WireError::backend_with_type("EngineTaskEpochClosed", source.to_string())
        }
    }
}

/// Trace and wire discriminator for live-query dispatch failures.
pub(super) fn query_error_type(source: &aion::QueryError) -> &'static str {
    match source {
        aion::QueryError::UnknownQuery(_) => "UnknownQuery",
        aion::QueryError::Timeout => "QueryTimeout",
        aion::QueryError::NotRunning(_) => "QueryNotRunning",
        aion::QueryError::Unknown(_) => "QueryUnknownWorkflow",
        aion::QueryError::ReplyDropped => "QueryReplyDropped",
        aion::QueryError::HandlerFailed { .. } => "QueryFailed",
        aion::QueryError::InvalidArguments { .. } => "QueryInvalidArguments",
        aion::QueryError::Engine(_) => "QueryEngine",
    }
}

/// Wire mapping for live-query dispatch failures.
pub(super) fn query_wire(query: &aion::QueryError, source: &EngineError) -> WireError {
    match query {
        aion::QueryError::UnknownQuery(_) => WireError::unknown_query(source.to_string()),
        aion::QueryError::Timeout => WireError::query_timeout(source.to_string()),
        aion::QueryError::NotRunning(_) | aion::QueryError::ReplyDropped => {
            WireError::not_running_with_type(query_error_type(query), source.to_string())
        }
        aion::QueryError::Unknown(_) => {
            WireError::not_found_with_type(query_error_type(query), source.to_string())
        }
        aion::QueryError::HandlerFailed { .. } => {
            WireError::query_failed(source.to_string()).with_error_type(query_error_type(query))
        }
        // The caller's own arguments were malformed: a request defect, not a
        // handler failure and not an engine fault. `invalid_input` is the
        // code that tells the caller to fix what it sent.
        aion::QueryError::InvalidArguments { .. } => {
            WireError::invalid_input(source.to_string()).with_error_type(query_error_type(query))
        }
        aion::QueryError::Engine(_) => {
            WireError::backend_with_type(query_error_type(query), source.to_string())
        }
    }
}

// --- Trace fields. The counterpart of the wire mappings above: `durability_wire`
// says what a caller is told, and `durability_trace_fields` says what an operator
// finds in the logs, and the two must name the same condition the same way. They
// live together for that reason — one of them sitting in `error.rs` was how a
// generic `Durability` label could drift away from a specific wire type without
// anything noticing.

/// A durability failure that is really a STORE failure keeps the store's own
/// trace fields; the rest are the engine's.
///
/// Extracted from [`engine_trace_fields`] because a nested match is a different
/// question from the flat dispatch around it, not to satisfy a line count.
pub(super) fn durability_trace_fields<'a>(
    durability: &'a DurabilityError,
    source: &'a EngineError,
) -> ErrorTraceFields<'a> {
    match durability {
        DurabilityError::Store(store) => store_trace_fields(store),
        DurabilityError::NonDeterminism(_)
        | DurabilityError::HistoryShape { .. }
        | DurabilityError::SearchAttribute(_) => simple_engine_fields("Durability", source),
        // Its own label rather than the generic `Durability`, and the SAME
        // label the engine-level variant gets below, so an operator searching
        // traces for "did this engine begin closing" finds every seam that
        // refused for that reason under one name instead of one name and a
        // generic one.
        DurabilityError::EngineTaskEpochClosed { .. } => {
            simple_engine_fields("EngineTaskEpochClosed", source)
        }
        // Its own label for the same reason: a refusal because the run's
        // generation was superseded is a fact about WHICH RUN the caller is
        // looking at, and folding it into the generic `Durability` label would
        // make it unsearchable in exactly the incident it describes.
        DurabilityError::RunSuperseded { .. } => simple_engine_fields("RunSuperseded", source),
    }
}

pub(super) fn simple_engine_fields<'a>(
    error_type: &'static str,
    source: &'a EngineError,
) -> ErrorTraceFields<'a> {
    ErrorTraceFields {
        error_type: Cow::Borrowed(error_type),
        store_error_type: None,
        reason: source,
    }
}

pub(super) fn store_trace_fields(source: &StoreError) -> ErrorTraceFields<'_> {
    ErrorTraceFields {
        error_type: Cow::Borrowed("StoreError"),
        store_error_type: Some(store_error_type(source)),
        reason: source,
    }
}