aion/workloop/

retire.rs

//! The retire-body invocation protocol (S3).
//!
//! # 🔴 THE CONTRACT THE EMITTER MUST TARGET, VERBATIM
//!
//! A workloop document that declares a `retire` block compiles to a SECOND
//! exported entry alongside the ordinary run entry:
//!
//! ```erlang
//! -export([run/1, retire/1]).
//!
//! retire(Carry) -> any().
//! ```
//!
//! - **Entry name**: [`RETIRE_ENTRY`] — `retire`, arity 1.
//! - **Argument**: the loop's CURRENT carry, byte-identical to the value the
//!   same generation's `run/1` was given. The retire body sees exactly the
//!   state the last iteration left, because a retirement that could not see
//!   the loop's final state could not clean it up.
//! - **Return**: ignored. The retirement RESULT is the operator-supplied
//!   payload on [`crate::Engine::retire_workloop`], not whatever the body
//!   returns — the body exists for its EFFECTS (draining a queue, releasing a
//!   lease, notifying a peer), and letting it also decide the terminal result
//!   would give one block two unrelated jobs.
//! - **Effects**: recorded through the loop's ONE Recorder. Every activity,
//!   timer, signal, child and hatch the body uses appends to the loop's own
//!   history exactly as an iteration's would.
//! - **Ordering**: the body runs to completion BEFORE the terminal
//!   `[LoopRetired + WorkflowCompleted]` batch is recorded. A retirement that
//!   recorded its terminal first would be asking a completed run to keep
//!   working, and the single-writer law would refuse the body's own appends.
//!
//! # 🔴 THE RETIREMENT IS THE LOOP'S FINAL GENERATION
//!
//! The body does not run inside the generation the loop was parked in. It runs
//! inside a generation opened for it, and both halves of that decision are
//! load-bearing.
//!
//! **The ordinal space.** Every positional durable command — `dispatch_activity`,
//! `spawn_child`, `hatch_detached` — keys on a counter that starts at ZERO and
//! resolves against the current run segment
//! (`NifContext::new_with_history_store` → `current_run_segment`). A body run
//! inside a generation that already dispatched an activity would have its own
//! first `dispatch_activity` fast-forward to that recorded activity and return
//! its result WITHOUT EXECUTING — a declared cleanup that appends nothing and
//! hands the author the loop's own stale data. `workflow.now()` would answer
//! the iteration's recorded clock for the same reason. The retirement
//! generation's segment is empty when the body starts, so every command it
//! issues can only resolve one way: live.
//!
//! **The single writer.** The body is published into the registry so its NIF
//! calls resolve to the loop's recorder, and a `Recorder` writes the WORKFLOW's
//! event stream — so a handle for the body plus a live handle for the loop is
//! two writers for one history (invariant 3). The publication therefore goes
//! through [`Registry::insert_sole_workflow_writer`], which proves the absence
//! of any other handle UNDER THE LOCK rather than asserting it in a comment.
//! An earlier version used `Registry::insert`, which is documented to REPLACE,
//! and dropped the displaced handle unexamined: retiring a loop that had not
//! yet parked left two Recorders live, and stopped the displaced process's pid
//! resolving at all — its next durable NIF stalled the full birth-wait budget
//! and then failed typed, which the SDKs treat as fatal.
//!
//! So a resident generation is STOOD DOWN first, explicitly and loudly: its
//! continuation is notified, its handle removed, and its process cancelled —
//! the same three steps an iteration close performs, because a retirement is
//! the same kind of generation boundary. Retirement stops the loop; that has
//! always been its meaning. What it must not do is stop it by accident.
//!
//! # 🔴 A MISSING ENTRY IS REFUSED LOUDLY, NEVER SKIPPED
//!
//! A workloop compiled before the emitter change exports no `retire/1`. The
//! engine must not treat that as "no retire body declared" — it cannot tell
//! the two apart, and silently skipping a declared cleanup is the failure
//! mode that loses a lease or strands a queue. So retirement REFUSES with a
//! diagnostic naming the module and the missing entry. That probe runs BEFORE
//! anything is stood down or recorded, so the refusal really does leave the
//! loop exactly as it was.
//!
//! # 🔴 THE RETIRE BODY MUST BE IDEMPOTENT, AND THAT IS THE AUTHOR'S JOB
//!
//! The body runs before any terminal is recorded, so a crash mid-retire
//! leaves the loop un-retired with no `LoopRetired` in history. The next
//! retirement attempt opens a FRESH retirement generation and runs the body
//! again from the top — deliberately, because the alternative is resolving the
//! second attempt's commands against the first attempt's partial record, which
//! is the silent-skip failure in another costume. Effects the body performed
//! are therefore re-performed on the retry. This is stated rather than
//! engineered around: recording a terminal before the cleanup finished would
//! claim a retirement that did not happen.

