aion-rs 0.31.0

//! The detached-hatch NIF bridge — `aion_flow_ffi:hatch_detached/3` (R13.1).
//!
//! A hatch starts a TOP-LEVEL workflow under a mandatory dedupe identity and
//! walks away: no lifecycle tie, no supervision edge, no awaited terminal.
//! Both document kinds reach it through this one native, and the identity is
//! derived ONCE for both — `hatch_workflow_id(namespace, type, key)`, a
//! `UUIDv5` over the caller's namespace, the target workflow type, and the
//! caller-declared key.
//!
//! # 🔴 THE DETERMINISM BOUNDARY IS WHY THIS IS A RESOLVED COMMAND
//!
//! It is tempting to argue that a hatch needs no replay resolution because
//! its identity is already deterministic: replay re-derives the same `UUIDv5`,
//! and `hatch_workflow` short-circuits on an identity that already has
//! history, so "nothing happens twice anyway".
//!
//! That argument is wrong, and it is wrong in the direction that costs
//! durability. A hatch STARTS A WORKFLOW — it is a side effect, and invariant
//! 2 says a side effect returns its RECORDED result on replay rather than
//! acting again. Leaning on the identity's dedupe instead would make every
//! replayed hatch perform a live store read and a live start attempt, so a
//! recovering run would re-drive the start path once per hatch per replay,
//! and the answer workflow code sees would come from a fresh store read
//! rather than from its own history. Replay would be observing the world, not
//! its record of the world.
//!
//! So the hatch resolves like `spawn_child` does: a positional
//! [`CorrelationKey::Hatch`] ordinal, matched against the run segment's n-th
//! recorded [`aion_core::Event::WorkflowHatched`]. On replay the recorded
//! event's `child_workflow_id` is returned and NOTHING is started. The `UUIDv5`
//! dedupe remains, but it is what settles a RACE between two live hatches of
//! the same identity — never what stands in for a recorded result.
//!
//! # 🔴 WHERE THE ANALOGY WITH `spawn_child` STOPS: THE ORDER
//!
//! It resolves like a spawn. It does not RECORD like one. `spawn_child` must
//! record before it starts because a child's id is fresh nondeterminism that
//! exists nowhere until it is written down; a hatch's id is a pure function of
//! the caller's own arguments, so a start that lands before its record is
//! re-derivable rather than orphaned. Recording first bought nothing here and
//! cost a live/replay disagreement on every failed start — see [`live_hatch`],
//! which states the whole argument.

use std::sync::Arc;

use aion_core::{ContentType, Payload, WorkflowId};
use beamr::native::ProcessContext;
use beamr::term::Term;
use beamr::term::binary_ref::BinaryRef;
use beamr::term::heap_borrow::HeapBorrow;
use chrono::Utc;

use crate::durability::{Command, CorrelationKey, Resolution, ResolveOutcome};
use crate::error::EngineError;
use crate::runtime::nif_child_engine::ChildNifBridge;
use crate::runtime::nif_context::{NifContext, NifContextError};
use crate::runtime::nif_result_term::{NifRefusal, error_result_term, ok_result_term};
use crate::runtime::nif_state::EngineNifState;
use crate::workloop::hatch::derive_identity;

/// NIF backing `aion_flow_ffi:hatch_detached/3`.
///
/// Arguments, in order — all UTF-8 binaries:
/// 1. `WorkflowType` — the target top-level workflow type to hatch.
/// 2. `Input` — the hatched workflow's input, as JSON text.
/// 3. `Key` — the caller-declared dedupe key within (namespace, type).
///
/// Returns `{ok, WorkflowIdBinary}` for both a fresh hatch and a duplicate —
/// a duplicate is a recorded NO-OP that resolves to the existing workflow, not
/// an error — and `{error, Reason}` for a refused declaration or an engine
/// fault.
pub(super) fn hatch_detached_impl(args: &[Term], ctx: &mut ProcessContext) -> Result<Term, Term> {
    match run_hatch(args, ctx) {
        Ok(term) => Ok(term),
        Err(refusal) => refusal.into_nif_result(),
    }
}

