Upstream valeriansaliou/sonic state of the touched files before 6afa0bb3c3

core/src/store/fst/backup.rs

@@ -0,0 +1,242 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use std::fs::File;
+use std::path::Path;
+use std::{fs, io};
+
+use fst::Streamer as _;
+
+use super::{FstStoreId, FstStorePathMode, FstStorePool};
+
+impl FstStorePool {
+    pub fn backup(&self, path: &Path) -> Result<(), io::Error> {
+        tracing::debug!("backing up all fst stores to path: {path:?}");
+
+        // Create backup directory (full path)
+        fs::create_dir_all(path)?;
+
+        // Proceed dump action (backup)
+        self.dump_action(
+            "backup",
+            FstStorePathMode::Permanent,
+            &self.fst_store_config.path,
+            path,
+            &Self::backup_item,
+        )
+    }
+
+    pub fn restore(&self, path: &Path) -> Result<(), io::Error> {
+        tracing::debug!("restoring all fst stores from path: {path:?}");
+
+        // Proceed dump action (restore)
+        self.dump_action(
+            "restore",
+            FstStorePathMode::Backup,
+            path,
+            &self.fst_store_config.path,
+            &Self::restore_item,
+        )
+    }
+
+    #[allow(clippy::type_complexity)]
+    fn dump_action(
+        &self,
+        action: &str,
+        path_mode: FstStorePathMode,
+        read_path: &Path,
+        write_path: &Path,
+        fn_item: &dyn Fn(&Self, &Path, &Path, &str, &str) -> Result<(), io::Error>,
+    ) -> Result<(), io::Error> {
+        let fst_extension = path_mode.extension();
+
+        // Iterate on FST collections.
+        'collections: for collection_entry in fs::read_dir(read_path)? {
+            let collection_entry = collection_entry?;
+
+            // Actual collection found?
+            let file_type = collection_entry.file_type()?;
+            if !file_type.is_dir() {
+                tracing::trace!(
+                    ?file_type,
+                    "Found non-directory entry in {read_path:?}, ignoring"
+                );
+                continue 'collections;
+            }
+
+            let file_name = collection_entry.file_name();
+            let Some(collection_hash) = file_name.to_str() else {
+                tracing::warn!(
+                    file_name_bytes = ?file_name.as_encoded_bytes(),
+                    "Found invalid entry in {read_path:?}, ignoring"
+                );
+                continue 'collections;
+            };
+
+            tracing::debug!("fst collection ongoing {action}: {collection_hash}");
+
+            // Create write folder for collection.
+            fs::create_dir_all(write_path.join(&collection_hash))?;
+
+            // Iterate on FST collection buckets.
+            let buckets_path = read_path.join(&collection_hash);
+            'buckets: for bucket_entry in fs::read_dir(&buckets_path)? {
+                let bucket_entry = bucket_entry?;
+
+                // Actual bucket found?
+                let file_type = collection_entry.file_type()?;
+                if !file_type.is_file() {
+                    tracing::trace!(
+                        ?file_type,
+                        "Found non-file entry in {buckets_path:?}, ignoring"
+                    );
+                    continue 'buckets;
+                }
+
+                let file_name = collection_entry.file_name();
+                let Some(bucket_file_name) = file_name.to_str() else {
+                    tracing::warn!(
+                        file_name_bytes = ?file_name.as_encoded_bytes(),
+                        "Found invalid entry in {buckets_path:?}, ignoring"
+                    );
+                    continue 'buckets;
+                };
+
+                let Some(bucket_hash) = bucket_file_name.strip_suffix(fst_extension) else {
+                    tracing::trace!(
+                        file_name_bytes = ?file_name.as_encoded_bytes(),
+                        "Ignoring {bucket_file_name:?}: wrong extension (expected: {fst_extension:?})"
+                    );
+                    continue 'buckets;
+                };
+
+                tracing::debug!("fst bucket ongoing {action}: {collection_hash}/{bucket_hash}");
+
+                fn_item(
+                    self,
+                    write_path,
+                    &bucket_entry.path(),
+                    collection_hash,
+                    bucket_hash,
+                )?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn backup_item(
+        &self,
+        backup_path: &Path,
+        _origin_path: &Path,
+        collection_hash: &str,
+        bucket_hash: &str,
+    ) -> Result<(), io::Error> {
+        use io::Write as _;
+
+        // Acquire access lock (in blocking write mode), and reference it in context
+        // Notice: this prevents store to be acquired from any context.
+        let _access = self.graph_access_lock.write().unwrap();
+
+        // Generate path to FST backup.
+        let fst_backup_path = {
+            let ext = FstStorePathMode::Backup.extension();
+            assert!(ext.starts_with("."));
+            backup_path
+                .join(collection_hash)
+                .join(format!("{bucket_hash}{ext}"))
+        };
+
+        let store_id = FstStoreId::try_from_hex(collection_hash, bucket_hash)?;
+
+        tracing::debug!("fst store {store_id} backing up to path: {fst_backup_path:?}");
+
+        // Erase any previously-existing FST backup.
+        fs::remove_file(&fst_backup_path).ok();
+
+        // Stream actual FST data to FST backup.
+        let backup_fst_file = File::create(&fst_backup_path)?;
+        let mut backup_fst_writer = io::BufWriter::new(backup_fst_file);
+
+        let origin_fst = (self.open(store_id))
+            .map_err(|error| io::Error::other(format!("Graph open failure: {error:?}")))?;
+
+        let mut origin_fst_stream = origin_fst.stream();
+
+        let mut count_words = 0;
+        while let Some(word) = origin_fst_stream.next() {
+            count_words += 1;
+
+            // Write word, and append a new line.
+            backup_fst_writer.write_all(word)?;
+            backup_fst_writer.write_all(b"\n")?;
+        }
+
+        tracing::info!(
+            "fst store {store_id} backed up to path: {fst_backup_path:?} ({count_words} words)"
+        );
+
+        Ok(())
+    }
+
+    fn restore_item(
+        &self,
+        _backup_path: &Path,
+        origin_path: &Path,
+        collection_name: &str,
+        bucket_name: &str,
+    ) -> Result<(), io::Error> {
+        use io::BufRead as _;
+
+        // Acquire access lock (in blocking write mode) to prevent store from
+        // being acquired from any context.
+        let _access = self.graph_access_lock.write().unwrap();
+
+        let store_id = FstStoreId::try_from_hex(collection_name, bucket_name)?;
+
+        tracing::debug!("fst store {store_id} restoring from path: {origin_path:?}");
+
+        // Force a FST store close.
+        self.close(store_id);
+
+        // Generate path to FST.
+        let fst_path = self
+            .fst_store_config
+            .store_path(store_id, FstStorePathMode::Permanent);
+
+        // Remove existing FST data?
+        if fst_path.exists() {
+            fs::remove_file(&fst_path)?;
+        }
+
+        // Stream backup words to restored FST.
+        let fst_writer = io::BufWriter::new(File::create(&fst_path)?);
+        let fst_backup_reader = io::BufReader::new(File::open(&origin_path)?);
+
+        let mut fst_builder = fst::SetBuilder::new(fst_writer).map_err(|error| {
+            io::Error::other(format!("Graph restore builder failure: {error:?}"))
+        })?;
+
+        for word in fst_backup_reader.lines() {
+            let word = word?;
+
+            fst_builder.insert(word).map_err(|error| {
+                io::Error::other(format!("Graph restore word insert failure: {error:?}"))
+            })?;
+        }
+
+        fst_builder.finish().map_err(|error| {
+            io::Error::other(format!("Graph restore finish failure: {error:?}"))
+        })?;
+
+        tracing::info!(
+            "fst store {store_id} restored to path: {fst_path:?} from backup: {origin_path:?}"
+        );
+
+        Ok(())
+    }
+}

core/src/store/kv/backup.rs

@@ -0,0 +1,170 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use std::path::Path;
+use std::{fs, io};
+
+use rocksdb::backup::{
+    BackupEngine as DBBackupEngine, BackupEngineOptions as DBBackupEngineOptions,
+    RestoreOptions as DBRestoreOptions,
+};
+
+use super::{KvStoreId, KvStorePool};
+
+impl KvStorePool {
+    pub fn backup(&self, path: &Path) -> Result<(), io::Error> {
+        tracing::debug!("backing up all kv stores to path: {path:?}");
+
+        // Create backup directory (full path)
+        fs::create_dir_all(path)?;
+
+        // Proceed dump action (backup)
+        self.dump_action(
+            "backup",
+            &self.kv_store_config.path,
+            path,
+            &Self::backup_item,
+        )
+    }
+
+    pub fn restore(&self, path: &Path) -> Result<(), io::Error> {
+        tracing::debug!("restoring all kv stores from path: {path:?}");
+
+        // Proceed dump action (restore)
+        self.dump_action(
+            "restore",
+            path,
+            &self.kv_store_config.path,
+            &Self::restore_item,
+        )
+    }
+
+    #[allow(clippy::type_complexity)]
+    fn dump_action(
+        &self,
+        action: &str,
+        read_path: &Path,
+        write_path: &Path,
+        fn_item: &dyn Fn(&Self, &Path, &Path, &str) -> Result<(), io::Error>,
+    ) -> Result<(), io::Error> {
+        // Iterate on KV collections.
+        for entry in fs::read_dir(read_path)? {
+            let Ok(collection) = entry else {
+                continue;
+            };
+
+            // Actual collection found?
+            if !collection.file_type().is_ok_and(|f| f.is_dir()) {
+                continue;
+            }
+
+            if let Some(collection_hash) = collection.file_name().to_str() {
+                tracing::debug!("kv collection ongoing {action}: {collection_hash}");
+
+                fn_item(self, write_path, &collection.path(), collection_hash)?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn backup_item(
+        &self,
+        backup_path: &Path,
+        _origin_path: &Path,
+        collection_hash: &str,
+    ) -> Result<(), io::Error> {
+        // Acquire access lock (in blocking write mode), and reference it in context
+        // Notice: this prevents store to be acquired from any context
+        let _access = self.store_access_lock.write().unwrap();
+
+        // Generate path to KV backup
+        let kv_backup_path = backup_path.join(collection_hash);
+
+        let store_id = KvStoreId::try_from_hex(collection_hash)?;
+
+        tracing::debug!("kv store {store_id} backing up to path: {kv_backup_path:?}");
+
+        // Erase any previously-existing KV backup
+        if kv_backup_path.exists() {
+            fs::remove_dir_all(&kv_backup_path)?;
+        }
+
+        // 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"))?;
+        let kv_backup_environment = rocksdb::Env::new()
+            .map_err(|_| io::Error::other("backup engine environment acquire failure"))?;
+
+        let mut kv_backup_engine = DBBackupEngine::open(&kv_backup_options, &kv_backup_environment)
+            .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"))?;
+
+        tracing::info!("kv store {store_id} backed up to path: {kv_backup_path:?}");
+
+        Ok(())
+    }
+
+    fn restore_item(
+        &self,
+        _backup_path: &Path,
+        origin_path: &Path,
+        collection_hash: &str,
+    ) -> Result<(), io::Error> {
+        // Acquire access lock (in blocking write mode), and reference it in context
+        // Notice: this prevents store to be acquired from any context
+        let _access = self.store_access_lock.write().unwrap();
+
+        let store_id = KvStoreId::try_from_hex(collection_hash)?;
+
+        tracing::debug!("kv store {store_id} restoring from path: {origin_path:?}");
+
+        // Force a KV store close
+        self.close(store_id, None);
+
+        // Generate path to KV
+        let kv_path = self.kv_store_config.store_path(store_id);
+
+        // Remove existing KV database data?
+        if kv_path.exists() {
+            fs::remove_dir_all(&kv_path)?;
+        }
+
+        // Create KV folder for collection
+        fs::create_dir_all(&kv_path)?;
+
+        // Initialize KV database backup engine
+        let kv_backup_options = DBBackupEngineOptions::new(&origin_path)
+            .map_err(|_| io::Error::other("backup engine options acquire failure"))?;
+        let kv_backup_environment = rocksdb::Env::new()
+            .map_err(|_| io::Error::other("backup engine environment acquire failure"))?;
+
+        let mut kv_backup_engine = DBBackupEngine::open(&kv_backup_options, &kv_backup_environment)
+            .map_err(|_| io::Error::other("backup engine failure"))?;
+
+        kv_backup_engine
+            .restore_from_latest_backup(&kv_path, &kv_path, &DBRestoreOptions::default())
+            .map_err(|_| io::Error::other("database restore failure"))?;
+
+        tracing::info!(
+            "kv store {store_id} restored to path: {kv_path:?} from backup: {origin_path:?}"
+        );
+
+        Ok(())
+    }
+}

server/tests/control_trigger.rs

@@ -0,0 +1,53 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2026, Rémi Bardon <remi@remibardon.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+mod common;
+
+use crate::common::prelude::*;
+
+#[test]
+fn trigger_consolidate() {
+    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 search =
+        SonicChannelSearchBlocking::connect(ctx.addr, "SecretPassword", &multiplexer).unwrap();
+
+    ingest
+        .push("collection", "bucket", "object", "foo bar")
+        .unwrap();
+
+    let terms = search.list("collection", "bucket").unwrap();
+    assert_eq!(terms.len(), 0);
+
+    () = control.trigger_consolidate().unwrap();
+
+    let terms = search.list("collection", "bucket").unwrap();
+    assert_eq!(terms.len(), 2);
+}
+
+#[test]
+#[ignore = "Not supported by sonic_client yet (FIXME)"]
+fn trigger_backup() {
+    todo!()
+}
+
+#[test]
+#[ignore = "Not supported by sonic_client yet (FIXME)"]
+fn trigger_restore() {
+    todo!()
+}