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        Event::WorkflowStarted { .. }
79        // A reopen returns the workflow to Running at the moment it is recorded.
80        | Event::WorkflowReopened { .. }
81        | Event::SearchAttributesUpdated { .. }
82        | Event::ActivityScheduled { .. }
83        | Event::ActivityStarted { .. }
84        | Event::ActivityCompleted { .. }
85        | Event::ActivityFailed { .. }
86        | Event::ActivityCancelled { .. }
87        | Event::TimerStarted { .. }
88        | Event::TimerFired { .. }
89        | Event::TimerCancelled { .. }
90        | Event::WithTimeoutCompleted { .. }
91        | Event::SignalReceived { .. }
92        | Event::SignalSent { .. }
93        | Event::ChildWorkflowStarted { .. }
94        | Event::ChildWorkflowCompleted { .. }
95        | Event::ChildWorkflowFailed { .. }
96        | Event::ChildWorkflowCancelled { .. }
97        | Event::ScheduleCreated { .. }
98        | Event::ScheduleUpdated { .. }
99        | Event::SchedulePaused { .. }
100        | Event::ScheduleResumed { .. }
101        | Event::ScheduleDeleted { .. }
102        | Event::ScheduleTriggered { .. } => WorkflowStatus::Running,
103    }
104}
105
106#[cfg(test)]
107mod tests {
108    use aion_core::{Event, EventEnvelope, Payload, WorkflowId, WorkflowStatus};
109
110    use super::SubscriptionSelector;
111
112    fn envelope(seq: u64) -> EventEnvelope {
113        EventEnvelope {
114            seq,
115            recorded_at: chrono::Utc::now(),
116            workflow_id: WorkflowId::new(uuid::Uuid::from_u128(1)),
117        }
118    }
119
120    fn payload() -> Result<Payload, aion_core::PayloadError> {
121        Payload::from_json(&serde_json::json!({ "label": "x" }))
122    }
123
124    fn signal(seq: u64) -> Result<Event, aion_core::PayloadError> {
125        Ok(Event::SignalReceived {
126            envelope: envelope(seq),
127            name: "ship".to_owned(),
128            payload: payload()?,
129        })
130    }
131
132    fn started(seq: u64) -> Result<Event, aion_core::PayloadError> {
133        Ok(Event::WorkflowStarted {
134            envelope: envelope(seq),
135            workflow_type: "checkout".to_owned(),
136            input: payload()?,
137            run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
138            parent_run_id: None,
139            package_version: aion_core::PackageVersion::new("a".repeat(64)),
140        })
141    }
142
143    fn completed(seq: u64) -> Result<Event, aion_core::PayloadError> {
144        Ok(Event::WorkflowCompleted {
145            envelope: envelope(seq),
146            result: payload()?,
147        })
148    }
149
150    fn failed(seq: u64) -> Event {
151        Event::WorkflowFailed {
152            envelope: envelope(seq),
153            error: aion_core::WorkflowError {
154                message: "boom".to_owned(),
155                details: None,
156            },
157        }
158    }
159
160    #[test]
161    fn unrestricted_selector_matches_everything() -> Result<(), Box<dyn std::error::Error>> {
162        let selector = SubscriptionSelector::unrestricted();
163
164        assert!(selector.matches(&signal(1)?, None));
165        assert!(selector.matches(&completed(2)?, Some("checkout")));
166        Ok(())
167    }
168
169    #[test]
170    fn type_selector_matches_only_the_recorded_type() -> Result<(), Box<dyn std::error::Error>> {
171        let selector = SubscriptionSelector {
172            workflow_type: Some("checkout".to_owned()),
173            status: None,
174        };
175
176        assert!(selector.matches(&signal(1)?, Some("checkout")));
177        assert!(!selector.matches(&signal(1)?, Some("fulfillment")));
178        assert!(
179            !selector.matches(&signal(1)?, None),
180            "a workflow with no recorded type never matches a type selector"
181        );
182        Ok(())
183    }
184
185    #[test]
186    fn status_selector_matches_per_event_kind() -> Result<(), Box<dyn std::error::Error>> {
187        let running = SubscriptionSelector {
188            workflow_type: None,
189            status: Some(WorkflowStatus::Running),
190        };
191        let completed_only = SubscriptionSelector {
192            workflow_type: None,
193            status: Some(WorkflowStatus::Completed),
194        };
195        let failed_only = SubscriptionSelector {
196            workflow_type: None,
197            status: Some(WorkflowStatus::Failed),
198        };
199
200        // Running matches every non-terminal event, including WorkflowStarted.
201        assert!(running.matches(&started(1)?, Some("checkout")));
202        assert!(running.matches(&signal(2)?, Some("checkout")));
203        assert!(!running.matches(&completed(3)?, Some("checkout")));
204
205        // Each terminal status matches exactly its terminal event kind.
206        assert!(completed_only.matches(&completed(3)?, Some("checkout")));
207        assert!(!completed_only.matches(&failed(3), Some("checkout")));
208        assert!(!completed_only.matches(&signal(2)?, Some("checkout")));
209        assert!(failed_only.matches(&failed(3), Some("checkout")));
210        assert!(!failed_only.matches(&completed(3)?, Some("checkout")));
211        Ok(())
212    }
213
214    #[test]
215    fn combined_selectors_and_together() -> Result<(), Box<dyn std::error::Error>> {
216        let selector = SubscriptionSelector {
217            workflow_type: Some("checkout".to_owned()),
218            status: Some(WorkflowStatus::Completed),
219        };
220
221        assert!(selector.matches(&completed(3)?, Some("checkout")));
222        assert!(
223            !selector.matches(&completed(3)?, Some("fulfillment")),
224            "matching status with mismatched type must not pass"
225        );
226        assert!(
227            !selector.matches(&signal(2)?, Some("checkout")),
228            "matching type with mismatched status must not pass"
229        );
230        Ok(())
231    }
232}
aion_server/stream/selector.rs

aion_server/stream/
selector.rs