fn run_hatch(args: &[Term], ctx: &mut ProcessContext) -> Result<Term, NifRefusal> {
    require_arity("hatch_detached", args, 3).map_err(|message| hatch_refusal(ctx, &message))?;
    let workflow_type = decode_string_arg(args[0], ctx.borrow_terms())
        .map_err(|error| format!("workflow_type:{error}"))
        .map_err(|message| hatch_refusal(ctx, &message))?;
    let input_text = decode_string_arg(args[1], ctx.borrow_terms())
        .map_err(|error| format!("input:{error}"))
        .map_err(|message| hatch_refusal(ctx, &message))?;
    let hatch_key = decode_string_arg(args[2], ctx.borrow_terms())
        .map_err(|error| format!("key:{error}"))
        .map_err(|message| hatch_refusal(ctx, &message))?;
    let input = Payload::new(ContentType::Json, input_text.into_bytes());

    let bridge = hatch_bridge(ctx).map_err(|message| hatch_refusal(ctx, &message))?;
    let pid = ctx
        .pid()
        .ok_or_else(|| "missing_caller_pid".to_owned())
        .map_err(|message| hatch_refusal(ctx, &message))?;
    // A hatch records `WorkflowHatched`; a query handler must stay read-only.
    let state = crate::runtime::nif_state::engine_nif_state(ctx)
        .map_err(|message| hatch_refusal(ctx, &message))?;
    crate::runtime::nif_query_pump::ensure_not_servicing_query(&state, pid, "hatch_detached")
        .map_err(|message| hatch_refusal(ctx, &message))?;

    let nif = new_context(&bridge, pid).map_err(|message| hatch_refusal(ctx, &message))?;
    let key = CorrelationKey::Hatch(nif.next_hatch_ordinal());
    let command = Command::HatchWorkflow {
        key,
        workflow_type: workflow_type.clone(),
        hatch_key: hatch_key.clone(),
        input: input.clone(),
    };

    resolve_hatch(ctx, &bridge, nif, workflow_type, hatch_key, input, command)
}

fn resolve_hatch(
    ctx: &mut ProcessContext,
    bridge: &Arc<ChildNifBridge>,
    mut nif: NifContext,
    workflow_type: String,
    hatch_key: String,
    input: Payload,
    command: Command,
) -> Result<Term, NifRefusal> {
    // OBSERVED seam, exactly as `spawn_child`'s is. The recorded
    // `WorkflowHatched` IS the answer this call returns to workflow code, and
    // the live branch records that same event and advances to the same
    // `recorded_at` before returning the same id — so a replayed pass and the
    // live pass it replays consume one event and see one identity.
    match nif
        .resolve_command_observed(command)
        .map_err(|error| hatch_refusal(ctx, &context_error(&error)))?
    {
        // 🔴 REPLAY RETURNS AND DOES NOT ACT. No store read, no start
        // attempt, no dedupe probe — the recorded event is the whole answer.
        ResolveOutcome::Recorded(Resolution::Hatched(hatched_id)) => {
            ok_result_term(ctx, hatched_id.to_string().as_bytes()).map_err(NifRefusal::Unbuildable)
        }
        ResolveOutcome::Recorded(other) => Err(hatch_refusal(
            ctx,
            &format!("unexpected_hatch_resolution:{other:?}"),
        )),
        ResolveOutcome::ResumeLive => {
            let namespace = nif.workflow_handle().namespace().to_owned();
            // 🔴 THE TYPE IS CHECKED BEFORE THE IDENTITY IS MINTED OR THE
            // EVENT RECORDED, EXACTLY AS `spawn_child` CHECKS IT.
            //
            // Everything after the record is an engine-internal obligation the
            // caller is told succeeded, so the last honest place to refuse is
            // here. Recording `WorkflowHatched` for a type this engine cannot
            // load leaves a durable event naming a workflow that does not
            // exist and that nothing will ever create — and, because the
            // recorded event is what replay returns, a replaying run would go
            // on being told the hatch succeeded forever.
            let package_version = bridge
                .routed_package_version(&workflow_type)
                .map_err(|error| hatch_refusal(ctx, &format!("hatch_version_resolution:{error}")))?
                .ok_or_else(|| {
                    hatch_refusal(
                        ctx,
                        &format!("hatch_workflow_type_not_loaded:{workflow_type}"),
                    )
                })?;
            let hatch_id = derive_identity(&namespace, &workflow_type, &hatch_key)
                .map_err(|error| hatch_refusal(ctx, &format!("identity_refused:{error}")))?;
            live_hatch(
                ctx,
                bridge,
                &mut nif,
                LiveHatch {
                    namespace,
                    workflow_type,
                    hatch_key,
                    hatch_id,
                    input,
                    package_version,
                },
            )
        }
    }
}

