fakecloud_efs/
persistence.rs

1//! Snapshot save/load for EFS state, with lifecycle reconciliation on restart.
2
3use std::sync::Arc;
4
5use tokio::sync::Mutex as AsyncMutex;
6
7use fakecloud_persistence::SnapshotStore;
8
9use crate::state::{EfsSnapshot, SharedEfsState, EFS_SNAPSHOT_SCHEMA_VERSION};
10
11#[derive(Debug, PartialEq, Eq)]
12pub enum LoadOutcome {
13    Empty,
14    Loaded(usize),
15}
16
17#[derive(Debug, thiserror::Error)]
18pub enum LoadError {
19    #[error("failed to read efs persistence snapshot: {0}")]
20    Io(String),
21    #[error("failed to parse efs persistence snapshot: {0}")]
22    Parse(String),
23    #[error("efs persistence schema too new: on-disk={on_disk}, max supported={supported}")]
24    SchemaTooNew { on_disk: u32, supported: u32 },
25}
26
27pub fn load_into(
28    store: &dyn SnapshotStore,
29    state: &SharedEfsState,
30) -> Result<LoadOutcome, LoadError> {
31    let Some(bytes) = store.load().map_err(|e| LoadError::Io(e.to_string()))? else {
32        return Ok(LoadOutcome::Empty);
33    };
34    let mut snapshot: EfsSnapshot =
35        serde_json::from_slice(&bytes).map_err(|e| LoadError::Parse(e.to_string()))?;
36    if snapshot.schema_version > EFS_SNAPSHOT_SCHEMA_VERSION {
37        return Err(LoadError::SchemaTooNew {
38            on_disk: snapshot.schema_version,
39            supported: EFS_SNAPSHOT_SCHEMA_VERSION,
40        });
41    }
42    // Reconcile any lifecycle transition that was in flight when the process
43    // stopped: a file system / mount target / access point left `creating`
44    // settles to `available` (there is no timer to resume).
45    for (_account_id, account) in snapshot.accounts.iter_mut() {
46        account.reconcile_lifecycle();
47    }
48    let accounts = snapshot.accounts.account_count();
49    *state.write() = snapshot.accounts;
50    Ok(LoadOutcome::Loaded(accounts))
51}
52
53pub async fn save_snapshot(
54    state: &SharedEfsState,
55    store: Option<Arc<dyn SnapshotStore>>,
56    lock: &AsyncMutex<()>,
57) {
58    let Some(store) = store else {
59        return;
60    };
61    let _guard = lock.lock().await;
62    let snapshot = EfsSnapshot {
63        schema_version: EFS_SNAPSHOT_SCHEMA_VERSION,
64        accounts: state.read().clone(),
65    };
66    let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
67        let bytes = serde_json::to_vec(&snapshot)
68            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
69        store.save(&bytes)
70    })
71    .await;
72    match join {
73        Ok(Ok(())) => {}
74        Ok(Err(err)) => tracing::error!(%err, "failed to write efs snapshot"),
75        Err(err) => tracing::error!(%err, "efs snapshot task panicked"),
76    }
77}
78
79#[cfg(test)]
80mod tests {
81    use super::*;
82    use crate::state::{EfsData, EfsSnapshot};
83    use fakecloud_core::multi_account::MultiAccountState;
84    use parking_lot::RwLock;
85    use std::sync::Mutex;
86
87    struct MemStore(Mutex<Option<Vec<u8>>>);
88    impl SnapshotStore for MemStore {
89        fn load(&self) -> std::io::Result<Option<Vec<u8>>> {
90            Ok(self.0.lock().unwrap().clone())
91        }
92        fn save(&self, bytes: &[u8]) -> std::io::Result<()> {
93            *self.0.lock().unwrap() = Some(bytes.to_vec());
94            Ok(())
95        }
96    }
97
98    fn state() -> SharedEfsState {
99        Arc::new(RwLock::new(MultiAccountState::new(
100            "000000000000",
101            "us-east-1",
102            "",
103        )))
104    }
105
106    #[test]
107    fn empty_store_is_empty() {
108        assert_eq!(
109            load_into(&MemStore(Mutex::new(None)), &state()).unwrap(),
110            LoadOutcome::Empty
111        );
112    }
113
114    #[test]
115    fn round_trip_restores_and_reconciles_lifecycle() {
116        let mut accounts: MultiAccountState<EfsData> =
117            MultiAccountState::new("000000000000", "us-east-1", "");
118        let data = accounts.get_or_create("111122223333");
119        data.file_systems.insert(
120            "fs-0123456789abcdef0".to_string(),
121            serde_json::json!({
122                "FileSystemId": "fs-0123456789abcdef0",
123                "LifeCycleState": "creating"
124            }),
125        );
126        let snap = EfsSnapshot {
127            schema_version: EFS_SNAPSHOT_SCHEMA_VERSION,
128            accounts,
129        };
130        let store = MemStore(Mutex::new(Some(serde_json::to_vec(&snap).unwrap())));
131        let restored = state();
132        assert_eq!(
133            load_into(&store, &restored).unwrap(),
134            LoadOutcome::Loaded(2)
135        );
136        let guard = restored.read();
137        let fs = guard
138            .get("111122223333")
139            .unwrap()
140            .file_systems
141            .get("fs-0123456789abcdef0")
142            .unwrap();
143        // Reconciled from creating -> available on load.
144        assert_eq!(fs.get("LifeCycleState").unwrap(), "available");
145    }
146
147    #[test]
148    fn rejects_future_schema() {
149        let accounts: MultiAccountState<EfsData> =
150            MultiAccountState::new("000000000000", "us-east-1", "");
151        let bytes = serde_json::to_vec(&serde_json::json!({
152            "schema_version": EFS_SNAPSHOT_SCHEMA_VERSION + 1,
153            "accounts": accounts,
154        }))
155        .unwrap();
156        let store = MemStore(Mutex::new(Some(bytes)));
157        assert!(matches!(
158            load_into(&store, &state()),
159            Err(LoadError::SchemaTooNew { .. })
160        ));
161    }
162}
fakecloud_efs/persistence.rs

fakecloud_efs/
persistence.rs
