fakecloud_managedblockchain/
persistence.rs

1//! Snapshot save/load for Amazon Managed Blockchain state.
2
3use std::sync::Arc;
4
5use tokio::sync::Mutex as AsyncMutex;
6
7use fakecloud_persistence::SnapshotStore;
8
9use crate::state::{
10    ManagedBlockchainSnapshot, SharedManagedBlockchainState,
11    MANAGEDBLOCKCHAIN_SNAPSHOT_SCHEMA_VERSION,
12};
13
14#[derive(Debug, PartialEq, Eq)]
15pub enum LoadOutcome {
16    Empty,
17    Loaded(usize),
18}
19
20#[derive(Debug, thiserror::Error)]
21pub enum LoadError {
22    #[error("failed to read managedblockchain persistence snapshot: {0}")]
23    Io(String),
24    #[error("failed to parse managedblockchain persistence snapshot: {0}")]
25    Parse(String),
26    #[error(
27        "managedblockchain persistence schema too new: on-disk={on_disk}, max supported={supported}"
28    )]
29    SchemaTooNew { on_disk: u32, supported: u32 },
30}
31
32pub fn load_into(
33    store: &dyn SnapshotStore,
34    state: &SharedManagedBlockchainState,
35) -> Result<LoadOutcome, LoadError> {
36    let Some(bytes) = store.load().map_err(|e| LoadError::Io(e.to_string()))? else {
37        return Ok(LoadOutcome::Empty);
38    };
39    let snapshot: ManagedBlockchainSnapshot =
40        serde_json::from_slice(&bytes).map_err(|e| LoadError::Parse(e.to_string()))?;
41    if snapshot.schema_version > MANAGEDBLOCKCHAIN_SNAPSHOT_SCHEMA_VERSION {
42        return Err(LoadError::SchemaTooNew {
43            on_disk: snapshot.schema_version,
44            supported: MANAGEDBLOCKCHAIN_SNAPSHOT_SCHEMA_VERSION,
45        });
46    }
47    let accounts = snapshot.accounts.account_count();
48    *state.write() = snapshot.accounts;
49    Ok(LoadOutcome::Loaded(accounts))
50}
51
52pub async fn save_snapshot(
53    state: &SharedManagedBlockchainState,
54    store: Option<Arc<dyn SnapshotStore>>,
55    lock: &AsyncMutex<()>,
56) {
57    let Some(store) = store else {
58        return;
59    };
60    let _guard = lock.lock().await;
61    let snapshot = ManagedBlockchainSnapshot {
62        schema_version: MANAGEDBLOCKCHAIN_SNAPSHOT_SCHEMA_VERSION,
63        accounts: state.read().clone(),
64    };
65    let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
66        let bytes = serde_json::to_vec(&snapshot)
67            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
68        store.save(&bytes)
69    })
70    .await;
71    match join {
72        Ok(Ok(())) => {}
73        Ok(Err(err)) => tracing::error!(%err, "failed to write managedblockchain snapshot"),
74        Err(err) => tracing::error!(%err, "managedblockchain snapshot task panicked"),
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81    use crate::state::ManagedBlockchainData;
82    use fakecloud_core::multi_account::MultiAccountState;
83    use parking_lot::RwLock;
84    use serde_json::json;
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() -> SharedManagedBlockchainState {
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_networks() {
116        let mut accounts: MultiAccountState<ManagedBlockchainData> =
117            MultiAccountState::new("000000000000", "us-east-1", "");
118        let data = accounts.get_or_create("111122223333");
119        data.networks
120            .insert("n-ABC".to_string(), json!({ "Id": "n-ABC", "Name": "net" }));
121        let snap = ManagedBlockchainSnapshot {
122            schema_version: MANAGEDBLOCKCHAIN_SNAPSHOT_SCHEMA_VERSION,
123            accounts,
124        };
125        let store = MemStore(Mutex::new(Some(serde_json::to_vec(&snap).unwrap())));
126        let restored = state();
127        assert_eq!(
128            load_into(&store, &restored).unwrap(),
129            LoadOutcome::Loaded(2)
130        );
131        let guard = restored.read();
132        assert!(guard
133            .get("111122223333")
134            .unwrap()
135            .networks
136            .contains_key("n-ABC"));
137    }
138
139    #[test]
140    fn rejects_future_schema() {
141        let accounts: MultiAccountState<ManagedBlockchainData> =
142            MultiAccountState::new("000000000000", "us-east-1", "");
143        let bytes = serde_json::to_vec(&serde_json::json!({
144            "schema_version": MANAGEDBLOCKCHAIN_SNAPSHOT_SCHEMA_VERSION + 1,
145            "accounts": accounts,
146        }))
147        .unwrap();
148        let store = MemStore(Mutex::new(Some(bytes)));
149        assert!(matches!(
150            load_into(&store, &state()),
151            Err(LoadError::SchemaTooNew { .. })
152        ));
153    }
154}
fakecloud_managedblockchain/persistence.rs

fakecloud_managedblockchain/
persistence.rs
