aion-rs 0.31.0

//! The ONE way a workflow's successor generation is opened (aion#213 R1).
//!
//! Continue-as-new reaches the engine by two doors — the operator API
//! ([`crate::lifecycle::continue_as_new`]) and the workflow's own
//! `continue_as_new` NIF, whose successor is opened by the exit monitor
//! ([`crate::lifecycle::completion`]) — and both used to open the successor by
//! calling the ordinary start path with the predecessor's workflow id. That
//! path derives a FRESH recorder's sequence head from an unlocked history read,
//! so each door minted a second append authority for a history whose first one
//! was still live and whose process was still executing. That is aion#213.
//!
//! Here the successor is opened THROUGH the predecessor's recorder, inside the
//! predecessor's own lock scope, as one durable batch. The workflow keeps one
//! recorder across the transition, so there is no second writer to race and no
//! head to re-read; the registry's handle is re-keyed onto the successor run
//! before the lock drops, so no reader ever sees the workflow with none or two.
//!
//! The shape is the one the workloop retirement boundary already uses
//! (`workloop::retire::open_retirement_generation`, #214): append through the
//! live handle's recorder, then hand the generation on.

use std::sync::Arc;

use aion_core::{Event, Payload, RunId, TimerId, WorkflowId};
use chrono::Utc;

use crate::EngineError;
use crate::durability::{
    ContinuationTerminal, ContinuedGeneration, OpeningGeneration, WorkflowStartRecord,
};
use crate::lifecycle::start::{
    StartWorkflowContext, abort_unmonitored_start, arm_declared_deadline_or_fail_start,
    deadline_fire_at, install_started_monitor,
};
use crate::registry::{SucceedingRunParts, WorkflowHandle};
use crate::supervision::spawn_workflow_with_policy;

/// Why this transition is opening a successor, and therefore what its batch
/// must contain.
pub(crate) enum ContinuationOrigin {
    /// The operator API path (`Engine::continue_as_new`): the predecessor's
    /// `WorkflowContinuedAsNew` has NOT been recorded, and this transition
    /// records it in the same batch that opens the successor.
    RecordsTheTerminal {
        /// Payload carried into the successor and recorded on the terminal.
        input: Payload,
        /// Workflow type override recorded on the terminal, when the caller
        /// supplied one.
        workflow_type: Option<String>,
    },
    /// The workflow-code path: the `continue_as_new` NIF already recorded the
    /// terminal and ended the predecessor's process, and the exit monitor is
    /// opening the successor from that durable terminal. Recording a second
    /// terminal for one run is not a thing this may do.
    TerminalAlreadyRecorded,
}

/// One continue-as-new transition, as its caller states it.
pub(crate) struct ContinuationRequest {
    /// The generation being continued.
    pub predecessor_run: RunId,
    /// Which half of the transition is already durable.
    pub origin: ContinuationOrigin,
    // NOTE: `origin` is consumed by the admission; everything below it is
    // still borrowed afterwards, which is why the admission takes it by value
    // and the rest of the request by reference.
    /// The workflow type the successor runs — already validated by the caller
    /// against the predecessor's own type.
    pub workflow_type: String,
    /// The successor's input: the value the predecessor's workflow code
    /// produced.
    pub input: Payload,
}

/// What a transition did.
pub(crate) enum ContinuationOutcome {
    /// The successor generation was opened by this call.
    ///
    /// Boxed because the other variant carries nothing and a `WorkflowHandle`
    /// is several hundred bytes: an unboxed pair would make every `Ok` of this
    /// function pay for the handle whether or not one exists.
    Opened(Box<WorkflowHandle>),
    /// The successor was ALREADY started in this history — by the other door,
    /// or by an earlier attempt of this one — so this call appended nothing.
    ///
    /// Idempotence is not a nicety here: the exit monitor, its retry ladder,
    /// and the startup sweep can all reach one predecessor terminal, and a
    /// second successor for one terminal would be a second live run of one
    /// workflow.
    AlreadyOpen,
}