/// Everything the live half of one hatch needs, gathered before any append.
struct LiveHatch {
    namespace: String,
    workflow_type: String,
    hatch_key: String,
    hatch_id: WorkflowId,
    input: Payload,
    package_version: aion_core::PackageVersion,
}

/// # 🔴 THE HATCH STARTS AND THEN RECORDS. `spawn_child` RECORDS AND THEN
/// STARTS. THE DIFFERENCE IS THE IDENTITY.
///
/// A child's `WorkflowId` is `WorkflowId::new_v4()` — recorded nondeterminism.
/// It exists nowhere until the parent writes it down, so the parent MUST
/// record first: a start-then-record child that crashed in the window would
/// leave a running workflow whose id no replay can ever re-derive. That is a
/// true orphan, and it is why `spawn_child` reports success after its record
/// and owns the repair in the background.
///
/// A hatch identity is `hatch_workflow_id(namespace, type, key)` — a `UUIDv5`,
/// a pure function of three values the caller already has. Every replay
/// re-derives it exactly. There is no orphan to protect against, and inheriting
/// the child's ordering bought the one thing it was supposed to prevent: the
/// record landed, `observe_recorded_at` advanced the seam, and then a start
/// failure returned an ERROR — while replay of that same position resolves the
/// recorded `WorkflowHatched` and returns OK. Live and replay disagreeing at
/// one recorded command is the determinism boundary (invariant 2), and it was
/// not repairable either: nothing in the tree re-drives a `WorkflowHatched`.
///
/// Starting first dissolves both. A crash between the start and the record
/// leaves a running detached workflow — which is exactly what a hatch is for —
/// and the caller's next replay re-derives the same id, finds its history
/// non-empty, records the hatch, and returns it. A start FAILURE now happens
/// with nothing recorded, so there is no recorded command for replay to
/// disagree with: the caller gets an error, and a retry re-attempts cleanly.
///
/// The dedupe probe stays ahead of the start, and the store's optimistic append
/// still settles two racing live hatches: the loser resolves to the winner's
/// workflow and records its own `WorkflowHatched` for its own ordinal.
fn live_hatch(
    ctx: &mut ProcessContext,
    bridge: &Arc<ChildNifBridge>,
    nif: &mut NifContext,
    hatch: LiveHatch,
) -> Result<Term, NifRefusal> {
    let LiveHatch {
        namespace,
        workflow_type,
        hatch_key,
        hatch_id,
        input,
        package_version,
    } = hatch;

    // The dedupe half is per-IDENTITY (this namespace/type/key has one
    // workflow); the record below is per-CALLER (this run hatched at this
    // ordinal). Both must hold, and they are different facts.
    let already_hatched = !bridge
        .tokio_handle()
        .block_on(bridge.store().read_history(&hatch_id))
        .map_err(|error| hatch_refusal(ctx, &format!("store:{error}")))?
        .is_empty();

    if !already_hatched {
        match bridge
            .tokio_handle()
            .block_on(bridge.start_hatched_under_recorded_id(
                &namespace,
                &workflow_type,
                hatch_id.clone(),
                input,
                package_version,
            )) {
            Ok(_handle) => {}
            // 🔴 THE DEDUPE RACE, SETTLED BY THE STORE. A concurrent hatch of
            // the same identity won the first append. The store's optimistic
            // concurrency IS the dedupe index — the loser resolves to the
            // existing workflow, never starts a second one.
            //
            // It is LOGGED, and that is not decoration. In this codebase a
            // `SequenceConflict` is defined as the signal of a double writer
            // (invariant 3), and this is the one place it is expected and
            // absorbed. An absorbed instance that left no line would be
            // indistinguishable from the defect the signal exists to report,
            // for anyone reading logs after one.
            Err(EngineError::Store(aion_store::StoreError::SequenceConflict { .. })) => {
                tracing::info!(
                    hatching_workflow_id = %nif.workflow_id(),
                    hatched_workflow_id = %hatch_id,
                    workflow_type = %workflow_type,
                    "hatch dedupe race: a concurrent hatch of this identity won the first \
                     append, so this call resolves to the existing workflow. This \
                     SequenceConflict is EXPECTED and absorbed here — it is the dedupe index \
                     doing its job, not the double-writer signal it means everywhere else"
                );
            }
            // 🔴 THE SAME RACE, CAUGHT ONE STEP EARLIER (aion#213). The winner
            // does not become durable and then visible in one instant: it
            // appends, spawns, and only then registers. A loser whose probe ran
            // before the winner's append sees the SequenceConflict above; a
            // loser whose probe ran after the winner's REGISTRATION is refused
            // by the start path's fresh-id guard instead, because the identity
            // now has a live writer whose recorder owns the head. Both mean the
            // same thing — this identity already has its workflow — and both
            // resolve to it. Absorbing only the first would make the hatch
            // dedupe fail intermittently, on the wider of the two windows.
            Err(EngineError::WorkflowIdAlreadyLive { .. }) => {
                tracing::info!(
                    hatching_workflow_id = %nif.workflow_id(),
                    hatched_workflow_id = %hatch_id,
                    workflow_type = %workflow_type,
                    "hatch dedupe race: a concurrent hatch of this identity is already live, so \
                     this call resolves to the existing workflow"
                );
            }
            Err(error) => {
                // NOTHING is recorded on this path, so live and replay cannot
                // disagree: there is no recorded command at this position for
                // a replay to resolve. The caller sees the failure, and a
                // retry re-attempts the start from a clean position.
                tracing::warn!(
                    hatching_workflow_id = %nif.workflow_id(),
                    hatched_workflow_id = %hatch_id,
                    workflow_type = %workflow_type,
                    error = %error,
                    "hatch failed to start; nothing was recorded, so the hatching run's \
                     history is unchanged and a retry re-attempts the start"
                );
                return Err(hatch_refusal(ctx, &format!("hatch_start_failed:{error}")));
            }
        }
    }

    // The workflow exists — freshly started, already there, or won by a racer.
    // Recording it now is what makes the answer this call returns replayable.
    let recorded_at = Utc::now();
    let recorded_id = hatch_id.clone();
    let recorded_key = hatch_key.clone();
    nif.block_on_recorder(move |recorder| {
        Box::pin(async move {
            recorder
                .record_workflow_hatched(recorded_at, recorded_id, recorded_key)
                .await
        })
    })
    .map_err(|error| hatch_refusal(ctx, &context_error(&error)))?;
    // Live half of the OBSERVED seam: this is the event a replayed hatch
    // resolves at this position.
    nif.observe_recorded_at(recorded_at);

    ok_result_term(ctx, hatch_id.to_string().as_bytes()).map_err(NifRefusal::Unbuildable)
}

