Skip to main content

aion/workloop/
retire.rs

1//! The retire-body invocation protocol (S3).
2//!
3//! # 🔴 THE CONTRACT THE EMITTER MUST TARGET, VERBATIM
4//!
5//! A workloop document that declares a `retire` block compiles to a SECOND
6//! exported entry alongside the ordinary run entry:
7//!
8//! ```erlang
9//! -export([run/1, retire/1]).
10//!
11//! retire(Carry) -> any().
12//! ```
13//!
14//! - **Entry name**: [`RETIRE_ENTRY`] — `retire`, arity 1.
15//! - **Argument**: the loop's CURRENT carry, byte-identical to the value the
16//!   same generation's `run/1` was given. The retire body sees exactly the
17//!   state the last iteration left, because a retirement that could not see
18//!   the loop's final state could not clean it up.
19//! - **Return**: ignored. The retirement RESULT is the operator-supplied
20//!   payload on [`crate::Engine::retire_workloop`], not whatever the body
21//!   returns — the body exists for its EFFECTS (draining a queue, releasing a
22//!   lease, notifying a peer), and letting it also decide the terminal result
23//!   would give one block two unrelated jobs.
24//! - **Effects**: recorded through the loop's ONE Recorder. Every activity,
25//!   timer, signal, child and hatch the body uses appends to the loop's own
26//!   history exactly as an iteration's would.
27//! - **Ordering**: the body runs to completion BEFORE the terminal
28//!   `[LoopRetired + WorkflowCompleted]` batch is recorded. A retirement that
29//!   recorded its terminal first would be asking a completed run to keep
30//!   working, and the single-writer law would refuse the body's own appends.
31//!
32//! # 🔴 THE RETIREMENT IS THE LOOP'S FINAL GENERATION
33//!
34//! The body does not run inside the generation the loop was parked in. It runs
35//! inside a generation opened for it, and both halves of that decision are
36//! load-bearing.
37//!
38//! **The ordinal space.** Every positional durable command — `dispatch_activity`,
39//! `spawn_child`, `hatch_detached` — keys on a counter that starts at ZERO and
40//! resolves against the current run segment
41//! (`NifContext::new_with_history_store` → `current_run_segment`). A body run
42//! inside a generation that already dispatched an activity would have its own
43//! first `dispatch_activity` fast-forward to that recorded activity and return
44//! its result WITHOUT EXECUTING — a declared cleanup that appends nothing and
45//! hands the author the loop's own stale data. `workflow.now()` would answer
46//! the iteration's recorded clock for the same reason. The retirement
47//! generation's segment is empty when the body starts, so every command it
48//! issues can only resolve one way: live.
49//!
50//! **The single writer.** The body is published into the registry so its NIF
51//! calls resolve to the loop's recorder, and a `Recorder` writes the WORKFLOW's
52//! event stream — so a handle for the body plus a live handle for the loop is
53//! two writers for one history (invariant 3). The publication therefore goes
54//! through [`Registry::insert_sole_workflow_writer`], which proves the absence
55//! of any other handle UNDER THE LOCK rather than asserting it in a comment.
56//! An earlier version used `Registry::insert`, which is documented to REPLACE,
57//! and dropped the displaced handle unexamined: retiring a loop that had not
58//! yet parked left two Recorders live, and stopped the displaced process's pid
59//! resolving at all — its next durable NIF stalled the full birth-wait budget
60//! and then failed typed, which the SDKs treat as fatal.
61//!
62//! So a resident generation is STOOD DOWN first, explicitly and loudly: its
63//! continuation is notified, its handle removed, and its process cancelled —
64//! the same three steps an iteration close performs, because a retirement is
65//! the same kind of generation boundary. Retirement stops the loop; that has
66//! always been its meaning. What it must not do is stop it by accident.
67//!
68//! # 🔴 A MISSING ENTRY IS REFUSED LOUDLY, NEVER SKIPPED
69//!
70//! A workloop compiled before the emitter change exports no `retire/1`. The
71//! engine must not treat that as "no retire body declared" — it cannot tell
72//! the two apart, and silently skipping a declared cleanup is the failure
73//! mode that loses a lease or strands a queue. So retirement REFUSES with a
74//! diagnostic naming the module and the missing entry. That probe runs BEFORE
75//! anything is stood down or recorded, so the refusal really does leave the
76//! loop exactly as it was.
77//!
78//! # 🔴 THE RETIRE BODY MUST BE IDEMPOTENT, AND THAT IS THE AUTHOR'S JOB
79//!
80//! The body runs before any terminal is recorded, so a crash mid-retire
81//! leaves the loop un-retired with no `LoopRetired` in history. The next
82//! retirement attempt opens a FRESH retirement generation and runs the body
83//! again from the top — deliberately, because the alternative is resolving the
84//! second attempt's commands against the first attempt's partial record, which
85//! is the silent-skip failure in another costume. Effects the body performed
86//! are therefore re-performed on the retry. This is stated rather than
87//! engineered around: recording a terminal before the cleanup finished would
88//! claim a retirement that did not happen.
89
90use std::sync::Arc;
91
92use aion_core::{Event, Payload, RunId, WorkflowId, WorkflowStatus};
93use aion_store::EventStore;
94use aion_store::visibility::VisibilityStore;
95use chrono::Utc;
96
97use super::error::WorkloopError;
98use crate::durability::{Recorder, WorkflowStartRecord};
99use crate::loader::WorkflowCatalog;
100use crate::registry::{
101    CompletionNotifier, HandleResidency, Registry, TerminalOutcome, WorkflowHandle,
102    WorkflowHandleParts,
103};
104use crate::runtime::{RuntimeHandle, RuntimeInput};
105
106/// The exported entry a compiled `retire` block lands on: `retire/1`.
107///
108/// A constant because three places must agree — the existence probe, the
109/// spawn, and the refusal diagnostic — and a name known in three places is a
110/// name that drifts.
111pub const RETIRE_ENTRY: &str = "retire";
112
113/// Arity of [`RETIRE_ENTRY`]. One argument: the loop's current carry.
114pub const RETIRE_ARITY: u32 = 1;
115
116/// Everything invoking one retire body needs.
117pub struct RetireInvocation<'a> {
118    /// The loop being retired.
119    pub loop_id: &'a WorkflowId,
120    /// The engine's runtime handle.
121    pub runtime: &'a Arc<RuntimeHandle>,
122    /// The catalog resolving the loop's deployed module.
123    pub catalog: &'a WorkflowCatalog,
124    /// The registry the retire process is published into.
125    pub registry: &'a Arc<Registry>,
126    /// Event store, for the loop's one-shot recorders.
127    pub store: &'a Arc<dyn EventStore>,
128    /// Visibility projection store, so the retirement generation's start
129    /// projects exactly as any other generation's does.
130    pub visibility_store: &'a Arc<dyn VisibilityStore>,
131}
132
133/// The refusal a workloop with no `retire/1` entry receives.
134///
135/// Names the module and the missing entry, because the operator's next action
136/// is to redeploy that module and they need to know which one.
137#[must_use]
138pub fn missing_entry_refusal(loop_id: &WorkflowId, module: &str) -> String {
139    format!(
140        "workloop {loop_id} declares a retire body but its deployed module `{module}` exports no \
141         `{RETIRE_ENTRY}/{RETIRE_ARITY}`. Refusing to retire: the engine cannot distinguish a \
142         module compiled before the retire entry existed from a loop that declared no cleanup at \
143         all, and silently skipping a declared retirement is how a lease is lost or a queue is \
144         stranded. Nothing was recorded and the loop is still running. Redeploy `{module}` from a \
145         toolchain that emits `{RETIRE_ENTRY}/{RETIRE_ARITY}`, then retire again"
146    )
147}
148
149/// Runs the loop's retire body to completion in a generation of its own, with
150/// its effects recorded through the loop's ONE Recorder.
151///
152/// In order: probe the deployed module's retire entry (refusing before
153/// anything changes), stand down any resident generation, open the retirement
154/// generation atomically, spawn [`RETIRE_ENTRY`] with the loop's carry,
155/// publish it as the workflow's SOLE writer, and await its exit. The
156/// publication is removed before returning, whichever way the body ended, so
157/// the caller can then record the terminal batch through a one-shot recorder
158/// without a second writer.
159///
160/// # Errors
161///
162/// Refuses an already-terminal run, a missing retire entry (with
163/// [`missing_entry_refusal`]), an unresolvable package, a failed spawn, a
164/// workflow that already has a writer, and a body that failed or was killed —
165/// a retirement whose cleanup did not complete must not record a terminal
166/// claiming it did.
167pub async fn run_retire_body(invocation: &RetireInvocation<'_>) -> Result<(), WorkloopError> {
168    let history = invocation.store.read_history(invocation.loop_id).await?;
169    refuse_if_terminal(invocation.loop_id, &history)?;
170
171    // The entry probe comes FIRST, before the stand-down and before any
172    // append: a module that cannot run the cleanup must leave the loop
173    // byte-identical, still running, still registered.
174    let generation = current_generation(invocation, &history)?;
175    if !invocation
176        .runtime
177        .module_exports_function(&generation.module, RETIRE_ENTRY)
178    {
179        return Err(WorkloopError::Engine {
180            reason: missing_entry_refusal(invocation.loop_id, &generation.module),
181        });
182    }
183
184    // 🔴 THE RETIREMENT GENERATION IS OPENED **BEFORE** THE RESIDENT ONE IS
185    // STOOD DOWN, AND THE ORDER IS THE WHOLE POINT.
186    //
187    // The stand-down cancels the loop's resident process. The process-exit
188    // monitor then wakes on that kill and — finding a run with no terminal and
189    // no newer lease in the registry — records `WorkflowFailed` for it. So a
190    // loop retired while an iteration was IN FLIGHT was killed and marked
191    // FAILED, its invariants fanned a loop-dead alarm, and the retirement's own
192    // append lost a sequence race with the monitor's. A retirement is not a
193    // failure, and the operator's verb must not manufacture one.
194    //
195    // An ITERATION CLOSE has always had this right: it records its terminal
196    // batch first and cancels the process second, so the monitor finds a run
197    // that already continued and stands down. Retirement is the same kind of
198    // generation boundary and now takes the same order. Opening the retirement
199    // generation records `WorkflowContinuedAsNew` for the current run, which IS
200    // that run's terminal — after it, the kill has nothing to report.
201    //
202    // It was only reachable once the sweep actually ran: before the workloop
203    // service had a production call site no loop ever iterated, so every test
204    // retired either a hand-seeded history or a parked loop, and a parked loop
205    // has no resident process for the stand-down to kill.
206    let (retirement_run, recorder) = open_retirement_generation(invocation, &generation).await?;
207
208    stand_down_resident_generation(
209        invocation.registry,
210        invocation.runtime,
211        invocation.loop_id,
212        &generation.carry,
213    )?;
214    let input =
215        RuntimeInput::from_payload(&generation.carry).map_err(|error| WorkloopError::Engine {
216            reason: format!("encoding the retire body's carry argument failed: {error}"),
217        })?;
218    let module = generation.module.clone();
219    let pid = invocation
220        .runtime
221        .spawn_workflow(&module, RETIRE_ENTRY, input)
222        .map_err(|error| WorkloopError::Engine {
223            reason: format!("spawning `{module}:{RETIRE_ENTRY}/{RETIRE_ARITY}` failed: {error}"),
224        })?;
225
226    // Everything from here owns the spawned process: any failure before the
227    // await must cancel it, or the retirement leaves a live, unmonitored,
228    // unregistered process on the loop's own module.
229    if let Err(error) = publish_retire_body(invocation, &retirement_run, pid, &generation, recorder)
230    {
231        cancel_orphaned_body(invocation.runtime, invocation.loop_id, pid, &error);
232        return Err(error);
233    }
234
235    let outcome = await_retire_body(invocation.runtime, invocation.loop_id, pid).await;
236
237    // Unpublish before the caller records the terminal batch, so the loop has
238    // exactly one writer again whichever way the body ended.
239    if let Err(error) = invocation
240        .registry
241        .remove(invocation.loop_id, &retirement_run)
242    {
243        tracing::warn!(
244            loop_id = %invocation.loop_id,
245            error = %error,
246            "removing the retire body's registry publication failed; the terminal batch below \
247             appends through a fresh one-shot recorder and a stale publication would make that \
248             a second writer"
249        );
250    }
251    outcome
252}
253
254/// The loop's current generation, as the retirement needs to see it.
255struct CurrentGeneration {
256    run_id: RunId,
257    workflow_type: String,
258    package_version: aion_core::PackageVersion,
259    loaded_version: aion_package::ContentHash,
260    module: String,
261    carry: Payload,
262}
263
264/// Refuses a retirement whose run already recorded a terminal.
265///
266/// Checked HERE — before the body runs — rather than only at the terminal
267/// append. Running the cleanup first and refusing afterwards means a second
268/// retirement of an already-retired loop re-releases a released lease and
269/// re-drains a drained queue, and only then reports that it should not have.
270/// The refusal must come before the effects, not after them.
271///
272/// # Errors
273///
274/// Returns [`WorkloopError::Engine`] naming the loop when its active run holds
275/// a terminal.
276pub fn refuse_if_terminal(loop_id: &WorkflowId, history: &[Event]) -> Result<(), WorkloopError> {
277    if aion_core::current_lease_terminal(history).is_some() {
278        return Err(WorkloopError::Engine {
279            reason: format!(
280                "cannot retire workloop {loop_id}: its run already recorded a terminal, so it \
281                 is not running and has nothing left to retire. No retire body was invoked — a \
282                 declared cleanup that ran a second time would release an already-released \
283                 lease and re-drain a drained queue"
284            ),
285        });
286    }
287    Ok(())
288}
289
290/// Ends a resident generation the way an iteration close ends one: notify the
291/// continuation, drop the registry entry, cancel the process.
292///
293/// # 🔴 THE PROCESS IS CANCELLED, NOT ORPHANED
294///
295/// Removing the handle alone leaves a runnable BEAM process whose pid no
296/// longer resolves to any handle. Its next durable NIF waits out the whole
297/// registration birth-wait budget and then fails typed, which the SDKs treat
298/// as `{badmatch, {error, _}}` — a workflow killed obscurely, minutes later,
299/// with no line anywhere saying a retirement did it. Cancelling says it now.
300fn stand_down_resident_generation(
301    registry: &Arc<Registry>,
302    runtime: &Arc<RuntimeHandle>,
303    loop_id: &WorkflowId,
304    carry: &Payload,
305) -> Result<(), WorkloopError> {
306    let handles = registry.list().map_err(|error| WorkloopError::Engine {
307        reason: format!("listing the registry to stand down {loop_id} failed: {error}"),
308    })?;
309    for handle in handles
310        .into_iter()
311        .filter(|handle| handle.workflow_id() == loop_id)
312    {
313        let run_id = handle.run_id().clone();
314        let pid = handle.pid();
315        tracing::info!(
316            %loop_id,
317            run_id = %run_id,
318            pid,
319            "retirement is standing down the loop's resident generation before running its \
320             declared retire body; the generation's own work stops here"
321        );
322        // The same notification an iteration close sends, carrying the same
323        // payload the retirement generation is about to be started with: a
324        // caller awaiting this generation learns it continued, and learns
325        // what it continued with.
326        handle.completion().notify(TerminalOutcome::ContinuedAsNew {
327            input: carry.clone(),
328            workflow_type: None,
329            parent_run_id: run_id.clone(),
330        });
331        registry
332            .remove(loop_id, &run_id)
333            .map_err(|error| WorkloopError::Engine {
334                reason: format!(
335                    "removing the resident generation's handle for {loop_id} run {run_id} \
336                     failed: {error}"
337                ),
338            })?;
339        runtime
340            .cancel_pid(pid)
341            .map_err(|error| WorkloopError::Engine {
342                reason: format!(
343                    "ending the resident generation's process {pid} for {loop_id} failed: \
344                     {error}. Refusing to run the retire body: that process is still runnable \
345                     on the loop's history and would be a second writer alongside the body"
346                ),
347            })?;
348    }
349    Ok(())
350}
351
352/// Records `[WorkflowContinuedAsNew, WorkflowStarted]` for the retirement,
353/// returning the retirement generation's run id.
354async fn open_retirement_generation(
355    invocation: &RetireInvocation<'_>,
356    generation: &CurrentGeneration,
357) -> Result<(RunId, Recorder), WorkloopError> {
358    let retirement_run = RunId::new_v4();
359    let start = WorkflowStartRecord {
360        workflow_type: generation.workflow_type.clone(),
361        input: generation.carry.clone(),
362        run_id: retirement_run.clone(),
363        parent_run_id: Some(generation.run_id.clone()),
364        parent_workflow_id: None,
365        package_version: generation.package_version.clone(),
366    };
367
368    // 🔴 THROUGH THE LOOP'S **ONE** RECORDER (invariant 3).
369    //
370    // A resident generation holds a live `Recorder` behind its registry
371    // handle, and that recorder owns the loop's tracked sequence head. A
372    // one-shot `resume_at` built from a freshly read history is a SECOND writer
373    // for the same workflow, and against a live loop it loses: the cadence
374    // sweep and the generation's own durable calls append between the read and
375    // the write, and the append fails with a `SequenceConflict` — which is the
376    // store telling us, correctly, that there were two writers.
377    //
378    // So the append goes through the live handle's recorder when the loop is
379    // resident, and through a one-shot only when it is genuinely parked and
380    // there is no other writer to be. This is the same rule
381    // `Engine::with_loop_recorder` follows for every other out-of-band append
382    // to a loop.
383    if let Some(handle) = live_handle(invocation)? {
384        let recorder = handle.recorder();
385        let mut recorder = recorder.lock().await;
386        let history = invocation.store.read_history(invocation.loop_id).await?;
387        refuse_if_terminal(invocation.loop_id, &history)?;
388        recorder
389            .record_workloop_retirement_generation(
390                Utc::now(),
391                generation.carry.clone(),
392                generation.run_id.clone(),
393                start,
394            )
395            .await?;
396        // The body's own recorder is built AFTER the stand-down removes this
397        // handle, from the head this append left behind, so the loop still has
398        // exactly one writer at every instant.
399        let head = recorder.head();
400        drop(recorder);
401        return Ok((
402            retirement_run.clone(),
403            Recorder::resume_at(
404                invocation.loop_id.clone(),
405                Arc::clone(invocation.store),
406                head,
407            )
408            .with_visibility(retirement_run, Arc::clone(invocation.visibility_store)),
409        ));
410    }
411
412    let history = invocation.store.read_history(invocation.loop_id).await?;
413    refuse_if_terminal(invocation.loop_id, &history)?;
414    let head = history.iter().map(Event::seq).max().unwrap_or_default();
415    let mut recorder = Recorder::resume_at(
416        invocation.loop_id.clone(),
417        Arc::clone(invocation.store),
418        head,
419    )
420    .with_visibility(
421        retirement_run.clone(),
422        Arc::clone(invocation.visibility_store),
423    );
424    recorder
425        .record_workloop_retirement_generation(
426            Utc::now(),
427            generation.carry.clone(),
428            generation.run_id.clone(),
429            start,
430        )
431        .await?;
432    Ok((retirement_run, recorder))
433}
434
435/// The loop's live registry handle, when a generation of it is resident.
436///
437/// # Errors
438///
439/// Propagates a registry read failure rather than treating it as "not
440/// resident": a poisoned registry cannot license the one-shot recorder path,
441/// because that path's whole precondition is that no other writer exists.
442fn live_handle(invocation: &RetireInvocation<'_>) -> Result<Option<WorkflowHandle>, WorkloopError> {
443    Ok(invocation
444        .registry
445        .list()
446        .map_err(|error| WorkloopError::Engine {
447            reason: format!(
448                "listing the registry to find {}'s live writer failed: {error}",
449                invocation.loop_id
450            ),
451        })?
452        .into_iter()
453        .find(|handle| handle.workflow_id() == invocation.loop_id))
454}
455
456/// Publishes the retire body as the loop's SOLE writer.
457fn publish_retire_body(
458    invocation: &RetireInvocation<'_>,
459    retirement_run: &RunId,
460    pid: crate::Pid,
461    generation: &CurrentGeneration,
462    recorder: Recorder,
463) -> Result<(), WorkloopError> {
464    // 🔴 THE HANDLE CARRIES THE RECORDER THAT OPENED THE GENERATION.
465    //
466    // Not a fresh `resume_at`: that would need the post-append head, and the
467    // only way to learn it is another whole-history read whose answer this
468    // recorder already holds. Carrying it forward is also the stricter
469    // reading of the one-writer law — the same instance that appended the
470    // retirement generation is the one the body appends through.
471    let handle = WorkflowHandle::new(WorkflowHandleParts {
472        workflow_id: invocation.loop_id.clone(),
473        run_id: retirement_run.clone(),
474        pid,
475        workflow_type: generation.workflow_type.clone(),
476        namespace: String::from("default"),
477        loaded_version: generation.loaded_version.clone(),
478        cached_status: WorkflowStatus::Running,
479        residency: HandleResidency::Resident,
480        recorder,
481        completion: CompletionNotifier::new(),
482    });
483    invocation
484        .registry
485        .insert_sole_workflow_writer((invocation.loop_id.clone(), retirement_run.clone()), handle)
486        .map_err(|error| WorkloopError::Engine {
487            reason: format!("publishing the retire body's handle failed: {error}"),
488        })
489}
490
491/// A spawned body that never became registered or monitored must not be left
492/// running: nothing would ever observe its exit, and it would go on calling
493/// durable NIFs against a loop that is being retired.
494fn cancel_orphaned_body(
495    runtime: &Arc<RuntimeHandle>,
496    loop_id: &WorkflowId,
497    pid: crate::Pid,
498    cause: &WorkloopError,
499) {
500    if let Err(error) = runtime.cancel_pid(pid) {
501        tracing::error!(
502            %loop_id,
503            pid,
504            cause = %cause,
505            error = %error,
506            "the retire body was spawned but could neither be published nor cancelled; it is \
507             running unmonitored on the loop's module and nothing will observe its exit"
508        );
509    }
510}
511
512/// Await the retire body's exit and classify it.
513async fn await_retire_body(
514    runtime: &Arc<RuntimeHandle>,
515    loop_id: &WorkflowId,
516    pid: crate::Pid,
517) -> Result<(), WorkloopError> {
518    let (sender, receiver) = tokio::sync::oneshot::channel();
519    // A monitor that cannot be armed leaves a RUNNING body nothing will ever
520    // observe the exit of — the same leak as a failed publication, one step
521    // later. It owns the process until the monitor is armed, so it cancels.
522    if let Err(error) = runtime.monitor_process(pid, move |outcome| {
523        // A closed receiver means the awaiting task is gone; nothing to
524        // report, and the send failure is not a fault of its own.
525        let _ = sender.send(outcome);
526    }) {
527        let failure = WorkloopError::Engine {
528            reason: format!("monitoring the retire body's process {pid} failed: {error}"),
529        };
530        cancel_orphaned_body(runtime, loop_id, pid, &failure);
531        return Err(failure);
532    }
533    let outcome = receiver.await.map_err(|_| WorkloopError::Engine {
534        reason: format!(
535            "the retire body's process {pid} exit was never reported; refusing to record a \
536             retirement whose cleanup cannot be shown to have completed"
537        ),
538    })?;
539    let outcome = outcome.map_err(|error| WorkloopError::Engine {
540        reason: format!("observing the retire body's exit failed: {error}"),
541    })?;
542    classify(pid, &outcome)
543}
544
545/// A retirement records its terminal only when the cleanup actually finished.
546fn classify(
547    pid: crate::Pid,
548    outcome: &crate::runtime::outcome::WorkflowProcessOutcome,
549) -> Result<(), WorkloopError> {
550    use crate::runtime::outcome::WorkflowProcessOutcome;
551    match outcome {
552        WorkflowProcessOutcome::Completed(_) => Ok(()),
553        // 🔴 THIS MESSAGE SAYS WHAT IS TRUE, NOT WHAT IS TIDY.
554        //
555        // It used to claim "Nothing is recorded and the loop is still
556        // running". Both halves were false. Everything the body did before
557        // failing IS recorded — that is the protocol's stated point — and the
558        // retirement generation the body ran in was opened before it started,
559        // so the loop's previous generation has already continued. Telling an
560        // operator that a failed retirement left no trace sends them looking
561        // for the wrong thing in the right history.
562        WorkflowProcessOutcome::Failed(error) => Err(WorkloopError::Engine {
563            reason: format!(
564                "the retire body (process {pid}) failed: {message}. No `LoopRetired` terminal \
565                 is recorded — that would claim a cleanup which did not finish — so the loop is \
566                 NOT retired and stays registered on the sweep set. What the body did before \
567                 failing IS in the loop's history, inside the retirement generation opened for \
568                 it; a further retirement attempt opens a fresh generation and runs the body \
569                 again from the top, so any effect it already performed will be performed twice \
570                 unless the body is idempotent",
571                message = error.message
572            ),
573        }),
574    }
575}
576
577/// The loop's current generation: its run, its recorded identity, its deployed
578/// module, and the carry the retire body receives.
579///
580/// # 🔴 THE CARRY IS THE GENERATION'S RECORDED INPUT, READ ONCE
581///
582/// S3's central claim is that the retire body's argument is byte-identical to
583/// what that generation's `run/1` was given. It is true because both come from
584/// the SAME field of the SAME event: the generation's `WorkflowStarted.input`.
585/// Nothing re-derives, re-encodes, or merges it on the way here.
586fn current_generation(
587    invocation: &RetireInvocation<'_>,
588    history: &[Event],
589) -> Result<CurrentGeneration, WorkloopError> {
590    let loop_id = invocation.loop_id;
591    let (run_id, workflow_type, package_version, carry) = history
592        .iter()
593        .rev()
594        .find_map(|event| match event {
595            Event::WorkflowStarted {
596                run_id,
597                workflow_type,
598                package_version,
599                input,
600                ..
601            } => Some((
602                run_id.clone(),
603                workflow_type.clone(),
604                package_version.clone(),
605                input.clone(),
606            )),
607            _ => None,
608        })
609        .ok_or_else(|| WorkloopError::Engine {
610            reason: format!("workloop {loop_id} has no recorded generation to retire"),
611        })?;
612    let loaded_version = crate::loader::parse_package_version(&workflow_type, &package_version)
613        .map_err(|error| WorkloopError::Engine {
614            reason: format!("resolving the retiring loop's package version failed: {error}"),
615        })?;
616    let loaded = invocation
617        .catalog
618        .get(&workflow_type, &loaded_version)
619        .map_err(|error| WorkloopError::Engine {
620            reason: format!("resolving the retiring loop's package failed: {error}"),
621        })?
622        .ok_or_else(|| WorkloopError::Engine {
623            reason: format!(
624                "workloop {loop_id} is pinned to package version {loaded_version} of \
625                 `{workflow_type}`, which is not loaded on this engine, so its retire body \
626                 cannot be reached"
627            ),
628        })?;
629    let module = loaded.deployed_entry_module().to_owned();
630    Ok(CurrentGeneration {
631        run_id,
632        workflow_type,
633        package_version,
634        loaded_version,
635        module,
636        carry,
637    })
638}