diff --git a/crates/kms/src/lib.rs b/crates/kms/src/lib.rs index de97e5349..161c43aff 100644 --- a/crates/kms/src/lib.rs +++ b/crates/kms/src/lib.rs @@ -89,7 +89,7 @@ pub use error::{KmsError, KmsUnavailableError, Result}; pub use manager::KmsManager; pub use service::{DataKey, ObjectEncryptionService}; pub use service_manager::{ - KmsServiceManager, KmsServiceStatus, get_global_encryption_service, get_global_kms_service_manager, + KmsServiceManager, KmsServiceStatus, KmsStartOutcome, get_global_encryption_service, get_global_kms_service_manager, init_global_kms_service_manager, }; pub use types::*; diff --git a/crates/kms/src/service_manager.rs b/crates/kms/src/service_manager.rs index b3a37ca22..d38f343fe 100644 --- a/crates/kms/src/service_manager.rs +++ b/crates/kms/src/service_manager.rs @@ -20,11 +20,12 @@ use crate::error::{KmsError, Result}; use crate::manager::KmsManager; use crate::service::ObjectEncryptionService; use arc_swap::ArcSwap; +use std::future::Future; use std::sync::{ Arc, OnceLock, atomic::{AtomicU64, Ordering}, }; -use tokio::sync::{Mutex, RwLock}; +use tokio::sync::Mutex; use tracing::{debug, error, info, warn}; const LOG_COMPONENT_KMS: &str = "kms"; @@ -44,6 +45,13 @@ pub enum KmsServiceStatus { Error(String), } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum KmsStartOutcome { + Started, + Restarted, + AlreadyRunning, +} + /// Service version information for zero-downtime reconfiguration #[derive(Clone)] struct ServiceVersion { @@ -55,16 +63,17 @@ struct ServiceVersion { manager: Arc, } +#[derive(Clone)] +struct RuntimeState { + config: Option, + status: KmsServiceStatus, + current_service: Option, +} + /// Dynamic KMS service manager with versioned services for zero-downtime reconfiguration pub struct KmsServiceManager { - /// Current service version (if running) - /// Uses ArcSwap for atomic, lock-free service switching - /// This allows instant atomic updates without blocking readers - current_service: ArcSwap>, - /// Current configuration - config: Arc>>, - /// Current status - status: Arc>, + /// Atomically published configuration, status, and current service. + state: ArcSwap, /// Version counter (monotonically increasing) version_counter: Arc, /// Mutex to protect lifecycle operations (start, stop, reconfigure) @@ -76,9 +85,11 @@ impl KmsServiceManager { /// Create a new KMS service manager (not configured) pub fn new() -> Self { Self { - current_service: ArcSwap::from_pointee(None), - config: Arc::new(RwLock::new(None)), - status: Arc::new(RwLock::new(KmsServiceStatus::NotConfigured)), + state: ArcSwap::from_pointee(RuntimeState { + config: None, + status: KmsServiceStatus::NotConfigured, + current_service: None, + }), version_counter: Arc::new(AtomicU64::new(0)), lifecycle_mutex: Arc::new(Mutex::new(())), } @@ -86,39 +97,65 @@ impl KmsServiceManager { /// Get current service status pub async fn get_status(&self) -> KmsServiceStatus { - self.status.read().await.clone() + self.state.load().status.clone() } /// Get current configuration (if any) pub async fn get_config(&self) -> Option { - self.config.read().await.clone() + self.state.load().config.clone() } /// Get configuration for status and management responses without static key material. pub async fn get_redacted_config(&self) -> Option { - let mut config = self.config.read().await.clone()?; + let mut config = self.state.load().config.clone()?; + Self::redact_config(&mut config); + Some(config) + } + + /// Get status and redacted configuration from the same published snapshot. + pub async fn get_redacted_state(&self) -> (KmsServiceStatus, Option) { + let state = self.state.load(); + let mut config = state.config.clone(); + if let Some(config) = &mut config { + Self::redact_config(config); + } + (state.status.clone(), config) + } + + fn redact_config(config: &mut KmsConfig) { if let BackendConfig::Static(static_config) = &mut config.backend_config { use zeroize::Zeroize; static_config.secret_key.zeroize(); } - Some(config) } /// Configure KMS with new configuration pub async fn configure(&self, new_config: KmsConfig) -> Result<()> { + self.configure_with_persistence(new_config, || async { Ok(()) }).await + } + + /// Configure KMS and publish the in-memory state only after persistence succeeds. + /// + /// The persistence callback runs under the lifecycle lock and must not call + /// another lifecycle method on this manager. + pub async fn configure_with_persistence(&self, new_config: KmsConfig, persist: Persist) -> Result<()> + where + Persist: FnOnce() -> PersistFuture, + PersistFuture: Future>, + { new_config.validate()?; - - // Update configuration - { - let mut config = self.config.write().await; - *config = Some(new_config.clone()); - } - - // Update status - { - let mut status = self.status.write().await; - *status = KmsServiceStatus::Configured; + let _guard = self.lifecycle_mutex.lock().await; + if self.state.load().current_service.is_some() { + return Err(KmsError::configuration_error( + "Cannot configure KMS while it is running; use reconfigure instead", + )); } + persist().await?; + self.state.store(Arc::new(RuntimeState { + config: Some(new_config), + status: KmsServiceStatus::Configured, + current_service: None, + })); debug!( event = EVENT_KMS_SERVICE_STATE, @@ -136,19 +173,35 @@ impl KmsServiceManager { self.start_internal().await } + /// Start or restart KMS with the running-state decision serialized with the lifecycle action. + pub async fn start_or_restart(&self, force: bool) -> Result { + let _guard = self.lifecycle_mutex.lock().await; + let running = self.state.load().current_service.is_some(); + if running && !force { + return Ok(KmsStartOutcome::AlreadyRunning); + } + self.start_internal().await?; + Ok(if running { + KmsStartOutcome::Restarted + } else { + KmsStartOutcome::Started + }) + } + /// Internal start implementation (called within lifecycle mutex) async fn start_internal(&self) -> Result<()> { - let config = { - let config_guard = self.config.read().await; - match config_guard.as_ref() { - Some(config) => config.clone(), - None => { - let err_msg = "Cannot start KMS: no configuration provided"; - error!("{}", err_msg); - let mut status = self.status.write().await; - *status = KmsServiceStatus::Error(err_msg.to_string()); - return Err(KmsError::configuration_error(err_msg)); - } + let state = self.state.load_full(); + let config = match state.config.as_ref() { + Some(config) => config.clone(), + None => { + let err_msg = "Cannot start KMS: no configuration provided"; + error!("{}", err_msg); + self.state.store(Arc::new(RuntimeState { + config: None, + status: KmsServiceStatus::Error(err_msg.to_string()), + current_service: None, + })); + return Err(KmsError::configuration_error(err_msg)); } }; @@ -161,17 +214,9 @@ impl KmsServiceManager { "KMS service starting" ); - match self.create_service_version(&config).await { + match self.create_healthy_service_version(&config).await { Ok(service_version) => { - // Atomically update to new service version (lock-free, instant) - // ArcSwap::store() is a true atomic operation using CAS - self.current_service.store(Arc::new(Some(service_version))); - - // Update status - { - let mut status = self.status.write().await; - *status = KmsServiceStatus::Running; - } + self.publish_running(config, service_version); debug!( event = EVENT_KMS_SERVICE_STATE, @@ -185,13 +230,24 @@ impl KmsServiceManager { Err(e) => { let err_msg = format!("Failed to create KMS backend: {e}"); error!("{}", err_msg); - let mut status = self.status.write().await; - *status = KmsServiceStatus::Error(err_msg.clone()); + if state.current_service.is_none() { + self.state.store(Arc::new(RuntimeState { + config: state.config.clone(), + status: KmsServiceStatus::Error(err_msg.clone()), + current_service: None, + })); + } Err(KmsError::backend_error(&err_msg)) } } } + /// Replace the running service without exposing a stopped interval. + pub async fn restart(&self) -> Result<()> { + let _guard = self.lifecycle_mutex.lock().await; + self.start_internal().await + } + /// Stop KMS service /// /// Note: This stops accepting new operations, but existing operations using @@ -213,15 +269,16 @@ impl KmsServiceManager { // Atomically clear current service version (lock-free, instant) // Note: Existing Arc references will keep the service alive until operations complete - self.current_service.store(Arc::new(None)); - - // Update status (keep configuration) - { - let mut status = self.status.write().await; - if !matches!(*status, KmsServiceStatus::NotConfigured) { - *status = KmsServiceStatus::Configured; - } - } + let state = self.state.load_full(); + self.state.store(Arc::new(RuntimeState { + config: state.config.clone(), + status: if state.config.is_some() { + KmsServiceStatus::Configured + } else { + KmsServiceStatus::NotConfigured + }, + current_service: None, + })); debug!( event = EVENT_KMS_SERVICE_STATE, @@ -244,6 +301,22 @@ impl KmsServiceManager { /// This ensures zero downtime during reconfiguration, even for long-running /// operations like encrypting large files. pub async fn reconfigure(&self, new_config: KmsConfig) -> Result<()> { + self.reconfigure_with_persistence(new_config, || async { Ok(()) }).await + } + + /// Reconfigure KMS after the candidate is healthy and persistence succeeds. + /// + /// The persistence callback runs under the lifecycle lock and must not call + /// another lifecycle method on this manager. + pub async fn reconfigure_with_persistence( + &self, + new_config: KmsConfig, + persist: Persist, + ) -> Result<()> + where + Persist: FnOnce() -> PersistFuture, + PersistFuture: Future>, + { let _guard = self.lifecycle_mutex.lock().await; debug!( @@ -255,29 +328,16 @@ impl KmsServiceManager { ); new_config.validate()?; - // Configure with new config - { - let mut config = self.config.write().await; - *config = Some(new_config.clone()); - } - // Create new service version without stopping old one // This allows existing operations to continue while new operations use new service - match self.create_service_version(&new_config).await { + match self.create_healthy_service_version(&new_config).await { Ok(new_service_version) => { // Get old version for logging (lock-free read) - let old_version = self.current_service.load().as_ref().as_ref().map(|sv| sv.version); + let old_version = self.state.load().current_service.as_ref().map(|sv| sv.version); - // Atomically switch to new service version (lock-free, instant CAS operation) - // This is a true atomic operation - no waiting for locks, instant switch - // Old service will be dropped when no more Arc references exist - self.current_service.store(Arc::new(Some(new_service_version.clone()))); + persist().await?; - // Update status - { - let mut status = self.status.write().await; - *status = KmsServiceStatus::Running; - } + self.publish_running(new_config, new_service_version.clone()); if let Some(old_ver) = old_version { info!( @@ -304,8 +364,6 @@ impl KmsServiceManager { Err(e) => { let err_msg = format!("Failed to reconfigure KMS: {e}"); error!("{}", err_msg); - let mut status = self.status.write().await; - *status = KmsServiceStatus::Error(err_msg.clone()); Err(KmsError::backend_error(&err_msg)) } } @@ -316,7 +374,7 @@ impl KmsServiceManager { /// Returns the manager from the current service version. /// Uses lock-free atomic load for optimal performance. pub async fn get_manager(&self) -> Option> { - self.current_service.load().as_ref().as_ref().map(|sv| sv.manager.clone()) + self.state.load().current_service.as_ref().map(|sv| sv.manager.clone()) } /// Get encryption service (if running) @@ -326,7 +384,7 @@ impl KmsServiceManager { /// This ensures new operations always use the latest service version, /// while existing operations continue using their Arc references. pub async fn get_encryption_service(&self) -> Option> { - self.current_service.load().as_ref().as_ref().map(|sv| sv.service.clone()) + self.state.load().current_service.as_ref().map(|sv| sv.service.clone()) } /// Get current service version number @@ -334,14 +392,16 @@ impl KmsServiceManager { /// Useful for monitoring and debugging. /// Uses lock-free atomic load. pub async fn get_service_version(&self) -> Option { - self.current_service.load().as_ref().as_ref().map(|sv| sv.version) + self.state.load().current_service.as_ref().map(|sv| sv.version) } /// Health check for the KMS service pub async fn health_check(&self) -> Result { - let manager = self.get_manager().await; - match manager { - Some(manager) => { + let checked_state = self.state.load_full(); + match checked_state.current_service.as_ref() { + Some(service_version) => { + let manager = service_version.manager.clone(); + let checked_version = service_version.version; // Perform health check on the backend match manager.health_check().await { Ok(healthy) => { @@ -352,9 +412,8 @@ impl KmsServiceManager { } Err(e) => { error!("KMS health check error: {}", e); - // Update status to error - let mut status = self.status.write().await; - *status = KmsServiceStatus::Error(format!("Health check failed: {e}")); + let _guard = self.lifecycle_mutex.lock().await; + self.mark_health_error_if_current(checked_version, &e); Err(e) } } @@ -413,6 +472,33 @@ impl KmsServiceManager { manager: kms_manager, }) } + + async fn create_healthy_service_version(&self, config: &KmsConfig) -> Result { + let service_version = self.create_service_version(config).await?; + if !service_version.manager.health_check().await? { + return Err(KmsError::backend_error("KMS backend health check failed")); + } + Ok(service_version) + } + + fn publish_running(&self, config: KmsConfig, service_version: ServiceVersion) { + self.state.store(Arc::new(RuntimeState { + config: Some(config), + status: KmsServiceStatus::Running, + current_service: Some(service_version), + })); + } + + fn mark_health_error_if_current(&self, checked_version: u64, error: &KmsError) { + let current = self.state.load_full(); + if current.current_service.as_ref().map(|version| version.version) == Some(checked_version) { + self.state.store(Arc::new(RuntimeState { + config: current.config.clone(), + status: KmsServiceStatus::Error(format!("Health check failed: {error}")), + current_service: current.current_service.clone(), + })); + } + } } impl Default for KmsServiceManager { @@ -445,6 +531,11 @@ pub async fn get_global_encryption_service() -> Option KmsConfig { + KmsConfig::static_kms(key_id.to_string(), BASE64_STANDARD.encode([fill; 32])) + } #[tokio::test] async fn configure_rejects_insecure_development_defaults_before_state_update() { @@ -462,8 +553,6 @@ mod tests { #[tokio::test] async fn redacted_config_omits_static_key_material() { - use base64::Engine as _; - let manager = KmsServiceManager::new(); let encoded_key = base64::engine::general_purpose::STANDARD.encode([0x5au8; 32]); manager @@ -477,4 +566,134 @@ mod tests { }; assert!(static_config.secret_key.is_empty()); } + + #[tokio::test] + async fn configure_persistence_failure_leaves_state_unchanged() { + let manager = KmsServiceManager::new(); + + let result = manager + .configure_with_persistence(static_config("key-a", 0x11), || async { Err(KmsError::backend_error("persist failed")) }) + .await; + + assert!(result.is_err()); + assert_eq!(manager.get_status().await, KmsServiceStatus::NotConfigured); + assert!(manager.get_config().await.is_none()); + assert!(manager.get_encryption_service().await.is_none()); + } + + #[tokio::test] + async fn configure_rejects_running_service_without_changing_snapshot() { + let manager = KmsServiceManager::new(); + manager.configure(static_config("key-a", 0x11)).await.expect("configure"); + manager.start().await.expect("start"); + let version = manager.get_service_version().await; + + let result = manager.configure(static_config("key-b", 0x22)).await; + + assert!(result.is_err()); + assert_eq!(manager.get_status().await, KmsServiceStatus::Running); + assert_eq!(manager.get_service_version().await, version); + assert_eq!( + manager.get_config().await.and_then(|config| config.default_key_id), + Some("key-a".to_string()) + ); + } + + #[tokio::test] + async fn reconfigure_persistence_failure_keeps_old_running_snapshot() { + let manager = KmsServiceManager::new(); + manager.configure(static_config("key-a", 0x11)).await.expect("configure"); + manager.start().await.expect("start"); + let old_version = manager.get_service_version().await; + let old_service = manager.get_encryption_service().await.expect("old service"); + + let result = manager + .reconfigure_with_persistence(static_config("key-b", 0x22), || async { + Err(KmsError::backend_error("persist failed")) + }) + .await; + + assert!(result.is_err()); + assert_eq!(manager.get_status().await, KmsServiceStatus::Running); + assert_eq!(manager.get_service_version().await, old_version); + assert_eq!( + manager.get_config().await.and_then(|config| config.default_key_id), + Some("key-a".to_string()) + ); + assert!(Arc::ptr_eq( + &old_service, + &manager.get_encryption_service().await.expect("old service remains") + )); + } + + #[tokio::test] + async fn reconfigure_candidate_failure_keeps_old_running_snapshot() { + let manager = KmsServiceManager::new(); + manager.configure(static_config("key-a", 0x11)).await.expect("configure"); + manager.start().await.expect("start"); + let old_version = manager.get_service_version().await; + let invalid_parent = tempfile::NamedTempFile::new().expect("temporary file"); + let invalid_config = KmsConfig::local(invalid_parent.path().join("keys")).with_insecure_development_defaults(); + + let result = manager.reconfigure(invalid_config).await; + + assert!(result.is_err()); + assert_eq!(manager.get_status().await, KmsServiceStatus::Running); + assert_eq!(manager.get_service_version().await, old_version); + assert_eq!( + manager.get_config().await.and_then(|config| config.default_key_id), + Some("key-a".to_string()) + ); + } + + #[tokio::test] + async fn restart_never_unpublishes_the_running_service() { + let manager = Arc::new(KmsServiceManager::new()); + manager.configure(static_config("key-a", 0x11)).await.expect("configure"); + manager.start().await.expect("start"); + let old_version = manager.get_service_version().await.expect("old version"); + let restarting = { + let manager = manager.clone(); + tokio::spawn(async move { manager.restart().await }) + }; + + while !restarting.is_finished() { + assert!(manager.get_encryption_service().await.is_some()); + tokio::task::yield_now().await; + } + restarting.await.expect("restart task").expect("restart"); + + assert!(manager.get_encryption_service().await.is_some()); + assert!(manager.get_service_version().await.expect("new version") > old_version); + assert_eq!(manager.get_status().await, KmsServiceStatus::Running); + } + + #[tokio::test] + async fn start_or_restart_decides_under_the_lifecycle_lock() { + let manager = KmsServiceManager::new(); + manager.configure(static_config("key-a", 0x11)).await.expect("configure"); + + assert_eq!(manager.start_or_restart(false).await.expect("initial start"), KmsStartOutcome::Started); + let first_version = manager.get_service_version().await.expect("first version"); + assert_eq!( + manager.start_or_restart(false).await.expect("already running"), + KmsStartOutcome::AlreadyRunning + ); + assert_eq!(manager.get_service_version().await, Some(first_version)); + assert_eq!(manager.start_or_restart(true).await.expect("forced restart"), KmsStartOutcome::Restarted); + assert!(manager.get_service_version().await.expect("restarted version") > first_version); + } + + #[tokio::test] + async fn stale_health_failure_cannot_poison_new_service_status() { + let manager = KmsServiceManager::new(); + manager.configure(static_config("key-a", 0x11)).await.expect("configure"); + manager.start().await.expect("start"); + let old_version = manager.get_service_version().await.expect("old version"); + manager.restart().await.expect("restart"); + + manager.mark_health_error_if_current(old_version, &KmsError::backend_error("stale failure")); + + assert_eq!(manager.get_status().await, KmsServiceStatus::Running); + } } diff --git a/rustfs/src/admin/handlers/kms_dynamic.rs b/rustfs/src/admin/handlers/kms_dynamic.rs index e4753542d..900e41bf1 100644 --- a/rustfs/src/admin/handlers/kms_dynamic.rs +++ b/rustfs/src/admin/handlers/kms_dynamic.rs @@ -363,36 +363,27 @@ impl Operation for ConfigureKmsHandler { // Convert request to KmsConfig let kms_config = configure_request.to_kms_config(); - // Configure the service - let (success, message, status) = match service_manager.configure(kms_config.clone()).await { + let persisted_config = kms_config.clone(); + let (success, message, status) = match service_manager + .configure_with_persistence(kms_config, || async move { + save_kms_config(&persisted_config) + .await + .map_err(|error| rustfs_kms::KmsError::backend_error(format!("Failed to persist KMS configuration: {error}"))) + }) + .await + { Ok(()) => { - // Persist the configuration to cluster storage - if let Err(e) = save_kms_config(&kms_config).await { - let error_msg = format!("KMS configured in memory but failed to persist: {e}"); - error!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "configure", - state = "persist_failed", - error = %e, - "admin kms dynamic state" - ); - let status = service_manager.get_status().await; - (false, error_msg, status) - } else { - let status = service_manager.get_status().await; - info!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "configure", - state = "configured", - status = ?status, - "admin kms dynamic state" - ); - (true, "KMS configured successfully".to_string(), status) - } + let status = service_manager.get_status().await; + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "configure", + state = "configured", + status = ?status, + "admin kms dynamic state" + ); + (true, "KMS configured successfully".to_string(), status) } Err(e) => { let error_msg = format!("Failed to configure KMS: {e}"); @@ -499,125 +490,61 @@ impl Operation for StartKmsHandler { ); let service_manager = kms_service_manager_from_context(); - - // Check if already running and force flag - let current_status = service_manager.get_status().await; - if matches!(current_status, KmsServiceStatus::Running) && !start_request.force.unwrap_or(false) { - warn!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "start", - state = "already_running", - "admin kms dynamic state" - ); - let response = StartKmsResponse { - success: false, - message: "KMS service is already running. Use force=true to restart.".to_string(), - status: current_status, - }; - let json_response = match serde_json::to_string(&response) { - Ok(json) => json, - Err(e) => { - error!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = EVENT_ADMIN_KMS_DYNAMIC_STATE, - operation = "start", - result = "response_serialize_failed", - error = %e, - "admin kms dynamic state" - ); - return Ok(S3Response::new(( - StatusCode::INTERNAL_SERVER_ERROR, - Body::from("Serialization error".to_string()), - ))); - } - }; - return Ok(S3Response::new((StatusCode::OK, Body::from(json_response)))); - } - - // Start the service (or restart if force=true) - let (success, message, status) = - if start_request.force.unwrap_or(false) && matches!(current_status, KmsServiceStatus::Running) { - // Force restart - match service_manager.stop().await { - Ok(()) => match service_manager.start().await { - Ok(()) => { - let status = service_manager.get_status().await; - info!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "restart", - state = "running", - status = ?status, - "admin kms dynamic state" - ); - (true, "KMS service restarted successfully".to_string(), status) - } - Err(e) => { - let error_msg = format!("Failed to restart KMS service: {e}"); - error!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "restart", - state = "start_failed", - error = %e, - "admin kms dynamic state" - ); - let status = service_manager.get_status().await; - (false, error_msg, status) - } - }, - Err(e) => { - let error_msg = format!("Failed to stop KMS service for restart: {e}"); - error!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "restart", - state = "stop_failed", - error = %e, - "admin kms dynamic state" - ); - let status = service_manager.get_status().await; - (false, error_msg, status) - } - } - } else { - // Normal start - match service_manager.start().await { - Ok(()) => { - let status = service_manager.get_status().await; - info!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "start", - state = "running", - status = ?status, - "admin kms dynamic state" - ); - (true, "KMS service started successfully".to_string(), status) - } - Err(e) => { - let error_msg = format!("Failed to start KMS service: {e}"); - error!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "start", - state = "start_failed", - error = %e, - "admin kms dynamic state" - ); - let status = service_manager.get_status().await; - (false, error_msg, status) - } - } - }; + let force = start_request.force.unwrap_or(false); + let (success, message, status) = match service_manager.start_or_restart(force).await { + Ok(rustfs_kms::KmsStartOutcome::Started) => { + let status = service_manager.get_status().await; + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "start", + state = "running", + status = ?status, + "admin kms dynamic state" + ); + (true, "KMS service started successfully".to_string(), status) + } + Ok(rustfs_kms::KmsStartOutcome::Restarted) => { + let status = service_manager.get_status().await; + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "restart", + state = "running", + status = ?status, + "admin kms dynamic state" + ); + (true, "KMS service restarted successfully".to_string(), status) + } + Ok(rustfs_kms::KmsStartOutcome::AlreadyRunning) => { + let status = service_manager.get_status().await; + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "start", + state = "already_running", + "admin kms dynamic state" + ); + (false, "KMS service is already running. Use force=true to restart.".to_string(), status) + } + Err(e) => { + let error_msg = format!("Failed to start or restart KMS service: {e}"); + error!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "start", + state = "start_failed", + error = %e, + "admin kms dynamic state" + ); + let status = service_manager.get_status().await; + (false, error_msg, status) + } + }; let response = StartKmsResponse { success, @@ -774,8 +701,7 @@ impl Operation for GetKmsStatusHandler { let service_manager = kms_service_manager_from_context(); - let status = service_manager.get_status().await; - let config = service_manager.get_redacted_config().await; + let (status, config) = service_manager.get_redacted_state().await; // Get backend type and health status let backend_type = config.as_ref().map(|c| c.backend.clone()); @@ -908,36 +834,27 @@ impl Operation for ReconfigureKmsHandler { // Convert request to KmsConfig let kms_config = configure_request.to_kms_config(); - // Reconfigure the service (stops, reconfigures, and starts) - let (success, message, status) = match service_manager.reconfigure(kms_config.clone()).await { + let persisted_config = kms_config.clone(); + let (success, message, status) = match service_manager + .reconfigure_with_persistence(kms_config, || async move { + save_kms_config(&persisted_config) + .await + .map_err(|error| rustfs_kms::KmsError::backend_error(format!("Failed to persist KMS configuration: {error}"))) + }) + .await + { Ok(()) => { - // Persist the configuration to cluster storage - if let Err(e) = save_kms_config(&kms_config).await { - let error_msg = format!("KMS reconfigured in memory but failed to persist: {e}"); - error!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "reconfigure", - state = "persist_failed", - error = %e, - "admin kms dynamic state" - ); - let status = service_manager.get_status().await; - (false, error_msg, status) - } else { - let status = service_manager.get_status().await; - info!( - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_KMS, - event = "kms_service_state", - operation = "reconfigure", - state = "reconfigured", - status = ?status, - "admin kms dynamic state" - ); - (true, "KMS reconfigured and restarted successfully".to_string(), status) - } + let status = service_manager.get_status().await; + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_KMS, + event = "kms_service_state", + operation = "reconfigure", + state = "reconfigured", + status = ?status, + "admin kms dynamic state" + ); + (true, "KMS reconfigured and restarted successfully".to_string(), status) } Err(e) => { let error_msg = format!("Failed to reconfigure KMS: {e}"); diff --git a/rustfs/src/app/context/global.rs b/rustfs/src/app/context/global.rs index bd1baf9a6..84118caba 100644 --- a/rustfs/src/app/context/global.rs +++ b/rustfs/src/app/context/global.rs @@ -18,13 +18,12 @@ use super::handles::{ IamHandle, KmsHandle, default_action_credential_interface, default_boot_time_interface, default_bucket_metadata_interface, default_bucket_monitor_interface, default_buffer_config_interface, default_deployment_id_interface, default_endpoints_interface, default_expiry_state_interface, default_federated_identity_interface, - default_internode_metrics_interface, default_kms_runtime_interface, default_local_node_name_interface, - default_lock_client_interface, default_lock_clients_interface, default_notification_system_interface, - default_notify_interface, default_outbound_tls_runtime_interface, default_performance_metrics_interface, - default_region_interface, default_replication_pool_interface, default_replication_stats_interface, - default_runtime_port_interface, default_s3select_db_interface, default_scanner_metrics_interface, - default_server_config_interface, default_storage_class_interface, default_tier_config_interface, - default_transition_state_interface, + default_internode_metrics_interface, default_local_node_name_interface, default_lock_client_interface, + default_lock_clients_interface, default_notification_system_interface, default_notify_interface, + default_outbound_tls_runtime_interface, default_performance_metrics_interface, default_region_interface, + default_replication_pool_interface, default_replication_stats_interface, default_runtime_port_interface, + default_s3select_db_interface, default_scanner_metrics_interface, default_server_config_interface, + default_storage_class_interface, default_tier_config_interface, default_transition_state_interface, }; use super::interfaces::{ ActionCredentialInterface, BootTimeInterface, BucketMetadataInterface, BucketMonitorInterface, BufferConfigInterface, @@ -80,6 +79,7 @@ pub struct AppContext { impl AppContext { pub fn new(object_store: Arc, iam: Arc, kms: Arc) -> Self { let object_data_cache = ObjectDataCacheAdapter::from_env_or_disabled(); + let kms_runtime = Arc::new(crate::app::context::handles::KmsRuntimeHandle::new(kms.handle())); // Let ecstore probe this cache inside get_object_reader, after // metadata resolution but before the erasure data read (backlog#802). crate::app::object_data_cache::register_object_data_cache_body_hook(Arc::clone(&object_data_cache)); @@ -94,7 +94,7 @@ impl AppContext { iam, federated_identity: default_federated_identity_interface(), kms, - kms_runtime: default_kms_runtime_interface(), + kms_runtime, outbound_tls_runtime: default_outbound_tls_runtime_interface(), notify: default_notify_interface(), notification_system: default_notification_system_interface(), diff --git a/rustfs/src/app/context/handles.rs b/rustfs/src/app/context/handles.rs index 7eb4e3f29..9e376e1a5 100644 --- a/rustfs/src/app/context/handles.rs +++ b/rustfs/src/app/context/handles.rs @@ -128,12 +128,19 @@ impl KmsInterface for KmsHandle { } /// Default KMS runtime interface adapter. -#[derive(Default)] -pub struct KmsRuntimeHandle; +pub struct KmsRuntimeHandle { + kms: Option>, +} + +impl KmsRuntimeHandle { + pub fn new(kms: Arc) -> Self { + Self { kms: Some(kms) } + } +} impl KmsRuntimeInterface for KmsRuntimeHandle { fn service_manager(&self) -> Option> { - runtime_sources::kms_service_manager() + self.kms.clone() } } @@ -485,7 +492,9 @@ pub fn default_notification_system_interface() -> Arc Arc { - Arc::new(KmsRuntimeHandle) + Arc::new(KmsRuntimeHandle { + kms: runtime_sources::kms_service_manager(), + }) } pub fn default_outbound_tls_runtime_interface() -> Arc { @@ -607,10 +616,10 @@ pub fn default_buffer_config_interface() -> Arc { #[cfg(test)] mod tests { use super::{ - ServerConfigHandle, default_federated_identity_interface, federated_identity_interface, - publish_default_federated_identity_service, runtime_sources, + KmsRuntimeHandle, KmsServiceManager, ServerConfigHandle, default_federated_identity_interface, + federated_identity_interface, publish_default_federated_identity_service, runtime_sources, }; - use crate::app::context::interfaces::ServerConfigInterface; + use crate::app::context::interfaces::{KmsRuntimeInterface, ServerConfigInterface}; use rustfs_config::server_config::Config; use rustfs_iam::{ federation::{FederatedIdentityRegistry, FederatedIdentityService, oidc::StandardOidcAdapter}, @@ -733,4 +742,19 @@ mod tests { "handle B must serve its own credentials" ); } + + #[test] + fn kms_runtime_handles_keep_injected_managers_isolated() { + let manager_a = Arc::new(KmsServiceManager::new()); + let manager_b = Arc::new(KmsServiceManager::new()); + let handle_a = KmsRuntimeHandle::new(manager_a.clone()); + let handle_b = KmsRuntimeHandle::new(manager_b.clone()); + + assert!(Arc::ptr_eq(&handle_a.service_manager().expect("manager A"), &manager_a)); + assert!(Arc::ptr_eq(&handle_b.service_manager().expect("manager B"), &manager_b)); + assert!(!Arc::ptr_eq( + &handle_a.service_manager().expect("manager A"), + &handle_b.service_manager().expect("manager B") + )); + } } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index bb6aac980..7525c6796 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -517,7 +517,7 @@ pub(crate) mod ecstore_object { }; } -#[cfg(test)] +#[cfg(all(test, feature = "rio-v2"))] pub(crate) mod ecstore_test_support { pub(crate) use rustfs_ecstore::api::bitrot::create_bitrot_reader; pub(crate) use rustfs_ecstore::api::disk::{DiskAPI, DiskOption, endpoint::Endpoint, new_disk};