aion_core/current_step.rs
1//! Which step a run is on, folded from its own history.
2//!
3//! There is no stored "current step" field anywhere and there deliberately is
4//! not one: exactly like [`crate::WorkflowStatus`], the answer is a PROJECTION
5//! of the authoritative event history, computed on read. A maintained
6//! projection would be a second copy of a fact history already holds, and the
7//! two would drift the first time an append path forgot to update it.
8//!
9//! # What "open" means here, precisely
10//!
11//! An activity ordinal is OPEN when the run's active segment records it as
12//! scheduled (and possibly dispatched) with no terminal event for it. The last
13//! event naming the ordinal decides:
14//!
15//! - [`Event::ActivityScheduled`] (re)opens it at the address the engine
16//! stamped — the task queue and node the dispatch really went to.
17//! - [`Event::ActivityStarted`] records the delivery attempt and the instant
18//! the ENGINE dispatched. It is NOT proof a worker took the work: the event
19//! is written at dispatch, atomically with its `ActivityScheduled`, before
20//! any worker is selected. That is why [`StepState::Dispatched`] is named for
21//! what happened rather than for what a reader might hope happened, and why
22//! liveness is answered by joining the live fleet, never by this fold.
23//! - `ActivityCompleted` / `ActivityFailed` / `ActivityCancelled` retire it.
24//! - [`Event::ActivityAdvisoryExhausted`] is deliberately NOT a terminal: it
25//! ACCOMPANIES an `ActivityFailed` and never replaces it, so treating it as
26//! one would retire an ordinal that its own failure already retired.
27//! - [`Event::WorkflowReopened`] SUPERSEDES the recorded terminal of every
28//! activity it names, which is the whole point of the event — those
29//! activities resolve to live re-dispatch on replay. The fold re-opens them
30//! as [`StepState::Reopened`]: they are open again, and nothing has been
31//! dispatched for them in this lease yet, so calling them `Dispatched` would
32//! claim a delivery that has not happened.
33//!
34//! # Fan-out means "the current step" is not always one step
35//!
36//! A fan-out has several ordinals open at once. [`current_step`] answers with
37//! the most recently advanced one because a reader asking "what is it doing"
38//! needs a single answer; [`open_steps`] returns all of them so a reader that
39//! must not lose the siblings never has to guess that they existed.
40
41use chrono::{DateTime, Utc};
42
43use crate::{ActivityId, Event, describe_live::OpenStep, describe_live::StepState};
44
45/// Working state for one activity ordinal while the segment is scanned.
46#[derive(Clone, Debug)]
47struct Open {
48 step: OpenStep,
49 /// Position of the last event that advanced this ordinal, so the most
50 /// recently advanced open ordinal can be identified without a second scan.
51 last_advanced_at: usize,
52}
53
54/// Every step the run's active segment records as open, in the order the
55/// ordinals were first opened.
56///
57/// The active segment is everything from the latest [`Event::WorkflowStarted`]
58/// — a continue-as-new starts a fresh segment, and the prior segment's
59/// activities belong to a run that no longer executes.
60///
61/// An `ActivityStarted` with no `ActivityScheduled` in the segment opens
62/// nothing: its address was never recorded in this segment, so there is no
63/// address to report and inventing one would be a guess.
64#[must_use]
65pub fn open_steps(history: &[Event]) -> Vec<OpenStep> {
66 fold(history).into_iter().map(|open| open.step).collect()
67}
68
69/// The single step a reader is told the run is on: the open step whose last
70/// history event is the most recent.
71///
72/// `None` for a run with nothing open — a completed run, a failed run, a run
73/// blocked on a timer or a signal. That absence is an answer, not a gap: the
74/// run is genuinely not inside an activity.
75#[must_use]
76pub fn current_step(history: &[Event]) -> Option<OpenStep> {
77 fold(history)
78 .into_iter()
79 .max_by_key(|open| open.last_advanced_at)
80 .map(|open| open.step)
81}
82
83/// The shared scan behind both readers.
84fn fold(history: &[Event]) -> Vec<Open> {
85 let segment_start = history
86 .iter()
87 .rposition(|event| matches!(event, Event::WorkflowStarted { .. }))
88 .unwrap_or(0);
89 let mut open: Vec<Open> = Vec::new();
90 for (position, event) in history[segment_start..].iter().enumerate() {
91 match event {
92 Event::ActivityScheduled {
93 envelope,
94 activity_id,
95 activity_type,
96 task_queue,
97 node,
98 ..
99 } => {
100 let entry = Open {
101 step: OpenStep {
102 activity_id: activity_id.clone(),
103 activity_type: activity_type.clone(),
104 task_queue: task_queue.clone(),
105 node: node.clone(),
106 scheduled_at: envelope.recorded_at,
107 state: StepState::Scheduled,
108 },
109 last_advanced_at: position,
110 };
111 replace_or_push(&mut open, entry);
112 }
113 Event::ActivityStarted {
114 envelope,
115 activity_id,
116 attempt,
117 } => {
118 if let Some(held) = find_mut(&mut open, activity_id) {
119 held.step.state = StepState::Dispatched {
120 attempt: *attempt,
121 dispatched_at: envelope.recorded_at,
122 };
123 held.last_advanced_at = position;
124 }
125 }
126 Event::ActivityCompleted { activity_id, .. }
127 | Event::ActivityFailed { activity_id, .. }
128 | Event::ActivityCancelled { activity_id, .. } => {
129 open.retain(|held| &held.step.activity_id != activity_id);
130 }
131 Event::WorkflowReopened {
132 envelope, reopened, ..
133 } => reopen(
134 &mut open,
135 history,
136 segment_start,
137 reopened,
138 envelope.recorded_at,
139 position,
140 ),
141 _ => {}
142 }
143 }
144 open
145}
146
147/// Re-open every activity a [`Event::WorkflowReopened`] names.
148///
149/// The reopened ordinal's terminal event has already retired it from `open`, so
150/// its recorded address has to be recovered from the segment's last
151/// `ActivityScheduled` for that ordinal. An ordinal with no scheduling in this
152/// segment is skipped for the same reason a bare `ActivityStarted` is: nothing
153/// recorded its address here, and inventing one would be a guess.
154fn reopen(
155 open: &mut Vec<Open>,
156 history: &[Event],
157 segment_start: usize,
158 reopened: &[ActivityId],
159 recorded_at: DateTime<Utc>,
160 position: usize,
161) {
162 for activity_id in reopened {
163 let Some(scheduled) = last_scheduling(&history[segment_start..], activity_id) else {
164 continue;
165 };
166 let entry = Open {
167 step: OpenStep {
168 state: StepState::Reopened {
169 reopened_at: recorded_at,
170 },
171 ..scheduled
172 },
173 last_advanced_at: position,
174 };
175 replace_or_push(open, entry);
176 }
177}
178
179/// The address the segment's most recent `ActivityScheduled` for `activity_id`
180/// stamped, projected as a freshly opened step.
181fn last_scheduling(segment: &[Event], activity_id: &ActivityId) -> Option<OpenStep> {
182 segment.iter().rev().find_map(|event| match event {
183 Event::ActivityScheduled {
184 envelope,
185 activity_id: scheduled_id,
186 activity_type,
187 task_queue,
188 node,
189 ..
190 } if scheduled_id == activity_id => Some(OpenStep {
191 activity_id: scheduled_id.clone(),
192 activity_type: activity_type.clone(),
193 task_queue: task_queue.clone(),
194 node: node.clone(),
195 scheduled_at: envelope.recorded_at,
196 state: StepState::Scheduled,
197 }),
198 _ => None,
199 })
200}
201
202/// Replace the held entry for `entry`'s ordinal, or append it when the ordinal
203/// is not held. Ordinals stay in first-opened order either way.
204fn replace_or_push(open: &mut Vec<Open>, entry: Open) {
205 match find_mut(open, &entry.step.activity_id) {
206 Some(held) => *held = entry,
207 None => open.push(entry),
208 }
209}
210
211/// The held entry for `activity_id`, if the ordinal is open.
212fn find_mut<'a>(open: &'a mut [Open], activity_id: &ActivityId) -> Option<&'a mut Open> {
213 open.iter_mut()
214 .find(|held| &held.step.activity_id == activity_id)
215}
216
217#[cfg(test)]
218#[path = "current_step_tests.rs"]
219mod tests;