fakecloud_iotwireless/
persistence.rs

1//! Snapshot save/load for AWS IoT Wireless control-plane state.
2
3use std::sync::Arc;
4
5use tokio::sync::Mutex as AsyncMutex;
6
7use fakecloud_persistence::SnapshotStore;
8
9use crate::state::{
10    IotWirelessSnapshot, SharedIotWirelessState, IOTWIRELESS_SNAPSHOT_SCHEMA_VERSION,
11};
12
13#[derive(Debug, PartialEq, Eq)]
14pub enum LoadOutcome {
15    Empty,
16    Loaded(usize),
17}
18
19#[derive(Debug, thiserror::Error)]
20pub enum LoadError {
21    #[error("failed to read iotwireless persistence snapshot: {0}")]
22    Io(String),
23    #[error("failed to parse iotwireless persistence snapshot: {0}")]
24    Parse(String),
25    #[error(
26        "iotwireless persistence schema too new: on-disk={on_disk}, max supported={supported}"
27    )]
28    SchemaTooNew { on_disk: u32, supported: u32 },
29}
30
31pub fn load_into(
32    store: &dyn SnapshotStore,
33    state: &SharedIotWirelessState,
34) -> Result<LoadOutcome, LoadError> {
35    let Some(bytes) = store.load().map_err(|e| LoadError::Io(e.to_string()))? else {
36        return Ok(LoadOutcome::Empty);
37    };
38    let snapshot: IotWirelessSnapshot =
39        serde_json::from_slice(&bytes).map_err(|e| LoadError::Parse(e.to_string()))?;
40    if snapshot.schema_version > IOTWIRELESS_SNAPSHOT_SCHEMA_VERSION {
41        return Err(LoadError::SchemaTooNew {
42            on_disk: snapshot.schema_version,
43            supported: IOTWIRELESS_SNAPSHOT_SCHEMA_VERSION,
44        });
45    }
46    let accounts = snapshot.accounts.account_count();
47    *state.write() = snapshot.accounts;
48    Ok(LoadOutcome::Loaded(accounts))
49}
50
51pub async fn save_snapshot(
52    state: &SharedIotWirelessState,
53    store: Option<Arc<dyn SnapshotStore>>,
54    lock: &AsyncMutex<()>,
55) {
56    let Some(store) = store else {
57        return;
58    };
59    let _guard = lock.lock().await;
60    let snapshot = IotWirelessSnapshot {
61        schema_version: IOTWIRELESS_SNAPSHOT_SCHEMA_VERSION,
62        accounts: state.read().clone(),
63    };
64    let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
65        let bytes = serde_json::to_vec(&snapshot)
66            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
67        store.save(&bytes)
68    })
69    .await;
70    match join {
71        Ok(Ok(())) => {}
72        Ok(Err(err)) => tracing::error!(%err, "failed to write iotwireless snapshot"),
73        Err(err) => tracing::error!(%err, "iotwireless snapshot task panicked"),
74    }
75}
76
77#[cfg(test)]
78mod tests {
79    use super::*;
80    use crate::state::IotWirelessData;
81    use fakecloud_core::multi_account::MultiAccountState;
82    use parking_lot::RwLock;
83    use serde_json::json;
84    use std::sync::Mutex;
85
86    struct MemStore(Mutex<Option<Vec<u8>>>);
87    impl SnapshotStore for MemStore {
88        fn load(&self) -> std::io::Result<Option<Vec<u8>>> {
89            Ok(self.0.lock().unwrap().clone())
90        }
91        fn save(&self, bytes: &[u8]) -> std::io::Result<()> {
92            *self.0.lock().unwrap() = Some(bytes.to_vec());
93            Ok(())
94        }
95    }
96
97    fn state() -> SharedIotWirelessState {
98        Arc::new(RwLock::new(MultiAccountState::new(
99            "000000000000",
100            "us-east-1",
101            "",
102        )))
103    }
104
105    #[test]
106    fn empty_store_is_empty() {
107        assert_eq!(
108            load_into(&MemStore(Mutex::new(None)), &state()).unwrap(),
109            LoadOutcome::Empty
110        );
111    }
112
113    #[test]
114    fn round_trip_restores_resources() {
115        let mut accounts: MultiAccountState<IotWirelessData> =
116            MultiAccountState::new("000000000000", "us-east-1", "");
117        let data = accounts.get_or_create("111122223333");
118        data.put_resource("destinations", "dest-1", json!({"Name": "dest-1"}));
119        let snap = IotWirelessSnapshot {
120            schema_version: IOTWIRELESS_SNAPSHOT_SCHEMA_VERSION,
121            accounts,
122        };
123        let store = MemStore(Mutex::new(Some(serde_json::to_vec(&snap).unwrap())));
124        let restored = state();
125        assert_eq!(
126            load_into(&store, &restored).unwrap(),
127            LoadOutcome::Loaded(2)
128        );
129        let guard = restored.read();
130        let acct = guard.get("111122223333").unwrap();
131        assert_eq!(
132            acct.get_resource("destinations", "dest-1").unwrap()["Name"],
133            "dest-1"
134        );
135    }
136
137    #[test]
138    fn rejects_future_schema() {
139        let accounts: MultiAccountState<IotWirelessData> =
140            MultiAccountState::new("000000000000", "us-east-1", "");
141        let bytes = serde_json::to_vec(&serde_json::json!({
142            "schema_version": IOTWIRELESS_SNAPSHOT_SCHEMA_VERSION + 1,
143            "accounts": accounts,
144        }))
145        .unwrap();
146        let store = MemStore(Mutex::new(Some(bytes)));
147        assert!(matches!(
148            load_into(&store, &state()),
149            Err(LoadError::SchemaTooNew { .. })
150        ));
151    }
152}
fakecloud_iotwireless/persistence.rs

fakecloud_iotwireless/
persistence.rs
