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

aion_server/stream/
selector.rs
