aion-rs 0.31.0

//! The run-scoped append guard: which generation of a workflow history a
//! recorder will still accept an append for (aion#213 R2).

use aion_core::RunId;

use super::Recorder;
use crate::durability::DurabilityError;

/// Whether the recorder will sequence an append offered on behalf of a
/// particular run.
///
/// A refusal is not a failure and is not an error: it is the correct outcome
/// for a late append from a generation the history has already moved past, and
/// the caller's own contract decides what to do with it (a workflow process
/// being torn down after continue-as-new simply stops).
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RunAdmission {
    /// The run is the generation this history is currently on; the append may
    /// be sequenced.
    Open,
    /// The run's generation is closed — a later generation has been started in
    /// this history — so NOTHING is sequenced for it.
    RefusedTerminal,
}

impl Recorder {
    /// Whether an append offered on behalf of `run_id` may be sequenced.
    ///
    /// # 🔴 A CLOSED GENERATION STILL HAS A LIVE PROCESS, AND THAT PROCESS STILL CALLS NIFS
    ///
    /// Continue-as-new does not stop the predecessor's BEAM process at the
    /// instant its terminal is recorded — the API path never touches the pid,
    /// and the NIF path's `cancel_pid` lands after the append. So a run that
    /// is durably over can still reach this recorder with a `TimerStarted` for
    /// a `sleep` it had already begun arming. Before aion#213 that append went
    /// through a SECOND recorder and broke the successor's sequence; with one
    /// recorder across the boundary it would instead land cleanly — appending
    /// a `TimerStarted` for a run AFTER its own terminal, into the successor's
    /// segment, where replay would read it as the successor's. Removing the
    /// sequence conflict without this guard would trade a loud failure for a
    /// silent corruption.
    ///
    /// The answer is the recorder's own tracked generation, which
    /// [`Recorder::follow_generation`] moves in the same batch that opens the
    /// successor — so the guard flips at exactly the instant the boundary
    /// commits, under the same lock, with no window on either side. When this
    /// recorder has no tracked generation at all (a schedule-coordinator
    /// recorder, or one built before any run was recorded through it) the
    /// question is settled against history instead, through
    /// [`crate::lifecycle::visibility::run_window`]: a run whose window stops
    /// short of the end of history has been superseded, and a run with no
    /// `WorkflowStarted` at all has no generation here to append into.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError::Store`] when the fallback history read
    /// fails. The read is NOT taken on the ordinary path: an append for the
    /// generation the recorder is on is admitted in O(1), so a workloop's
    /// unbounded history is never re-read per append.
    pub async fn admit_run_append(&self, run_id: &RunId) -> Result<RunAdmission, DurabilityError> {
        if let Some(current) = &self.run_id {
            return Ok(if current == run_id {
                RunAdmission::Open
            } else {
                RunAdmission::RefusedTerminal
            });
        }
        let history = self.store.read_history(&self.workflow_id).await?;
        Ok(run_generation_admission(&history, run_id))
    }
}

/// Whether `run_id`'s generation is still the open one in `history`.
///
/// The window [`crate::lifecycle::visibility::run_window`] returns ends where
/// the NEXT generation's `WorkflowStarted` begins, so a window that reaches the
/// end of history is the live generation and a shorter one has been superseded.
fn run_generation_admission(history: &[aion_core::Event], run_id: &RunId) -> RunAdmission {
    match crate::lifecycle::visibility::run_window(history, run_id) {
        Some(window) if window.len() == history.len() => RunAdmission::Open,
        // A superseded generation, or a run this history never started at all
        // — neither may append into the generation that is live now.
        _ => RunAdmission::RefusedTerminal,
    }
}

#[cfg(test)]
#[path = "run_guard_tests.rs"]
mod tests;