/// Open the successor generation of `predecessor` through its own recorder.
///
/// The whole transition — the terminal (when this door owes one), the
/// predecessor's declared-deadline retirement, the successor's
/// `WorkflowStarted`, the successor's own declared deadline, the process
/// spawn, and the registry re-key — happens under ONE hold of the
/// predecessor's recorder lock. Nothing between those steps can observe the
/// workflow with two writers, with two handles, or with none.
///
/// # The one window this DOES leave, and why every exit from it is a refusal
///
/// Between the durable batch and the re-key, history already names the
/// successor while the registry still names the predecessor — and a BEAM
/// process is spawned inside that interval. It cannot be closed by ordering
/// (the handle needs the pid, and the pid needs the spawn), so what closes it
/// is that every actor which can arrive there refuses rather than guesses:
/// the timer bridge's append is refused by the recorder's run guard on the
/// stale run; an arm for the predecessor's pid is admitted but its fire no-ops
/// on `timer_disposition_in_active_segment`; a signal is refused by the
/// signal router's terminal check; and a start under this id is refused by
/// `workflow_identity`. The window is stated here rather than left for a
/// reader to rediscover, because "safe by four independent refusals" is a
/// property that has to be maintained, not assumed.
///
/// # Errors
///
/// Returns [`EngineError::WorkflowNotFound`] when the replacement type is not
/// loaded, [`EngineError::Runtime`] when the predecessor already recorded a
/// terminal it may not continue from or still has pending work, and the typed
/// durability, runtime, supervision, and registry errors of the steps it
/// performs.
pub(crate) async fn open_successor_generation(
    context: &StartWorkflowContext,
    predecessor: &WorkflowHandle,
    request: ContinuationRequest,
) -> Result<ContinuationOutcome, EngineError> {
    let ContinuationRequest {
        predecessor_run,
        origin,
        workflow_type,
        input,
    } = request;
    // The pinned resolution is held for the whole transition, exactly as the
    // start path holds it: from here until the registry re-key lands, unload
    // verification sees this version as in use.
    //
    // D1: a continuation is the upgrade path for a long-lived workflow, so the
    // successor resolves the LATEST loaded version rather than inheriting the
    // predecessor's pin, and records it durably in its own `WorkflowStarted`.
    let pinned = crate::lifecycle::start::admission::resolve_contract(
        &context.catalog,
        &workflow_type,
        None,
    )?;
    let loaded = pinned.workflow();
    let workflow_id = predecessor.workflow_id().clone();
    let successor_run = RunId::new_v4();

    let recorder = predecessor.recorder();
    let mut recorder = recorder.lock().await;
    // ONE deterministic clock read for the whole boundary: the successor's
    // `WorkflowStarted.recorded_at` AND its deadline's `fire_at` derive from
    // it, so a replay or an adoption re-arm computes the identical fire time
    // (never a second live-clock read).
    //
    // 🔴 READ INSIDE THE LOCK, LIKE EVERY OTHER WRITER ON THIS PATH. The wait
    // above is not nominal: a predecessor append can win this mutex and hold it
    // across a store round trip — which is exactly what
    // `the_transition_and_a_concurrent_predecessor_append_are_serialised_never_conflicting`
    // exercises. A clock read taken before the wait would stamp the four
    // boundary events EARLIER than the event immediately preceding them in the
    // same history, and would compute the successor's deadline `fire_at` from
    // that earlier instant. Nothing workflow-visible moves backwards either way
    // (`advance_workflow_now` is a `fetch_max` and the successor gets a fresh
    // cell), so this is history hygiene rather than a correctness bug — but a
    // recorded timestamp that predates the event before it is a lie in the
    // durable record, and every other terminal writer on this path already
    // reads its clock under the lock.
    let recorded_at = Utc::now();

    // History inspection and the batch are atomic under the recorder lock: a
    // concurrent cancel, completion, or freshly resolving activity records
    // through this same recorder, so a check taken outside the lock could be
    // stale by the time the batch lands.
    let history = context.store.read_history(&workflow_id).await?;
    let TransitionAdmission::Append(terminal) =
        admit_transition(context, &history, &workflow_id, &predecessor_run, origin)?
    else {
        return Ok(ContinuationOutcome::AlreadyOpen);
    };

    let deadline = successor_deadline(&successor_run, loaded.declared_timeout(), recorded_at)?;
    let armed_deadline = recorder
        .record_continue_as_new_boundary(
            recorded_at,
            ContinuedGeneration {
                run_id: predecessor_run.clone(),
                terminal,
                // D5, in the SAME batch as the terminal rather than after it:
                // the id comes from history and is scoped to exactly this
                // predecessor run, so the successor's fresh deadline — which
                // this batch arms two events later — is untouched. Read from
                // the history taken under this lock, so a retirement already
                // completed by the NIF path is a clean no-op here.
                outstanding_deadline: crate::time::outstanding_deadline_timer(
                    &history,
                    &predecessor_run,
                ),
            },
            OpeningGeneration {
                start: WorkflowStartRecord {
                    workflow_type: loaded.workflow_type().to_owned(),
                    input: input.clone(),
                    run_id: successor_run.clone(),
                    parent_run_id: Some(predecessor_run.clone()),
                    // A continuation is not a spawn: the successor's parent
                    // link is the run it continues, never a parent workflow.
                    parent_workflow_id: None,
                    package_version: crate::loader::package_version_of(loaded.version()),
                },
                deadline,
            },
        )
        .await?;

    // From here the durable transition has COMMITTED and the recorder has
    // followed the successor. Everything below establishes the successor's
    // live presence; a failure leaves a durably-started run whose process
    // startup recovery re-installs, which is why each step says what it leaves
    // behind rather than pretending it cannot fail.
    let successor = publish_successor(
        context,
        predecessor,
        loaded,
        SuccessorProcess {
            input: &input,
            predecessor_run: &predecessor_run,
            successor_run: successor_run.clone(),
        },
    )?;
    drop(recorder);

    arm_declared_deadline_or_fail_start(context, &workflow_id, &successor, armed_deadline).await?;
    install_started_monitor(context, &successor)?;
    // The projection the recorder wrote is the authority; this re-projection
    // is the same idempotent upsert the start path performs, and it heals a
    // recorder that carries no visibility store of its own.
    super::visibility::upsert_workflow_visibility(
        Arc::clone(&context.store),
        Arc::clone(&context.visibility_store),
        &workflow_id,
        &successor_run,
    )
    .await?;

    Ok(ContinuationOutcome::Opened(Box::new(successor)))
}

