diff --git a/crates/ecstore/src/error.rs b/crates/ecstore/src/error.rs index e20d1e6be..7bb84b8a0 100644 --- a/crates/ecstore/src/error.rs +++ b/crates/ecstore/src/error.rs @@ -160,6 +160,14 @@ pub enum StorageError { NotModified, #[error("Invalid range specified: {0}")] InvalidRangeSpec(String), + #[error("Namespace lock quorum unavailable for {mode} lock on {bucket}/{object}: required {required}, achieved {achieved}")] + NamespaceLockQuorumUnavailable { + mode: &'static str, + bucket: String, + object: String, + required: usize, + achieved: usize, + }, // ── Generic ────────────────────────────────────────────────────── #[error("Unexpected error")] @@ -204,6 +212,7 @@ impl StorageError { | StorageError::ErasureWriteQuorum | StorageError::InsufficientReadQuorum(_, _) | StorageError::InsufficientWriteQuorum(_, _) + | StorageError::NamespaceLockQuorumUnavailable { .. } ) } } @@ -433,6 +442,19 @@ impl Clone for StorageError { StorageError::NotModified => StorageError::NotModified, StorageError::InvalidPartNumber(a) => StorageError::InvalidPartNumber(*a), StorageError::InvalidRangeSpec(a) => StorageError::InvalidRangeSpec(a.clone()), + StorageError::NamespaceLockQuorumUnavailable { + mode, + bucket, + object, + required, + achieved, + } => StorageError::NamespaceLockQuorumUnavailable { + mode, + bucket: bucket.clone(), + object: object.clone(), + required: *required, + achieved: *achieved, + }, } } } @@ -505,6 +527,7 @@ impl StorageError { StorageError::InvalidRangeSpec(_) => 0x3D, StorageError::NotModified => 0x3E, StorageError::InvalidPartNumber(_) => 0x3F, + StorageError::NamespaceLockQuorumUnavailable { .. } => 0x42, } } @@ -579,6 +602,13 @@ impl StorageError { 0x3D => Some(StorageError::InvalidRangeSpec(Default::default())), 0x3E => Some(StorageError::NotModified), 0x3F => Some(StorageError::InvalidPartNumber(Default::default())), + 0x42 => Some(StorageError::NamespaceLockQuorumUnavailable { + mode: "write", + bucket: Default::default(), + object: Default::default(), + required: Default::default(), + achieved: Default::default(), + }), _ => None, } } @@ -984,6 +1014,17 @@ mod tests { assert_eq!(StorageError::DecommissionAlreadyRunning.to_u32(), 0x30); assert_eq!(StorageError::RebalanceAlreadyRunning.to_u32(), 0x40); assert_eq!(StorageError::OperationCanceled.to_u32(), 0x41); + assert_eq!( + StorageError::NamespaceLockQuorumUnavailable { + mode: "write", + bucket: "bucket".into(), + object: "object".into(), + required: 3, + achieved: 2, + } + .to_u32(), + 0x42 + ); } #[test] @@ -998,6 +1039,10 @@ mod tests { assert!(matches!(StorageError::from_u32(0x30), Some(StorageError::DecommissionAlreadyRunning))); assert!(matches!(StorageError::from_u32(0x40), Some(StorageError::RebalanceAlreadyRunning))); assert!(matches!(StorageError::from_u32(0x41), Some(StorageError::OperationCanceled))); + assert!(matches!( + StorageError::from_u32(0x42), + Some(StorageError::NamespaceLockQuorumUnavailable { .. }) + )); // Test invalid code returns None assert!(StorageError::from_u32(0xFF).is_none()); diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index db6f23065..6ca6978d2 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -622,12 +622,7 @@ impl ObjectIO for SetDisks { .await? .get_read_lock(get_lock_acquire_timeout()) .await - .map_err(|e| { - Error::other(format!( - "Failed to acquire read lock: {}", - self.format_lock_error_from_error(bucket, object, "read", &e) - )) - })?; + .map_err(|e| self.map_namespace_lock_error(bucket, object, "read", e))?; // Record lock acquisition for deadlock detection let _lock_id = record_lock_acquire(bucket, object, "read"); @@ -756,12 +751,12 @@ impl ObjectIO for SetDisks { if let Some(http_preconditions) = opts.http_preconditions.clone() { if !opts.no_lock { let ns_lock = self.new_ns_lock(bucket, object).await?; - object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| { - StorageError::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?); + object_lock_guard = Some( + ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + ); } if let Some(err) = self.check_write_precondition(bucket, object, opts).await { @@ -975,12 +970,12 @@ impl ObjectIO for SetDisks { if !opts.no_lock && object_lock_guard.is_none() { let ns_lock = self.new_ns_lock(bucket, object).await?; - object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| { - StorageError::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?); + object_lock_guard = Some( + ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + ); } let (online_disks, _, op_old_dir) = Self::rename_data( @@ -1398,12 +1393,7 @@ impl ObjectOperations for SetDisks { .await? .get_write_lock(get_lock_acquire_timeout()) .await - .map_err(|e| { - Error::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(src_bucket, src_object, "write", &e) - )) - })?; + .map_err(|e| self.map_namespace_lock_error(src_bucket, src_object, "write", e))?; let disks = self.get_disks_internal().await; @@ -1785,12 +1775,7 @@ impl ObjectOperations for SetDisks { .await? .get_write_lock(get_lock_acquire_timeout()) .await - .map_err(|e| { - Error::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?, + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, ) } else { None @@ -1931,12 +1916,7 @@ impl ObjectOperations for SetDisks { .await? .get_read_lock(get_lock_acquire_timeout()) .await - .map_err(|e| { - Error::other(format!( - "Failed to acquire read lock: {}", - self.format_lock_error_from_error(bucket, object, "read", &e) - )) - })?, + .map_err(|e| self.map_namespace_lock_error(bucket, object, "read", e))?, ) } else { None @@ -1990,12 +1970,7 @@ impl ObjectOperations for SetDisks { .await? .get_write_lock(get_lock_acquire_timeout()) .await - .map_err(|e| { - Error::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?, + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, ) } else { None @@ -2527,12 +2502,7 @@ impl SetDisks { .await? .get_write_lock(get_lock_acquire_timeout()) .await - .map_err(|e| { - Error::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?, + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, ) } else { None @@ -3067,12 +3037,12 @@ impl MultipartOperations for SetDisks { if let Some(http_preconditions) = opts.http_preconditions.clone() { let object_lock_guard = if !opts.no_lock { let ns_lock = self.new_ns_lock(bucket, object).await?; - Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| { - StorageError::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?) + Some( + ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + ) } else { None }; @@ -3249,12 +3219,12 @@ impl MultipartOperations for SetDisks { if let Some(http_preconditions) = opts.http_preconditions.clone() { if !opts.no_lock { let ns_lock = self.new_ns_lock(bucket, object).await?; - object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| { - StorageError::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?); + object_lock_guard = Some( + ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + ); } if let Some(err) = self.check_write_precondition(bucket, object, opts).await { @@ -3599,12 +3569,12 @@ impl MultipartOperations for SetDisks { if !opts.no_lock && object_lock_guard.is_none() { let ns_lock = self.new_ns_lock(bucket, object).await?; - object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| { - StorageError::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?); + object_lock_guard = Some( + ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + ); } self.cleanup_multipart_path(&parts).await; @@ -3685,12 +3655,12 @@ impl HealOperations for SetDisks { ) -> Result<(HealResultItem, Option)> { let _write_lock_guard = if !opts.no_lock { let ns_lock = self.new_ns_lock(bucket, object).await?; - Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| { - StorageError::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?) + Some( + ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + ) } else { None }; diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index 9fca89c46..90d13ab7d 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/heal.rs @@ -39,12 +39,12 @@ impl SetDisks { let write_lock_guard = if !opts.no_lock { let ns_lock = self.new_ns_lock(bucket, object).await?; - Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| { - StorageError::other(format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - )) - })?) + Some( + ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?, + ) } else { None }; @@ -713,13 +713,7 @@ impl SetDisks { .await? .get_write_lock(get_lock_acquire_timeout()) .await - .map_err(|e| { - let message = format!( - "Failed to acquire write lock: {}", - self.format_lock_error_from_error(bucket, object, "write", &e) - ); - DiskError::other(message) - })?; + .map_err(|e| DiskError::other(self.map_namespace_lock_error(bucket, object, "write", e).to_string()))?; self.heal_object_dir_locked(bucket, object, dry_run, remove).await } diff --git a/crates/ecstore/src/set_disk/lock.rs b/crates/ecstore/src/set_disk/lock.rs index 4a6cc7d5d..aa536c426 100644 --- a/crates/ecstore/src/set_disk/lock.rs +++ b/crates/ecstore/src/set_disk/lock.rs @@ -50,6 +50,30 @@ impl SetDisks { } } + pub(super) fn map_namespace_lock_error( + &self, + bucket: &str, + object: &str, + mode: &'static str, + err: rustfs_lock::error::LockError, + ) -> StorageError { + match err { + rustfs_lock::error::LockError::QuorumNotReached { required, achieved } => { + StorageError::NamespaceLockQuorumUnavailable { + mode, + bucket: bucket.to_string(), + object: object.to_string(), + required, + achieved, + } + } + other => StorageError::other(format!( + "Failed to acquire {mode} lock: {}", + self.format_lock_error_from_error(bucket, object, mode, &other) + )), + } + } + pub(super) async fn get_disks_internal(&self) -> Vec> { let rl = self.disks.read().await; diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index 2d0b9adda..e651f926e 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -526,7 +526,8 @@ async fn health_check(method: Method, uri: Uri) -> Response { let readiness_report = collect_dependency_readiness().await; let storage_ready = readiness_report.readiness.storage_ready; let iam_ready = readiness_report.readiness.iam_ready; - let health = health_check_state(storage_ready, iam_ready, probe); + let lock_quorum_ready = readiness_report.readiness.lock_quorum_ready; + let health = health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe); let builder = Response::builder() .status(health.status_code) @@ -543,6 +544,7 @@ async fn health_check(method: Method, uri: Uri) -> Response { health, storage_ready, iam_ready, + lock_quorum_ready, &readiness_report.degraded_reasons, "rustfs-console", Some(uptime), diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index 2fa880ca5..e76a472e2 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -61,8 +61,13 @@ pub(crate) async fn collect_dependency_readiness() -> crate::server::DependencyR collect_runtime_dependency_readiness_report().await } -pub(crate) fn health_check_state(storage_ready: bool, iam_ready: bool, probe: HealthProbe) -> HealthCheckState { - let ready = storage_ready && iam_ready; +pub(crate) fn health_check_state( + storage_ready: bool, + iam_ready: bool, + lock_quorum_ready: bool, + probe: HealthProbe, +) -> HealthCheckState { + let ready = storage_ready && iam_ready && lock_quorum_ready; let status = if ready { "ok" } else { "degraded" }; let status_code = match probe { @@ -85,7 +90,7 @@ pub(crate) fn health_minimal_response_enabled() -> bool { ) } -pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool) -> Value { +pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool, lock_quorum_ready: bool) -> Value { json!({ "storage": { "status": if storage_ready { "connected" } else { "disconnected" }, @@ -94,6 +99,10 @@ pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool) -> V "iam": { "status": if iam_ready { "connected" } else { "disconnected" }, "ready": iam_ready, + }, + "lock": { + "status": if lock_quorum_ready { "connected" } else { "disconnected" }, + "ready": lock_quorum_ready, } }) } @@ -119,6 +128,7 @@ pub(crate) fn build_health_payload( health: HealthCheckState, storage_ready: bool, iam_ready: bool, + lock_quorum_ready: bool, degraded_reasons: &[crate::server::ReadinessDegradedReason], service: &str, uptime: Option, @@ -136,7 +146,7 @@ pub(crate) fn build_health_payload( "service": service, "timestamp": jiff::Zoned::now().to_string(), "version": env!("CARGO_PKG_VERSION"), - "details": build_component_details(storage_ready, iam_ready), + "details": build_component_details(storage_ready, iam_ready, lock_quorum_ready), "degradedReasons": build_degraded_reasons(degraded_reasons), }); @@ -152,10 +162,19 @@ pub(crate) fn build_health_response( probe: HealthProbe, storage_ready: bool, iam_ready: bool, + lock_quorum_ready: bool, degraded_reasons: &[crate::server::ReadinessDegradedReason], ) -> S3Response<(StatusCode, Body)> { - let health = health_check_state(storage_ready, iam_ready, probe); - let health_info = build_health_payload(health, storage_ready, iam_ready, degraded_reasons, "rustfs-endpoint", None); + let health = health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe); + let health_info = build_health_payload( + health, + storage_ready, + iam_ready, + lock_quorum_ready, + degraded_reasons, + "rustfs-endpoint", + None, + ); let mut headers = HeaderMap::new(); headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); @@ -195,6 +214,7 @@ impl Operation for HealthCheckHandler { probe, readiness_report.readiness.storage_ready, readiness_report.readiness.iam_ready, + readiness_report.readiness.lock_quorum_ready, &readiness_report.degraded_reasons, )) } @@ -207,7 +227,7 @@ mod tests { #[test] fn test_readiness_state_ready() { - let state = health_check_state(true, true, HealthProbe::Readiness); + let state = health_check_state(true, true, true, HealthProbe::Readiness); assert_eq!(state.status_code, StatusCode::OK); assert_eq!(state.status, "ok"); assert!(state.ready); @@ -215,7 +235,7 @@ mod tests { #[test] fn test_readiness_state_storage_not_ready() { - let state = health_check_state(false, true, HealthProbe::Readiness); + let state = health_check_state(false, true, true, HealthProbe::Readiness); assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE); assert_eq!(state.status, "degraded"); assert!(!state.ready); @@ -223,7 +243,7 @@ mod tests { #[test] fn test_liveness_state_iam_not_ready() { - let state = health_check_state(true, false, HealthProbe::Liveness); + let state = health_check_state(true, false, true, HealthProbe::Liveness); assert_eq!(state.status_code, StatusCode::OK); assert_eq!(state.status, "degraded"); assert!(!state.ready); @@ -231,7 +251,15 @@ mod tests { #[test] fn test_readiness_state_iam_not_ready() { - let state = health_check_state(true, false, HealthProbe::Readiness); + let state = health_check_state(true, false, true, HealthProbe::Readiness); + assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(state.status, "degraded"); + assert!(!state.ready); + } + + #[test] + fn test_readiness_state_lock_not_ready() { + let state = health_check_state(true, true, false, HealthProbe::Readiness); assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE); assert_eq!(state.status, "degraded"); assert!(!state.ready); @@ -239,12 +267,14 @@ mod tests { #[test] fn test_health_check_component_details() { - let details = build_component_details(true, false); + let details = build_component_details(true, false, false); assert_eq!(details["storage"]["status"], "connected"); assert_eq!(details["storage"]["ready"], true); assert_eq!(details["iam"]["status"], "disconnected"); assert_eq!(details["iam"]["ready"], false); + assert_eq!(details["lock"]["status"], "disconnected"); + assert_eq!(details["lock"]["ready"], false); } #[test] @@ -265,6 +295,7 @@ mod tests { HealthProbe::Readiness, false, true, + true, &[crate::server::ReadinessDegradedReason::StorageQuorumUnavailable], ); assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE); @@ -272,7 +303,7 @@ mod tests { #[test] fn test_build_health_response_readiness_returns_200_when_deps_ready() { - let resp = build_health_response(Method::GET, HealthProbe::Readiness, true, true, &[]); + let resp = build_health_response(Method::GET, HealthProbe::Readiness, true, true, true, &[]); assert_eq!(resp.output.0, StatusCode::OK); } @@ -283,6 +314,7 @@ mod tests { HealthProbe::Liveness, false, false, + false, &[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], ); assert_eq!(resp.output.0, StatusCode::OK); @@ -295,6 +327,7 @@ mod tests { HealthProbe::Readiness, false, false, + false, &[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], ); assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE); @@ -302,12 +335,13 @@ mod tests { #[test] fn test_build_health_payload_minimal_mode_returns_status_and_ready_only() { - let health = health_check_state(true, false, HealthProbe::Readiness); + let health = health_check_state(true, false, true, HealthProbe::Readiness); with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("true"), || { let payload = build_health_payload( health, true, false, + true, &[crate::server::ReadinessDegradedReason::IamNotReady], "rustfs-endpoint", Some(123), @@ -323,11 +357,12 @@ mod tests { #[test] fn test_build_health_payload_includes_degraded_reasons() { - let health = health_check_state(false, false, HealthProbe::Readiness); + let health = health_check_state(false, false, false, HealthProbe::Readiness); let payload = build_health_payload( health, false, false, + false, &[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], "rustfs-endpoint", None, diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 14a2d8352..dfc40fe51 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -217,6 +217,7 @@ impl From for ApiError { StorageError::BucketExists(_) => S3ErrorCode::BucketAlreadyOwnedByYou, StorageError::StorageFull => S3ErrorCode::ServiceUnavailable, StorageError::SlowDown => S3ErrorCode::SlowDown, + StorageError::NamespaceLockQuorumUnavailable { .. } => S3ErrorCode::ServiceUnavailable, StorageError::DecommissionNotStarted => S3ErrorCode::InvalidRequest, StorageError::DecommissionAlreadyRunning => S3ErrorCode::InvalidRequest, StorageError::RebalanceAlreadyRunning => S3ErrorCode::InvalidRequest, @@ -413,6 +414,16 @@ mod tests { (StorageError::BucketExists("test".into()), S3ErrorCode::BucketAlreadyOwnedByYou), (StorageError::StorageFull, S3ErrorCode::ServiceUnavailable), (StorageError::SlowDown, S3ErrorCode::SlowDown), + ( + StorageError::NamespaceLockQuorumUnavailable { + mode: "write", + bucket: "bucket".into(), + object: "object".into(), + required: 3, + achieved: 2, + }, + S3ErrorCode::ServiceUnavailable, + ), (StorageError::DecommissionNotStarted, S3ErrorCode::InvalidRequest), (StorageError::DecommissionAlreadyRunning, S3ErrorCode::InvalidRequest), (StorageError::RebalanceAlreadyRunning, S3ErrorCode::InvalidRequest), @@ -460,6 +471,23 @@ mod tests { assert!(s3_error.source().is_some()); } + #[test] + fn test_namespace_lock_quorum_failure_maps_to_service_unavailable_status() { + let api_error: ApiError = StorageError::NamespaceLockQuorumUnavailable { + mode: "write", + bucket: "bucket".into(), + object: "object".into(), + required: 3, + achieved: 2, + } + .into(); + + let s3_error: S3Error = api_error.into(); + + assert_eq!(*s3_error.code(), S3ErrorCode::ServiceUnavailable); + assert_eq!(s3_error.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE)); + } + #[test] fn test_api_error_to_s3_error_without_source() { let api_error = ApiError { diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 46b3a8733..12a8cb451 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -642,7 +642,8 @@ where let readiness_report = collect_dependency_readiness_report().await; let storage_ready = readiness_report.readiness.storage_ready; let iam_ready = readiness_report.readiness.iam_ready; - let health = health_check_state(storage_ready, iam_ready, probe); + let lock_quorum_ready = readiness_report.readiness.lock_quorum_ready; + let health = health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe); let body = if method == Method::HEAD { Bytes::new() } else { @@ -650,6 +651,7 @@ where health, storage_ready, iam_ready, + lock_quorum_ready, &readiness_report.degraded_reasons, "rustfs-endpoint", None, diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 14704ca29..a7880b515 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -20,6 +20,8 @@ use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; use metrics::{counter, gauge}; use rustfs_common::GlobalReadiness; +use rustfs_ecstore::global::is_dist_erasure; +use rustfs_ecstore::global::{get_global_endpoints_opt, get_global_lock_clients}; use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::store_api::StorageAPI; use rustfs_iam::get_global_iam_sys; @@ -47,13 +49,18 @@ const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_ pub struct DependencyReadiness { pub storage_ready: bool, pub iam_ready: bool, + pub lock_quorum_ready: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ReadinessDegradedReason { StorageQuorumUnavailable, IamNotReady, + LockQuorumUnavailable, StorageAndIamUnavailable, + StorageAndLockUnavailable, + IamAndLockUnavailable, + StorageIamAndLockUnavailable, } impl ReadinessDegradedReason { @@ -61,7 +68,11 @@ impl ReadinessDegradedReason { match self { ReadinessDegradedReason::StorageQuorumUnavailable => "storage_quorum_unavailable", ReadinessDegradedReason::IamNotReady => "iam_not_ready", + ReadinessDegradedReason::LockQuorumUnavailable => "lock_quorum_unavailable", ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable", + ReadinessDegradedReason::StorageAndLockUnavailable => "storage_and_lock_unavailable", + ReadinessDegradedReason::IamAndLockUnavailable => "iam_and_lock_unavailable", + ReadinessDegradedReason::StorageIamAndLockUnavailable => "storage_iam_and_lock_unavailable", } } } @@ -211,6 +222,14 @@ struct StorageReadinessCacheEntry { storage_ready: bool, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct LockQuorumStatus { + pub ready: bool, + pub connected_clients: usize, + pub total_clients: usize, + pub required_quorum: usize, +} + const DISK_STATE_OK: &str = "ok"; const DISK_STATE_UNFORMATTED: &str = "unformatted"; const RUNTIME_STATE_RETURNING: &str = "returning"; @@ -346,17 +365,21 @@ fn storage_ready_from_runtime_state(info: &StorageInfo) -> bool { }) } -fn degraded_reasons(storage_ready: bool, iam_ready_raw: bool) -> Vec { - match (storage_ready, iam_ready_raw) { - (true, true) => Vec::new(), - (false, false) => vec![ReadinessDegradedReason::StorageAndIamUnavailable], - (false, true) => vec![ReadinessDegradedReason::StorageQuorumUnavailable], - (true, false) => vec![ReadinessDegradedReason::IamNotReady], +fn degraded_reasons(storage_ready: bool, iam_ready_raw: bool, lock_quorum_ready: bool) -> Vec { + match (storage_ready, iam_ready_raw, lock_quorum_ready) { + (true, true, true) => Vec::new(), + (false, false, false) => vec![ReadinessDegradedReason::StorageIamAndLockUnavailable], + (false, false, true) => vec![ReadinessDegradedReason::StorageAndIamUnavailable], + (false, true, false) => vec![ReadinessDegradedReason::StorageAndLockUnavailable], + (true, false, false) => vec![ReadinessDegradedReason::IamAndLockUnavailable], + (false, true, true) => vec![ReadinessDegradedReason::StorageQuorumUnavailable], + (true, false, true) => vec![ReadinessDegradedReason::IamNotReady], + (true, true, false) => vec![ReadinessDegradedReason::LockQuorumUnavailable], } } fn record_readiness_report(report: &DependencyReadinessReport) { - let ready = report.readiness.storage_ready && report.readiness.iam_ready; + let ready = report.readiness.storage_ready && report.readiness.iam_ready && report.readiness.lock_quorum_ready; gauge!(METRIC_RUNTIME_READINESS_READY).set(if ready { 1.0 } else { 0.0 }); for reason in &report.degraded_reasons { counter!(METRIC_RUNTIME_READINESS_DEGRADED_TOTAL, "reason" => reason.as_str()).increment(1); @@ -376,13 +399,15 @@ pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport update_storage_readiness_cache(computed).await; computed }; + let lock_quorum_status = collect_lock_quorum_status_uncached().await; let readiness = DependencyReadiness { storage_ready, iam_ready: iam_ready_raw, + lock_quorum_ready: lock_quorum_status.ready, }; let report = DependencyReadinessReport { - degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw), + degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw, readiness.lock_quorum_ready), readiness, }; record_readiness_report(&report); @@ -392,13 +417,15 @@ pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport async fn collect_dependency_readiness_uncached() -> DependencyReadiness { let iam_ready_raw = get_global_iam_sys().is_some_and(|sys| sys.is_ready()); let storage_ready = collect_storage_readiness_uncached().await; + let lock_quorum_status = collect_lock_quorum_status_uncached().await; let readiness = DependencyReadiness { storage_ready, iam_ready: iam_ready_raw, + lock_quorum_ready: lock_quorum_status.ready, }; let report = DependencyReadinessReport { - degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw), + degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw, readiness.lock_quorum_ready), readiness, }; record_readiness_report(&report); @@ -414,6 +441,115 @@ async fn collect_storage_readiness_uncached() -> bool { } } +fn set_lock_quorum_status( + online_hosts: &HashSet, + set_endpoints: &[rustfs_ecstore::disk::endpoint::Endpoint], +) -> LockQuorumStatus { + let total_clients = set_endpoints + .iter() + .map(rustfs_ecstore::disk::endpoint::Endpoint::host_port) + .filter(|host| !host.is_empty()) + .collect::>(); + let total_clients_len = total_clients.len(); + if total_clients_len == 0 { + return LockQuorumStatus::default(); + } + + let connected_clients = total_clients.iter().filter(|host| online_hosts.contains(*host)).count(); + let required_quorum = if total_clients_len > 1 { + (total_clients_len / 2) + 1 + } else { + 1 + }; + + LockQuorumStatus { + ready: connected_clients >= required_quorum, + connected_clients, + total_clients: total_clients_len, + required_quorum, + } +} + +fn aggregate_lock_quorum_status( + pool_endpoints: &rustfs_ecstore::endpoints::EndpointServerPools, + online_hosts: &HashSet, +) -> LockQuorumStatus { + let mut connected_clients = 0usize; + let mut total_clients = 0usize; + let mut required_quorum = 0usize; + + for pool in pool_endpoints.as_ref() { + for set_idx in 0..pool.set_count { + let set_endpoints = pool + .endpoints + .as_ref() + .iter() + .filter(|endpoint| endpoint.set_idx == set_idx as i32) + .cloned() + .collect::>(); + + let status = set_lock_quorum_status(online_hosts, &set_endpoints); + if status.total_clients == 0 { + return LockQuorumStatus::default(); + } + + connected_clients += status.connected_clients; + total_clients += status.total_clients; + required_quorum += status.required_quorum; + + if !status.ready { + return LockQuorumStatus { + ready: false, + connected_clients, + total_clients, + required_quorum, + }; + } + } + } + + if total_clients == 0 { + LockQuorumStatus::default() + } else { + LockQuorumStatus { + ready: true, + connected_clients, + total_clients, + required_quorum, + } + } +} + +async fn collect_lock_quorum_status_uncached() -> LockQuorumStatus { + if !is_dist_erasure().await { + return LockQuorumStatus { + ready: true, + connected_clients: 1, + total_clients: 1, + required_quorum: 1, + }; + } + + let Some(pool_endpoints) = get_global_endpoints_opt() else { + return LockQuorumStatus::default(); + }; + let Some(lock_clients) = get_global_lock_clients() else { + return LockQuorumStatus::default(); + }; + + let online_hosts = futures::future::join_all(lock_clients.iter().map(|(host, client)| { + let host = host.clone(); + let client = client.clone(); + async move { (host, client.is_online().await) } + })) + .await + .into_iter() + .filter_map(|(host, online)| online.then_some(host)) + .collect::>(); + + aggregate_lock_quorum_status(&pool_endpoints, &online_hosts) +} + pub async fn wait_for_runtime_readiness_with( max_wait: Duration, poll_interval: Duration, @@ -429,17 +565,18 @@ where loop { let readiness = load_readiness().await; - if readiness.storage_ready && readiness.iam_ready { + if readiness.storage_ready && readiness.iam_ready && readiness.lock_quorum_ready { on_ready(readiness); return Ok(()); } if tokio::time::Instant::now() >= startup_deadline { let reason = format!( - "startup readiness timed out after {}s: storage_ready={}, iam_ready={}", + "startup readiness timed out after {}s: storage_ready={}, iam_ready={}, lock_quorum_ready={}", max_wait.as_secs(), readiness.storage_ready, - readiness.iam_ready + readiness.iam_ready, + readiness.lock_quorum_ready ); return Err(std::io::Error::other(reason)); } @@ -448,6 +585,7 @@ where target: "rustfs::server::readiness", storage_ready = readiness.storage_ready, iam_ready = readiness.iam_ready, + lock_quorum_ready = readiness.lock_quorum_ready, "Runtime readiness has not reached write quorum yet; delaying ready state publication" ); tokio::time::sleep(poll_interval).await; @@ -479,6 +617,7 @@ mod tests { future::ready(DependencyReadiness { storage_ready: false, iam_ready: false, + lock_quorum_ready: false, }) }, |_| { @@ -494,6 +633,34 @@ mod tests { assert_eq!(state_manager.current_state(), ServiceState::Starting); } + #[tokio::test] + async fn wait_for_runtime_readiness_with_does_not_publish_ready_when_lock_quorum_is_not_reached() { + let readiness = GlobalReadiness::new(); + let state_manager = ServiceStateManager::new(); + + let err = wait_for_runtime_readiness_with( + Duration::ZERO, + Duration::from_millis(1), + || { + future::ready(DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: false, + }) + }, + |_| { + readiness.mark_stage(rustfs_common::SystemStage::FullReady); + state_manager.update(ServiceState::Ready); + }, + ) + .await + .expect_err("lock quorum should block readiness publication"); + + assert!(err.to_string().contains("lock_quorum_ready=false")); + assert!(!readiness.is_ready()); + assert_eq!(state_manager.current_state(), ServiceState::Starting); + } + #[test] fn storage_ready_from_runtime_state_returns_false_when_all_disks_faulty() { let info = StorageInfo { @@ -605,4 +772,126 @@ mod tests { "if any set fails write quorum, readiness must be false" ); } + + #[test] + fn aggregate_lock_quorum_status_requires_each_set_to_meet_quorum() { + use rustfs_ecstore::disk::endpoint::Endpoint; + use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; + + let endpoints = vec![ + Endpoint { + url: url::Url::parse("http://node1:9000/data1").unwrap(), + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: url::Url::parse("http://node2:9000/data2").unwrap(), + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 1, + }, + Endpoint { + url: url::Url::parse("http://node3:9000/data3").unwrap(), + is_local: false, + pool_idx: 0, + set_idx: 1, + disk_idx: 0, + }, + Endpoint { + url: url::Url::parse("http://node4:9000/data4").unwrap(), + is_local: false, + pool_idx: 0, + set_idx: 1, + disk_idx: 1, + }, + ]; + let pools = EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: String::new(), + platform: String::new(), + }]); + + let status = aggregate_lock_quorum_status( + &pools, + &["node1:9000", "node2:9000", "node3:9000", "node4:9000"] + .into_iter() + .map(str::to_string) + .collect::>(), + ); + + assert!(status.ready); + assert_eq!(status.connected_clients, 4); + assert_eq!(status.total_clients, 4); + assert_eq!(status.required_quorum, 4); + } + + #[test] + fn aggregate_lock_quorum_status_fails_when_any_set_loses_quorum() { + use rustfs_ecstore::disk::endpoint::Endpoint; + use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; + + let endpoints = vec![ + Endpoint { + url: url::Url::parse("http://node1:9000/data1").unwrap(), + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: url::Url::parse("http://node2:9000/data2").unwrap(), + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 1, + }, + Endpoint { + url: url::Url::parse("http://node3:9000/data3").unwrap(), + is_local: false, + pool_idx: 0, + set_idx: 1, + disk_idx: 0, + }, + Endpoint { + url: url::Url::parse("http://node4:9000/data4").unwrap(), + is_local: false, + pool_idx: 0, + set_idx: 1, + disk_idx: 1, + }, + ]; + let pools = EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: String::new(), + platform: String::new(), + }]); + + let status = + aggregate_lock_quorum_status(&pools, &["node1:9000"].into_iter().map(str::to_string).collect::>()); + + assert!(!status.ready); + } + + #[test] + fn degraded_reasons_include_lock_quorum_failures() { + assert_eq!(degraded_reasons(true, true, false), vec![ReadinessDegradedReason::LockQuorumUnavailable]); + assert_eq!( + degraded_reasons(false, true, false), + vec![ReadinessDegradedReason::StorageAndLockUnavailable] + ); + assert_eq!(degraded_reasons(true, false, false), vec![ReadinessDegradedReason::IamAndLockUnavailable]); + assert_eq!( + degraded_reasons(false, false, false), + vec![ReadinessDegradedReason::StorageIamAndLockUnavailable] + ); + } }