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    #[serde(default)]
154    pub encryption_configuration: Option<Value>,
155}
156
157#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
158pub enum StateMachineType {
159    Standard,
160    Express,
161}
162
163impl StateMachineType {
164    pub fn as_str(&self) -> &'static str {
165        match self {
166            Self::Standard => "STANDARD",
167            Self::Express => "EXPRESS",
168        }
169    }
170
171    pub fn parse(s: &str) -> Option<Self> {
172        match s {
173            "STANDARD" => Some(Self::Standard),
174            "EXPRESS" => Some(Self::Express),
175            _ => None,
176        }
177    }
178}
179
180#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
181pub enum StateMachineStatus {
182    Active,
183    Deleting,
184}
185
186impl StateMachineStatus {
187    pub fn as_str(&self) -> &'static str {
188        match self {
189            Self::Active => "ACTIVE",
190            Self::Deleting => "DELETING",
191        }
192    }
193}
194
195#[derive(Debug, Clone, Serialize, Deserialize)]
196pub struct Execution {
197    pub execution_arn: String,
198    pub state_machine_arn: String,
199    pub state_machine_name: String,
200    pub name: String,
201    pub status: ExecutionStatus,
202    pub input: Option<String>,
203    pub output: Option<String>,
204    pub start_date: DateTime<Utc>,
205    pub stop_date: Option<DateTime<Utc>>,
206    pub error: Option<String>,
207    pub cause: Option<String>,
208    pub history_events: Vec<HistoryEvent>,
209    /// Parent execution ARN when this execution was started by another
210    /// state machine via `arn:aws:states:::states:startExecution[.sync]`.
211    /// `None` for top-level executions started by external callers.
212    #[serde(default)]
213    pub parent_execution_arn: Option<String>,
214    /// True when this execution was created by `StartSyncExecution`
215    /// (EXPRESS state machines only). Distinguishes it from regular
216    /// async executions in introspection endpoints.
217    #[serde(default)]
218    pub is_sync: bool,
219    /// Billed duration in milliseconds, populated on terminal state for
220    /// sync executions. Mirrors `billingDetails.billedDurationInMilliseconds`
221    /// from the StartSyncExecution response.
222    #[serde(default)]
223    pub billed_duration_ms: Option<i64>,
224    /// Billed memory in MB for sync executions. Mirrors
225    /// `billingDetails.billedMemoryUsedInMB`.
226    #[serde(default)]
227    pub billed_memory_mb: Option<i64>,
228}
229
230#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
231pub enum ExecutionStatus {
232    Running,
233    Succeeded,
234    Failed,
235    TimedOut,
236    Aborted,
237    PendingRedrive,
238}
239
240impl ExecutionStatus {
241    pub fn as_str(&self) -> &'static str {
242        match self {
243            Self::Running => "RUNNING",
244            Self::Succeeded => "SUCCEEDED",
245            Self::Failed => "FAILED",
246            Self::TimedOut => "TIMED_OUT",
247            Self::Aborted => "ABORTED",
248            Self::PendingRedrive => "PENDING_REDRIVE",
249        }
250    }
251}
252
253#[derive(Debug, Clone, Serialize, Deserialize)]
254pub struct HistoryEvent {
255    pub id: i64,
256    pub event_type: String,
257    pub timestamp: DateTime<Utc>,
258    pub previous_event_id: i64,
259    pub details: Value,
260}
261
262impl StepFunctionsState {
263    pub fn new(account_id: &str, region: &str) -> Self {
264        Self {
265            account_id: account_id.to_string(),
266            region: region.to_string(),
267            state_machines: BTreeMap::new(),
268            executions: BTreeMap::new(),
269            activities: BTreeMap::new(),
270            state_machine_versions: BTreeMap::new(),
271            state_machine_aliases: BTreeMap::new(),
272            map_runs: BTreeMap::new(),
273            task_tokens: BTreeMap::new(),
274        }
275    }
276
277    pub fn reset(&mut self) {
278        self.state_machines.clear();
279        self.executions.clear();
280        self.activities.clear();
281        self.state_machine_versions.clear();
282        self.state_machine_aliases.clear();
283        self.map_runs.clear();
284        self.task_tokens.clear();
285    }
286
287    pub fn state_machine_arn(&self, region: &str, name: &str) -> String {
288        format!(
289            "arn:aws:states:{}:{}:stateMachine:{}",
290            region, self.account_id, name
291        )
292    }
293
294    pub fn execution_arn(
295        &self,
296        region: &str,
297        state_machine_name: &str,
298        execution_name: &str,
299    ) -> String {
300        format!(
301            "arn:aws:states:{}:{}:execution:{}:{}",
302            region, self.account_id, state_machine_name, execution_name
303        )
304    }
305}
306
307#[cfg(test)]
308mod tests {
309    use super::*;
310
311    #[test]
312    fn state_machine_type_as_str() {
313        assert_eq!(StateMachineType::Standard.as_str(), "STANDARD");
314        assert_eq!(StateMachineType::Express.as_str(), "EXPRESS");
315    }
316
317    #[test]
318    fn state_machine_type_parse() {
319        assert_eq!(
320            StateMachineType::parse("STANDARD"),
321            Some(StateMachineType::Standard)
322        );
323        assert_eq!(
324            StateMachineType::parse("EXPRESS"),
325            Some(StateMachineType::Express)
326        );
327        assert_eq!(StateMachineType::parse("bogus"), None);
328    }
329
330    #[test]
331    fn state_machine_status_as_str() {
332        assert_eq!(StateMachineStatus::Active.as_str(), "ACTIVE");
333        assert_eq!(StateMachineStatus::Deleting.as_str(), "DELETING");
334    }
335
336    #[test]
337    fn execution_status_as_str() {
338        assert_eq!(ExecutionStatus::Running.as_str(), "RUNNING");
339        assert_eq!(ExecutionStatus::Succeeded.as_str(), "SUCCEEDED");
340        assert_eq!(ExecutionStatus::Failed.as_str(), "FAILED");
341        assert_eq!(ExecutionStatus::TimedOut.as_str(), "TIMED_OUT");
342        assert_eq!(ExecutionStatus::Aborted.as_str(), "ABORTED");
343        assert_eq!(ExecutionStatus::PendingRedrive.as_str(), "PENDING_REDRIVE");
344    }
345
346    #[test]
347    fn state_machine_arn_format() {
348        let state = StepFunctionsState::new("123456789012", "us-east-1");
349        assert_eq!(
350            state.state_machine_arn("us-east-1", "my-sm"),
351            "arn:aws:states:us-east-1:123456789012:stateMachine:my-sm"
352        );
353    }
354
355    #[test]
356    fn execution_arn_format() {
357        let state = StepFunctionsState::new("123456789012", "us-east-1");
358        assert_eq!(
359            state.execution_arn("us-east-1", "sm", "exec-1"),
360            "arn:aws:states:us-east-1:123456789012:execution:sm:exec-1"
361        );
362    }
363
364    #[test]
365    fn arns_use_request_region_not_server_default() {
366        // Server default region is us-east-1, but a client whose credential
367        // scope resolves to eu-central-1 must get eu-central-1 ARNs.
368        let state = StepFunctionsState::new("123456789012", "us-east-1");
369        assert_eq!(
370            state.state_machine_arn("eu-central-1", "my-sm"),
371            "arn:aws:states:eu-central-1:123456789012:stateMachine:my-sm"
372        );
373        assert_eq!(
374            state.execution_arn("eu-central-1", "sm", "exec-1"),
375            "arn:aws:states:eu-central-1:123456789012:execution:sm:exec-1"
376        );
377    }
378
379    #[test]
380    fn state_reset_clears_all() {
381        let mut state = StepFunctionsState::new("123456789012", "us-east-1");
382        state.state_machines.insert(
383            "x".to_string(),
384            StateMachine {
385                name: "sm".to_string(),
386                arn: "arn:aws:states:us-east-1:123:stateMachine:sm".to_string(),
387                definition: "{}".to_string(),
388                role_arn: "r".to_string(),
389                machine_type: StateMachineType::Standard,
390                status: StateMachineStatus::Active,
391                creation_date: Utc::now(),
392                update_date: Utc::now(),
393                tags: BTreeMap::new(),
394                revision_id: "v1".to_string(),
395                logging_configuration: None,
396                tracing_configuration: None,
397                description: String::new(),
398                encryption_configuration: None,
399            },
400        );
401        state.reset();
402        assert!(state.state_machines.is_empty());
403        assert!(state.executions.is_empty());
404    }
405}
fakecloud_stepfunctions/state.rs

fakecloud_stepfunctions/
state.rs