/// What the admission decided, under the predecessor's recorder lock.
enum TransitionAdmission {
    /// Append the boundary, carrying this terminal — `None` when the
    /// predecessor's terminal is already durable.
    Append(Option<ContinuationTerminal>),
    /// The successor is ALREADY started; append nothing.
    AlreadyOpen,
}

/// What [`publish_successor`] needs about the run it is giving a process to.
struct SuccessorProcess<'a> {
    /// The successor's input, spawned into its process.
    input: &'a Payload,
    /// The generation whose registry slot the successor takes over.
    predecessor_run: &'a RunId,
    /// The successor's own run identifier.
    successor_run: RunId,
}

/// Everything that must be true, under the predecessor's recorder lock, before
/// a boundary may be appended — and the terminal the batch owes, if any.
///
/// [`TransitionAdmission::AlreadyOpen`] means the successor is already started
/// and the caller must append nothing.
///
/// # Errors
///
/// Returns [`EngineError::WorkflowWriterHeld`] when another run of this
/// workflow holds the registry's handle, and [`EngineError::Runtime`] when the
/// run already recorded a terminal it may not continue from or still has
/// pending work.
fn admit_transition(
    context: &StartWorkflowContext,
    history: &[Event],
    workflow_id: &WorkflowId,
    predecessor_run: &RunId,
    origin: ContinuationOrigin,
) -> Result<TransitionAdmission, EngineError> {
    if successor_already_started(history, predecessor_run) {
        return Ok(TransitionAdmission::AlreadyOpen);
    }
    // 🔴 THE PREDECESSOR MUST STILL BE THE WORKFLOW'S WRITER, AND THE PROOF IS
    // TAKEN UNDER ITS OWN LOCK. The caller appends through the recorder the
    // predecessor's handle carries; if some OTHER run of this workflow now
    // holds the registry's handle, that run's recorder is the live one and
    // ours is stale — appending through it would be the very second writer
    // this module exists to remove. No handle at all is fine: nothing else is
    // writing, and the re-key publishes the successor as the sole writer.
    if let Some(incumbent) = context.registry.sole_handle(workflow_id)?
        && incumbent.run_id() != predecessor_run
    {
        return Err(EngineError::WorkflowWriterHeld {
            workflow_id: workflow_id.to_string(),
            holder_run_id: incumbent.run_id().to_string(),
            holder_pid: incumbent.pid(),
        });
    }
    match origin {
        ContinuationOrigin::RecordsTheTerminal {
            input,
            workflow_type,
        } => {
            refuse_continuation_from_a_terminal_run(history, workflow_id, predecessor_run)?;
            // 🔴 SCOPED TO THIS GENERATION'S SEGMENT, for the reason
            // `workloop/iteration.rs` states at its own call of this helper:
            // `guard_no_pending_work` accumulates unsettled activities and
            // children by forward-scanning whatever slice it is handed, and
            // `WorkflowStarted` sits in its NO-OP arm — a generation boundary
            // does not clear the pending sets. Continue-as-new produces exactly
            // the same unbounded generation chain in one history, so handed the
            // WHOLE history an activity left unsettled by generation 7 (a
            // timeout teardown, a reopen, a kill) would refuse the continuation
            // of generation 4,000 forever, naming work from generations ago —
            // and the scan would grow without bound with the chain's age.
            //
            // Pending work is run-scoped by nature: a generation can only
            // settle what it started. The terminal check on the line above is
            // already run-scoped, so this makes the pair consistent.
            super::continue_as_new::guard_no_pending_work(aion_core::run_segment(
                history,
                predecessor_run,
            ))?;
            Ok(TransitionAdmission::Append(Some(ContinuationTerminal {
                input,
                workflow_type,
            })))
        }
        ContinuationOrigin::TerminalAlreadyRecorded => Ok(TransitionAdmission::Append(None)),
    }
}

