aion-rs 0.31.0

//! The activity-lease record [`Engine`] exposes to the server's handoff seam
//! (WA-010 R2).
//!
//! The server's worker-selection path holds no Recorder — invariant 3 gives
//! each workflow exactly one, owned by the engine — so the lease fact reaches
//! the run's Recorder the way a signal does ([`Engine::signal`]): a registered
//! run (resident or suspended) through its registry handle, a paused-and-not-
//! resident run or an idle workloop through a one-shot recorder resumed at the
//! recorded head, and a run inside its registration birth window by waiting
//! the handle out. No path here touches `EventStore::append` directly.

use aion_core::{ActivityId, Event, RunId, WorkerAttribution, WorkflowId, WorkflowStatus};
use chrono::Utc;

use crate::EngineError;
use crate::durability::Recorder;
use crate::registry::WorkflowHandle;

use super::api::{Engine, workflow_not_found};
use super::delegated::run_has_terminal_history;

impl Engine {
    /// Record that `worker` accepted `attempt` of `activity_id` on `run`.
    ///
    /// The lease is informational: nothing is delivered to the workflow
    /// process and nothing wakes, because a lease changes no wait the run is
    /// parked on — it names who is doing the work. It is appended through the
    /// run's single Recorder, serialising with every timer, signal and
    /// completion arrival for that run, so concurrent arrivals never race the
    /// sequence head.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::ActivityLeaseAfterTerminal`] when the run has
    /// already reached a terminal event (nothing recorded),
    /// [`EngineError::WorkflowNotFound`] when the `(workflow, run)` pair is
    /// unknown or its handle never appears within the registration birth
    /// window, and the store or durability error when the append fails.
    pub async fn record_activity_lease(
        &self,
        id: &WorkflowId,
        run: &RunId,
        activity_id: ActivityId,
        attempt: u32,
        worker: WorkerAttribution,
    ) -> Result<(), EngineError> {
        if let Some(handle) = self.registry().get(id, run)? {
            return record_through_handle(&handle, activity_id, attempt, worker).await;
        }

        let history = self.store().read_history(id).await?;
        if run_has_terminal_history(&history, run) {
            return Err(lease_after_terminal(id, run, activity_id, attempt));
        }

        // Paused-but-not-resident (crashed while paused, #204) and idle
        // workloops (registered on the cadence service, no resident process)
        // have no live handle and are deliberately excluded from respawn, so
        // waiting the birth window out would only time out. Record through a
        // one-shot recorder resumed at the recorded head, exactly as a signal
        // to such a run is recorded.
        let segment = aion_core::run_segment(&history, run);
        let paused = matches!(
            aion_core::status_from_events(segment),
            WorkflowStatus::Paused
        );
        let idle_workloop = match &self.workloop {
            Some(workloop) => workloop.store.get_workloop(id).await?.is_some(),
            None => false,
        };
        if paused || idle_workloop {
            let head = history.last().map(Event::seq).unwrap_or_default();
            let mut recorder = Recorder::resume_at(id.clone(), self.store(), head)
                .with_visibility(run.clone(), self.visibility_store());
            recorder
                .record_activity_leased(Utc::now(), activity_id, attempt, worker)
                .await?;
            return Ok(());
        }

        let handle = self
            .handle_after_birth_window(id, run, &history)
            .await?
            .ok_or_else(|| workflow_not_found(id, run))?;
        record_through_handle(&handle, activity_id, attempt, worker).await
    }
}

/// Append the lease under the handle's recorder lock.
///
/// The terminal check and the append are atomic under that lock: the exit
/// monitor records terminal events through the same recorder, so a lease
/// racing a completion either lands before the terminal (and is a true fact
/// about the attempt that then completed) or is refused after it — never
/// behind it.
async fn record_through_handle(
    handle: &WorkflowHandle,
    activity_id: ActivityId,
    attempt: u32,
    worker: WorkerAttribution,
) -> Result<(), EngineError> {
    let recorder = handle.recorder();
    let mut recorder = recorder.lock().await;
    let history = recorder.read_history().await?;
    if run_has_terminal_history(&history, handle.run_id()) {
        return Err(lease_after_terminal(
            handle.workflow_id(),
            handle.run_id(),
            activity_id,
            attempt,
        ));
    }
    recorder
        .record_activity_leased(Utc::now(), activity_id, attempt, worker)
        .await?;
    Ok(())
}

fn lease_after_terminal(
    id: &WorkflowId,
    run: &RunId,
    activity_id: ActivityId,
    attempt: u32,
) -> EngineError {
    EngineError::ActivityLeaseAfterTerminal {
        workflow_id: id.clone(),
        run_id: run.clone(),
        activity_id,
        attempt,
    }
}