use std::sync::Arc;

use aion_core::{Event, Payload, RunId, WorkflowId, WorkflowStatus};
use aion_store::EventStore;
use aion_store::visibility::VisibilityStore;
use chrono::Utc;

use super::error::WorkloopError;
use crate::durability::{Recorder, WorkflowStartRecord};
use crate::loader::WorkflowCatalog;
use crate::registry::{
    CompletionNotifier, HandleResidency, Registry, TerminalOutcome, WorkflowHandle,
    WorkflowHandleParts,
};
use crate::runtime::{RuntimeHandle, RuntimeInput};

/// The exported entry a compiled `retire` block lands on: `retire/1`.
///
/// A constant because three places must agree — the existence probe, the
/// spawn, and the refusal diagnostic — and a name known in three places is a
/// name that drifts.
pub const RETIRE_ENTRY: &str = "retire";

/// Arity of [`RETIRE_ENTRY`]. One argument: the loop's current carry.
pub const RETIRE_ARITY: u32 = 1;

/// Everything invoking one retire body needs.
pub struct RetireInvocation<'a> {
    /// The loop being retired.
    pub loop_id: &'a WorkflowId,
    /// The engine's runtime handle.
    pub runtime: &'a Arc<RuntimeHandle>,
    /// The catalog resolving the loop's deployed module.
    pub catalog: &'a WorkflowCatalog,
    /// The registry the retire process is published into.
    pub registry: &'a Arc<Registry>,
    /// Event store, for the loop's one-shot recorders.
    pub store: &'a Arc<dyn EventStore>,
    /// Visibility projection store, so the retirement generation's start
    /// projects exactly as any other generation's does.
    pub visibility_store: &'a Arc<dyn VisibilityStore>,
}

/// The refusal a workloop with no `retire/1` entry receives.
///
/// Names the module and the missing entry, because the operator's next action
/// is to redeploy that module and they need to know which one.
#[must_use]
pub fn missing_entry_refusal(loop_id: &WorkflowId, module: &str) -> String {
    format!(
        "workloop {loop_id} declares a retire body but its deployed module `{module}` exports no \
         `{RETIRE_ENTRY}/{RETIRE_ARITY}`. Refusing to retire: the engine cannot distinguish a \
         module compiled before the retire entry existed from a loop that declared no cleanup at \
         all, and silently skipping a declared retirement is how a lease is lost or a queue is \
         stranded. Nothing was recorded and the loop is still running. Redeploy `{module}` from a \
         toolchain that emits `{RETIRE_ENTRY}/{RETIRE_ARITY}`, then retire again"
    )
}

