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, StepState, 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/// The activities the run's active segment records as dispatched and not yet
57/// terminated, in the order the ordinals were opened.
58///
59/// This is the DISPATCHED slice of [`aion_core::open_steps`] — the single fold
60/// that answers "what is this run's history holding open", shared with the live
61/// describe join rather than reimplemented here. Keeping one fold is the point:
62/// two scans of the same events, each with its own idea of what a terminal is,
63/// would sooner or later disagree, and an operator would be reading whichever
64/// one happened to be wired to the surface they opened.
65///
66/// A step the fold reports as `Scheduled` or `Reopened` is deliberately NOT
67/// here. Nothing has been dispatched for it, so there is no delivery for the
68/// fleet to fail to serve, and classifying one would report a queue problem
69/// against work that has not been handed to a queue.
70#[must_use]
71pub fn open_activities_in_active_segment(history: &[Event]) -> Vec<OpenActivity> {
72 aion_core::open_steps(history)
73 .into_iter()
74 .filter_map(|step| match step.state {
75 StepState::Dispatched {
76 attempt,
77 dispatched_at,
78 } => Some(OpenActivity {
79 activity_id: step.activity_id,
80 activity_type: step.activity_type,
81 task_queue: step.task_queue,
82 node: step.node,
83 attempt,
84 dispatched_at,
85 }),
86 StepState::Scheduled | StepState::Reopened { .. } => None,
87 })
88 .collect()
89}
90
91/// The live fleet seams an in-flight activity's reachability is judged against.
92///
93/// Exactly the three the dispatcher's selection wait holds: the connected-worker
94/// registry (the census), the deployed queue declarations (the structural
95/// verdict), and the parked-dispatch state (whether this process is waiting).
96pub struct ActivityReachability<'a> {
97 /// Live connected-worker registry.
98 pub registry: &'a ConnectedWorkerRegistry,
99 /// Deployed queue declarations.
100 pub declarations: &'a QueueDeclarationSource,
101 /// Live parked-dispatch state.
102 pub state: &'a QueueServiceState,
103}
104
105impl ActivityReachability<'_> {
106 /// Every in-flight activity in `history` the fleet cannot currently serve.
107 ///
108 /// Returns an EMPTY vector for a run whose in-flight activities all have a
109 /// live compatible worker, and for a run with no in-flight activity at all.
110 ///
111 /// # Errors
112 ///
113 /// Returns [`ServerError`] when the connected-worker registry or the parked
114 /// dispatch state cannot be read. Never guesses: an unreadable fleet is
115 /// reported, not rendered as "everything is fine".
116 pub fn unserved(
117 &self,
118 namespace: &str,
119 workflow_id: &WorkflowId,
120 history: &[Event],
121 ) -> Result<Vec<UnservedActivity>, ServerError> {
122 let mut unserved = Vec::new();
123 for open in open_activities_in_active_segment(history) {
124 let address = ServiceAddress {
125 namespace: namespace.to_owned(),
126 task_queue: open.task_queue,
127 activity_type: open.activity_type,
128 node: open.node,
129 };
130 let census = self.registry.pool_census(
131 &address.namespace,
132 &address.task_queue,
133 &address.activity_type,
134 address.node.as_deref(),
135 )?;
136 let declaration = self.declarations.declaration_for(&address.task_queue);
137 // `None` is the healthy answer: a live compatible worker can take
138 // this dispatch, so there is nothing to report and nothing is
139 // reported.
140 let Some(reason) = classify(declaration, &census) else {
141 continue;
142 };
143 let dispatch_parked = self.state.is_unserved(workflow_id, &open.activity_id)?;
144 unserved.push(UnservedActivity {
145 activity_id: open.activity_id,
146 reason: reason.as_str().to_owned(),
147 detail: reason.explain(&address),
148 activity_type: address.activity_type,
149 task_queue: address.task_queue,
150 node: address.node,
151 attempt: open.attempt,
152 dispatched_at: open.dispatched_at,
153 workers_in_pool: count(census.workers_in_pool),
154 workers_serving_activity: count(census.workers_serving_activity),
155 compatible_workers: count(census.compatible_workers),
156 dispatch_parked,
157 });
158 }
159 Ok(unserved)
160 }
161}
162
163/// Widen a census count to the wire width, saturating rather than wrapping.
164fn count(value: usize) -> u64 {
165 u64::try_from(value).unwrap_or(u64::MAX)
166}
167
168#[cfg(test)]
169#[path = "reachability_tests.rs"]
170mod tests;