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

fakecloud_stepfunctions/
state.rs