/// Runs the loop's retire body to completion in a generation of its own, with
/// its effects recorded through the loop's ONE Recorder.
///
/// In order: probe the deployed module's retire entry (refusing before
/// anything changes), stand down any resident generation, open the retirement
/// generation atomically, spawn [`RETIRE_ENTRY`] with the loop's carry,
/// publish it as the workflow's SOLE writer, and await its exit. The
/// publication is removed before returning, whichever way the body ended, so
/// the caller can then record the terminal batch through a one-shot recorder
/// without a second writer.
///
/// # Errors
///
/// Refuses an already-terminal run, a missing retire entry (with
/// [`missing_entry_refusal`]), an unresolvable package, a failed spawn, a
/// workflow that already has a writer, and a body that failed or was killed —
/// a retirement whose cleanup did not complete must not record a terminal
/// claiming it did.
pub async fn run_retire_body(invocation: &RetireInvocation<'_>) -> Result<(), WorkloopError> {
    let history = invocation.store.read_history(invocation.loop_id).await?;
    refuse_if_terminal(invocation.loop_id, &history)?;

    // The entry probe comes FIRST, before the stand-down and before any
    // append: a module that cannot run the cleanup must leave the loop
    // byte-identical, still running, still registered.
    let generation = current_generation(invocation, &history)?;
    if !invocation
        .runtime
        .module_exports_function(&generation.module, RETIRE_ENTRY)
    {
        return Err(WorkloopError::Engine {
            reason: missing_entry_refusal(invocation.loop_id, &generation.module),
        });
    }

    // 🔴 THE RETIREMENT GENERATION IS OPENED **BEFORE** THE RESIDENT ONE IS
    // STOOD DOWN, AND THE ORDER IS THE WHOLE POINT.
    //
    // The stand-down cancels the loop's resident process. The process-exit
    // monitor then wakes on that kill and — finding a run with no terminal and
    // no newer lease in the registry — records `WorkflowFailed` for it. So a
    // loop retired while an iteration was IN FLIGHT was killed and marked
    // FAILED, its invariants fanned a loop-dead alarm, and the retirement's own
    // append lost a sequence race with the monitor's. A retirement is not a
    // failure, and the operator's verb must not manufacture one.
    //
    // An ITERATION CLOSE has always had this right: it records its terminal
    // batch first and cancels the process second, so the monitor finds a run
    // that already continued and stands down. Retirement is the same kind of
    // generation boundary and now takes the same order. Opening the retirement
    // generation records `WorkflowContinuedAsNew` for the current run, which IS
    // that run's terminal — after it, the kill has nothing to report.
    //
    // It was only reachable once the sweep actually ran: before the workloop
    // service had a production call site no loop ever iterated, so every test
    // retired either a hand-seeded history or a parked loop, and a parked loop
    // has no resident process for the stand-down to kill.
    let (retirement_run, recorder) = open_retirement_generation(invocation, &generation).await?;

    stand_down_resident_generation(
        invocation.registry,
        invocation.runtime,
        invocation.loop_id,
        &generation.carry,
    )?;
    let input =
        RuntimeInput::from_payload(&generation.carry).map_err(|error| WorkloopError::Engine {
            reason: format!("encoding the retire body's carry argument failed: {error}"),
        })?;
    let module = generation.module.clone();
    let pid = invocation
        .runtime
        .spawn_workflow(&module, RETIRE_ENTRY, input)
        .map_err(|error| WorkloopError::Engine {
            reason: format!("spawning `{module}:{RETIRE_ENTRY}/{RETIRE_ARITY}` failed: {error}"),
        })?;

    // Everything from here owns the spawned process: any failure before the
    // await must cancel it, or the retirement leaves a live, unmonitored,
    // unregistered process on the loop's own module.
    if let Err(error) = publish_retire_body(invocation, &retirement_run, pid, &generation, recorder)
    {
        cancel_orphaned_body(invocation.runtime, invocation.loop_id, pid, &error);
        return Err(error);
    }

    let outcome = await_retire_body(invocation.runtime, invocation.loop_id, pid).await;

    // Unpublish before the caller records the terminal batch, so the loop has
    // exactly one writer again whichever way the body ended.
    if let Err(error) = invocation
        .registry
        .remove(invocation.loop_id, &retirement_run)
    {
        tracing::warn!(
            loop_id = %invocation.loop_id,
            error = %error,
            "removing the retire body's registry publication failed; the terminal batch below \
             appends through a fresh one-shot recorder and a stale publication would make that \
             a second writer"
        );
    }
    outcome
}

/// The loop's current generation, as the retirement needs to see it.
struct CurrentGeneration {
    run_id: RunId,
    workflow_type: String,
    package_version: aion_core::PackageVersion,
    loaded_version: aion_package::ContentHash,
    module: String,
    carry: Payload,
}

/// Refuses a retirement whose run already recorded a terminal.
///
/// Checked HERE — before the body runs — rather than only at the terminal
/// append. Running the cleanup first and refusing afterwards means a second
/// retirement of an already-retired loop re-releases a released lease and
/// re-drains a drained queue, and only then reports that it should not have.
/// The refusal must come before the effects, not after them.
///
/// # Errors
///
/// Returns [`WorkloopError::Engine`] naming the loop when its active run holds
/// a terminal.
pub fn refuse_if_terminal(loop_id: &WorkflowId, history: &[Event]) -> Result<(), WorkloopError> {
    if aion_core::current_lease_terminal(history).is_some() {
        return Err(WorkloopError::Engine {
            reason: format!(
                "cannot retire workloop {loop_id}: its run already recorded a terminal, so it \
                 is not running and has nothing left to retire. No retire body was invoked — a \
                 declared cleanup that ran a second time would release an already-released \
                 lease and re-drain a drained queue"
            ),
        });
    }
    Ok(())
}

