aion-rs 0.10.0

//! Reserved workflow-deadline timer identity and the deadline-handler seam.
//!
//! A declared workflow timeout arms a single reserved timer named
//! `deadline:{run_id}`. The reserved prefix is engine-minted only — the author
//! NIF choke point (`decode_timer_id_arg`) refuses any author-supplied name that
//! starts with it — so a `deadline:` timer in history is always the engine's
//! per-run deadline and never a workflow-authored timer.
//!
//! When such a timer elapses, [`TimerService`](crate::time::TimerService)
//! demuxes it out of the generic `TimerFired` path and hands it to the
//! registered [`DeadlineHandler`], which records `WorkflowTimedOut` and tears
//! the run down engine-side. Routing is inversion of control precisely so the
//! timer machinery never needs a strong handle back to the engine (the
//! documented `RuntimeHandle`/`EngineNifState` cycle-avoidance): the engine
//! registers a handler holding whatever weak references it needs at construction
//! time.

use aion_core::{Event, RunId, TimerCancelCause, TimerId};
use chrono::Utc;
use uuid::Uuid;

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

/// Reserved name prefix for a workflow's declared-timeout deadline timer.
///
/// Engine-minted only: `decode_timer_id_arg` refuses author-supplied names under
/// this prefix, so a `deadline:` timer is always the per-run workflow deadline.
pub const DEADLINE_TIMER_PREFIX: &str = "deadline:";

/// Closed-set descriptor token recorded on the `WorkflowTimedOut` terminal for a
/// declared workflow timeout, consistent with the `deadline:` timer id family.
///
/// User-visible in `one_motion` output and replay terminals; the sole value a
/// declared-workflow-timeout deadline records.
pub const WORKFLOW_TIMEOUT_DESCRIPTOR: &str = "workflow";

/// The reserved deadline timer id for `run_id`: `deadline:{run_id}`.
///
/// # Errors
///
/// Returns [`aion_core::IdError`] only if timer-name construction rejects the
/// composed name; the name is always non-empty, so this never fails in practice
/// and the `Result` exists solely to keep the helper total without an `unwrap`.
pub fn deadline_timer_id(run_id: &RunId) -> Result<TimerId, aion_core::IdError> {
    TimerId::named(format!("{DEADLINE_TIMER_PREFIX}{run_id}"))
}

/// Whether `timer_id` is a reserved workflow-deadline timer.
#[must_use]
pub fn is_deadline_timer(timer_id: &TimerId) -> bool {
    timer_id
        .name()
        .is_some_and(|name| name.starts_with(DEADLINE_TIMER_PREFIX))
}

/// The run id encoded in a `deadline:{run_id}` timer, if `timer_id` is a
/// well-formed reserved deadline timer.
///
/// Returns `None` for a non-deadline timer or a deadline-prefixed name whose
/// suffix is not a valid run identifier.
#[must_use]
pub fn deadline_run_id(timer_id: &TimerId) -> Option<RunId> {
    let suffix = timer_id.name()?.strip_prefix(DEADLINE_TIMER_PREFIX)?;
    Uuid::parse_str(suffix).ok().map(RunId::new)
}

/// The still-live reserved deadline timer for `run_id`, derived purely from
/// recorded history — or `None` when the run never armed a deadline (or its
/// deadline was already fired/cancelled).
///
/// LAW 1: this constructs NO deadline id. It reads the id only from a
/// `TimerStarted` event whose id is a reserved deadline timer for exactly
/// `run_id` (matched by the run encoded in the id, not by a minted candidate),
/// so a timeout-less run — which recorded no such event — yields `None` and
/// touches no deadline object of any kind. Liveness is last-event-wins: a later
/// `TimerFired`/`TimerCancelled` for the same id retires it.
#[must_use]
pub fn outstanding_deadline_timer(history: &[Event], run_id: &RunId) -> Option<TimerId> {
    let mut live: Option<TimerId> = None;
    for event in history {
        let (timer_id, started) = match event {
            Event::TimerStarted { timer_id, .. } => (timer_id, true),
            Event::TimerFired { timer_id, .. } | Event::TimerCancelled { timer_id, .. } => {
                (timer_id, false)
            }
            _ => continue,
        };
        if !is_deadline_timer(timer_id) || deadline_run_id(timer_id).as_ref() != Some(run_id) {
            continue;
        }
        live = if started {
            Some(timer_id.clone())
        } else {
            None
        };
    }
    live
}