/// Give the durably-started successor a process and the workflow's single
/// registry handle, under the caller's still-held recorder lock.
///
/// 🔴 THE RE-KEY IS INSIDE THAT LOCK, AND THE HANDLE SHARES THE RECORDER. The
/// successor handle carries the very `Arc<Mutex<Recorder>>` the caller is
/// holding, so the workflow's single append authority is continuous across the
/// boundary: a late predecessor append blocks on that lock and is then refused
/// by the recorder's run guard, instead of being sequenced against a rival
/// recorder's stale head.
///
/// # Errors
///
/// Returns supervision, runtime, and registry failures. A spawn that cannot be
/// published has its process aborted here rather than left running unmonitored
/// on a history it would keep writing to.
fn publish_successor(
    context: &StartWorkflowContext,
    predecessor: &WorkflowHandle,
    loaded: &crate::loader::LoadedWorkflow,
    process: SuccessorProcess<'_>,
) -> Result<WorkflowHandle, EngineError> {
    let SuccessorProcess {
        input,
        predecessor_run,
        successor_run,
    } = process;
    context
        .supervision
        .ensure_type_supervisor(loaded.workflow_type())?;
    let runtime_input = crate::runtime::RuntimeInput::from_payload(input)?;
    let pid = spawn_workflow_with_policy(
        &context.runtime,
        loaded.deployed_entry_module(),
        loaded.entry_function(),
        runtime_input,
    )?;
    if let Err(error) = context
        .supervision
        .place_workflow(loaded.workflow_type(), pid)
    {
        return Err(abort_unmonitored_start(&context.runtime, pid, error));
    }

    let successor = predecessor.succeeding_run(SucceedingRunParts {
        run_id: successor_run.clone(),
        pid,
        loaded_version: loaded.version().clone(),
    });
    if let Err(error) = context.registry.rekey_generation(
        predecessor.workflow_id(),
        predecessor_run,
        successor_run,
        successor.clone(),
    ) {
        tracing::error!(
            workflow_id = %predecessor.workflow_id(),
            predecessor_run = %predecessor_run,
            successor_run = %successor.run_id(),
            error = %error,
            "the continue-as-new successor is durably started but could not be published; its \
             process is being aborted and startup recovery re-installs the run"
        );
        return Err(abort_unmonitored_start(&context.runtime, pid, error));
    }
    Ok(successor)
}

/// Whether a successor for `predecessor_run` is already in `history`.
///
/// The link is the successor's recorded `parent_run_id`, which is exactly what
/// a continuation writes and nothing else does — a spawned child carries a
/// `parent_workflow_id` instead — so this identifies the successor of THIS
/// generation rather than any later one.
fn successor_already_started(history: &[Event], predecessor_run: &RunId) -> bool {
    history.iter().any(|event| {
        matches!(
            event,
            Event::WorkflowStarted {
                parent_run_id: Some(existing),
                ..
            } if existing == predecessor_run
        )
    })
}

/// Refuse an API-path continuation of a run that already recorded a terminal.
fn refuse_continuation_from_a_terminal_run(
    history: &[Event],
    workflow_id: &WorkflowId,
    run: &RunId,
) -> Result<(), EngineError> {
    if super::completion::terminal_outcome_from_history(history, run).is_some() {
        return Err(EngineError::Runtime {
            reason: format!(
                "continue_as_new rejected: workflow {workflow_id} run {run} already recorded a \
                 terminal event"
            ),
        });
    }
    Ok(())
}

/// The successor's declared-timeout deadline: its reserved timer id and the
/// instant it fires, derived from the SAME `started_at` its `WorkflowStarted`
/// records so a replay or adoption re-arm computes the identical fire time.
///
/// LAW 1: a package with no declared timeout yields `None` and records no
/// deadline object of any kind.
fn successor_deadline(
    run_id: &RunId,
    declared_timeout: Option<std::time::Duration>,
    started_at: chrono::DateTime<Utc>,
) -> Result<Option<(TimerId, chrono::DateTime<Utc>)>, EngineError> {
    let Some(timeout) = declared_timeout else {
        return Ok(None);
    };
    let fire_at = deadline_fire_at(started_at, timeout)?;
    let deadline_id =
        crate::time::deadline_timer_id(run_id).map_err(|error| EngineError::Runtime {
            reason: format!("failed to mint deadline timer id: {error}"),
        })?;
    Ok(Some((deadline_id, fire_at)))
}