/// Ends a resident generation the way an iteration close ends one: notify the
/// continuation, drop the registry entry, cancel the process.
///
/// # 🔴 THE PROCESS IS CANCELLED, NOT ORPHANED
///
/// Removing the handle alone leaves a runnable BEAM process whose pid no
/// longer resolves to any handle. Its next durable NIF waits out the whole
/// registration birth-wait budget and then fails typed, which the SDKs treat
/// as `{badmatch, {error, _}}` — a workflow killed obscurely, minutes later,
/// with no line anywhere saying a retirement did it. Cancelling says it now.
fn stand_down_resident_generation(
    registry: &Arc<Registry>,
    runtime: &Arc<RuntimeHandle>,
    loop_id: &WorkflowId,
    carry: &Payload,
) -> Result<(), WorkloopError> {
    let handles = registry.list().map_err(|error| WorkloopError::Engine {
        reason: format!("listing the registry to stand down {loop_id} failed: {error}"),
    })?;
    for handle in handles
        .into_iter()
        .filter(|handle| handle.workflow_id() == loop_id)
    {
        let run_id = handle.run_id().clone();
        let pid = handle.pid();
        tracing::info!(
            %loop_id,
            run_id = %run_id,
            pid,
            "retirement is standing down the loop's resident generation before running its \
             declared retire body; the generation's own work stops here"
        );
        // The same notification an iteration close sends, carrying the same
        // payload the retirement generation is about to be started with: a
        // caller awaiting this generation learns it continued, and learns
        // what it continued with.
        handle.completion().notify(TerminalOutcome::ContinuedAsNew {
            input: carry.clone(),
            workflow_type: None,
            parent_run_id: run_id.clone(),
        });
        registry
            .remove(loop_id, &run_id)
            .map_err(|error| WorkloopError::Engine {
                reason: format!(
                    "removing the resident generation's handle for {loop_id} run {run_id} \
                     failed: {error}"
                ),
            })?;
        runtime
            .cancel_pid(pid)
            .map_err(|error| WorkloopError::Engine {
                reason: format!(
                    "ending the resident generation's process {pid} for {loop_id} failed: \
                     {error}. Refusing to run the retire body: that process is still runnable \
                     on the loop's history and would be a second writer alongside the body"
                ),
            })?;
    }
    Ok(())
}

/// Records `[WorkflowContinuedAsNew, WorkflowStarted]` for the retirement,
/// returning the retirement generation's run id.
async fn open_retirement_generation(
    invocation: &RetireInvocation<'_>,
    generation: &CurrentGeneration,
) -> Result<(RunId, Recorder), WorkloopError> {
    let retirement_run = RunId::new_v4();
    let start = WorkflowStartRecord {
        workflow_type: generation.workflow_type.clone(),
        input: generation.carry.clone(),
        run_id: retirement_run.clone(),
        parent_run_id: Some(generation.run_id.clone()),
        parent_workflow_id: None,
        package_version: generation.package_version.clone(),
    };

    // 🔴 THROUGH THE LOOP'S **ONE** RECORDER (invariant 3).
    //
    // A resident generation holds a live `Recorder` behind its registry
    // handle, and that recorder owns the loop's tracked sequence head. A
    // one-shot `resume_at` built from a freshly read history is a SECOND writer
    // for the same workflow, and against a live loop it loses: the cadence
    // sweep and the generation's own durable calls append between the read and
    // the write, and the append fails with a `SequenceConflict` — which is the
    // store telling us, correctly, that there were two writers.
    //
    // So the append goes through the live handle's recorder when the loop is
    // resident, and through a one-shot only when it is genuinely parked and
    // there is no other writer to be. This is the same rule
    // `Engine::with_loop_recorder` follows for every other out-of-band append
    // to a loop.
    if let Some(handle) = live_handle(invocation)? {
        let recorder = handle.recorder();
        let mut recorder = recorder.lock().await;
        let history = invocation.store.read_history(invocation.loop_id).await?;
        refuse_if_terminal(invocation.loop_id, &history)?;
        recorder
            .record_workloop_retirement_generation(
                Utc::now(),
                generation.carry.clone(),
                generation.run_id.clone(),
                start,
            )
            .await?;
        // The body's own recorder is built AFTER the stand-down removes this
        // handle, from the head this append left behind, so the loop still has
        // exactly one writer at every instant.
        let head = recorder.head();
        drop(recorder);
        return Ok((
            retirement_run.clone(),
            Recorder::resume_at(
                invocation.loop_id.clone(),
                Arc::clone(invocation.store),
                head,
            )
            .with_visibility(retirement_run, Arc::clone(invocation.visibility_store)),
        ));
    }

    let history = invocation.store.read_history(invocation.loop_id).await?;
    refuse_if_terminal(invocation.loop_id, &history)?;
    let head = history.iter().map(Event::seq).max().unwrap_or_default();
    let mut recorder = Recorder::resume_at(
        invocation.loop_id.clone(),
        Arc::clone(invocation.store),
        head,
    )
    .with_visibility(
        retirement_run.clone(),
        Arc::clone(invocation.visibility_store),
    );
    recorder
        .record_workloop_retirement_generation(
            Utc::now(),
            generation.carry.clone(),
            generation.run_id.clone(),
            start,
        )
        .await?;
    Ok((retirement_run, recorder))
}