/// Cancels a run's still-live declared-timeout deadline as part of a terminal
/// transition, under the caller's already-held recorder lock.
///
/// This is the ONE deadline-retirement primitive EVERY terminal writer shares —
/// `complete`, `fail`, `cancel`, and continue-as-new on both the API and NIF
/// paths — so D5 ("a recorded terminal permanently retires the run's deadline")
/// is enforced uniformly instead of re-implemented, or forgotten, per writer.
///
/// It derives the deadline id purely from `history` (never minted): a
/// timeout-less run recorded no deadline `TimerStarted`, so it retires nothing
/// and touches no deadline object (LAW 1). It is IDEMPOTENT — a run whose
/// deadline was already cancelled (or never armed) has no outstanding deadline,
/// so a re-entry records nothing. That idempotence is what makes a terminal
/// transition RESUMABLE: because the terminal append and this cancellation are
/// two separate durable writes, a crash between them leaves the deadline
/// outstanding, and the next terminal writer — or the process-exit monitor
/// re-encountering the run's own terminal — completes the cancellation. Without
/// it, whole-history `outstanding_future_timers` recovery would keep re-arming a
/// predecessor deadline after failover.
///
/// `history` MUST be the history read under the same recorder lock. On the happy
/// path it is read BEFORE the terminal append so the deadline still reads live
/// and is retired; on a resume it is read after the terminal, where an
/// already-cancelled deadline is a clean no-op and an uncancelled one is
/// completed.
///
/// # Errors
///
/// Returns [`DurabilityError`] when the `TimerCancelled` append fails; the caller
/// propagates it so the interrupted transition is retried rather than silently
/// leaving the deadline live.
pub async fn retire_run_deadline(
    recorder: &mut Recorder,
    history: &[Event],
    run_id: &RunId,
) -> Result<(), DurabilityError> {
    if let Some(deadline_id) = outstanding_deadline_timer(history, run_id) {
        recorder
            .record_timer_cancelled(Utc::now(), deadline_id, TimerCancelCause::WorkflowIntent)
            .await?;
    }
    Ok(())
}

/// Errors surfaced by a [`DeadlineHandler`] to the timer service.
///
/// Deliberately message-only: the handler lives engine-side and produces the
/// engine's own typed errors, which the timer service (in a lower crate layer)
/// only needs to observe and propagate as a fire failure.
#[derive(thiserror::Error, Debug)]
#[error("deadline handler failed: {0}")]
pub struct DeadlineHandlerError(pub String);

/// Engine-side handler invoked when a workflow's declared-timeout deadline
/// elapses.
///
/// The timer service demuxes a reserved `deadline:{run_id}` fire to this seam
/// instead of recording a generic `TimerFired`. The implementation records
/// `WorkflowTimedOut` for the run under the per-handle recorder lock (losing
/// cleanly to any concurrent terminal) and tears the run down.
#[async_trait::async_trait]
pub trait DeadlineHandler: Send + Sync {
    /// Drive `run_id` of `workflow_id` to a `WorkflowTimedOut` terminal.
    ///
    /// # Errors
    ///
    /// Returns [`DeadlineHandlerError`] when the terminal cannot be recorded or
    /// the run cannot be torn down; the timer service surfaces it as a fire
    /// failure.
    async fn on_deadline_elapsed(
        &self,
        workflow_id: aion_core::WorkflowId,
        run_id: RunId,
    ) -> Result<(), DeadlineHandlerError>;
}

#[cfg(test)]
mod tests {
    use aion_core::{Event, EventEnvelope, RunId, TimerCancelCause, TimerId, WorkflowId};
    use uuid::Uuid;

    use super::{
        deadline_run_id, deadline_timer_id, is_deadline_timer, outstanding_deadline_timer,
    };

