Implemented distributed deployment functionality:

1. Each instance synchronizes the latest data at intervals defined by
   the configured mounts_monitor_interval time, with a default value of
   5 seconds.
2. For the mysql storage backend, used GET_LOCK and RELEASE_LOCK to implement
   distributed locks
3. For the file storage backend, used lockfile to implement distributed locks.
4. Removed unnecessary mutex locks in mysql storage backend.

Author: wa5i

.github/workflows/rust.yml

@@ -100,21 +100,17 @@ jobs:
         my-cnf: |
           skip-ssl
           port=3306
-    - name: Set macOS Environment Variables
-      if: runner.os == 'macOS'
-      env:
-        MACOSX_DEPLOYMENT_TARGET: "11.0"
-      run: |
-          export MACOSX_DEPLOYMENT_TARGET=11.0
     - name: install diesel_cli
       run: cargo install diesel_cli --no-default-features --features mysql
     - name: init database
       run: diesel setup --database-url mysql://root:password@127.0.0.1:3306/vault
-    - name: Build
-      run: cargo build --features storage_mysql --verbose
     - name: ulimit -n
       run: ulimit -n 65535
-    - name: Run tests
+    - name: Run tests for macOS
+      if: runner.os == 'macOS'
+      run: cargo test --features storage_mysql --lib --bins --tests --verbose
+    - name: Run tests for non-macOS
+      if: runner.os != 'macOS'
       run: cargo test --features storage_mysql --verbose
     - name: Build crate doc
       run: cargo doc --no-deps

Cargo.toml

@@ -78,6 +78,7 @@ stretto = "0.8"
 priority-queue = "2.1"
 crossbeam-channel = "0.5"
 maybe-async = { version = "0.2", optional = false }
+lockfile = "0.4.0"
 
 # optional dependencies
 openssl = { version = "*", optional = true }

src/cli/command/server.rs

