Add backup + restore commands (need to be implemented)

Signed-off-by: Valerian Saliou <valerian@valeriansaliou.name>

Author: valeriansaliou

PROTOCOL.md

@@ -169,15 +169,16 @@ _The Sonic Channel Control mode is used for administration purposes. Once in thi
 
 **➡️ Available commands:**
 
-* `TRIGGER`: trigger an action (syntax: `TRIGGER [<action>]?`; time complexity: `O(1)`)
+* `TRIGGER`: trigger an action (syntax: `TRIGGER [<action>]? [<data>]?`; time complexity: `O(1)`)
 * `INFO`: get server information (syntax: `INFO`; time complexity: `O(1)`)
 * `PING`: ping server (syntax: `PING`; time complexity: `O(1)`)
 * `HELP`: show help (syntax: `HELP [<manual>]?`; time complexity: `O(1)`)
 * `QUIT`: stop connection (syntax: `QUIT`; time complexity: `O(1)`)
 
 **⏩ Syntax terminology:**
 
-* `<action>`: action to be triggered (available actions: `consolidate`);
+* `<action>`: action to be triggered (available actions: `consolidate`, `backup`, `restore`);
+* `<data>`: additional data to provide to the action (required for: `backup`, `restore`);
 * `<manual>`: help manual to be shown (available manuals: `commands`);
 
 **⬇️ Control flow example (via `telnet`):**

src/channel/command.rs

@@ -7,6 +7,7 @@
 use hashbrown::HashMap;
 use rand::distributions::Alphanumeric;
 use rand::{thread_rng, Rng};
+use std::path::Path;
 use std::str::{self, SplitWhitespace};
 use std::vec::Vec;
 
@@ -15,6 +16,7 @@ use super::statistics::ChannelStatistics;
 use crate::query::builder::{QueryBuilder, QueryBuilderResult};
 use crate::query::types::{QueryGenericLang, QuerySearchLimit, QuerySearchOffset};
 use crate::store::fst::StoreFSTPool;
+use crate::store::kv::StoreKVPool;
 use crate::store::operation::StoreOperationDispatch;
 use crate::APP_CONF;
 
