fakecloud_swf/
state.rs

1//! Account-partitioned, serializable state for Amazon SWF (`swf`).
2//!
3//! Everything an SWF caller can create lives here: domains (with their
4//! `REGISTERED` / `DEPRECATED` status), registered activity and workflow types
5//! (versioned, deprecatable), and workflow executions. Each execution carries
6//! its full ordered event history plus the in-flight decision and activity
7//! tasks that drive the real decider/worker state machine. Configuration blobs
8//! (a type's `default*` timeouts, an execution's timeouts / child policy) are
9//! stored as their already-output-valid wire JSON so a `Describe*` echoes
10//! exactly what its `Register*` / `Start*` persisted.
11//!
12//! Types are keyed by `name\u{1}version` (see [`crate::shared::type_key`]);
13//! executions by their minted `runId`. Task tokens index back to the owning
14//! execution so a `Respond*` / `RecordActivityTaskHeartbeat` resolves in O(1).
15//! Tags live in an ARN-keyed side map so `ListTagsForResource` has a single
16//! source of truth.
17
18use std::collections::BTreeMap;
19use std::sync::Arc;
20
21use parking_lot::RwLock;
22use serde::{Deserialize, Serialize};
23use serde_json::{Map, Value};
24
25use fakecloud_core::multi_account::{AccountState, MultiAccountState};
26
27pub const SWF_SNAPSHOT_SCHEMA_VERSION: u32 = 1;
28
29/// A registered SWF domain.
30#[derive(Debug, Clone, Serialize, Deserialize)]
31pub struct Domain {
32    pub name: String,
33    /// `REGISTERED` or `DEPRECATED`.
34    pub status: String,
35    #[serde(default)]
36    pub description: Option<String>,
37    /// The retention period in days, stored as the wire string SWF uses.
38    pub retention_days: String,
39    pub arn: String,
40}
41
42/// A registered activity or workflow type.
43#[derive(Debug, Clone, Serialize, Deserialize)]
44pub struct TypeRecord {
45    pub name: String,
46    pub version: String,
47    /// `REGISTERED` or `DEPRECATED`.
48    pub status: String,
49    #[serde(default)]
50    pub description: Option<String>,
51    pub creation_date: f64,
52    #[serde(default)]
53    pub deprecation_date: Option<f64>,
54    /// The `default*` configuration members, stored as their wire JSON object
55    /// so `Describe*Type`'s `configuration` echoes verbatim.
56    #[serde(default)]
57    pub configuration: Map<String, Value>,
58}
59
60/// The lifecycle state of a scheduled/started activity task inside an execution.
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62pub enum ActivityState {
63    Scheduled,
64    Started,
65    Closed,
66}
67
68/// One activity task tracked by an execution.
69#[derive(Debug, Clone, Serialize, Deserialize)]
70pub struct ActivityRecord {
71    pub activity_id: String,
72    pub activity_type_name: String,
73    pub activity_type_version: String,
74    #[serde(default)]
75    pub input: Option<String>,
76    pub task_list: String,
77    pub scheduled_event_id: i64,
78    #[serde(default)]
79    pub started_event_id: Option<i64>,
80    pub state: ActivityState,
81    pub task_token: String,
82}
83
84/// A workflow execution and everything the state machine needs to drive it.
85#[derive(Debug, Clone, Serialize, Deserialize)]
86pub struct Execution {
87    pub domain: String,
88    pub workflow_id: String,
89    pub run_id: String,
90    pub workflow_type_name: String,
91    pub workflow_type_version: String,
92    pub task_list: String,
93    #[serde(default)]
94    pub task_priority: Option<String>,
95    /// `OPEN` or `CLOSED`.
96    pub status: String,
97    /// `COMPLETED` / `FAILED` / `CANCELED` / `TERMINATED` / `TIMED_OUT` /
98    /// `CONTINUED_AS_NEW` once closed.
99    #[serde(default)]
100    pub close_status: Option<String>,
101    pub start_timestamp: f64,
102    #[serde(default)]
103    pub close_timestamp: Option<f64>,
104    #[serde(default)]
105    pub tag_list: Vec<String>,
106    #[serde(default)]
107    pub input: Option<String>,
108    pub child_policy: String,
109    pub task_start_to_close_timeout: String,
110    pub execution_start_to_close_timeout: String,
111    #[serde(default)]
112    pub lambda_role: Option<String>,
113    #[serde(default)]
114    pub cancel_requested: bool,
115    #[serde(default)]
116    pub latest_execution_context: Option<String>,
117
118    /// The ordered event history, each entry an already-output-valid
119    /// `HistoryEvent` wire object.
120    #[serde(default)]
121    pub events: Vec<Value>,
122    /// The next `eventId` to assign (1-based, monotonic).
123    pub next_event_id: i64,
124
125    /// A decision task is scheduled and awaiting a poll.
126    #[serde(default)]
127    pub decision_scheduled: bool,
128    /// The `eventId` of the outstanding `DecisionTaskScheduled` event.
129    #[serde(default)]
130    pub decision_scheduled_event_id: Option<i64>,
131    /// The `eventId` of the `DecisionTaskStarted` for an in-flight decision.
132    #[serde(default)]
133    pub decision_started_event_id: Option<i64>,
134    /// The `eventId` of the previous `DecisionTaskStarted` (for the next poll).
135    #[serde(default)]
136    pub previous_started_event_id: i64,
137    /// The task token handed out for the in-flight decision task, if any.
138    #[serde(default)]
139    pub decision_task_token: Option<String>,
140
141    /// Activity tasks tracked by this execution, in schedule order.
142    #[serde(default)]
143    pub activities: Vec<ActivityRecord>,
144}
145
146/// Per-account Amazon SWF state.
147#[derive(Debug, Clone, Default, Serialize, Deserialize)]
148pub struct SwfData {
149    #[serde(default)]
150    pub account_id: String,
151    #[serde(default)]
152    pub region: String,
153
154    /// Domains keyed by name.
155    #[serde(default)]
156    pub domains: BTreeMap<String, Domain>,
157    /// Activity types keyed by `"{domain}\u{1}{name}\u{1}{version}"`.
158    #[serde(default)]
159    pub activity_types: BTreeMap<String, TypeRecord>,
160    /// Workflow types keyed by `"{domain}\u{1}{name}\u{1}{version}"`.
161    #[serde(default)]
162    pub workflow_types: BTreeMap<String, TypeRecord>,
163    /// Workflow executions keyed by `runId`.
164    #[serde(default)]
165    pub executions: BTreeMap<String, Execution>,
166    /// Decision task token -> owning `runId`.
167    #[serde(default)]
168    pub decision_tokens: BTreeMap<String, String>,
169    /// Activity task token -> `(runId, activityId)`.
170    #[serde(default)]
171    pub activity_tokens: BTreeMap<String, ActivityRef>,
172    /// Resource tags keyed by domain ARN.
173    #[serde(default)]
174    pub tags: BTreeMap<String, BTreeMap<String, String>>,
175}
176
177/// A back-reference from an activity task token to its execution and activity.
178#[derive(Debug, Clone, Serialize, Deserialize)]
179pub struct ActivityRef {
180    pub run_id: String,
181    pub activity_id: String,
182}
183
184impl AccountState for SwfData {
185    fn new_for_account(account_id: &str, region: &str, _endpoint: &str) -> Self {
186        Self {
187            account_id: account_id.to_string(),
188            region: region.to_string(),
189            ..Default::default()
190        }
191    }
192}
193
194pub type SharedSwfState = Arc<RwLock<MultiAccountState<SwfData>>>;
195
196#[derive(Debug, Serialize, Deserialize)]
197pub struct SwfSnapshot {
198    pub schema_version: u32,
199    pub accounts: MultiAccountState<SwfData>,
200}
201
202impl SwfData {
203    /// Composite key for a type record within a domain.
204    pub fn type_map_key(domain: &str, name: &str, version: &str) -> String {
205        format!("{domain}\u{1}{name}\u{1}{version}")
206    }
207
208    /// Find the currently-open execution for a `workflow_id` (SWF allows only
209    /// one open execution per workflow id at a time), preferring an exact
210    /// `run_id` when supplied.
211    pub fn find_execution<'a>(
212        &'a self,
213        workflow_id: &str,
214        run_id: Option<&str>,
215    ) -> Option<&'a Execution> {
216        if let Some(rid) = run_id.filter(|r| !r.is_empty()) {
217            return self
218                .executions
219                .get(rid)
220                .filter(|e| e.workflow_id == workflow_id);
221        }
222        // Prefer an open execution; fall back to any (most recent by start).
223        self.executions
224            .values()
225            .filter(|e| e.workflow_id == workflow_id)
226            .max_by(|a, b| {
227                (a.status == "OPEN", a.start_timestamp)
228                    .partial_cmp(&(b.status == "OPEN", b.start_timestamp))
229                    .unwrap_or(std::cmp::Ordering::Equal)
230            })
231    }
232
233    pub fn find_execution_mut<'a>(
234        &'a mut self,
235        workflow_id: &str,
236        run_id: Option<&str>,
237    ) -> Option<&'a mut Execution> {
238        let rid = match run_id.filter(|r| !r.is_empty()) {
239            Some(rid) => Some(rid.to_string()),
240            None => self
241                .executions
242                .values()
243                .filter(|e| e.workflow_id == workflow_id)
244                .max_by(|a, b| {
245                    (a.status == "OPEN", a.start_timestamp)
246                        .partial_cmp(&(b.status == "OPEN", b.start_timestamp))
247                        .unwrap_or(std::cmp::Ordering::Equal)
248                })
249                .map(|e| e.run_id.clone()),
250        }?;
251        self.executions
252            .get_mut(&rid)
253            .filter(|e| e.workflow_id == workflow_id)
254    }
255}
256
257impl Execution {
258    /// Allocate the next monotonic event id.
259    pub fn alloc_event_id(&mut self) -> i64 {
260        let id = self.next_event_id;
261        self.next_event_id += 1;
262        id
263    }
264}
265
266#[cfg(test)]
267mod tests {
268    use super::*;
269
270    #[test]
271    fn find_execution_prefers_open() {
272        let mut d = SwfData::new_for_account("000000000000", "us-east-1", "");
273        d.executions
274            .insert("run-old".into(), exec("wf-1", "run-old", "CLOSED", 1.0));
275        d.executions
276            .insert("run-new".into(), exec("wf-1", "run-new", "OPEN", 2.0));
277        assert_eq!(d.find_execution("wf-1", None).unwrap().run_id, "run-new");
278        assert_eq!(
279            d.find_execution("wf-1", Some("run-old")).unwrap().run_id,
280            "run-old"
281        );
282    }
283
284    fn exec(wf: &str, run: &str, status: &str, ts: f64) -> Execution {
285        Execution {
286            domain: "d".into(),
287            workflow_id: wf.into(),
288            run_id: run.into(),
289            workflow_type_name: "t".into(),
290            workflow_type_version: "1".into(),
291            task_list: "tl".into(),
292            task_priority: None,
293            status: status.into(),
294            close_status: None,
295            start_timestamp: ts,
296            close_timestamp: None,
297            tag_list: vec![],
298            input: None,
299            child_policy: "TERMINATE".into(),
300            task_start_to_close_timeout: "10".into(),
301            execution_start_to_close_timeout: "3600".into(),
302            lambda_role: None,
303            cancel_requested: false,
304            latest_execution_context: None,
305            events: vec![],
306            next_event_id: 1,
307            decision_scheduled: false,
308            decision_scheduled_event_id: None,
309            decision_started_event_id: None,
310            previous_started_event_id: 0,
311            decision_task_token: None,
312            activities: vec![],
313        }
314    }
315}
fakecloud_swf/state.rs

fakecloud_swf/
state.rs