    type TestResult = Result<(), Box<dyn std::error::Error>>;

    fn envelope(seq: u64, workflow_id: &WorkflowId) -> EventEnvelope {
        EventEnvelope {
            seq,
            recorded_at: chrono::Utc::now(),
            workflow_id: workflow_id.clone(),
        }
    }

    #[test]
    fn outstanding_deadline_timer_is_none_for_a_run_without_a_deadline() -> TestResult {
        // LAW 1: a run whose history has no deadline `TimerStarted` of its own
        // yields no deadline object of any kind — not even a constructed candidate
        // id. Neither an ordinary author timer nor ANOTHER run's deadline matches.
        let workflow_id = WorkflowId::new_v4();
        let run_id = RunId::new_v4();
        let other_run = RunId::new_v4();
        let history = vec![
            Event::TimerStarted {
                envelope: envelope(1, &workflow_id),
                timer_id: deadline_timer_id(&other_run)?,
                fire_at: chrono::Utc::now(),
            },
            Event::TimerStarted {
                envelope: envelope(2, &workflow_id),
                timer_id: TimerId::anonymous(3),
                fire_at: chrono::Utc::now(),
            },
        ];
        assert_eq!(outstanding_deadline_timer(&history, &run_id), None);
        Ok(())
    }

    #[test]
    fn outstanding_deadline_timer_tracks_liveness_and_scopes_to_the_run() -> TestResult {
        let workflow_id = WorkflowId::new_v4();
        let this_run = RunId::new_v4();
        let other_run = RunId::new_v4();
        let this_deadline = deadline_timer_id(&this_run)?;
        let other_deadline = deadline_timer_id(&other_run)?;

        // Armed for this run, plus another run's deadline that must be ignored.
        let armed = vec![
            Event::TimerStarted {
                envelope: envelope(1, &workflow_id),
                timer_id: this_deadline.clone(),
                fire_at: chrono::Utc::now(),
            },
            Event::TimerStarted {
                envelope: envelope(2, &workflow_id),
                timer_id: other_deadline,
                fire_at: chrono::Utc::now(),
            },
        ];
        assert_eq!(
            outstanding_deadline_timer(&armed, &this_run),
            Some(this_deadline.clone())
        );

        // A later cancel retires it (last-event-wins).
        let mut cancelled = armed;
        cancelled.push(Event::TimerCancelled {
            envelope: envelope(3, &workflow_id),
            timer_id: this_deadline,
            cause: TimerCancelCause::WorkflowIntent,
        });
        assert_eq!(outstanding_deadline_timer(&cancelled, &this_run), None);
        Ok(())
    }

    #[test]
    fn deadline_timer_id_round_trips_to_its_run() -> TestResult {
        let run_id = RunId::new(Uuid::from_u128(42));
        let timer_id = deadline_timer_id(&run_id)?;
        assert!(is_deadline_timer(&timer_id));
        assert_eq!(deadline_run_id(&timer_id), Some(run_id));
        Ok(())
    }

    #[test]
    fn author_named_timer_is_not_a_deadline() -> TestResult {
        let timer_id = TimerId::named("review-deadline")?;
        assert!(!is_deadline_timer(&timer_id));
        assert_eq!(deadline_run_id(&timer_id), None);
        Ok(())
    }

    #[test]
    fn anonymous_timer_is_not_a_deadline() {
        let timer_id = TimerId::anonymous(7);
        assert!(!is_deadline_timer(&timer_id));
        assert_eq!(deadline_run_id(&timer_id), None);
    }

    #[test]
    fn deadline_prefixed_but_malformed_suffix_has_no_run() -> TestResult {
        // A `deadline:`-prefixed name whose suffix is not a UUID is detected as
        // a deadline timer (prefix match) but yields no run id — the timer
        // service turns this into a typed refusal, never a silent generic fire.
        let timer_id = TimerId::named("deadline:not-a-uuid")?;
        assert!(is_deadline_timer(&timer_id));
        assert_eq!(deadline_run_id(&timer_id), None);
        Ok(())
    }
}