@@ -64,7 +66,8 @@ lazy_static! {
         vec!["PUSH", "POP", "COUNT", "FLUSHC", "FLUSHB", "FLUSHO", "PING", "HELP", "QUIT"];
     pub static ref COMMANDS_MODE_CONTROL: Vec<&'static str> =
         vec!["TRIGGER", "INFO", "PING", "HELP", "QUIT"];
-    pub static ref CONTROL_TRIGGER_ACTIONS: Vec<&'static str> = vec!["consolidate"];
+    pub static ref CONTROL_TRIGGER_ACTIONS: Vec<&'static str> =
+        vec!["consolidate", "backup", "restore"];
     static ref MANUAL_MODE_SEARCH: HashMap<&'static str, &'static Vec<&'static str>> =
         [("commands", &*COMMANDS_MODE_SEARCH)]
             .iter()
@@ -776,26 +779,60 @@ impl ChannelCommandIngest {
 
 impl ChannelCommandControl {
     pub fn dispatch_trigger(mut parts: SplitWhitespace) -> ChannelResult {
-        match (parts.next(), parts.next()) {
-            (None, _) => Ok(vec![ChannelCommandResponse::Result(format!(
+        match (parts.next(), parts.next(), parts.next()) {
+            (None, _, _) => Ok(vec![ChannelCommandResponse::Result(format!(
                 "actions({})",
                 CONTROL_TRIGGER_ACTIONS.join(", ")
             ))]),
-            (Some(action_key), next_part) => {
-                if next_part.is_none() {
-                    let action_key_lower = action_key.to_lowercase();
+            (Some(action_key), data_part, last_part) => {
+                let action_key_lower = action_key.to_lowercase();
 
-                    match action_key_lower.as_str() {
-                        "consolidate" => {
+                match action_key_lower.as_str() {
+                    "consolidate" => {
+                        if data_part.is_none() {
                             // Force a FST consolidate
                             StoreFSTPool::consolidate(true);
 
                             Ok(vec![ChannelCommandResponse::Ok])
+                        } else {
+                            Err(ChannelCommandError::InvalidFormat("TRIGGER consolidate"))
                         }
-                        _ => Err(ChannelCommandError::NotFound),
                     }
-                } else {
-                    Err(ChannelCommandError::InvalidFormat("TRIGGER [<action>]?"))
+                    "backup" => {
+                        match (data_part, last_part) {
+                            (Some(path), None) => {
+                                // Proceed KV + FST backup
+                                let path = Path::new(path);
+
+                                if StoreKVPool::backup(path).is_ok()
+                                    && StoreFSTPool::backup(path).is_ok()
+                                {
+                                    Ok(vec![ChannelCommandResponse::Ok])
+                                } else {
+                                    Err(ChannelCommandError::InternalError)
+                                }
+                            }
+                            _ => Err(ChannelCommandError::InvalidFormat("TRIGGER backup <path>")),
+                        }
+                    }
+                    "restore" => {
+                        match (data_part, last_part) {
+                            (Some(path), None) => {
+                                // Proceed KV + FST restore
+                                let path = Path::new(path);
+
+                                if StoreKVPool::restore(path).is_ok()
+                                    && StoreFSTPool::restore(path).is_ok()
+                                {
+                                    Ok(vec![ChannelCommandResponse::Ok])
+                                } else {
+                                    Err(ChannelCommandError::InternalError)
+                                }
+                            }
+                            _ => Err(ChannelCommandError::InvalidFormat("TRIGGER restore <path>")),
+                        }
+                    }
+                    _ => Err(ChannelCommandError::NotFound),
                 }
             }
         }

src/store/fst.rs

@@ -18,7 +18,7 @@ use std::fmt;
 use std::fs::{self, File};
 use std::io::BufWriter;
 use std::iter::FromIterator;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::str;
 use std::sync::{Arc, Mutex, RwLock};
 use std::thread;
@@ -137,6 +137,28 @@ impl StoreFSTPool {
         )
     }
 
+    pub fn backup(path: &Path) -> Result<(), ()> {
+        debug!("backing up all fst stores to path: {:?}", path);
+
+        // TODO: read all FSTs one-by-one
+        // TODO: stream each FST word to a .fst.bck file, for each fst
+
+        Ok(())
+    }
+
+    pub fn restore(path: &Path) -> Result<(), ()> {
+        debug!("restoring all fst stores from path: {:?}", path);
+
+        // TODO: <iteratively do this>:
+        // -- TODO: iteratively lock; for each FST file being processed (access + write locks; \
+        //      same as destroy lock)
+        // -- TODO: read all FST backups one-by-one
+        // -- TODO: re-build actual ordered FST
+        // -- TODO: close opened FST + nuke their consolidate (if they appear in the backup)
+
+        Ok(())
+    }
+
     pub fn consolidate(force: bool) {
         debug!("scanning for fst store pool items to consolidate");
 

src/store/kv.rs

@@ -13,7 +13,7 @@ use rocksdb::{
 use std::fmt;
 use std::fs;
 use std::io::Cursor;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::SystemTime;
 use std::vec::Drain;
@@ -122,6 +122,26 @@ impl StoreKVPool {
             &*STORE_ACCESS_LOCK,
         )
     }
+
+    pub fn backup(path: &Path) -> Result<(), ()> {
+        debug!("backing up all kv stores to path: {:?}", path);
+
+        // TODO
+
+        Ok(())
+    }
+
+    pub fn restore(path: &Path) -> Result<(), ()> {
+        debug!("restoring all kv stores from path: {:?}", path);
+
+        // TODO: <iteratively do this>:
+        // -- TODO: lock for each KV dump being processed (access + write locks; same as destroy \
+        //      lock)
+        // -- TODO: read all KV dumps one-by-one, and read by RocksDB manager
+        // -- TODO: close opened KV (if they appear in the backup)
+
+        Ok(())
+    }
 }
 
 impl StoreGenericPool<StoreKVKey, StoreKV, StoreKVBuilder> for StoreKVPool {}