fakecloud_iotdata/
persistence.rs

1//! Snapshot save/load for AWS IoT Data Plane state.
2
3use std::sync::Arc;
4
5use tokio::sync::Mutex as AsyncMutex;
6
7use fakecloud_persistence::SnapshotStore;
8
9use crate::state::{IotDataSnapshot, SharedIotDataState, IOTDATA_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 iotdata persistence snapshot: {0}")]
20    Io(String),
21    #[error("failed to parse iotdata persistence snapshot: {0}")]
22    Parse(String),
23    #[error("iotdata 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: &SharedIotDataState,
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 snapshot: IotDataSnapshot =
35        serde_json::from_slice(&bytes).map_err(|e| LoadError::Parse(e.to_string()))?;
36    if snapshot.schema_version > IOTDATA_SNAPSHOT_SCHEMA_VERSION {
37        return Err(LoadError::SchemaTooNew {
38            on_disk: snapshot.schema_version,
39            supported: IOTDATA_SNAPSHOT_SCHEMA_VERSION,
40        });
41    }
42    let accounts = snapshot.accounts.account_count();
43    *state.write() = snapshot.accounts;
44    Ok(LoadOutcome::Loaded(accounts))
45}
46
47pub async fn save_snapshot(
48    state: &SharedIotDataState,
49    store: Option<Arc<dyn SnapshotStore>>,
50    lock: &AsyncMutex<()>,
51) {
52    let Some(store) = store else {
53        return;
54    };
55    let _guard = lock.lock().await;
56    let snapshot = IotDataSnapshot {
57        schema_version: IOTDATA_SNAPSHOT_SCHEMA_VERSION,
58        accounts: state.read().clone(),
59    };
60    let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
61        let bytes = serde_json::to_vec(&snapshot)
62            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
63        store.save(&bytes)
64    })
65    .await;
66    match join {
67        Ok(Ok(())) => {}
68        Ok(Err(err)) => tracing::error!(%err, "failed to write iotdata snapshot"),
69        Err(err) => tracing::error!(%err, "iotdata snapshot task panicked"),
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76    use crate::state::IotDataData;
77    use fakecloud_core::multi_account::MultiAccountState;
78    use parking_lot::RwLock;
79    use serde_json::json;
80    use std::collections::BTreeMap;
81    use std::sync::Mutex;
82
83    struct MemStore(Mutex<Option<Vec<u8>>>);
84    impl SnapshotStore for MemStore {
85        fn load(&self) -> std::io::Result<Option<Vec<u8>>> {
86            Ok(self.0.lock().unwrap().clone())
87        }
88        fn save(&self, bytes: &[u8]) -> std::io::Result<()> {
89            *self.0.lock().unwrap() = Some(bytes.to_vec());
90            Ok(())
91        }
92    }
93
94    fn state() -> SharedIotDataState {
95        Arc::new(RwLock::new(MultiAccountState::new(
96            "000000000000",
97            "us-east-1",
98            "",
99        )))
100    }
101
102    #[test]
103    fn empty_store_is_empty() {
104        assert_eq!(
105            load_into(&MemStore(Mutex::new(None)), &state()).unwrap(),
106            LoadOutcome::Empty
107        );
108    }
109
110    #[test]
111    fn round_trip_restores_shadows_and_retained() {
112        let mut accounts: MultiAccountState<IotDataData> =
113            MultiAccountState::new("000000000000", "us-east-1", "");
114        let data = accounts.get_or_create("111122223333");
115        let mut named = BTreeMap::new();
116        named.insert("".to_string(), json!({ "version": 3 }));
117        data.shadows.insert("sensor-1".to_string(), named);
118        data.retained
119            .insert("a/b".to_string(), json!({ "topic": "a/b", "qos": 1 }));
120        let snap = IotDataSnapshot {
121            schema_version: IOTDATA_SNAPSHOT_SCHEMA_VERSION,
122            accounts,
123        };
124        let store = MemStore(Mutex::new(Some(serde_json::to_vec(&snap).unwrap())));
125        let restored = state();
126        assert_eq!(
127            load_into(&store, &restored).unwrap(),
128            LoadOutcome::Loaded(2)
129        );
130        let guard = restored.read();
131        let acct = guard.get("111122223333").unwrap();
132        assert_eq!(acct.shadows["sensor-1"][""]["version"], 3);
133        assert_eq!(acct.retained["a/b"]["qos"], 1);
134    }
135
136    #[test]
137    fn rejects_future_schema() {
138        let accounts: MultiAccountState<IotDataData> =
139            MultiAccountState::new("000000000000", "us-east-1", "");
140        let bytes = serde_json::to_vec(&serde_json::json!({
141            "schema_version": IOTDATA_SNAPSHOT_SCHEMA_VERSION + 1,
142            "accounts": accounts,
143        }))
144        .unwrap();
145        let store = MemStore(Mutex::new(Some(bytes)));
146        assert!(matches!(
147            load_into(&store, &state()),
148            Err(LoadError::SchemaTooNew { .. })
149        ));
150    }
151}
fakecloud_iotdata/persistence.rs

fakecloud_iotdata/
persistence.rs
