From 07a9d15ab2def3e6462bf196551e5f0ad11708f0 Mon Sep 17 00:00:00 2001 From: Hauser Date: Thu, 17 Sep 2026 14:53:55 +0800 Subject: [PATCH] fix(readiness): smooth pool metadata timeout flaps (#7956) Keep recent writable observations during transient pool metadata inspection timeouts while preserving confirmed write blocks and fail-closed behavior. Consolidate the runtime observation cache into one ordered state machine and expose timeout metrics. Co-authored-by: zhi22915 --- crates/ecstore/src/cluster/rpc/remote_disk.rs | 5 + crates/ecstore/src/disk/disk_store.rs | 24 ++ crates/ecstore/src/disk/mod.rs | 8 + crates/ecstore/src/store/init.rs | 4 +- docs/architecture/readiness-matrix.md | 5 +- rustfs/src/server/health.rs | 16 + rustfs/src/server/readiness.rs | 297 +++++++++++++++++- 7 files changed, 354 insertions(+), 5 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index c1451332b..d5e41c03e 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -1170,6 +1170,11 @@ impl RemoteDisk { self.health.force_runtime_state_for_test(state); } + #[cfg(test)] + pub fn force_offline_for_test(&self) { + self.health.force_offline_for_test(); + } + /// Same as [`DiskHealthTracker::reset_for_store_init_retry`]: undo a transient faulty mark before another format load attempt. pub fn reset_health_for_store_init_retry(&self) { self.health.reset_for_store_init_retry(&self.endpoint); diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 45390eefa..8c4fb2a72 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -30,6 +30,8 @@ use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_madmin::{info_commands::DiskMetrics, metrics::TimedAction}; #[cfg(not(test))] use std::sync::OnceLock; +#[cfg(test)] +use std::sync::atomic::AtomicBool; use std::{ collections::HashMap, path::PathBuf, @@ -628,6 +630,8 @@ pub struct DiskHealthTracker { /// Authoritative atomically published runtime/status pair. state_snapshot: AtomicU64, transition_lock: std::sync::Mutex<()>, + #[cfg(test)] + test_forced_offline: AtomicBool, } fn pack_health_state(runtime_state: RuntimeDriveHealthState, status: u32) -> u64 { @@ -975,6 +979,8 @@ impl DiskHealthTracker { last_capacity_probe_unix_secs: AtomicI64::new(0), state_snapshot: AtomicU64::new(pack_health_state(RuntimeDriveHealthState::Online, DISK_HEALTH_OK)), transition_lock: std::sync::Mutex::new(()), + #[cfg(test)] + test_forced_offline: AtomicBool::new(false), } } @@ -1045,6 +1051,13 @@ impl DiskHealthTracker { self.publish_state(state, status); } + #[cfg(test)] + pub fn force_offline_for_test(&self) { + self.test_forced_offline.store(true, Ordering::Release); + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + self.publish_state(RuntimeDriveHealthState::Offline, DISK_HEALTH_FAULTY); + } + pub fn swap_ok_to_faulty(&self) -> bool { let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let (_, status) = unpack_health_state(self.state_snapshot.load(Ordering::Acquire)); @@ -1123,6 +1136,8 @@ impl DiskHealthTracker { /// Remote disks are marked faulty on timeout/network errors; the init loop retries with the /// same [`DiskStore`] handles, which would otherwise fail immediately at `is_faulty()`. pub fn reset_for_store_init_retry(&self, endpoint: &Endpoint) { + #[cfg(test)] + self.test_forced_offline.store(false, Ordering::Release); self.reset_for_store_init_retry_at(endpoint, current_unix_time()); } @@ -1142,6 +1157,10 @@ impl DiskHealthTracker { } pub fn mark_recovery_success(&self, endpoint: &Endpoint, reason: &'static str) -> bool { + #[cfg(test)] + if self.test_forced_offline.load(Ordering::Acquire) { + return false; + } let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let current = self.runtime_state(); let next = match current { @@ -1440,6 +1459,11 @@ impl LocalDiskWrapper { self.health.force_runtime_state_for_test(state); } + #[cfg(test)] + pub fn force_offline_for_test(&self) { + self.health.force_offline_for_test(); + } + /// Same as [`DiskHealthTracker::reset_for_store_init_retry`]: undo a transient faulty mark before another format load attempt. pub fn reset_health_for_store_init_retry(&self) { self.health.reset_for_store_init_retry(&self.disk.endpoint()); diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 9a5534e39..75f748a3a 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -979,6 +979,14 @@ impl Disk { Disk::Remote(remote_disk) => remote_disk.force_runtime_state_for_test(state), } } + + #[cfg(test)] + pub fn force_offline_for_test(&self) { + match self { + Disk::Local(local_disk) => local_disk.force_offline_for_test(), + Disk::Remote(remote_disk) => remote_disk.force_offline_for_test(), + } + } } #[derive(Debug)] diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 92ffcc3f6..27f6f5dcd 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -4090,7 +4090,7 @@ mod tests { .collect::>(); for disk in &disks { disk.close().await.expect("fault injection should stop per-disk monitoring"); - disk.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Offline); + disk.force_offline_for_test(); } // Sets has an independent endpoint monitor that renews missing slots. @@ -4127,7 +4127,7 @@ mod tests { .collect::>(); for disk in &disks { disk.close().await.expect("fault injection should stop per-disk monitoring"); - disk.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Offline); + disk.force_offline_for_test(); } set.connect_disks().await; for disk in &disks { diff --git a/docs/architecture/readiness-matrix.md b/docs/architecture/readiness-matrix.md index a6fccc77c..b050be8c0 100644 --- a/docs/architecture/readiness-matrix.md +++ b/docs/architecture/readiness-matrix.md @@ -43,6 +43,9 @@ an unknown or unsupported peer-health snapshot degrades readiness with - A blocked pool metadata writer degrades node and cluster-write readiness with `pool_meta_write_blocked`. Metadata save-gate inspection is bounded to 100 ms; contention reports `pool_metadata_check_timeout` without installing a block. + Node readiness keeps the last confirmed writable observation for the normal + readiness-cache TTL during a transient inspection timeout, while preserving a + confirmed block and failing closed when no fresh observation exists. - The authenticated cluster snapshot extends its existing node-local metadata gate inspection with safe reason, failure phase, and original block time. It distinguishes timeout from a block and changes no admission or recovery @@ -76,7 +79,7 @@ For a healthy IAM and metadata writer in a four-node, one-drive-per-node EC 2+2 | 1 | false | false | true | 503 / false | | All restored | true | true | true | 200 / true | -The node path reads local disk-handle health and reuses the same reachable-host observation as its lock dependency, including the existing `RUSTFS_HEALTH_READINESS_CACHE_TTL_MS` cache. Only `Online` drives count; a reachable host with a `Returning` drive does not yet prove data I/O has recovered. It does not call cluster `storage_info`, local `disk_info`, or add disk-info RPCs. The entire storage inventory snapshot has a separate 100 ms wait budget; expiry reports `storage_readiness_check_timeout` and fails closed. Pool metadata inspection retains its own 100 ms budget. These observations are not an atomic cluster snapshot and do not bypass the existing lock-probe timing or cache policy. +The node path reads local disk-handle health and reuses the same reachable-host observation as its lock dependency, including the existing `RUSTFS_HEALTH_READINESS_CACHE_TTL_MS` cache. Only `Online` drives count; a reachable host with a `Returning` drive does not yet prove data I/O has recovered. It does not call cluster `storage_info`, local `disk_info`, or add disk-info RPCs. The entire storage inventory snapshot has a separate 100 ms wait budget; expiry reports `storage_readiness_check_timeout` and fails closed. Pool metadata inspection retains its own 100 ms budget; a timeout is counted by `rustfs_pool_metadata_check_timeouts_total` and uses the last confirmed node-local gate state only within the same cache TTL. These observations are not an atomic cluster snapshot and do not bypass the existing lock-probe timing or cache policy. The new fields are additive. Their absence in an older response is not evidence of storage quorum. Minimal responses still contain only the existing top-level fields, liveness remains dependency-independent, and HEAD responses remain bodyless. diff --git a/rustfs/src/server/health.rs b/rustfs/src/server/health.rs index ee1747263..b035a3f9a 100644 --- a/rustfs/src/server/health.rs +++ b/rustfs/src/server/health.rs @@ -531,6 +531,22 @@ mod tests { }); } + #[test] + #[serial] + fn writable_node_readiness_keeps_200_for_transient_pool_metadata_timeout() { + let mut report = ready_report(); + report.degraded_reasons = vec![ReadinessDegradedReason::PoolMetadataCheckTimeout]; + + with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false"), || { + let parts = build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&report), "rustfs", None, None); + + assert_eq!(parts.status_code, StatusCode::OK); + let payload = parts.payload.expect("GET readiness body"); + assert_eq!(payload["ready"], true); + assert_eq!(payload["degradedReasons"], json!(["pool_metadata_check_timeout"])); + }); + } + #[tokio::test] async fn readiness_collects_object_stalls_and_recovers_on_completion() { let object_traffic_health = ObjectTrafficHealth::enabled_for_test(Duration::ZERO); diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 1232935f5..d00401679 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -38,7 +38,7 @@ use std::task::{Context, Poll}; use std::time::Duration; use std::{ collections::{HashMap, HashSet}, - sync::OnceLock, + sync::{Mutex as StdMutex, OnceLock}, time::Instant, }; use tokio::sync::Mutex; @@ -75,6 +75,7 @@ fn startup_runtime_readiness_max_wait() -> Duration { } const METRIC_RUNTIME_READINESS_READY: &str = "rustfs_runtime_readiness_ready"; const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_degraded_total"; +const METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL: &str = "rustfs_pool_metadata_check_timeouts_total"; pub use crate::shared_types::{DependencyReadiness, DependencyReadinessReport, ReadinessDegradedReason, StorageReadinessDetails}; @@ -266,6 +267,66 @@ struct StorageWriteReadinessStatus { pool_metadata_reason: Option, } +#[derive(Debug, Clone, Copy, Default)] +struct NodePoolMetadataReadinessCache { + entry: Option, +} + +#[derive(Debug, Clone, Copy)] +struct NodePoolMetadataReadinessCacheEntry { + cached_at: Instant, + observed_at: Instant, + status: StorageWriteReadinessStatus, +} + +impl NodePoolMetadataReadinessCache { + fn observe( + &mut self, + observed: StorageWriteReadinessStatus, + observed_at: Instant, + now: Instant, + ttl: Duration, + ) -> StorageWriteReadinessStatus { + if ttl.is_zero() { + self.entry = None; + return observed; + } + + if observed.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetadataCheckTimeout) { + let Some(previous) = self.entry else { + return observed; + }; + if now.saturating_duration_since(previous.cached_at) > ttl { + return observed; + } + + // A confirmed write block must never be hidden by a later timeout. + if previous.status.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetaWriteBlocked) { + return previous.status; + } + + // A transient inspection outage is diagnostic, not proof that the + // writer became unsafe. Keep the last confirmed writable state. + if previous.status.ready { + return StorageWriteReadinessStatus { + ready: true, + pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout), + }; + } + return observed; + } + + if self.entry.is_none_or(|entry| entry.observed_at <= observed_at) { + self.entry = Some(NodePoolMetadataReadinessCacheEntry { + cached_at: now, + observed_at, + status: observed, + }); + } + observed + } +} + fn pool_metadata_write_readiness(result: Result<(), StorageError>) -> StorageWriteReadinessStatus { match result { Ok(()) => StorageWriteReadinessStatus { @@ -342,6 +403,11 @@ fn health_cluster_timeout() -> Duration { ) } +fn node_pool_metadata_readiness_cache() -> &'static StdMutex { + static CACHE: OnceLock> = OnceLock::new(); + CACHE.get_or_init(|| StdMutex::new(NodePoolMetadataReadinessCache::default())) +} + fn storage_readiness_cache() -> &'static Mutex> { static CACHE: OnceLock>> = OnceLock::new(); CACHE.get_or_init(|| Mutex::new(None)) @@ -446,6 +512,26 @@ async fn update_storage_readiness_cache(status: StorageWriteReadinessStatus) { }); } +fn apply_node_pool_metadata_timeout_policy( + observed: StorageWriteReadinessStatus, + observed_at: Instant, +) -> StorageWriteReadinessStatus { + if observed.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetadataCheckTimeout) { + counter!(METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL).increment(1); + } + let mut cache = node_pool_metadata_readiness_cache() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + cache.observe(observed, observed_at, Instant::now(), health_readiness_cache_ttl()) +} + +#[cfg(test)] +fn reset_node_pool_metadata_readiness_cache() { + *node_pool_metadata_readiness_cache() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = NodePoolMetadataReadinessCache::default(); +} + async fn load_cached_lock_quorum_status() -> Option { let ttl = health_readiness_cache_ttl(); if ttl.is_zero() { @@ -724,6 +810,9 @@ fn record_readiness_report(report: &DependencyReadinessReport) { && report.readiness.peer_health_ready; gauge!(METRIC_RUNTIME_READINESS_READY).set(if ready { 1.0 } else { 0.0 }); for reason in &report.degraded_reasons { + if ready && *reason == ReadinessDegradedReason::PoolMetadataCheckTimeout { + continue; + } counter!(METRIC_RUNTIME_READINESS_DEGRADED_TOTAL, "reason" => reason.as_str()).increment(1); } } @@ -788,7 +877,8 @@ pub async fn collect_node_readiness_report() -> DependencyReadinessReport { let mut details = StorageReadinessDetails::default(); let mut storage_check_timed_out = false; if let Some(store) = runtime_sources::current_object_store_handle() { - storage = pool_metadata_write_readiness(store.pool_meta_write_status().await); + let pool_metadata_status = store.pool_meta_write_status().await; + storage = apply_node_pool_metadata_timeout_policy(pool_metadata_write_readiness(pool_metadata_status), Instant::now()); details.pool_metadata_write_ready = storage.ready; match node_storage_snapshot(store.as_ref(), &lock_observation.online_hosts).await { Ok(info) => { @@ -1172,6 +1262,8 @@ mod tests { use super::*; use crate::storage_api::server::readiness::{DiskOption, new_disk}; + use metrics_util::MetricKind; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use rustfs_madmin::{BackendInfo, Disk}; use serial_test::serial; use std::future; @@ -2276,6 +2368,207 @@ mod tests { assert_eq!(report.degraded_reasons[0].as_str(), "pool_metadata_check_timeout"); } + #[test] + #[serial] + fn sticky_pool_metadata_timeout_uses_dedicated_metric_without_degraded_count() { + reset_node_pool_metadata_readiness_cache(); + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + with_var(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"), || { + let writable = StorageWriteReadinessStatus { + ready: true, + pool_metadata_reason: None, + }; + assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable); + let sticky = apply_node_pool_metadata_timeout_policy( + StorageWriteReadinessStatus { + ready: false, + pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout), + }, + Instant::now(), + ); + assert!(sticky.ready); + record_readiness_report(&DependencyReadinessReport { + readiness: DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + peer_health_ready: true, + }, + degraded_reasons: vec![ReadinessDegradedReason::PoolMetadataCheckTimeout], + storage_details: None, + }); + }); + }); + reset_node_pool_metadata_readiness_cache(); + + let entries = snapshotter.snapshot().into_vec(); + let ready = entries.iter().find_map(|(composite, _, _, value)| { + (composite.kind() == MetricKind::Gauge && composite.key().name() == METRIC_RUNTIME_READINESS_READY).then_some(value) + }); + assert!(matches!(ready, Some(DebugValue::Gauge(value)) if value.into_inner() == 1.0)); + + let degraded = entries.iter().find_map(|(composite, _, _, value)| { + (composite.kind() == MetricKind::Counter + && composite.key().name() == METRIC_RUNTIME_READINESS_DEGRADED_TOTAL + && composite + .key() + .labels() + .any(|label| label.key() == "reason" && label.value() == "pool_metadata_check_timeout")) + .then_some(value) + }); + assert!(degraded.is_none()); + + let timeout = entries.iter().find_map(|(composite, _, _, value)| { + (composite.kind() == MetricKind::Counter && composite.key().name() == METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL) + .then_some(value) + }); + assert!(matches!(timeout, Some(DebugValue::Counter(1)))); + } + + #[tokio::test] + #[serial] + async fn node_pool_metadata_timeout_preserves_recent_writable_observation() { + reset_node_pool_metadata_readiness_cache(); + async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async { + let writable = StorageWriteReadinessStatus { + ready: true, + pool_metadata_reason: None, + }; + assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable); + + let timed_out = apply_node_pool_metadata_timeout_policy( + StorageWriteReadinessStatus { + ready: false, + pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout), + }, + Instant::now(), + ); + + assert!(timed_out.ready); + assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout)); + }) + .await; + reset_node_pool_metadata_readiness_cache(); + } + + #[tokio::test] + #[serial] + async fn node_pool_metadata_timeout_fails_closed_with_empty_cache() { + reset_node_pool_metadata_readiness_cache(); + async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async { + let timed_out = apply_node_pool_metadata_timeout_policy( + StorageWriteReadinessStatus { + ready: false, + pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout), + }, + Instant::now(), + ); + + assert!(!timed_out.ready); + assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout)); + }) + .await; + reset_node_pool_metadata_readiness_cache(); + } + + #[test] + fn node_pool_metadata_timeout_fails_closed_after_cache_ttl() { + let mut cache = NodePoolMetadataReadinessCache::default(); + let observed_at = Instant::now(); + let ttl = Duration::from_secs(60); + let writable = StorageWriteReadinessStatus { + ready: true, + pool_metadata_reason: None, + }; + assert_eq!(cache.observe(writable, observed_at, observed_at, ttl), writable); + + let timed_out = cache.observe( + StorageWriteReadinessStatus { + ready: false, + pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout), + }, + observed_at + Duration::from_secs(61), + observed_at + Duration::from_secs(61), + ttl, + ); + + assert!(!timed_out.ready); + assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout)); + } + + #[tokio::test] + #[serial] + async fn node_pool_metadata_timeout_never_hides_confirmed_block() { + reset_node_pool_metadata_readiness_cache(); + async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async { + let blocked = StorageWriteReadinessStatus { + ready: false, + pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetaWriteBlocked), + }; + let newer_observation = Instant::now(); + assert_eq!(apply_node_pool_metadata_timeout_policy(blocked, newer_observation), blocked); + assert_eq!( + apply_node_pool_metadata_timeout_policy( + StorageWriteReadinessStatus { + ready: true, + pool_metadata_reason: None, + }, + newer_observation + .checked_sub(Duration::from_secs(1)) + .expect("test instant has elapsed time"), + ), + StorageWriteReadinessStatus { + ready: true, + pool_metadata_reason: None, + } + ); + + let timed_out = apply_node_pool_metadata_timeout_policy( + StorageWriteReadinessStatus { + ready: false, + pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout), + }, + Instant::now(), + ); + + assert_eq!(timed_out, blocked); + }) + .await; + reset_node_pool_metadata_readiness_cache(); + } + + #[tokio::test] + #[serial] + async fn node_pool_metadata_timeout_fails_closed_without_fresh_observation() { + reset_node_pool_metadata_readiness_cache(); + async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async { + let writable = StorageWriteReadinessStatus { + ready: true, + pool_metadata_reason: None, + }; + assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable); + + async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("0"))], async { + let timed_out = apply_node_pool_metadata_timeout_policy( + StorageWriteReadinessStatus { + ready: false, + pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout), + }, + Instant::now(), + ); + + assert!(!timed_out.ready); + assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout)); + }) + .await; + }) + .await; + reset_node_pool_metadata_readiness_cache(); + } + #[test] fn degraded_reasons_report_pool_meta_write_blocked() { let readiness = DependencyReadiness {