fix(core): Make TRIGGER backup work while stores are in use

`TRIGGER backup` answered `ERR internal_error` in two common cases:

- a KV store was open in the pool (e.g. right after a `PUSH`): the backup
  opened the database a second time, which RocksDB refuses as its `LOCK`
  file is held. The pooled store is now backed up instead, after a flush.
- the KV or FST store directory did not exist yet (nothing ingested, or
  nothing consolidated yet): listing it failed. There is nothing to back up
  in that case.

Also enable the `trigger_backup` test, which covers both cases.

Fixes #274

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",

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:?}");
 

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::<Vec<_>>();
+    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]