aion-rs 0.31.0

//! Generation boundaries: the atomic appends that end one run of a workflow
//! history and open the next one in it.
//!
//! Three transitions share this shape — a workloop iteration close, a workloop
//! retirement, and continue-as-new — and all three are the same durable
//! sentence: *this generation is over, that one has begun*. They live together
//! because the half that is easy to get wrong is not the batch, it is what
//! happens AFTER it: the recorder must follow the successor
//! ([`Recorder::follow_generation`]), or the loop grows one phantom running
//! row per window (#214) and, worse, a second recorder gets minted for the
//! same history (aion#213).
//!
//! Split out of `recorder.rs` as a named module rather than left inline
//! because that file is far past the 500-line production cap this codebase
//! holds itself to, and because the three boundaries only converge if they sit
//! where a reader can see all of them at once.

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

use super::{Recorder, WorkflowStartRecord};
use crate::durability::DurabilityError;

/// The `WorkflowContinuedAsNew` a boundary records for the outgoing
/// generation.
pub struct ContinuationTerminal {
    /// Payload carried into the successor, recorded on the terminal.
    pub input: Payload,
    /// Workflow type override recorded on the terminal, when the caller gave
    /// one.
    pub workflow_type: Option<String>,
}

/// The predecessor half of a continue-as-new boundary: the run that ends, the
/// terminal that ends it, and the declared-timeout deadline that terminal
/// retires.
pub struct ContinuedGeneration {
    /// The run this boundary closes.
    pub run_id: RunId,
    /// The terminal to record, or `None` when `WorkflowContinuedAsNew` is
    /// ALREADY durable for this run.
    ///
    /// The workflow-code path records its own terminal inside the
    /// `continue_as_new` NIF and ends its process; the successor is opened
    /// later, by the exit monitor. That boundary must not record a second
    /// terminal — a run has exactly one — so it opens the successor (and
    /// completes any owed deadline retirement) in a batch with no terminal in
    /// it.
    pub terminal: Option<ContinuationTerminal>,
    /// The predecessor's still-outstanding declared-timeout deadline, derived
    /// from history by the caller and retired in the SAME batch (D5). `None`
    /// for a run that declared no timeout, or whose deadline was already
    /// retired — LAW 1: nothing is minted here.
    pub outstanding_deadline: Option<TimerId>,
}

/// The successor half of a continue-as-new boundary: the generation the
/// transition opens, and the deadline it arms.
pub struct OpeningGeneration {
    /// Identity recorded on the successor's `WorkflowStarted`.
    pub start: WorkflowStartRecord,
    /// The successor's own declared-timeout deadline: its reserved timer id
    /// and the instant it fires. `None` for a package that declares no
    /// timeout, which records no deadline event of any kind (LAW 1).
    pub deadline: Option<(TimerId, DateTime<Utc>)>,
}

