diff --git a/core/src/store/fst/backup.rs b/core/src/store/fst/backup.rs index f733aa6..8c9e607 100644 --- a/core/src/store/fst/backup.rs +++ b/core/src/store/fst/backup.rs @@ -20,6 +20,12 @@ impl FstStorePool { // Create backup directory (full path) fs::create_dir_all(path)?; + // NOTE: The FST store directory gets created when the first FST is + // consolidated, so there is nothing to back up until then. + if !self.fst_store_config.path.exists() { + return Ok(()); + } + // Proceed dump action (backup) self.dump_action( "backup", diff --git a/core/src/store/kv/backup.rs b/core/src/store/kv/backup.rs index 90a360e..dd5bf12 100644 --- a/core/src/store/kv/backup.rs +++ b/core/src/store/kv/backup.rs @@ -6,6 +6,7 @@ // License: Mozilla Public License v2.0 (MPL v2.0) use std::path::Path; +use std::sync::Arc; use std::{fs, io}; use rocksdb::backup::{ @@ -22,6 +23,12 @@ impl KvStorePool { // Create backup directory (full path) fs::create_dir_all(path)?; + // NOTE: The KV store directory gets created when the first KV store + // is opened, so there is nothing to back up until then. + if !self.kv_store_config.path.exists() { + return Ok(()); + } + // Proceed dump action (backup) self.dump_action( "backup", @@ -97,10 +104,6 @@ impl KvStorePool { // Create backup folder for collection fs::create_dir_all(backup_path.join(collection_hash))?; - let origin_kv = self - .open(store_id, |_| {}) - .map_err(|_| io::Error::other("database open failure"))?; - // Initialize KV database backup engine let kv_backup_options = DBBackupEngineOptions::new(&kv_backup_path) .map_err(|_| io::Error::other("backup engine options acquire failure"))?; @@ -111,9 +114,23 @@ impl KvStorePool { .map_err(|_| io::Error::other("backup engine failure"))?; // Proceed actual KV database backup - kv_backup_engine - .create_new_backup(&origin_kv) - .map_err(|_| io::Error::other("database backup failure"))?; + // NOTE: RocksDB refuses to open a database which is already open in + // this process (its `LOCK` file is held), so back up the pooled + // store if there is one. Flush it first, as its memtables may hold + // writes not persisted yet (e.g. if the write-ahead log is disabled). + let pooled_store = self.read().unwrap().get(&store_id).map(Arc::clone); + + match pooled_store { + Some(store) => kv_backup_engine.create_new_backup_flush(&store.database, true), + None => { + let origin_kv = self + .open(store_id, |_| {}) + .map_err(|_| io::Error::other("database open failure"))?; + + kv_backup_engine.create_new_backup(&origin_kv) + } + } + .map_err(|_| io::Error::other("database backup failure"))?; tracing::info!("kv store {store_id} backed up to path: {kv_backup_path:?}"); diff --git a/server/tests/control_trigger.rs b/server/tests/control_trigger.rs index ef4ed63..56ad4b1 100644 --- a/server/tests/control_trigger.rs +++ b/server/tests/control_trigger.rs @@ -41,9 +41,52 @@ fn trigger_consolidate() { } #[test] -#[ignore = "Not supported by sonic_client yet (FIXME)"] fn trigger_backup() { - todo!() + let ctx = start_empty(|command| { + command + .env("SONIC_STORE__FST__GRAPH__CONSOLIDATE_AFTER", "3600") + .env("SONIC_STORE__FST__POOL__INACTIVE_AFTER", "3700") + .env("SONIC_STORE__KV__DATABASE__FLUSH_AFTER", "3600") + .env("SONIC_STORE__KV__POOL__INACTIVE_AFTER", "3700") + }); + + let multiplexer = SonicMultiplexer::new().unwrap(); + + let ingest = + SonicChannelIngestBlocking::connect(ctx.addr, "SecretPassword", &multiplexer).unwrap(); + let control = + SonicChannelControlBlocking::connect(ctx.addr, "SecretPassword", &multiplexer).unwrap(); + + let backup_path = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")) + .join("test-backups") + .join(ctx.id.to_string()); + if backup_path.exists() { + std::fs::remove_dir_all(&backup_path).unwrap(); + } + + // NOTE: Nothing was pushed yet, so the store directories do not exist. + () = control + .trigger_backup(backup_path.to_str().unwrap()) + .unwrap(); + + // NOTE: Pushing opens the KV store, which then stays open in the pool + // while the backup runs, and leaves the FST changes pending. + ingest + .push("collection", "bucket", "object", "foo bar") + .unwrap(); + + () = control + .trigger_backup(backup_path.to_str().unwrap()) + .unwrap(); + + let kv_backups = std::fs::read_dir(backup_path.join("kv")) + .unwrap() + .map(|entry| entry.unwrap().path()) + .collect::>(); + assert_eq!(kv_backups.len(), 1); + assert!(std::fs::read_dir(&kv_backups[0]).unwrap().next().is_some()); + + std::fs::remove_dir_all(&backup_path).unwrap(); } #[test]