fn hatch_refusal(ctx: &mut ProcessContext, message: &str) -> NifRefusal {
    NifRefusal::reported(error_result_term(ctx, &format!("hatch_detached:{message}")))
}

fn hatch_bridge(ctx: &ProcessContext) -> Result<Arc<ChildNifBridge>, String> {
    let state = crate::runtime::nif_state::engine_nif_state(ctx)?;
    hatch_bridge_from_state(&state)
}

fn hatch_bridge_from_state(state: &EngineNifState) -> Result<Arc<ChildNifBridge>, String> {
    let slot = match state.child_bridge.read() {
        Ok(slot) => slot.clone(),
        Err(poisoned) => poisoned.into_inner().clone(),
    };
    slot.ok_or_else(|| "no_engine_nif_bridge_configured".to_owned())
}

fn new_context(bridge: &ChildNifBridge, pid: u64) -> Result<NifContext, String> {
    NifContext::new_with_history_store(
        pid,
        bridge.registry(),
        bridge.tokio_handle(),
        Some(bridge.store()),
        bridge.watch_backoff(),
    )
    .map_err(|error| context_error(&error))
}

fn context_error(error: &NifContextError) -> String {
    error.error_reason()
}

fn decode_string_arg(term: Term, heap: HeapBorrow<'_>) -> Result<String, String> {
    let bin = BinaryRef::new(term).ok_or_else(|| "argument is not a binary".to_owned())?;
    String::from_utf8(bin.as_bytes(heap).to_vec())
        .map_err(|_| "argument is not valid UTF-8".to_owned())
}

fn require_arity(name: &str, args: &[Term], expected: usize) -> Result<(), String> {
    if args.len() == expected {
        Ok(())
    } else {
        Err(format!(
            "{name}: expected {expected} arguments, got {}",
            args.len()
        ))
    }
}