aion/engine/
api_activity_lease.rs

1//! The activity-lease record [`Engine`] exposes to the server's handoff seam
2//! (WA-010 R2).
3//!
4//! The server's worker-selection path holds no Recorder — invariant 3 gives
5//! each workflow exactly one, owned by the engine — so the lease fact reaches
6//! the run's Recorder the way a signal does ([`Engine::signal`]): a registered
7//! run (resident or suspended) through its registry handle, a paused-and-not-
8//! resident run or an idle workloop through a one-shot recorder resumed at the
9//! recorded head, and a run inside its registration birth window by waiting
10//! the handle out. No path here touches `EventStore::append` directly.
11
12use aion_core::{ActivityId, Event, RunId, WorkerAttribution, WorkflowId, WorkflowStatus};
13use chrono::Utc;
14
15use crate::EngineError;
16use crate::durability::Recorder;
17use crate::registry::WorkflowHandle;
18
19use super::api::{Engine, workflow_not_found};
20use super::delegated::run_has_terminal_history;
21
22impl Engine {
23    /// Record that `worker` accepted `attempt` of `activity_id` on `run`.
24    ///
25    /// The lease is informational: nothing is delivered to the workflow
26    /// process and nothing wakes, because a lease changes no wait the run is
27    /// parked on — it names who is doing the work. It is appended through the
28    /// run's single Recorder, serialising with every timer, signal and
29    /// completion arrival for that run, so concurrent arrivals never race the
30    /// sequence head.
31    ///
32    /// # Errors
33    ///
34    /// Returns [`EngineError::ActivityLeaseAfterTerminal`] when the run has
35    /// already reached a terminal event (nothing recorded),
36    /// [`EngineError::WorkflowNotFound`] when the `(workflow, run)` pair is
37    /// unknown or its handle never appears within the registration birth
38    /// window, and the store or durability error when the append fails.
39    pub async fn record_activity_lease(
40        &self,
41        id: &WorkflowId,
42        run: &RunId,
43        activity_id: ActivityId,
44        attempt: u32,
45        worker: WorkerAttribution,
46    ) -> Result<(), EngineError> {
47        if let Some(handle) = self.registry().get(id, run)? {
48            return record_through_handle(&handle, activity_id, attempt, worker).await;
49        }
50
51        let history = self.store().read_history(id).await?;
52        if run_has_terminal_history(&history, run) {
53            return Err(lease_after_terminal(id, run, activity_id, attempt));
54        }
55
56        // Paused-but-not-resident (crashed while paused, #204) and idle
57        // workloops (registered on the cadence service, no resident process)
58        // have no live handle and are deliberately excluded from respawn, so
59        // waiting the birth window out would only time out. Record through a
60        // one-shot recorder resumed at the recorded head, exactly as a signal
61        // to such a run is recorded.
62        let segment = aion_core::run_segment(&history, run);
63        let paused = matches!(
64            aion_core::status_from_events(segment),
65            WorkflowStatus::Paused
66        );
67        let idle_workloop = match &self.workloop {
68            Some(workloop) => workloop.store.get_workloop(id).await?.is_some(),
69            None => false,
70        };
71        if paused || idle_workloop {
72            let head = history.last().map(Event::seq).unwrap_or_default();
73            let mut recorder = Recorder::resume_at(id.clone(), self.store(), head)
74                .with_visibility(run.clone(), self.visibility_store());
75            recorder
76                .record_activity_leased(Utc::now(), activity_id, attempt, worker)
77                .await?;
78            return Ok(());
79        }
80
81        let handle = self
82            .handle_after_birth_window(id, run, &history)
83            .await?
84            .ok_or_else(|| workflow_not_found(id, run))?;
85        record_through_handle(&handle, activity_id, attempt, worker).await
86    }
87}
88
89/// Append the lease under the handle's recorder lock.
90///
91/// The terminal check and the append are atomic under that lock: the exit
92/// monitor records terminal events through the same recorder, so a lease
93/// racing a completion either lands before the terminal (and is a true fact
94/// about the attempt that then completed) or is refused after it — never
95/// behind it.
96async fn record_through_handle(
97    handle: &WorkflowHandle,
98    activity_id: ActivityId,
99    attempt: u32,
100    worker: WorkerAttribution,
101) -> Result<(), EngineError> {
102    let recorder = handle.recorder();
103    let mut recorder = recorder.lock().await;
104    let history = recorder.read_history().await?;
105    if run_has_terminal_history(&history, handle.run_id()) {
106        return Err(lease_after_terminal(
107            handle.workflow_id(),
108            handle.run_id(),
109            activity_id,
110            attempt,
111        ));
112    }
113    recorder
114        .record_activity_leased(Utc::now(), activity_id, attempt, worker)
115        .await?;
116    Ok(())
117}
118
119fn lease_after_terminal(
120    id: &WorkflowId,
121    run: &RunId,
122    activity_id: ActivityId,
123    attempt: u32,
124) -> EngineError {
125    EngineError::ActivityLeaseAfterTerminal {
126        workflow_id: id.clone(),
127        run_id: run.clone(),
128        activity_id,
129        attempt,
130    }
131}
aion/engine/api_activity_lease.rs

aion/engine/
api_activity_lease.rs