impl Recorder {
    /// Records a continue-as-new boundary ATOMICALLY as ONE batch:
    /// `[WorkflowContinuedAsNew?, TimerCancelled(predecessor deadline)?,
    /// WorkflowStarted(successor), TimerStarted(successor deadline)?]`
    /// (aion#213 R1).
    ///
    /// The two optional halves are optional for opposite reasons. The terminal
    /// is absent only when the workflow's own `continue_as_new` NIF already
    /// recorded it (see [`ContinuedGeneration::terminal`]); the deadline
    /// events are absent when the run in question declares no timeout, in
    /// which case NO deadline object of any kind is touched (LAW 1).
    ///
    /// # 🔴 WHY THE SUCCESSOR'S START IS IN THE PREDECESSOR'S BATCH
    ///
    /// It used to be four appends across two recorders: the terminal and the
    /// deadline retirement under the predecessor's recorder lock, then — lock
    /// released — `start_workflow_with_options` reading the history head
    /// UNLOCKED and seeding a SECOND recorder for the same workflow id. The
    /// predecessor's process is still alive across that gap, so its in-flight
    /// `sleep` arm appended through the old recorder, advanced the durable
    /// head past the value the new one had just read, and the successor's
    /// first append died on `SequenceConflict { expected: 4, found: 5 }`. That
    /// is aion#213, and it is not a race that can be narrowed: any window at
    /// all between the head read and the first append is enough.
    ///
    /// One batch under one recorder removes the window rather than shrinking
    /// it. There is no second recorder to seed, no head to re-read, and no
    /// interval in which the two halves of the transition can be observed
    /// apart — a crash either leaves the predecessor still running or leaves
    /// the successor durably started, never a terminal with no successor.
    ///
    /// The deadline events ride the SAME batch for the same reason: the
    /// predecessor's retirement (D5) and the successor's arming are part of
    /// the transition, and splitting them off would reintroduce exactly the
    /// crash windows `retire_run_deadline`'s repair chain exists to mop up.
    ///
    /// Returns the successor deadline's ARMING DESCRIPTOR — its timer id, its
    /// fire instant, and the history sequence its `TimerStarted` landed at,
    /// which is the identity the durable timer row is scheduled with — or
    /// `None` when the successor declares no timeout. Returned as one value
    /// rather than a bare sequence so a caller can never pair a recorded
    /// deadline with a missing sequence, or arm a deadline that was never
    /// recorded.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or
    /// the sequence tracker cannot advance after a successful append.
    pub async fn record_continue_as_new_boundary(
        &mut self,
        recorded_at: DateTime<Utc>,
        continued: ContinuedGeneration,
        opening: OpeningGeneration,
    ) -> Result<Option<(TimerId, DateTime<Utc>, u64)>, DurabilityError> {
        let ContinuedGeneration {
            run_id: parent_run_id,
            terminal,
            outstanding_deadline,
        } = continued;
        let OpeningGeneration { start, deadline } = opening;
        let WorkflowStartRecord {
            workflow_type: successor_type,
            input: successor_input,
            run_id,
            parent_run_id: successor_parent_run,
            parent_workflow_id,
            package_version,
        } = start;
        let successor_run_id = run_id.clone();

        // Envelopes are minted in batch order from the tracked head, each
        // from the last — never from a second read of the sequence — so the
        // whole boundary occupies one contiguous run of sequence numbers.
        let mut envelope = self.next_envelope(recorded_at)?;
        let mut batch = Vec::with_capacity(4);
        if let Some(ContinuationTerminal {
            input,
            workflow_type,
        }) = terminal
        {
            batch.push(Event::WorkflowContinuedAsNew {
                envelope: envelope.clone(),
                input,
                workflow_type,
                parent_run_id,
            });
            envelope = self.envelope_after(&envelope, recorded_at)?;
        }
        if let Some(deadline_id) = outstanding_deadline {
            batch.push(Event::TimerCancelled {
                envelope: envelope.clone(),
                timer_id: deadline_id,
                // The predecessor's run asked to continue; retiring its
                // deadline is that intent carried out, not engine teardown.
                cause: TimerCancelCause::WorkflowIntent,
            });
            envelope = self.envelope_after(&envelope, recorded_at)?;
        }
        batch.push(Event::WorkflowStarted {
            envelope: envelope.clone(),
            workflow_type: successor_type,
            input: successor_input,
            run_id,
            parent_run_id: successor_parent_run,
            parent_workflow_id,
            package_version,
        });
        let armed = match deadline {
            Some((deadline_id, fire_at)) => {
                let envelope = self.envelope_after(&envelope, recorded_at)?;
                let armed_seq = envelope.seq;
                batch.push(Event::TimerStarted {
                    envelope,
                    timer_id: deadline_id.clone(),
                    fire_at,
                });
                Some((deadline_id, fire_at, armed_seq))
            }
            None => None,
        };

        self.durable_append(&batch).await?;
        self.follow_generation(successor_run_id).await;
        Ok(armed)
    }

    /// Point this recorder at the generation the batch just opened.
    ///
    /// The predecessor's row is projected FIRST — from its own window it now
    /// carries the terminal that was just appended — and only then does the
    /// projection follow the successor, whose row is the one running row this
    /// history has. Projected once, under the predecessor's run id, the
    /// successor's `WorkflowStarted` read as the predecessor still running:
    /// one phantom running row per generation (#214).
    ///
    /// Moving `run_id` is not bookkeeping either. It is what the run-scoped
    /// append guard reads: from here on, an append offered on behalf of the
    /// predecessor is an append for a generation this recorder has left
    /// behind, and is refused rather than sequenced (aion#213 R2).
    ///
    /// Visibility failures are non-fatal on both projections — the durable
    /// batch has already committed, and reconciliation repairs a row — but the
    /// run move is not optional and happens regardless.
    async fn follow_generation(&mut self, successor_run_id: RunId) {
        self.upsert_visibility_projection_nonfatal().await;
        self.run_id = Some(successor_run_id.clone());
        self.retarget_visibility(successor_run_id);
        self.upsert_visibility_projection_nonfatal().await;
    }

