fakecloud_stepfunctions/
state.rs

1use std::collections::BTreeMap;
2use std::sync::Arc;
3
4use chrono::{DateTime, Utc};
5use parking_lot::RwLock;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9pub type SharedStepFunctionsState =
10    Arc<RwLock<fakecloud_core::multi_account::MultiAccountState<StepFunctionsState>>>;
11
12impl fakecloud_core::multi_account::AccountState for StepFunctionsState {
13    fn new_for_account(account_id: &str, region: &str, _endpoint: &str) -> Self {
14        Self::new(account_id, region)
15    }
16}
17
18pub const STEPFUNCTIONS_SNAPSHOT_SCHEMA_VERSION: u32 = 2;
19
20#[derive(Debug, Serialize, Deserialize)]
21pub struct StepFunctionsSnapshot {
22    pub schema_version: u32,
23    #[serde(default)]
24    pub accounts: Option<fakecloud_core::multi_account::MultiAccountState<StepFunctionsState>>,
25    #[serde(default)]
26    pub state: Option<StepFunctionsState>,
27}
28
29#[derive(Debug, Clone, Serialize, Deserialize)]
30pub struct StepFunctionsState {
31    pub account_id: String,
32    pub region: String,
33    /// State machines keyed by ARN.
34    #[serde(default)]
35    pub state_machines: BTreeMap<String, StateMachine>,
36    /// Executions keyed by execution ARN.
37    #[serde(default)]
38    pub executions: BTreeMap<String, Execution>,
39    #[serde(default)]
40    pub activities: BTreeMap<String, Activity>,
41    #[serde(default)]
42    pub state_machine_versions: BTreeMap<String, StateMachineVersion>,
43    #[serde(default)]
44    pub state_machine_aliases: BTreeMap<String, StateMachineAlias>,
45    #[serde(default)]
46    pub map_runs: BTreeMap<String, MapRun>,
47    /// Pending task tokens issued for sync activities + their outcome.
48    #[serde(default)]
49    pub task_tokens: BTreeMap<String, TaskTokenState>,
50}
51
52#[derive(Debug, Clone, Serialize, Deserialize)]
53pub struct Activity {
54    pub name: String,
55    pub arn: String,
56    pub creation_date: DateTime<Utc>,
57    pub tags: BTreeMap<String, String>,
58}
59
60#[derive(Debug, Clone, Serialize, Deserialize)]
61pub struct StateMachineVersion {
62    pub state_machine_arn: String,
63    pub version: i64,
64    pub revision_id: String,
65    pub description: String,
66    pub creation_date: DateTime<Utc>,
67}
68
69#[derive(Debug, Clone, Serialize, Deserialize)]
70pub struct StateMachineAlias {
71    pub name: String,
72    pub arn: String,
73    pub description: String,
74    pub routing_configuration: Vec<AliasRoute>,
75    pub creation_date: DateTime<Utc>,
76    pub update_date: DateTime<Utc>,
77}
78
79#[derive(Debug, Clone, Serialize, Deserialize)]
80pub struct AliasRoute {
81    pub state_machine_version_arn: String,
82    pub weight: i32,
83}
84
85#[derive(Debug, Clone, Serialize, Deserialize)]
86pub struct MapRun {
87    pub map_run_arn: String,
88    pub execution_arn: String,
89    pub max_concurrency: i32,
90    pub tolerated_failure_percentage: f64,
91    pub tolerated_failure_count: i64,
92    pub status: String,
93    pub start_date: DateTime<Utc>,
94    pub stop_date: Option<DateTime<Utc>>,
95    /// Total number of items the distributed Map iterated over.
96    #[serde(default)]
97    pub total_count: i64,
98    /// Iterations that completed successfully.
99    #[serde(default)]
100    pub succeeded_count: i64,
101    /// Iterations that failed.
102    #[serde(default)]
103    pub failed_count: i64,
104}
105
106#[derive(Debug, Clone, Serialize, Deserialize)]
107pub struct TaskTokenState {
108    pub activity_arn: String,
109    /// PENDING (waiting for `GetActivityTask` to dequeue) /
110    /// IN_PROGRESS (worker has picked it up) /
111    /// SUCCEEDED / FAILED / TIMED_OUT.
112    pub status: String,
113    pub output: Option<String>,
114    pub error: Option<String>,
115    pub cause: Option<String>,
116    /// Input the state machine wanted the worker to process. `None`
117    /// for tokens minted by external `GetActivityTask` callers without
118    /// any associated activity execution (legacy synthetic path).
119    #[serde(default)]
120    pub input: Option<String>,
121    #[serde(default = "default_now")]
122    pub created_at: DateTime<Utc>,
123    #[serde(default)]
124    pub last_heartbeat_at: Option<DateTime<Utc>>,
125    /// Per AWS docs: state machine fails the task if no heartbeat in
126    /// this many seconds while the worker is running.
127    #[serde(default)]
128    pub heartbeat_seconds: Option<i64>,
129    /// Overall timeout for the task; counted from `created_at`.
130    #[serde(default)]
131    pub timeout_seconds: Option<i64>,
132}
133
134fn default_now() -> DateTime<Utc> {
135    Utc::now()
136}
137
138#[derive(Debug, Clone, Serialize, Deserialize)]
139pub struct StateMachine {
140    pub name: String,
141    pub arn: String,
142    pub definition: String,
143    pub role_arn: String,
144    pub machine_type: StateMachineType,
145    pub status: StateMachineStatus,
146    pub creation_date: DateTime<Utc>,
147    pub update_date: DateTime<Utc>,
148    pub tags: BTreeMap<String, String>,
149    pub revision_id: String,
150    pub logging_configuration: Option<Value>,
151    pub tracing_configuration: Option<Value>,
152    pub description: String,
153}
154
155#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
156pub enum StateMachineType {
157    Standard,
158    Express,
159}
160
161impl StateMachineType {
162    pub fn as_str(&self) -> &'static str {
163        match self {
164            Self::Standard => "STANDARD",
165            Self::Express => "EXPRESS",
166        }
167    }
168
169    pub fn parse(s: &str) -> Option<Self> {
170        match s {
171            "STANDARD" => Some(Self::Standard),
172            "EXPRESS" => Some(Self::Express),
173            _ => None,
174        }
175    }
176}
177
178#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
179pub enum StateMachineStatus {
180    Active,
181    Deleting,
182}
183
184impl StateMachineStatus {
185    pub fn as_str(&self) -> &'static str {
186        match self {
187            Self::Active => "ACTIVE",
188            Self::Deleting => "DELETING",
189        }
190    }
191}
192
193#[derive(Debug, Clone, Serialize, Deserialize)]
194pub struct Execution {
195    pub execution_arn: String,
196    pub state_machine_arn: String,
197    pub state_machine_name: String,
198    pub name: String,
199    pub status: ExecutionStatus,
200    pub input: Option<String>,
201    pub output: Option<String>,
202    pub start_date: DateTime<Utc>,
203    pub stop_date: Option<DateTime<Utc>>,
204    pub error: Option<String>,
205    pub cause: Option<String>,
206    pub history_events: Vec<HistoryEvent>,
207    /// Parent execution ARN when this execution was started by another
208    /// state machine via `arn:aws:states:::states:startExecution[.sync]`.
209    /// `None` for top-level executions started by external callers.
210    #[serde(default)]
211    pub parent_execution_arn: Option<String>,
212    /// True when this execution was created by `StartSyncExecution`
213    /// (EXPRESS state machines only). Distinguishes it from regular
214    /// async executions in introspection endpoints.
215    #[serde(default)]
216    pub is_sync: bool,
217    /// Billed duration in milliseconds, populated on terminal state for
218    /// sync executions. Mirrors `billingDetails.billedDurationInMilliseconds`
219    /// from the StartSyncExecution response.
220    #[serde(default)]
221    pub billed_duration_ms: Option<i64>,
222    /// Billed memory in MB for sync executions. Mirrors
223    /// `billingDetails.billedMemoryUsedInMB`.
224    #[serde(default)]
225    pub billed_memory_mb: Option<i64>,
226}
227
228#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
229pub enum ExecutionStatus {
230    Running,
231    Succeeded,
232    Failed,
233    TimedOut,
234    Aborted,
235    PendingRedrive,
236}
237
238impl ExecutionStatus {
239    pub fn as_str(&self) -> &'static str {
240        match self {
241            Self::Running => "RUNNING",
242            Self::Succeeded => "SUCCEEDED",
243            Self::Failed => "FAILED",
244            Self::TimedOut => "TIMED_OUT",
245            Self::Aborted => "ABORTED",
246            Self::PendingRedrive => "PENDING_REDRIVE",
247        }
248    }
249}
250
251#[derive(Debug, Clone, Serialize, Deserialize)]
252pub struct HistoryEvent {
253    pub id: i64,
254    pub event_type: String,
255    pub timestamp: DateTime<Utc>,
256    pub previous_event_id: i64,
257    pub details: Value,
258}
259
260impl StepFunctionsState {
261    pub fn new(account_id: &str, region: &str) -> Self {
262        Self {
263            account_id: account_id.to_string(),
264            region: region.to_string(),
265            state_machines: BTreeMap::new(),
266            executions: BTreeMap::new(),
267            activities: BTreeMap::new(),
268            state_machine_versions: BTreeMap::new(),
269            state_machine_aliases: BTreeMap::new(),
270            map_runs: BTreeMap::new(),
271            task_tokens: BTreeMap::new(),
272        }
273    }
274
275    pub fn reset(&mut self) {
276        self.state_machines.clear();
277        self.executions.clear();
278        self.activities.clear();
279        self.state_machine_versions.clear();
280        self.state_machine_aliases.clear();
281        self.map_runs.clear();
282        self.task_tokens.clear();
283    }
284
285    pub fn state_machine_arn(&self, name: &str) -> String {
286        format!(
287            "arn:aws:states:{}:{}:stateMachine:{}",
288            self.region, self.account_id, name
289        )
290    }
291
292    pub fn execution_arn(&self, state_machine_name: &str, execution_name: &str) -> String {
293        format!(
294            "arn:aws:states:{}:{}:execution:{}:{}",
295            self.region, self.account_id, state_machine_name, execution_name
296        )
297    }
298}
299
300#[cfg(test)]
301mod tests {
302    use super::*;
303
304    #[test]
305    fn state_machine_type_as_str() {
306        assert_eq!(StateMachineType::Standard.as_str(), "STANDARD");
307        assert_eq!(StateMachineType::Express.as_str(), "EXPRESS");
308    }
309
310    #[test]
311    fn state_machine_type_parse() {
312        assert_eq!(
313            StateMachineType::parse("STANDARD"),
314            Some(StateMachineType::Standard)
315        );
316        assert_eq!(
317            StateMachineType::parse("EXPRESS"),
318            Some(StateMachineType::Express)
319        );
320        assert_eq!(StateMachineType::parse("bogus"), None);
321    }
322
323    #[test]
324    fn state_machine_status_as_str() {
325        assert_eq!(StateMachineStatus::Active.as_str(), "ACTIVE");
326        assert_eq!(StateMachineStatus::Deleting.as_str(), "DELETING");
327    }
328
329    #[test]
330    fn execution_status_as_str() {
331        assert_eq!(ExecutionStatus::Running.as_str(), "RUNNING");
332        assert_eq!(ExecutionStatus::Succeeded.as_str(), "SUCCEEDED");
333        assert_eq!(ExecutionStatus::Failed.as_str(), "FAILED");
334        assert_eq!(ExecutionStatus::TimedOut.as_str(), "TIMED_OUT");
335        assert_eq!(ExecutionStatus::Aborted.as_str(), "ABORTED");
336        assert_eq!(ExecutionStatus::PendingRedrive.as_str(), "PENDING_REDRIVE");
337    }
338
339    #[test]
340    fn state_machine_arn_format() {
341        let state = StepFunctionsState::new("123456789012", "us-east-1");
342        assert_eq!(
343            state.state_machine_arn("my-sm"),
344            "arn:aws:states:us-east-1:123456789012:stateMachine:my-sm"
345        );
346    }
347
348    #[test]
349    fn execution_arn_format() {
350        let state = StepFunctionsState::new("123456789012", "us-east-1");
351        assert_eq!(
352            state.execution_arn("sm", "exec-1"),
353            "arn:aws:states:us-east-1:123456789012:execution:sm:exec-1"
354        );
355    }
356
357    #[test]
358    fn state_reset_clears_all() {
359        let mut state = StepFunctionsState::new("123456789012", "us-east-1");
360        state.state_machines.insert(
361            "x".to_string(),
362            StateMachine {
363                name: "sm".to_string(),
364                arn: "arn:aws:states:us-east-1:123:stateMachine:sm".to_string(),
365                definition: "{}".to_string(),
366                role_arn: "r".to_string(),
367                machine_type: StateMachineType::Standard,
368                status: StateMachineStatus::Active,
369                creation_date: Utc::now(),
370                update_date: Utc::now(),
371                tags: BTreeMap::new(),
372                revision_id: "v1".to_string(),
373                logging_configuration: None,
374                tracing_configuration: None,
375                description: String::new(),
376            },
377        );
378        state.reset();
379        assert!(state.state_machines.is_empty());
380        assert!(state.executions.is_empty());
381    }
382}
fakecloud_stepfunctions/state.rs

fakecloud_stepfunctions/
state.rs
