aion_server/worker/queue_service/reachability.rs
1//! Whether an in-flight activity can still reach a worker.
2//!
3//! `ActivityStarted` is recorded by the engine at DISPATCH time, atomically with
4//! its `ActivityScheduled` and before any worker has leased the work (see
5//! `aion::durability::recorder::fan_out` and the single-dispatch seam in
6//! `aion::runtime::nif_activity_dispatch`). So a started, unterminated activity
7//! means "the engine handed this to the fleet", not "someone is working on it",
8//! and a dispatch to a task queue nobody serves is byte-for-byte
9//! indistinguishable in history from one a worker is executing right now. The
10//! projected status is `Running` for both, correctly — status is a projection of
11//! history, and history genuinely contains no terminal event.
12//!
13//! What history alone cannot say, the live fleet can. This module joins the two:
14//! the run's own recorded dispatch addresses against the SAME poller census and
15//! the SAME [`classify`] verdict the dispatcher's selection wait uses
16//! ([`super::wait`]). There is no second notion of "is anyone serving this" here
17//! — a disagreement between what an operator reads and what a dispatch does
18//! would be worse than the silence it replaces.
19//!
20//! The healthy case is silent by construction: [`classify`] returns `None` for
21//! an address with a live compatible worker, so a run whose activity is genuinely
22//! being executed produces no entry at all.
23
24use aion_core::{ActivityId, Event, UnservedActivity, WorkflowId};
25use chrono::{DateTime, Utc};
26
27use super::census::classify;
28use super::declarations::QueueDeclarationSource;
29use super::state::QueueServiceState;
30use super::taxonomy::ServiceAddress;
31use crate::error::ServerError;
32use crate::worker::registry::ConnectedWorkerRegistry;
33
34/// One activity the active run segment records as dispatched and unterminated.
35///
36/// The address fields are the ones the engine STAMPED on the dispatch
37/// (`ActivityScheduled` carries the resolved `task_queue` and `node`), so this is
38/// the address the dispatch really went to, not a re-resolution that could
39/// disagree with it.
40#[derive(Clone, Debug, Eq, PartialEq)]
41pub struct OpenActivity {
42 /// Activity ordinal recorded in history.
43 pub activity_id: ActivityId,
44 /// Activity type the dispatch needs served.
45 pub activity_type: String,
46 /// Task queue stamped on the recorded `ActivityScheduled`.
47 pub task_queue: String,
48 /// Node affinity stamped on the recorded `ActivityScheduled`, if any.
49 pub node: Option<String>,
50 /// One-based delivery attempt stamped on the recorded `ActivityStarted`.
51 pub attempt: u32,
52 /// `recorded_at` of the `ActivityStarted` — when the ENGINE dispatched.
53 pub dispatched_at: DateTime<Utc>,
54}
55
56/// Working state for one activity ordinal while the segment is scanned.
57#[derive(Clone, Debug)]
58struct Open {
59 activity_id: ActivityId,
60 activity_type: String,
61 task_queue: String,
62 node: Option<String>,
63 attempt: u32,
64 dispatched_at: DateTime<Utc>,
65 started: bool,
66}
67
68/// The activities the run's active segment records as dispatched and not yet
69/// terminated, in dispatch order.
70///
71/// Scans forward from the latest `WorkflowStarted` (the active run segment) and
72/// lets the *last* event for each activity ordinal decide: an
73/// `ActivityScheduled` (re)opens the ordinal at its stamped address, an
74/// `ActivityStarted` records the delivery attempt and the dispatch instant, and
75/// `ActivityCompleted` / `ActivityFailed` / `ActivityCancelled` retire it. An
76/// ordinal re-dispatched after a crash — the shape a restart's replay produces,
77/// and the shape the 24-day exhibit has four of — is correctly reported once, at
78/// its most recent dispatch, rather than once per re-arm.
79///
80/// `ActivityAdvisoryExhausted` is deliberately not a terminal: it ACCOMPANIES an
81/// `ActivityFailed` and never replaces it, so retiring on it would retire the
82/// ordinal twice.
83///
84/// An `ActivityStarted` with no `ActivityScheduled` in the segment is not
85/// reported. Its address was never recorded in this segment, so there is nothing
86/// to classify, and inventing one would be the guess this module exists to avoid.
87#[must_use]
88pub fn open_activities_in_active_segment(history: &[Event]) -> Vec<OpenActivity> {
89 let segment_start = history
90 .iter()
91 .rposition(|event| matches!(event, Event::WorkflowStarted { .. }))
92 .unwrap_or(0);
93 let mut open: Vec<Open> = Vec::new();
94 for event in &history[segment_start..] {
95 match event {
96 Event::ActivityScheduled {
97 envelope,
98 activity_id,
99 activity_type,
100 task_queue,
101 node,
102 ..
103 } => {
104 let entry = Open {
105 activity_id: activity_id.clone(),
106 activity_type: activity_type.clone(),
107 task_queue: task_queue.clone(),
108 node: node.clone(),
109 attempt: 0,
110 dispatched_at: envelope.recorded_at,
111 started: false,
112 };
113 match open
114 .iter_mut()
115 .find(|held| &held.activity_id == activity_id)
116 {
117 Some(held) => *held = entry,
118 None => open.push(entry),
119 }
120 }
121 Event::ActivityStarted {
122 envelope,
123 activity_id,
124 attempt,
125 } => {
126 if let Some(held) = open
127 .iter_mut()
128 .find(|held| &held.activity_id == activity_id)
129 {
130 held.attempt = *attempt;
131 held.dispatched_at = envelope.recorded_at;
132 held.started = true;
133 }
134 }
135 Event::ActivityCompleted { activity_id, .. }
136 | Event::ActivityFailed { activity_id, .. }
137 | Event::ActivityCancelled { activity_id, .. } => {
138 open.retain(|held| &held.activity_id != activity_id);
139 }
140 _ => {}
141 }
142 }
143 open.into_iter()
144 .filter(|held| held.started)
145 .map(|held| OpenActivity {
146 activity_id: held.activity_id,
147 activity_type: held.activity_type,
148 task_queue: held.task_queue,
149 node: held.node,
150 attempt: held.attempt,
151 dispatched_at: held.dispatched_at,
152 })
153 .collect()
154}
155
156/// The live fleet seams an in-flight activity's reachability is judged against.
157///
158/// Exactly the three the dispatcher's selection wait holds: the connected-worker
159/// registry (the census), the deployed queue declarations (the structural
160/// verdict), and the parked-dispatch state (whether this process is waiting).
161pub struct ActivityReachability<'a> {
162 /// Live connected-worker registry.
163 pub registry: &'a ConnectedWorkerRegistry,
164 /// Deployed queue declarations.
165 pub declarations: &'a QueueDeclarationSource,
166 /// Live parked-dispatch state.
167 pub state: &'a QueueServiceState,
168}
169
170impl ActivityReachability<'_> {
171 /// Every in-flight activity in `history` the fleet cannot currently serve.
172 ///
173 /// Returns an EMPTY vector for a run whose in-flight activities all have a
174 /// live compatible worker, and for a run with no in-flight activity at all.
175 ///
176 /// # Errors
177 ///
178 /// Returns [`ServerError`] when the connected-worker registry or the parked
179 /// dispatch state cannot be read. Never guesses: an unreadable fleet is
180 /// reported, not rendered as "everything is fine".
181 pub fn unserved(
182 &self,
183 namespace: &str,
184 workflow_id: &WorkflowId,
185 history: &[Event],
186 ) -> Result<Vec<UnservedActivity>, ServerError> {
187 let mut unserved = Vec::new();
188 for open in open_activities_in_active_segment(history) {
189 let address = ServiceAddress {
190 namespace: namespace.to_owned(),
191 task_queue: open.task_queue,
192 activity_type: open.activity_type,
193 node: open.node,
194 };
195 let census = self.registry.pool_census(
196 &address.namespace,
197 &address.task_queue,
198 &address.activity_type,
199 address.node.as_deref(),
200 )?;
201 let declaration = self.declarations.declaration_for(&address.task_queue);
202 // `None` is the healthy answer: a live compatible worker can take
203 // this dispatch, so there is nothing to report and nothing is
204 // reported.
205 let Some(reason) = classify(declaration, &census) else {
206 continue;
207 };
208 let dispatch_parked = self.state.is_unserved(workflow_id, &open.activity_id)?;
209 unserved.push(UnservedActivity {
210 activity_id: open.activity_id,
211 reason: reason.as_str().to_owned(),
212 detail: reason.explain(&address),
213 activity_type: address.activity_type,
214 task_queue: address.task_queue,
215 node: address.node,
216 attempt: open.attempt,
217 dispatched_at: open.dispatched_at,
218 workers_in_pool: count(census.workers_in_pool),
219 workers_serving_activity: count(census.workers_serving_activity),
220 compatible_workers: count(census.compatible_workers),
221 dispatch_parked,
222 });
223 }
224 Ok(unserved)
225 }
226}
227
228/// Widen a census count to the wire width, saturating rather than wrapping.
229fn count(value: usize) -> u64 {
230 u64::try_from(value).unwrap_or(u64::MAX)
231}
232
233#[cfg(test)]
234#[path = "reachability_tests.rs"]
235mod tests;