    /// Records a workloop iteration boundary ATOMICALLY (R3.1): one batch
    /// appends `IterationClosed { routes, health_samples }`, the existing
    /// `WorkflowContinuedAsNew` terminal carrying the carry payload, and the
    /// successor generation's `WorkflowStarted` — so no crash window can leave
    /// a closed iteration without its terminal, or a continued loop without
    /// its successor generation (which would read as loop death to the
    /// cadence dead-man and as a pending continuation to the exit monitor).
    /// The successor is recorded WITHOUT any process being spawned; the
    /// cadence service wakes it when its window fires (R13.3).
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_workloop_iteration_boundary(
        &mut self,
        recorded_at: DateTime<Utc>,
        routes: Vec<String>,
        health_samples: Vec<aion_core::HealthSample>,
        carry: Payload,
        parent_run_id: RunId,
        successor: WorkflowStartRecord,
    ) -> Result<(), DurabilityError> {
        let closed_envelope = self.next_envelope(recorded_at)?;
        let continued_envelope = self.envelope_after(&closed_envelope, recorded_at)?;
        let started_envelope = self.envelope_after(&continued_envelope, recorded_at)?;
        let WorkflowStartRecord {
            workflow_type,
            input,
            run_id,
            parent_run_id: successor_parent_run,
            parent_workflow_id,
            package_version,
        } = successor;
        let recorded_run_id = run_id.clone();
        let batch = [
            Event::IterationClosed {
                envelope: closed_envelope,
                routes,
                health_samples,
            },
            Event::WorkflowContinuedAsNew {
                envelope: continued_envelope,
                input: carry,
                workflow_type: None,
                parent_run_id,
            },
            Event::WorkflowStarted {
                envelope: started_envelope,
                workflow_type,
                input,
                run_id,
                parent_run_id: successor_parent_run,
                parent_workflow_id,
                package_version,
            },
        ];
        self.durable_append(&batch).await?;
        self.follow_generation(recorded_run_id).await;
        Ok(())
    }

    /// Opens the loop's FINAL generation for a declared retire body (S3)
    /// ATOMICALLY: one batch appends the current generation's
    /// `WorkflowContinuedAsNew` terminal, carrying the loop's carry, and the
    /// retirement generation's `WorkflowStarted` under a fresh run id with
    /// that same carry as its input.
    ///
    /// # 🔴 WHY A RETIREMENT OPENS A GENERATION AT ALL
    ///
    /// The retire body is a fresh execution, and every ordinal-positional
    /// durable command it issues — `dispatch_activity`, `spawn_child`,
    /// `hatch_detached` — keys on a counter that starts at ZERO and is
    /// resolved against the CURRENT RUN SEGMENT. Run the body inside the
    /// generation the loop was already in and its first activity resolves to
    /// that generation's first recorded activity: the cleanup returns the
    /// iteration's stale result, executes nothing, and appends nothing. A
    /// declared cleanup that silently does not run is the exact failure the
    /// retire protocol exists to prevent, so the body is given a segment of
    /// its own, where the only correct answer to every command is "run it".
    ///
    /// The pair is atomic for the same reason the iteration boundary's is: a
    /// crash between the halves would leave a generation with no terminal and
    /// no successor — a loop that reads as neither continued nor dead.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the append or the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_workloop_retirement_generation(
        &mut self,
        recorded_at: DateTime<Utc>,
        carry: Payload,
        parent_run_id: RunId,
        retirement: WorkflowStartRecord,
    ) -> Result<(), DurabilityError> {
        let continued_envelope = self.next_envelope(recorded_at)?;
        let started_envelope = self.envelope_after(&continued_envelope, recorded_at)?;
        let WorkflowStartRecord {
            workflow_type,
            input,
            run_id,
            parent_run_id: retirement_parent_run,
            parent_workflow_id,
            package_version,
        } = retirement;
        let recorded_run_id = run_id.clone();
        let batch = [
            Event::WorkflowContinuedAsNew {
                envelope: continued_envelope,
                input: carry,
                workflow_type: None,
                parent_run_id,
            },
            Event::WorkflowStarted {
                envelope: started_envelope,
                workflow_type,
                input,
                run_id,
                parent_run_id: retirement_parent_run,
                parent_workflow_id,
                package_version,
            },
        ];
        self.durable_append(&batch).await?;
        self.follow_generation(recorded_run_id).await;
        Ok(())
    }
}