Add backup/restore tests for Memory and ScyllaDB backends (Phase 0b)

linera-core/src/worker.rs

@@ -81,7 +81,7 @@ pub const DEFAULT_EXECUTION_STATE_CACHE_SIZE: usize = 10_000;
 #[path = "unit_tests/worker_tests.rs"]
 mod worker_tests;
 
-#[cfg(all(test, feature = "rocksdb"))]
+#[cfg(test)]
 #[path = "unit_tests/worker_backup_tests.rs"]
 mod worker_backup_tests;
 

linera-storage/src/db_storage.rs

@@ -1592,8 +1592,8 @@ where
     Database: linera_views::backends::DatabaseBackup,
 {
     /// Backs up the underlying database to the given directory.
-    pub fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
-        self.database.backup_to(dir)
+    pub async fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
+        self.database.backup_to(dir).await
     }
 }
 

linera-views/src/backends/journaling.rs

@@ -460,6 +460,15 @@ where
     }
 }
 
+#[cfg(with_testing)]
+impl<D: crate::backends::DatabaseBackup + Sync> crate::backends::DatabaseBackup
+    for JournalingKeyValueDatabase<D>
+{
+    async fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
+        self.database.backup_to(dir).await
+    }
+}
+
 impl<S> JournalingKeyValueStore<S> {
     /// Creates a new journaling store.
     pub fn new(store: S) -> Self {

linera-views/src/backends/lru_caching.rs

@@ -501,8 +501,10 @@ where
 }
 
 #[cfg(with_testing)]
-impl<D: crate::backends::DatabaseBackup> crate::backends::DatabaseBackup for LruCachingDatabase<D> {
-    fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
-        self.database.backup_to(dir)
+impl<D: crate::backends::DatabaseBackup + Sync> crate::backends::DatabaseBackup
+    for LruCachingDatabase<D>
+{
+    async fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
+        self.database.backup_to(dir).await
     }
 }

linera-views/src/backends/memory.rs

@@ -356,6 +356,32 @@ impl TestKeyValueDatabase for MemoryDatabase {
     }
 }
 
+#[cfg(with_testing)]
+impl crate::backends::DatabaseBackup for MemoryDatabase {
+    async fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
+        use std::collections::BTreeMap;
+        let databases = MEMORY_DATABASES
+            .lock()
+            .expect("MEMORY_DATABASES lock should not be poisoned");
+        let namespace_map = databases
+            .databases
+            .get(&self.namespace)
+            .ok_or_else(|| anyhow::anyhow!("namespace not found: {}", self.namespace))?;
+        let snapshot: BTreeMap<Vec<u8>, BTreeMap<Vec<u8>, Vec<u8>>> = namespace_map
+            .iter()
+            .map(|(root_key, store_map)| {
+                let store_map = store_map
+                    .read()
+                    .expect("MemoryStore lock should not be poisoned");
+                (root_key.clone(), store_map.clone())
+            })
+            .collect();
+        let encoded = bcs::to_bytes(&snapshot)?;
+        std::fs::write(dir.join("memory_backup.bcs"), encoded)?;
+        Ok(())
+    }
+}
+
 /// The error type for [`MemoryStore`].
 #[derive(Error, Debug)]
 pub enum MemoryStoreError {

linera-views/src/backends/metering.rs

@@ -585,8 +585,10 @@ where
 }
 
 #[cfg(with_testing)]
-impl<D: crate::backends::DatabaseBackup> crate::backends::DatabaseBackup for MeteredDatabase<D> {
-    fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
-        self.database.backup_to(dir)
+impl<D: crate::backends::DatabaseBackup + Sync> crate::backends::DatabaseBackup
+    for MeteredDatabase<D>
+{
+    async fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
+        self.database.backup_to(dir).await
     }
 }

linera-views/src/backends/mod.rs

@@ -24,8 +24,11 @@ pub mod rocks_db;
 pub mod indexed_db;
 
 #[cfg(with_testing)]
-/// Creates a RocksDB backup of the underlying database into a directory.
+/// Serializes the contents of a database namespace to disk for test backup/restore.
 pub trait DatabaseBackup {
-    /// Writes a RocksDB backup snapshot into `dir`.
-    fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()>;
+    /// Writes a snapshot of the namespace into `dir`.
+    fn backup_to(
+        &self,
+        dir: &std::path::Path,
+    ) -> impl std::future::Future<Output = anyhow::Result<()>> + Send;
 }

linera-views/src/backends/rocks_db.rs

@@ -794,7 +794,7 @@ pub type RocksDbDatabase = LruCachingDatabase<ValueSplittingDatabase<RocksDbData
 
 #[cfg(with_testing)]
 impl crate::backends::DatabaseBackup for RocksDbDatabaseInternal {
-    fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
+    async fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
         use rocksdb::{
             backup::{BackupEngine, BackupEngineOptions},
             Env,

linera-views/src/backends/scylla_db.rs

@@ -1304,6 +1304,39 @@ impl ScyllaDbDatabaseInternal {
     }
 }
 
+#[cfg(with_testing)]
+impl crate::backends::DatabaseBackup for ScyllaDbDatabaseInternal {
+    async fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
+        use std::collections::BTreeMap;
+
+        use futures::StreamExt as _;
+
+        let namespace = &self.store.namespace;
+        let statement = self
+            .store
+            .session
+            .prepare(format!(
+                "SELECT root_key, k, v FROM {KEYSPACE}.\"{namespace}\""
+            ))
+            .await
+            .map_err(|e| anyhow::anyhow!("{e}"))?;
+        let rows = Box::pin(self.store.session.execute_iter(statement, &[]))
+            .await
+            .map_err(|e| anyhow::anyhow!("{e}"))?;
+        let mut snapshot: BTreeMap<Vec<u8>, BTreeMap<Vec<u8>, Vec<u8>>> = BTreeMap::new();
+        let mut rows = rows
+            .rows_stream::<(Vec<u8>, Vec<u8>, Vec<u8>)>()
+            .map_err(|e| anyhow::anyhow!("{e}"))?;
+        while let Some(row) = rows.next().await {
+            let (root_key, k, v) = row.map_err(|e| anyhow::anyhow!("{e}"))?;
+            snapshot.entry(root_key).or_default().insert(k, v);
+        }
+        let encoded = bcs::to_bytes(&snapshot).map_err(|e| anyhow::anyhow!("{e}"))?;
+        std::fs::write(dir.join("scylladb_backup.bcs"), encoded)?;
+        Ok(())
+    }
+}
+
 #[cfg(with_testing)]
 impl TestKeyValueDatabase for JournalingKeyValueDatabase<ScyllaDbDatabaseInternal> {
     async fn new_test_config(

linera-views/src/backends/value_splitting.rs

@@ -361,11 +361,11 @@ where
 }
 
 #[cfg(with_testing)]
-impl<D: crate::backends::DatabaseBackup> crate::backends::DatabaseBackup
+impl<D: crate::backends::DatabaseBackup + Sync> crate::backends::DatabaseBackup
     for ValueSplittingDatabase<D>
 {
-    fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
-        self.database.backup_to(dir)
+    async fn backup_to(&self, dir: &std::path::Path) -> anyhow::Result<()> {
+        self.database.backup_to(dir).await
     }
 }