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        | Event::ActivityCancelled { .. }
91        | Event::TimerStarted { .. }
92        | Event::TimerFired { .. }
93        | Event::TimerCancelled { .. }
94        | Event::WithTimeoutCompleted { .. }
95        | Event::SignalReceived { .. }
96        | Event::SignalSent { .. }
97        | Event::ChildWorkflowStarted { .. }
98        | Event::ChildWorkflowCompleted { .. }
99        | Event::ChildWorkflowFailed { .. }
100        | Event::ChildWorkflowCancelled { .. }
101        | Event::ScheduleCreated { .. }
102        | Event::ScheduleUpdated { .. }
103        | Event::SchedulePaused { .. }
104        | Event::ScheduleResumed { .. }
105        | Event::ScheduleDeleted { .. }
106        | Event::ScheduleTriggered { .. } => WorkflowStatus::Running,
107    }
108}
109
110#[cfg(test)]
111mod tests {
112    use aion_core::{Event, EventEnvelope, Payload, WorkflowId, WorkflowStatus};
113
114    use super::SubscriptionSelector;
115
116    fn envelope(seq: u64) -> EventEnvelope {
117        EventEnvelope {
118            seq,
119            recorded_at: chrono::Utc::now(),
120            workflow_id: WorkflowId::new(uuid::Uuid::from_u128(1)),
121        }
122    }
123
124    fn payload() -> Result<Payload, aion_core::PayloadError> {
125        Payload::from_json(&serde_json::json!({ "label": "x" }))
126    }
127
128    fn signal(seq: u64) -> Result<Event, aion_core::PayloadError> {
129        Ok(Event::SignalReceived {
130            envelope: envelope(seq),
131            name: "ship".to_owned(),
132            payload: payload()?,
133        })
134    }
135
136    fn started(seq: u64) -> Result<Event, aion_core::PayloadError> {
137        Ok(Event::WorkflowStarted {
138            envelope: envelope(seq),
139            workflow_type: "checkout".to_owned(),
140            input: payload()?,
141            run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
142            parent_run_id: None,
143            package_version: aion_core::PackageVersion::new("a".repeat(64)),
144        })
145    }
146
147    fn completed(seq: u64) -> Result<Event, aion_core::PayloadError> {
148        Ok(Event::WorkflowCompleted {
149            envelope: envelope(seq),
150            result: payload()?,
151        })
152    }
153
154    fn failed(seq: u64) -> Event {
155        Event::WorkflowFailed {
156            envelope: envelope(seq),
157            error: aion_core::WorkflowError {
158                message: "boom".to_owned(),
159                details: None,
160            },
161        }
162    }
163
164    #[test]
165    fn unrestricted_selector_matches_everything() -> Result<(), Box<dyn std::error::Error>> {
166        let selector = SubscriptionSelector::unrestricted();
167
168        assert!(selector.matches(&signal(1)?, None));
169        assert!(selector.matches(&completed(2)?, Some("checkout")));
170        Ok(())
171    }
172
173    #[test]
174    fn type_selector_matches_only_the_recorded_type() -> Result<(), Box<dyn std::error::Error>> {
175        let selector = SubscriptionSelector {
176            workflow_type: Some("checkout".to_owned()),
177            status: None,
178        };
179
180        assert!(selector.matches(&signal(1)?, Some("checkout")));
181        assert!(!selector.matches(&signal(1)?, Some("fulfillment")));
182        assert!(
183            !selector.matches(&signal(1)?, None),
184            "a workflow with no recorded type never matches a type selector"
185        );
186        Ok(())
187    }
188
189    #[test]
190    fn status_selector_matches_per_event_kind() -> Result<(), Box<dyn std::error::Error>> {
191        let running = SubscriptionSelector {
192            workflow_type: None,
193            status: Some(WorkflowStatus::Running),
194        };
195        let completed_only = SubscriptionSelector {
196            workflow_type: None,
197            status: Some(WorkflowStatus::Completed),
198        };
199        let failed_only = SubscriptionSelector {
200            workflow_type: None,
201            status: Some(WorkflowStatus::Failed),
202        };
203
204        // Running matches every non-terminal event, including WorkflowStarted.
205        assert!(running.matches(&started(1)?, Some("checkout")));
206        assert!(running.matches(&signal(2)?, Some("checkout")));
207        assert!(!running.matches(&completed(3)?, Some("checkout")));
208
209        // Each terminal status matches exactly its terminal event kind.
210        assert!(completed_only.matches(&completed(3)?, Some("checkout")));
211        assert!(!completed_only.matches(&failed(3), Some("checkout")));
212        assert!(!completed_only.matches(&signal(2)?, Some("checkout")));
213        assert!(failed_only.matches(&failed(3), Some("checkout")));
214        assert!(!failed_only.matches(&completed(3)?, Some("checkout")));
215        Ok(())
216    }
217
218    fn timed_out(seq: u64) -> Event {
219        Event::WorkflowTimedOut {
220            envelope: envelope(seq),
221            timeout: "workflow".to_owned(),
222        }
223    }
224
225    #[test]
226    fn status_selector_projects_workflow_timed_out_to_timed_out()
227    -> Result<(), Box<dyn std::error::Error>> {
228        // The ops-console stream selector must surface a `WorkflowTimedOut`
229        // terminal as `TimedOut`: a `TimedOut` subscription admits it, a
230        // `Running` subscription rejects it (it is terminal), and a `Failed`
231        // subscription does not confuse it for a failure.
232        let timed_out_only = SubscriptionSelector {
233            workflow_type: None,
234            status: Some(WorkflowStatus::TimedOut),
235        };
236        let running = SubscriptionSelector {
237            workflow_type: None,
238            status: Some(WorkflowStatus::Running),
239        };
240        let failed_only = SubscriptionSelector {
241            workflow_type: None,
242            status: Some(WorkflowStatus::Failed),
243        };
244
245        assert!(timed_out_only.matches(&timed_out(4), Some("checkout")));
246        assert!(!timed_out_only.matches(&completed(3)?, Some("checkout")));
247        assert!(!running.matches(&timed_out(4), Some("checkout")));
248        assert!(!failed_only.matches(&timed_out(4), Some("checkout")));
249        Ok(())
250    }
251
252    #[test]
253    fn combined_selectors_and_together() -> Result<(), Box<dyn std::error::Error>> {
254        let selector = SubscriptionSelector {
255            workflow_type: Some("checkout".to_owned()),
256            status: Some(WorkflowStatus::Completed),
257        };
258
259        assert!(selector.matches(&completed(3)?, Some("checkout")));
260        assert!(
261            !selector.matches(&completed(3)?, Some("fulfillment")),
262            "matching status with mismatched type must not pass"
263        );
264        assert!(
265            !selector.matches(&signal(2)?, Some("checkout")),
266            "matching type with mismatched status must not pass"
267        );
268        Ok(())
269    }
270}
aion_server/stream/selector.rs

aion_server/stream/
selector.rs
