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

fakecloud_stepfunctions/
state.rs