@@ -150,12 +150,10 @@ impl Server {
 
         let backend = storage::new_backend(storage.stype.as_str(), &storage.config).unwrap();
 
-        let barrier = storage::barrier_aes_gcm::AESGCMBarrier::new(Arc::clone(&backend));
-
         let metrics_manager = Arc::new(RwLock::new(MetricsManager::new(config.collection_interval)));
         let system_metrics = Arc::clone(&metrics_manager.read().unwrap().system_metrics);
 
-        let core = Arc::new(RwLock::new(Core { physical: backend, barrier: Arc::new(barrier), ..Default::default() }));
+        let core = Arc::new(RwLock::new(Core::new(backend)));
 
         {
             let mut c = core.write()?;

src/cli/config.rs

@@ -40,6 +40,8 @@ pub struct Config {
     pub collection_interval: u64,
     #[serde(default = "default_hmac_level")]
     pub mount_entry_hmac_level: MountEntryHMACLevel,
+    #[serde(default = "default_mounts_monitor_interval")]
+    pub mounts_monitor_interval: u64,
 }
 
 #[derive(Debug, Copy, Clone, Serialize, Deserialize, PartialEq)]
@@ -58,6 +60,10 @@ fn default_collection_interval() -> u64 {
     15
 }
 
+fn default_mounts_monitor_interval() -> u64 {
+    5
+}
+
 /// A struct that contains several configurable options for networking stuffs
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Listener {

src/core.rs

@@ -8,9 +8,8 @@
 //! of RustyVault.
 
 use std::{
-    collections::HashMap,
     ops::{Deref, DerefMut},
-    sync::{Arc, Mutex, RwLock},
+    sync::{Arc, RwLock},
 };
 
 use as_any::Downcast;
@@ -30,7 +29,9 @@ use crate::{
         pki::PkiModule,
         policy::PolicyModule,
     },
-    mount::MountTable,
+    mount::{
+        MountTable, MountsMonitor, MountsRouter, CORE_MOUNT_CONFIG_PATH, LOGICAL_BARRIER_PREFIX, SYSTEM_BARRIER_PREFIX,
+    },
     router::Router,
     shamir::{ShamirSecret, SHAMIR_OVERHEAD},
     storage::{
@@ -70,48 +71,79 @@ pub struct Core {
     pub physical: Arc<dyn PhysicalBackend>,
     pub barrier: Arc<dyn SecurityBarrier>,
     pub system_view: Option<Arc<BarrierView>>,
-    pub mounts: Arc<MountTable>,
-    pub router: Arc<Router>,
     pub handlers: RwLock<Vec<Arc<dyn Handler>>>,
     pub auth_handlers: Arc<RwLock<Vec<Arc<dyn AuthHandler>>>>,
-    pub logical_backends: Mutex<HashMap<String, Arc<LogicalBackendNewFunc>>>,
+    pub router: Arc<Router>,
+    pub mounts_router: Arc<MountsRouter>,
     pub module_manager: ModuleManager,
     pub sealed: bool,
     pub unseal_key_shares: Vec<Vec<u8>>,
     pub hmac_key: Vec<u8>,
     pub mount_entry_hmac_level: MountEntryHMACLevel,
+    pub mounts_monitor: Option<MountsMonitor>,
+    pub mounts_monitor_interval: u64,
 }
 
 impl Default for Core {
     fn default() -> Self {
         let backend: Arc<dyn PhysicalBackend> = Arc::new(physical::mock::MockBackend::new());
-        let barrier = barrier_aes_gcm::AESGCMBarrier::new(Arc::clone(&backend));
+        let barrier = Arc::new(barrier_aes_gcm::AESGCMBarrier::new(Arc::clone(&backend)));
+        let barrier_cloned = Arc::clone(&barrier);
         let router = Arc::new(Router::new());
 
         Core {
             self_ref: None,
             physical: backend,
-            barrier: Arc::new(barrier),
+            barrier: barrier_cloned,
             system_view: None,
-            mounts: Arc::new(MountTable::new()),
             router: Arc::clone(&router),
+            mounts_router: Arc::new(MountsRouter::new(
+                Arc::new(MountTable::new(CORE_MOUNT_CONFIG_PATH)),
+                Arc::clone(&router),
+                barrier,
+                LOGICAL_BARRIER_PREFIX,
+                "",
+            )),
             handlers: RwLock::new(vec![router]),
             auth_handlers: Arc::new(RwLock::new(Vec::new())),
-            logical_backends: Mutex::new(HashMap::new()),
             module_manager: ModuleManager::new(),
             sealed: true,
             unseal_key_shares: Vec::new(),
             hmac_key: Vec::new(),
             mount_entry_hmac_level: MountEntryHMACLevel::None,
+            mounts_monitor: None,
+            mounts_monitor_interval: 5,
         }
     }
 }
 
 #[maybe_async::maybe_async]
 impl Core {
+    pub fn new(backend: Arc<dyn PhysicalBackend>) -> Self {
+        let barrier = Arc::new(barrier_aes_gcm::AESGCMBarrier::new(Arc::clone(&backend)));
+        let barrier_cloned = Arc::clone(&barrier);
+        let router = Arc::new(Router::new());
+
+        Core {
+            physical: backend,
+            barrier: barrier_cloned,
+            router: Arc::clone(&router),
+            mounts_router: Arc::new(MountsRouter::new(
+                Arc::new(MountTable::new(CORE_MOUNT_CONFIG_PATH)),
+                Arc::clone(&router),
+                barrier,
+                LOGICAL_BARRIER_PREFIX,
+                "",
+            )),
+            handlers: RwLock::new(vec![router]),
+            ..Default::default()
+        }
+    }
+
     pub fn config(&mut self, core: Arc<RwLock<Core>>, config: Option<&Config>) -> Result<(), RvError> {
         if let Some(conf) = config {
             self.mount_entry_hmac_level = conf.mount_entry_hmac_level;
+            self.mounts_monitor_interval = conf.mounts_monitor_interval;
         }
 
         self.module_manager.set_default_modules(Arc::clone(&core))?;
@@ -141,6 +173,8 @@ impl Core {
         let cert_module = CertModule::new(self);
         self.module_manager.add_module(Arc::new(RwLock::new(Box::new(cert_module))))?;
 
+        self.mounts_monitor = Some(MountsMonitor::new(core, self.mounts_monitor_interval));
+
         let handlers = { self.handlers.read()?.clone() };
         for handler in handlers.iter() {
             match handler.post_config(self, config) {
@@ -246,27 +280,15 @@ impl Core {
     }
 
     pub fn get_logical_backend(&self, logical_type: &str) -> Result<Arc<LogicalBackendNewFunc>, RvError> {
-        let logical_backends = self.logical_backends.lock().unwrap();
-        if let Some(backend) = logical_backends.get(logical_type) {
-            Ok(backend.clone())
-        } else {
-            Err(RvError::ErrCoreLogicalBackendNoExist)
-        }
+        self.mounts_router.get_backend(logical_type)
     }
 
     pub fn add_logical_backend(&self, logical_type: &str, backend: Arc<LogicalBackendNewFunc>) -> Result<(), RvError> {
-        let mut logical_backends = self.logical_backends.lock().unwrap();
-        if logical_backends.contains_key(logical_type) {
-            return Err(RvError::ErrCoreLogicalBackendExist);
-        }
-        logical_backends.insert(logical_type.to_string(), backend);
-        Ok(())
+        self.mounts_router.add_backend(logical_type, backend)
     }
 
     pub fn delete_logical_backend(&self, logical_type: &str) -> Result<(), RvError> {
-        let mut logical_backends = self.logical_backends.lock().unwrap();
-        logical_backends.remove(logical_type);
-        Ok(())
+        self.mounts_router.delete_backend(logical_type)
     }
 
     pub fn add_handler(&self, handler: Arc<dyn Handler>) -> Result<(), RvError> {
@@ -397,16 +419,27 @@ impl Core {
 
         // Perform initial setup
         self.hmac_key = self.barrier.derive_hmac_key()?;
-        self.mounts.load_or_default(self.barrier.as_storage(), Some(&self.hmac_key), self.mount_entry_hmac_level)?;
+        self.mounts_router.load_or_default(
+            self.barrier.as_storage(),
+            Some(&self.hmac_key),
+            self.mount_entry_hmac_level,
+        )?;
 
-        self.setup_mounts()?;
+        self.mounts_router.setup(self.self_ref.as_ref().unwrap())?;
+
+        self.system_view = Some(Arc::new(BarrierView::new(self.barrier.clone(), SYSTEM_BARRIER_PREFIX)));
 
         self.module_manager.init(self)?;
 
+        self.mounts_monitor.as_ref().unwrap().add_mounts_router(self.mounts_router.clone());
+        self.mounts_monitor.as_mut().unwrap().start();
+
         Ok(())
     }
 
     fn pre_seal(&mut self) -> Result<(), RvError> {
+        self.mounts_monitor.as_ref().unwrap().remove_mounts_router(self.mounts_router.clone());
+        self.mounts_monitor.as_mut().unwrap().stop();
         self.module_manager.cleanup(self)?;
         self.unload_mounts()?;
         Ok(())
@@ -546,10 +579,10 @@ impl Core {
 
 #[cfg(test)]
 mod test {
-    use crate::test_utils::test_rusty_vault_init;
+    use crate::test_utils::init_test_rusty_vault;
 
     #[test]
     fn test_core_init() {
-        let _ = test_rusty_vault_init("test_core_init");
+        let _ = init_test_rusty_vault("test_core_init");
     }
 }

src/errors.rs

@@ -173,6 +173,12 @@ pub enum RvError {
     ErrCredentailInvalid,
     #[error("Credentail is not config.")]
     ErrCredentailNotConfig,
+    #[error("Storage backend doesn't require a lock.")]
+    ErrStorageBackendLockless,
+    #[error("Storage backend lock failed.")]
+    ErrStorageBackendLockFailed,
+    #[error("Storage backend unlock failed.")]
+    ErrStorageBackendUnlockFailed,
     #[error("Some IO error happened, {:?}", .source)]
     IO {
         #[from]
@@ -295,6 +301,12 @@ pub enum RvError {
         source: std::string::FromUtf8Error,
     },
 
+    #[error("Some lockfile error happened, {:?}", .source)]
+    LockfileError {
+        #[from]
+        source: lockfile::Error,
+    },
+
     /// Database Errors Begin
     ///
     #[error("Database type is not support now. Please try postgressql or mysql again.")]

src/logical/backend.rs

@@ -261,7 +261,7 @@ mod test {
     use crate::{
         logical::{field::FieldTrait, Field, FieldType, PathOperation},
         new_fields, new_fields_internal, new_path, new_path_internal, new_secret, new_secret_internal, storage,
-        test_utils::test_backend,
+        test_utils::new_test_backend,
     };
 
     struct MyTest;
@@ -304,7 +304,7 @@ mod test {
 
     #[test]
     fn test_logical_backend_api() {
-        let backend = test_backend("test_logical_backend_api");
+        let backend = new_test_backend("test_logical_backend_api");
 
         let t = MyTest::new();
 
@@ -461,7 +461,7 @@ mod test {
 
     #[test]
     fn test_logical_path_field() {
-        let backend = test_backend("test_logical_path_field");
+        let backend = new_test_backend("test_logical_path_field");
 
         let barrier = storage::barrier_aes_gcm::AESGCMBarrier::new(Arc::clone(&backend));
 

src/modules/auth/expiration.rs

@@ -687,14 +687,14 @@ mod mod_expiration_tests {
         mount::{MountEntry, MOUNT_TABLE_TYPE},
         new_fields, new_fields_internal, new_logical_backend, new_logical_backend_internal, new_path,
         new_path_internal, new_secret, new_secret_internal,
-        test_utils::{test_rusty_vault_init, NoopBackend},
+        test_utils::{init_test_rusty_vault, NoopBackend},
     };
 
     macro_rules! mock_expiration_manager {
         () => {{
             let name = format!("{}_{}", file!(), line!()).replace("/", "_").replace("\\", "_").replace(".", "_");
-            println!("test_rusty_vault_init, name: {}", name);
-            let (_, core) = test_rusty_vault_init(&name);
+            println!("init_test_rusty_vault, name: {}", name);
+            let (_, core) = init_test_rusty_vault(&name);
             let core_cloned = core.clone();
             let core_locked = core_cloned.read().unwrap();