aion_server/stream/
selector.rs

1//! Server-side selector filtering for filtered subscriptions.
2//!
3//! `FilteredSubscription` advertises optional `workflow_type` and `status`
4//! selectors. The engine's `EventFilter` has no type or status dimension, so
5//! the selection runs at the socket seam, after the namespace gate proved
6//! ownership and resolved the workflow's recorded type from the same durable
7//! read.
8//!
9//! Selector semantics (documented in `docs/API.md`):
10//!
11//! - `workflow_type` matches when the event's workflow has that recorded type
12//!   at the time the namespace gate resolved it: the initial durable read
13//!   returns the head-of-history `WorkflowStarted` type at read time, which on
14//!   a continue-as-new chain can briefly run ahead of an older delivered event
15//!   (a one-event-loop forward-skew window) until the stream's own
16//!   `WorkflowStarted` refresh self-heals the cached type. A workflow whose
17//!   history records no started run never matches a type selector.
18//! - `status` matches per event kind: each terminal lifecycle event matches
19//!   exactly its projected status (`WorkflowCompleted` → `Completed`,
20//!   `WorkflowFailed` → `Failed`, `WorkflowCancelled` → `Cancelled`,
21//!   `WorkflowTimedOut` → `TimedOut`, `WorkflowContinuedAsNew` →
22//!   `ContinuedAsNew`); every non-terminal event — including
23//!   `WorkflowStarted` — matches `Running`.
24//! - When both selectors are present they AND together.
25
26use aion_core::{Event, WorkflowStatus};
27
28/// Validated subscription selectors applied before frame encoding.
29#[derive(Clone, Debug, Default, Eq, PartialEq)]
30pub struct SubscriptionSelector {
31    /// Deliver only events of workflows with this recorded type.
32    pub workflow_type: Option<String>,
33    /// Deliver only events whose kind projects to this status.
34    pub status: Option<WorkflowStatus>,
35}
36
37impl SubscriptionSelector {
38    /// Selector that admits every event (per-workflow and firehose
39    /// subscriptions carry no selectors).
40    #[must_use]
41    pub const fn unrestricted() -> Self {
42        Self {
43            workflow_type: None,
44            status: None,
45        }
46    }
47
48    /// Decide whether an event passes the selector. `workflow_type` is the
49    /// event's workflow's recorded type as resolved by the namespace gate.
50    #[must_use]
51    pub fn matches(&self, event: &Event, workflow_type: Option<&str>) -> bool {
52        if let Some(selected_type) = &self.workflow_type {
53            // No recorded type (no started run) can never satisfy a type
54            // selector — absence is not a wildcard.
55            if workflow_type != Some(selected_type.as_str()) {
56                return false;
57            }
58        }
59        if let Some(selected_status) = self.status {
60            if event_status(event) != selected_status {
61                return false;
62            }
63        }
64        true
65    }
66}
67
68/// Status projected by a single event's kind: terminal lifecycle events
69/// project exactly their terminal status; every other event belongs to a
70/// running workflow at the moment it was recorded.
71const fn event_status(event: &Event) -> WorkflowStatus {
72    match event {
73        Event::WorkflowCompleted { .. } => WorkflowStatus::Completed,
74        Event::WorkflowFailed { .. } => WorkflowStatus::Failed,
75        Event::WorkflowCancelled { .. } => WorkflowStatus::Cancelled,
76        Event::WorkflowTimedOut { .. } => WorkflowStatus::TimedOut,
77        Event::WorkflowContinuedAsNew { .. } => WorkflowStatus::ContinuedAsNew,
78        // A pause projects Paused at the moment it is recorded (#204).
79        Event::WorkflowPaused { .. } => WorkflowStatus::Paused,
80        Event::WorkflowStarted { .. }
81        // A reopen returns the workflow to Running at the moment it is recorded.
82        | Event::WorkflowReopened { .. }
83        // A resume returns the workflow to Running at the moment it is recorded.
84        | Event::WorkflowResumed { .. }
85        | Event::SearchAttributesUpdated { .. }
86        | Event::ActivityScheduled { .. }
87        | Event::ActivityStarted { .. }
88        | Event::ActivityCompleted { .. }
89        | Event::ActivityFailed { .. }
90        // A side channel exhausted its budget; the workflow it warns about
91        // is still running.
92        | Event::ActivityAdvisoryExhausted { .. }
93        | Event::ActivityCancelled { .. }
94        | Event::TimerStarted { .. }
95        | Event::TimerFired { .. }
96        | Event::TimerCancelled { .. }
97        | Event::WithTimeoutCompleted { .. }
98        | Event::SignalReceived { .. }
99        | Event::SignalSent { .. }
100        | Event::ChildWorkflowStarted { .. }
101        | Event::ChildWorkflowCompleted { .. }
102        | Event::ChildWorkflowFailed { .. }
103        | Event::ChildWorkflowCancelled { .. }
104        | Event::ScheduleCreated { .. }
105        | Event::ScheduleUpdated { .. }
106        | Event::SchedulePaused { .. }
107        | Event::ScheduleResumed { .. }
108        | Event::ScheduleDeleted { .. }
109        | Event::ScheduleTriggered { .. } => WorkflowStatus::Running,
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use aion_core::{Event, EventEnvelope, Payload, WorkflowId, WorkflowStatus};
116
117    use super::SubscriptionSelector;
118
119    fn envelope(seq: u64) -> EventEnvelope {
120        EventEnvelope {
121            seq,
122            recorded_at: chrono::Utc::now(),
123            workflow_id: WorkflowId::new(uuid::Uuid::from_u128(1)),
124        }
125    }
126
127    fn payload() -> Result<Payload, aion_core::PayloadError> {
128        Payload::from_json(&serde_json::json!({ "label": "x" }))
129    }
130
131    fn signal(seq: u64) -> Result<Event, aion_core::PayloadError> {
132        Ok(Event::SignalReceived {
133            envelope: envelope(seq),
134            name: "ship".to_owned(),
135            payload: payload()?,
136        })
137    }
138
139    fn started(seq: u64) -> Result<Event, aion_core::PayloadError> {
140        Ok(Event::WorkflowStarted {
141            envelope: envelope(seq),
142            workflow_type: "checkout".to_owned(),
143            input: payload()?,
144            run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
145            parent_run_id: None,
146            package_version: aion_core::PackageVersion::new("a".repeat(64)),
147        })
148    }
149
150    fn completed(seq: u64) -> Result<Event, aion_core::PayloadError> {
151        Ok(Event::WorkflowCompleted {
152            envelope: envelope(seq),
153            result: payload()?,
154        })
155    }
156
157    fn failed(seq: u64) -> Event {
158        Event::WorkflowFailed {
159            envelope: envelope(seq),
160            error: aion_core::WorkflowError {
161                message: "boom".to_owned(),
162                details: None,
163            },
164        }
165    }
166
167    #[test]
168    fn unrestricted_selector_matches_everything() -> Result<(), Box<dyn std::error::Error>> {
169        let selector = SubscriptionSelector::unrestricted();
170
171        assert!(selector.matches(&signal(1)?, None));
172        assert!(selector.matches(&completed(2)?, Some("checkout")));
173        Ok(())
174    }
175
176    #[test]
177    fn type_selector_matches_only_the_recorded_type() -> Result<(), Box<dyn std::error::Error>> {
178        let selector = SubscriptionSelector {
179            workflow_type: Some("checkout".to_owned()),
180            status: None,
181        };
182
183        assert!(selector.matches(&signal(1)?, Some("checkout")));
184        assert!(!selector.matches(&signal(1)?, Some("fulfillment")));
185        assert!(
186            !selector.matches(&signal(1)?, None),
187            "a workflow with no recorded type never matches a type selector"
188        );
189        Ok(())
190    }
191
192    #[test]
193    fn status_selector_matches_per_event_kind() -> Result<(), Box<dyn std::error::Error>> {
194        let running = SubscriptionSelector {
195            workflow_type: None,
196            status: Some(WorkflowStatus::Running),
197        };
198        let completed_only = SubscriptionSelector {
199            workflow_type: None,
200            status: Some(WorkflowStatus::Completed),
201        };
202        let failed_only = SubscriptionSelector {
203            workflow_type: None,
204            status: Some(WorkflowStatus::Failed),
205        };
206
207        // Running matches every non-terminal event, including WorkflowStarted.
208        assert!(running.matches(&started(1)?, Some("checkout")));
209        assert!(running.matches(&signal(2)?, Some("checkout")));
210        assert!(!running.matches(&completed(3)?, Some("checkout")));
211
212        // Each terminal status matches exactly its terminal event kind.
213        assert!(completed_only.matches(&completed(3)?, Some("checkout")));
214        assert!(!completed_only.matches(&failed(3), Some("checkout")));
215        assert!(!completed_only.matches(&signal(2)?, Some("checkout")));
216        assert!(failed_only.matches(&failed(3), Some("checkout")));
217        assert!(!failed_only.matches(&completed(3)?, Some("checkout")));
218        Ok(())
219    }
220
221    fn timed_out(seq: u64) -> Event {
222        Event::WorkflowTimedOut {
223            envelope: envelope(seq),
224            timeout: "workflow".to_owned(),
225        }
226    }
227
228    #[test]
229    fn status_selector_projects_workflow_timed_out_to_timed_out()
230    -> Result<(), Box<dyn std::error::Error>> {
231        // The ops-console stream selector must surface a `WorkflowTimedOut`
232        // terminal as `TimedOut`: a `TimedOut` subscription admits it, a
233        // `Running` subscription rejects it (it is terminal), and a `Failed`
234        // subscription does not confuse it for a failure.
235        let timed_out_only = SubscriptionSelector {
236            workflow_type: None,
237            status: Some(WorkflowStatus::TimedOut),
238        };
239        let running = SubscriptionSelector {
240            workflow_type: None,
241            status: Some(WorkflowStatus::Running),
242        };
243        let failed_only = SubscriptionSelector {
244            workflow_type: None,
245            status: Some(WorkflowStatus::Failed),
246        };
247
248        assert!(timed_out_only.matches(&timed_out(4), Some("checkout")));
249        assert!(!timed_out_only.matches(&completed(3)?, Some("checkout")));
250        assert!(!running.matches(&timed_out(4), Some("checkout")));
251        assert!(!failed_only.matches(&timed_out(4), Some("checkout")));
252        Ok(())
253    }
254
255    #[test]
256    fn combined_selectors_and_together() -> Result<(), Box<dyn std::error::Error>> {
257        let selector = SubscriptionSelector {
258            workflow_type: Some("checkout".to_owned()),
259            status: Some(WorkflowStatus::Completed),
260        };
261
262        assert!(selector.matches(&completed(3)?, Some("checkout")));
263        assert!(
264            !selector.matches(&completed(3)?, Some("fulfillment")),
265            "matching status with mismatched type must not pass"
266        );
267        assert!(
268            !selector.matches(&signal(2)?, Some("checkout")),
269            "matching type with mismatched status must not pass"
270        );
271        Ok(())
272    }
273}
aion_server/stream/selector.rs

aion_server/stream/
selector.rs