/// The loop's live registry handle, when a generation of it is resident.
///
/// # Errors
///
/// Propagates a registry read failure rather than treating it as "not
/// resident": a poisoned registry cannot license the one-shot recorder path,
/// because that path's whole precondition is that no other writer exists.
fn live_handle(invocation: &RetireInvocation<'_>) -> Result<Option<WorkflowHandle>, WorkloopError> {
    // aion#213: resolved as the loop's SOLE handle, never a first match. This
    // function's own name says "the live writer", and `open_retirement_generation`
    // appends the retirement boundary through the recorder it returns — so two
    // handles here is the invariant-3 breach, not a set to choose from, and a
    // refusal is the only honest answer.
    invocation
        .registry
        .sole_handle(invocation.loop_id)
        .map_err(|error| WorkloopError::Engine {
            reason: format!(
                "resolving {}'s live writer failed: {error}",
                invocation.loop_id
            ),
        })
}

/// Publishes the retire body as the loop's SOLE writer.
fn publish_retire_body(
    invocation: &RetireInvocation<'_>,
    retirement_run: &RunId,
    pid: crate::Pid,
    generation: &CurrentGeneration,
    recorder: Recorder,
) -> Result<(), WorkloopError> {
    // 🔴 THE HANDLE CARRIES THE RECORDER THAT OPENED THE GENERATION.
    //
    // Not a fresh `resume_at`: that would need the post-append head, and the
    // only way to learn it is another whole-history read whose answer this
    // recorder already holds. Carrying it forward is also the stricter
    // reading of the one-writer law — the same instance that appended the
    // retirement generation is the one the body appends through.
    let handle = WorkflowHandle::new(WorkflowHandleParts {
        workflow_id: invocation.loop_id.clone(),
        run_id: retirement_run.clone(),
        pid,
        workflow_type: generation.workflow_type.clone(),
        namespace: String::from("default"),
        loaded_version: generation.loaded_version.clone(),
        cached_status: WorkflowStatus::Running,
        residency: HandleResidency::Resident,
        recorder,
        completion: CompletionNotifier::new(),
    });
    invocation
        .registry
        .insert_sole_workflow_writer((invocation.loop_id.clone(), retirement_run.clone()), handle)
        .map_err(|error| WorkloopError::Engine {
            reason: format!("publishing the retire body's handle failed: {error}"),
        })
}

/// A spawned body that never became registered or monitored must not be left
/// running: nothing would ever observe its exit, and it would go on calling
/// durable NIFs against a loop that is being retired.
fn cancel_orphaned_body(
    runtime: &Arc<RuntimeHandle>,
    loop_id: &WorkflowId,
    pid: crate::Pid,
    cause: &WorkloopError,
) {
    if let Err(error) = runtime.cancel_pid(pid) {
        tracing::error!(
            %loop_id,
            pid,
            cause = %cause,
            error = %error,
            "the retire body was spawned but could neither be published nor cancelled; it is \
             running unmonitored on the loop's module and nothing will observe its exit"
        );
    }
}

