fakecloud_dsql/
ticker.rs

1//! Background lifecycle advancement for DSQL clusters and streams.
2//!
3//! AWS reports a freshly created cluster as `CREATING` and only later flips it
4//! to `ACTIVE`; a deleted cluster passes through `DELETING` before it vanishes.
5//! Streams follow the same `CREATING -> ACTIVE` / `DELETING -> DELETED` shape.
6//! This ticker mirrors that observable behavior so clients that poll `GetCluster`
7//! until `ACTIVE` (the documented pattern, and what the Terraform/SDK waiters do)
8//! actually converge, instead of seeing a cluster stuck in `CREATING` forever.
9
10use std::sync::Arc;
11use std::time::Duration;
12
13use chrono::Utc;
14use tokio::sync::Mutex as AsyncMutex;
15
16use fakecloud_persistence::SnapshotStore;
17
18use crate::persistence::save_snapshot;
19use crate::state::SharedDsqlState;
20
21/// How long a cluster/stream lingers in a transient state before advancing.
22/// Short enough that tests and interactive use converge quickly, long enough
23/// that a `CREATING` status is actually observable on the create response.
24const TRANSITION_AFTER: chrono::Duration = chrono::Duration::milliseconds(400);
25
26/// Poll interval for the advancement loop.
27const TICK: Duration = Duration::from_millis(200);
28
29/// Spawn the lifecycle loop. Advances transient statuses and removes clusters
30/// that have finished deleting, persisting after any change.
31pub fn spawn(
32    state: SharedDsqlState,
33    store: Option<Arc<dyn SnapshotStore>>,
34    lock: Arc<AsyncMutex<()>>,
35) {
36    tokio::spawn(async move {
37        loop {
38            tokio::time::sleep(TICK).await;
39            let changed = advance(&state);
40            if changed {
41                save_snapshot(&state, store.clone(), &lock).await;
42            }
43        }
44    });
45}
46
47/// Advance every account's clusters/streams one step. Returns whether anything
48/// changed (so the caller only persists on real transitions).
49pub fn advance(state: &SharedDsqlState) -> bool {
50    let now = Utc::now();
51    let mut changed = false;
52    let mut accounts = state.write();
53    for (_id, acct) in accounts.iter_mut() {
54        let mut to_remove = Vec::new();
55        for (cid, cluster) in acct.clusters.iter_mut() {
56            let elapsed = now.signed_duration_since(cluster.creation_time);
57            match cluster.status.as_str() {
58                "CREATING" | "PENDING_SETUP" if elapsed >= TRANSITION_AFTER => {
59                    cluster.status = "ACTIVE".to_string();
60                    changed = true;
61                }
62                "DELETING" | "PENDING_DELETE" => {
63                    // DeleteCluster stamped `deleted_at`; remove only once the
64                    // grace window has elapsed, so the DELETING state is
65                    // observable and persisted (not gone on the next tick).
66                    let deleting_for = cluster
67                        .deleted_at
68                        .map(|t| now.signed_duration_since(t))
69                        .unwrap_or(TRANSITION_AFTER);
70                    if deleting_for >= TRANSITION_AFTER {
71                        to_remove.push(cid.clone());
72                        changed = true;
73                    }
74                }
75                _ => {}
76            }
77            // Advance / reap streams within the cluster.
78            let mut streams_to_remove = Vec::new();
79            for (sid, stream) in cluster.streams.iter_mut() {
80                match stream.status.as_str() {
81                    "CREATING"
82                        if now.signed_duration_since(stream.creation_time) >= TRANSITION_AFTER =>
83                    {
84                        stream.status = "ACTIVE".to_string();
85                        changed = true;
86                    }
87                    "DELETING" => {
88                        let deleting_for = stream
89                            .deleted_at
90                            .map(|t| now.signed_duration_since(t))
91                            .unwrap_or(TRANSITION_AFTER);
92                        if deleting_for >= TRANSITION_AFTER {
93                            streams_to_remove.push(sid.clone());
94                            changed = true;
95                        }
96                    }
97                    _ => {}
98                }
99            }
100            for sid in streams_to_remove {
101                cluster.streams.remove(&sid);
102            }
103        }
104        for cid in to_remove {
105            acct.clusters.remove(&cid);
106        }
107    }
108    changed
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114    use crate::state::{cluster_arn, cluster_endpoint, Cluster, EncryptionDetails};
115    use fakecloud_core::multi_account::MultiAccountState;
116    use parking_lot::RwLock;
117    use std::collections::BTreeMap;
118
119    fn cluster(status: &str, age_ms: i64) -> Cluster {
120        Cluster {
121            identifier: "abcdefghij0123456789klmnop".into(),
122            arn: cluster_arn("us-east-1", "000000000000", "abcdefghij0123456789klmnop"),
123            status: status.into(),
124            creation_time: Utc::now() - chrono::Duration::milliseconds(age_ms),
125            deletion_protection_enabled: true,
126            multi_region_properties: None,
127            encryption_details: EncryptionDetails {
128                encryption_type: "AWS_OWNED_KMS_KEY".into(),
129                kms_key_arn: None,
130                encryption_status: "ENABLED".into(),
131            },
132            endpoint: cluster_endpoint("abcdefghij0123456789klmnop", "us-east-1"),
133            tags: BTreeMap::new(),
134            policy: None,
135            policy_version: 0,
136            client_token: None,
137            deleted_at: None,
138            streams: BTreeMap::new(),
139        }
140    }
141
142    fn state_with(c: Cluster) -> SharedDsqlState {
143        let mut accounts: MultiAccountState<crate::state::DsqlState> =
144            MultiAccountState::new("000000000000", "us-east-1", "");
145        accounts
146            .default_mut()
147            .clusters
148            .insert(c.identifier.clone(), c);
149        Arc::new(RwLock::new(accounts))
150    }
151
152    #[test]
153    fn creating_advances_to_active_after_window() {
154        let state = state_with(cluster("CREATING", 1000));
155        assert!(advance(&state));
156        assert_eq!(
157            state.read().get("000000000000").unwrap().clusters["abcdefghij0123456789klmnop"].status,
158            "ACTIVE"
159        );
160    }
161
162    #[test]
163    fn creating_stays_within_window() {
164        let state = state_with(cluster("CREATING", 0));
165        assert!(!advance(&state));
166    }
167
168    #[test]
169    fn deleting_removed_after_grace_window() {
170        // deleted_at older than the window: reaped this tick.
171        let mut c = cluster("DELETING", 0);
172        c.deleted_at = Some(Utc::now() - chrono::Duration::milliseconds(1000));
173        let state = state_with(c);
174        assert!(advance(&state));
175        assert!(state
176            .read()
177            .get("000000000000")
178            .unwrap()
179            .clusters
180            .is_empty());
181    }
182
183    #[test]
184    fn deleting_survives_within_grace_window() {
185        // Just-requested deletion stays observable until the window elapses.
186        let mut c = cluster("DELETING", 0);
187        c.deleted_at = Some(Utc::now());
188        let state = state_with(c);
189        assert!(!advance(&state));
190        assert_eq!(
191            state.read().get("000000000000").unwrap().clusters["abcdefghij0123456789klmnop"].status,
192            "DELETING"
193        );
194    }
195}
fakecloud_dsql/ticker.rs

fakecloud_dsql/
ticker.rs