/// Await the retire body's exit and classify it.
async fn await_retire_body(
    runtime: &Arc<RuntimeHandle>,
    loop_id: &WorkflowId,
    pid: crate::Pid,
) -> Result<(), WorkloopError> {
    let (sender, receiver) = tokio::sync::oneshot::channel();
    // A monitor that cannot be armed leaves a RUNNING body nothing will ever
    // observe the exit of — the same leak as a failed publication, one step
    // later. It owns the process until the monitor is armed, so it cancels.
    if let Err(error) = runtime.monitor_process(pid, move |outcome| {
        // A closed receiver means the awaiting task is gone; nothing to
        // report, and the send failure is not a fault of its own.
        let _ = sender.send(outcome);
    }) {
        let failure = WorkloopError::Engine {
            reason: format!("monitoring the retire body's process {pid} failed: {error}"),
        };
        cancel_orphaned_body(runtime, loop_id, pid, &failure);
        return Err(failure);
    }
    let outcome = receiver.await.map_err(|_| WorkloopError::Engine {
        reason: format!(
            "the retire body's process {pid} exit was never reported; refusing to record a \
             retirement whose cleanup cannot be shown to have completed"
        ),
    })?;
    let outcome = outcome.map_err(|error| WorkloopError::Engine {
        reason: format!("observing the retire body's exit failed: {error}"),
    })?;
    classify(pid, &outcome)
}

/// A retirement records its terminal only when the cleanup actually finished.
fn classify(
    pid: crate::Pid,
    outcome: &crate::runtime::outcome::WorkflowProcessOutcome,
) -> Result<(), WorkloopError> {
    use crate::runtime::outcome::WorkflowProcessOutcome;
    match outcome {
        WorkflowProcessOutcome::Completed(_) => Ok(()),
        // 🔴 THIS MESSAGE SAYS WHAT IS TRUE, NOT WHAT IS TIDY.
        //
        // It used to claim "Nothing is recorded and the loop is still
        // running". Both halves were false. Everything the body did before
        // failing IS recorded — that is the protocol's stated point — and the
        // retirement generation the body ran in was opened before it started,
        // so the loop's previous generation has already continued. Telling an
        // operator that a failed retirement left no trace sends them looking
        // for the wrong thing in the right history.
        WorkflowProcessOutcome::Failed(error) => Err(WorkloopError::Engine {
            reason: format!(
                "the retire body (process {pid}) failed: {message}. No `LoopRetired` terminal \
                 is recorded — that would claim a cleanup which did not finish — so the loop is \
                 NOT retired and stays registered on the sweep set. What the body did before \
                 failing IS in the loop's history, inside the retirement generation opened for \
                 it; a further retirement attempt opens a fresh generation and runs the body \
                 again from the top, so any effect it already performed will be performed twice \
                 unless the body is idempotent",
                message = error.message
            ),
        }),
    }
}

/// The loop's current generation: its run, its recorded identity, its deployed
/// module, and the carry the retire body receives.
///
/// # 🔴 THE CARRY IS THE GENERATION'S RECORDED INPUT, READ ONCE
///
/// S3's central claim is that the retire body's argument is byte-identical to
/// what that generation's `run/1` was given. It is true because both come from
/// the SAME field of the SAME event: the generation's `WorkflowStarted.input`.
/// Nothing re-derives, re-encodes, or merges it on the way here.
fn current_generation(
    invocation: &RetireInvocation<'_>,
    history: &[Event],
) -> Result<CurrentGeneration, WorkloopError> {
    let loop_id = invocation.loop_id;
    let (run_id, workflow_type, package_version, carry) = history
        .iter()
        .rev()
        .find_map(|event| match event {
            Event::WorkflowStarted {
                run_id,
                workflow_type,
                package_version,
                input,
                ..
            } => Some((
                run_id.clone(),
                workflow_type.clone(),
                package_version.clone(),
                input.clone(),
            )),
            _ => None,
        })
        .ok_or_else(|| WorkloopError::Engine {
            reason: format!("workloop {loop_id} has no recorded generation to retire"),
        })?;
    let loaded_version = crate::loader::parse_package_version(&workflow_type, &package_version)
        .map_err(|error| WorkloopError::Engine {
            reason: format!("resolving the retiring loop's package version failed: {error}"),
        })?;
    let loaded = invocation
        .catalog
        .get(&workflow_type, &loaded_version)
        .map_err(|error| WorkloopError::Engine {
            reason: format!("resolving the retiring loop's package failed: {error}"),
        })?
        .ok_or_else(|| WorkloopError::Engine {
            reason: format!(
                "workloop {loop_id} is pinned to package version {loaded_version} of \
                 `{workflow_type}`, which is not loaded on this engine, so its retire body \
                 cannot be reached"
            ),
        })?;
    let module = loaded.deployed_entry_module().to_owned();
    Ok(CurrentGeneration {
        run_id,
        workflow_type,
        package_version,
        loaded_version,
        module,
        carry,
    })
}