From eef8f4325185962fd34279b537fcef5b3d85d233 Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 24 Jun 2026 15:48:06 +0800 Subject: [PATCH] fix: harden lock timeouts, health, and heal paths (#3815) --- crates/config/src/constants/object.rs | 13 ++ crates/ecstore/src/rpc/remote_disk.rs | 2 +- crates/ecstore/src/rpc/remote_locker.rs | 136 +++++++++------- crates/heal/src/heal/erasure_healer.rs | 31 +--- crates/heal/src/heal/manager.rs | 32 +++- crates/heal/src/heal/storage.rs | 58 +++++-- crates/heal/src/heal/task.rs | 59 +++++-- crates/scanner/src/scanner_folder.rs | 207 +++++++++++++++++------- rustfs/src/admin/console.rs | 7 +- rustfs/src/admin/handlers/health.rs | 16 +- rustfs/src/server/layer.rs | 6 +- rustfs/src/server/mod.rs | 1 + rustfs/src/server/readiness.rs | 11 ++ 13 files changed, 400 insertions(+), 179 deletions(-) diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index d77189431..bb0d98857 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -281,6 +281,19 @@ pub const ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT: &str = "RUSTFS_OBJECT_LOCK_ACQUIRE_TI /// Default lock acquisition timeout: 5 seconds. pub const DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT: u64 = 5; +/// Environment variable for remote namespace lock RPC transport timeout in milliseconds. +/// +/// This timeout bounds the internode RPC call itself. It is intentionally +/// separate from `RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT` and the distributed lock +/// per-attempt acquire budget so short lock-contention windows do not become +/// aggressive network deadlines. +/// +/// Default: 3000 milliseconds (can be overridden by `RUSTFS_OBJECT_LOCK_RPC_TIMEOUT_MS`). +pub const ENV_OBJECT_LOCK_RPC_TIMEOUT_MS: &str = "RUSTFS_OBJECT_LOCK_RPC_TIMEOUT_MS"; + +/// Default remote lock RPC transport timeout: 3000 milliseconds. +pub const DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS: u64 = 3000; + /// Environment variable to enable object namespace lock diagnostics. /// /// When enabled, RustFS emits slow lock acquisition and long lock hold diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index a96ed8635..ee3675bfc 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -568,7 +568,7 @@ impl RemoteDisk { { // Check if disk is faulty if self.health.is_faulty() { - warn!( + debug!( event = EVENT_REMOTE_DISK_HEALTH, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REMOTE_DISK, diff --git a/crates/ecstore/src/rpc/remote_locker.rs b/crates/ecstore/src/rpc/remote_locker.rs index 8f616da34..dec62b75c 100644 --- a/crates/ecstore/src/rpc/remote_locker.rs +++ b/crates/ecstore/src/rpc/remote_locker.rs @@ -38,6 +38,8 @@ pub struct RemoteClient { } impl RemoteClient { + const RPC_TIMEOUT_GRACE: Duration = Duration::from_millis(500); + pub fn new(endpoint: String) -> Self { Self { addr: endpoint } } @@ -122,12 +124,15 @@ impl RemoteClient { resources.join(", ") } - fn rpc_timeout(timeout_duration: Duration) -> Duration { - if timeout_duration.is_zero() { - Duration::from_millis(1) - } else { - timeout_duration - } + fn rpc_timeout(lock_wait_budget: Duration) -> Duration { + let configured = Duration::from_millis( + rustfs_utils::get_env_u64( + rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, + rustfs_config::DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS, + ) + .max(1), + ); + configured.max(lock_wait_budget.saturating_add(Self::RPC_TIMEOUT_GRACE)) } async fn execute_rpc( @@ -221,7 +226,6 @@ impl RemoteClient { .iter() .map(|request| request.acquire_timeout) .max() - .map(Self::rpc_timeout) .unwrap_or_else(|| Duration::from_millis(1)) } @@ -590,7 +594,8 @@ mod tests { } #[tokio::test] - async fn test_remote_client_acquire_lock_respects_request_timeout_and_evicts_connection() { + #[serial_test::serial] + async fn test_remote_client_acquire_lock_uses_rpc_timeout_and_evicts_connection() { ensure_test_rpc_secret(); let Some((addr, accept_task)) = spawn_hanging_listener().await else { return; @@ -598,35 +603,44 @@ mod tests { cache_lazy_channel(&addr).await; assert!(GLOBAL_CONN_MAP.read().await.contains_key(&addr)); - let client = RemoteClient::new(addr.clone()); - let request = test_lock_request(Duration::from_millis(50)); - let started_at = tokio::time::Instant::now(); + temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async { + let client = RemoteClient::new(addr.clone()); + let request = test_lock_request(Duration::from_millis(5)); + let started_at = tokio::time::Instant::now(); - let response = client.acquire_lock(&request).await.unwrap(); + let response = client.acquire_lock(&request).await.unwrap(); + let elapsed = started_at.elapsed(); - assert!( - started_at.elapsed() < Duration::from_secs(1), - "remote lock RPC should honor request timeout" - ); - assert!(!response.success, "timed out lock acquisition should fail"); - assert!( - response - .error - .as_deref() - .is_some_and(|error| error.contains("Remote lock RPC timed out")), - "expected remote RPC timeout marker, got {:?}", - response.error - ); - assert!( - !GLOBAL_CONN_MAP.read().await.contains_key(&addr), - "timeout should evict cached connection" - ); + assert!( + elapsed >= Duration::from_millis(40), + "remote lock RPC should use configured transport timeout, got {elapsed:?}" + ); + assert!( + elapsed < Duration::from_secs(1), + "test RPC timeout should keep the test fast, got {elapsed:?}" + ); + assert!(!response.success, "timed out lock acquisition should fail"); + assert!( + response + .error + .as_deref() + .is_some_and(|error| error.contains("Remote lock RPC timed out")), + "expected remote RPC timeout marker, got {:?}", + response.error + ); + assert!( + !GLOBAL_CONN_MAP.read().await.contains_key(&addr), + "transport timeout should evict cached connection" + ); + }) + .await; accept_task.abort(); } #[tokio::test] - async fn test_remote_client_acquire_locks_batch_respects_request_timeout_and_evicts_connection() { + #[serial_test::serial] + async fn test_remote_client_acquire_locks_batch_uses_rpc_timeout_and_evicts_connection() { ensure_test_rpc_secret(); let Some((addr, accept_task)) = spawn_hanging_listener().await else { return; @@ -634,37 +648,49 @@ mod tests { cache_lazy_channel(&addr).await; assert!(GLOBAL_CONN_MAP.read().await.contains_key(&addr)); - let client = RemoteClient::new(addr.clone()); - let requests = vec![test_lock_request(Duration::from_millis(50))]; - let started_at = tokio::time::Instant::now(); + temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async { + let client = RemoteClient::new(addr.clone()); + let requests = vec![test_lock_request(Duration::from_millis(5))]; + let started_at = tokio::time::Instant::now(); - let responses = client.acquire_locks_batch(&requests).await.unwrap(); + let responses = client.acquire_locks_batch(&requests).await.unwrap(); + let elapsed = started_at.elapsed(); - assert!( - started_at.elapsed() < Duration::from_secs(1), - "remote batch lock RPC should honor request timeout" - ); - assert_eq!(responses.len(), 1); - assert!(!responses[0].success, "timed out batch lock acquisition should fail"); - assert!( - responses[0] - .error - .as_deref() - .is_some_and(|error| error.contains("Remote lock RPC timed out")), - "expected remote RPC timeout marker, got {:?}", - responses[0].error - ); - assert!( - !GLOBAL_CONN_MAP.read().await.contains_key(&addr), - "batch timeout should evict cached connection" - ); + assert!( + elapsed >= Duration::from_millis(40), + "remote batch lock RPC should use configured transport timeout, got {elapsed:?}" + ); + assert!( + elapsed < Duration::from_secs(1), + "test RPC timeout should keep the test fast, got {elapsed:?}" + ); + assert_eq!(responses.len(), 1); + assert!(!responses[0].success, "timed out batch lock acquisition should fail"); + assert!( + responses[0] + .error + .as_deref() + .is_some_and(|error| error.contains("Remote lock RPC timed out")), + "expected remote RPC timeout marker, got {:?}", + responses[0].error + ); + assert!( + !GLOBAL_CONN_MAP.read().await.contains_key(&addr), + "batch transport timeout should evict cached connection" + ); + }) + .await; accept_task.abort(); } #[test] - fn test_remote_client_zero_timeout_is_clamped() { - assert_eq!(RemoteClient::rpc_timeout(Duration::ZERO), Duration::from_millis(1)); - assert_eq!(RemoteClient::rpc_timeout(Duration::from_millis(25)), Duration::from_millis(25)); + #[serial_test::serial] + fn test_remote_client_rpc_timeout_uses_configured_floor_and_grace() { + temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"), || { + assert_eq!(RemoteClient::rpc_timeout(Duration::ZERO), Duration::from_millis(500)); + assert_eq!(RemoteClient::rpc_timeout(Duration::from_millis(25)), Duration::from_millis(525)); + assert_eq!(RemoteClient::rpc_timeout(Duration::from_secs(1)), Duration::from_millis(1500)); + }); } } diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index ff8e1ddde..3496178f7 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -15,7 +15,7 @@ use crate::heal::{ progress::HealProgress, resume::{CheckpointManager, ResumeManager, ResumeUtils}, - storage::HealStorageAPI, + storage::{HealStorageAPI, next_heal_listing_token}, }; use crate::{Error, Result}; use futures::{StreamExt, future::join_all, stream::FuturesUnordered}; @@ -677,20 +677,7 @@ impl ErasureSetHealer { break; } - continuation_token = next_token; - if continuation_token.is_none() { - warn!( - target: "rustfs::heal::erasure_healer", - event = EVENT_HEAL_ERASURE_BUCKET_STATE, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_ERASURE_HEALER, - set_disk_id, - bucket, - state = "missing_continuation_token", - "Erasure set bucket listing truncated without continuation token" - ); - break; - } + continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?; } Ok(()) @@ -827,19 +814,7 @@ impl ErasureSetHealer { break; } - continuation_token = next_token; - if continuation_token.is_none() { - warn!( - target: "rustfs::heal::erasure_healer", - event = EVENT_HEAL_ERASURE_BUCKET_STATE, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_ERASURE_HEALER, - bucket, - state = "missing_continuation_token", - "Erasure set bucket listing truncated without continuation token" - ); - break; - } + continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?; } // 7. final progress update diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 0d3c07920..2362b1e5b 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -487,14 +487,18 @@ fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool { HealType::Cluster => false, HealType::Object { bucket, object, .. } | HealType::Metadata { bucket, object } - | HealType::ECDecode { bucket, object, .. } => heal_path == bucket || heal_path == format!("{bucket}/{object}"), + | HealType::ECDecode { bucket, object, .. } => heal_path_matches_bucket_child(heal_path, bucket, object), HealType::Bucket { bucket } => heal_path == bucket, - HealType::Prefix { bucket, prefix } => heal_path == bucket || heal_path == format!("{bucket}/{prefix}"), + HealType::Prefix { bucket, prefix } => heal_path_matches_bucket_child(heal_path, bucket, prefix), HealType::ErasureSet { set_disk_id, .. } => heal_path == set_disk_id, HealType::MRF { meta_path } => heal_path == meta_path.trim_matches('/'), } } +fn heal_path_matches_bucket_child(heal_path: &str, bucket: &str, child: &str) -> bool { + heal_path == bucket || heal_path == format!("{bucket}/{child}").trim_matches('/') +} + fn publish_active_heal_count(active_heals: &HashMap>) { crate::set_heal_active_tasks(active_heals.len()); } @@ -3041,6 +3045,30 @@ mod tests { assert!(retry_request_for_result(&task, &result).is_none()); } + #[test] + fn test_heal_type_matches_path_normalizes_prefix_trailing_slash() { + let heal_type = HealType::Prefix { + bucket: "bucket".to_string(), + prefix: "logs/".to_string(), + }; + + assert!(heal_type_matches_path(&heal_type, "bucket")); + assert!(heal_type_matches_path(&heal_type, "bucket/logs")); + assert!(heal_type_matches_path(&heal_type, "bucket/logs/")); + } + + #[test] + fn test_heal_type_matches_path_normalizes_object_trailing_slash() { + let heal_type = HealType::Object { + bucket: "bucket".to_string(), + object: "object/".to_string(), + version_id: None, + }; + + assert!(heal_type_matches_path(&heal_type, "bucket/object")); + assert!(heal_type_matches_path(&heal_type, "bucket/object/")); + } + async fn insert_retrying_request(manager: &HealManager, request: HealRequest) -> CancellationToken { let task_id = request.id.clone(); let cancel_token = CancellationToken::new(); diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index dcd385769..9b44828ec 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -34,6 +34,21 @@ const EVENT_HEAL_STORAGE_OBJECT_VERIFY: &str = "heal_storage_object_verify"; const EVENT_HEAL_STORAGE_ADMIN_OP: &str = "heal_storage_admin_op"; const EVENT_HEAL_STORAGE_REPAIR_OP: &str = "heal_storage_repair_op"; +pub(crate) fn next_heal_listing_token( + bucket: &str, + prefix: &str, + next_token: Option, + is_truncated: bool, +) -> Result> { + if !is_truncated { + return Ok(None); + } + + next_token.map(Some).ok_or_else(|| Error::TaskExecutionFailed { + message: format!("Object listing for {bucket}/{prefix} was truncated without continuation token"), + }) +} + /// Disk status for heal operations #[derive(Debug, Clone, PartialEq, Eq)] pub enum DiskStatus { @@ -1086,21 +1101,7 @@ impl HealStorageAPI for ECStoreHealStorage { break; } - continuation_token = next_token; - if continuation_token.is_none() { - warn!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "list_objects_for_heal", - bucket, - prefix, - state = "missing_continuation_token", - "Heal storage object listing truncated without continuation token" - ); - break; - } + continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?; } debug!( @@ -1233,7 +1234,32 @@ impl HealStorageAPI for ECStoreHealStorage { #[cfg(test)] mod tests { use super::super::StorageError; - use super::{is_transient_object_exists_error, is_transient_object_exists_message}; + use super::{is_transient_object_exists_error, is_transient_object_exists_message, next_heal_listing_token}; + + #[test] + fn next_heal_listing_token_returns_none_for_complete_page() { + assert_eq!( + next_heal_listing_token("bucket", "prefix", None, false).expect("complete page should not fail"), + None + ); + } + + #[test] + fn next_heal_listing_token_returns_token_for_truncated_page() { + assert_eq!( + next_heal_listing_token("bucket", "prefix", Some("token-1".to_string()), true) + .expect("truncated page with token should continue"), + Some("token-1".to_string()) + ); + } + + #[test] + fn next_heal_listing_token_fails_for_truncated_page_without_token() { + let err = next_heal_listing_token("bucket", "prefix", None, true).expect_err("truncated page without token must fail"); + + assert!(matches!(err, super::Error::TaskExecutionFailed { .. })); + assert!(err.to_string().contains("truncated without continuation token")); + } #[test] fn transient_object_exists_message_matches_lock_quorum_failures() { diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 1dbf53647..de6ca2551 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -12,7 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::heal::{ErasureSetHealer, progress::HealProgress, storage::HealStorageAPI}; +use crate::heal::{ + ErasureSetHealer, + progress::HealProgress, + storage::{HealStorageAPI, next_heal_listing_token}, +}; use crate::{Error, Result}; use metrics::{counter, histogram}; use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode}; @@ -1329,20 +1333,7 @@ impl HealTask { break; } - continuation_token = next_token; - if continuation_token.is_none() { - warn!( - target: "rustfs::heal::task", - event = EVENT_HEAL_BUCKET_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - bucket, - result = "missing_continuation_token", - "Heal bucket listing truncated without continuation token" - ); - break; - } + continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?; } if failed > 0 { @@ -2129,6 +2120,7 @@ mod tests { object_heal_opts: Mutex>, format_no_heal_required: Mutex, listed_prefixes: Mutex>, + truncate_without_token: Mutex, } #[async_trait::async_trait] @@ -2246,6 +2238,10 @@ mod tests { continuation_token: Option<&str>, ) -> Result<(Vec, Option, bool)> { self.listed_prefixes.lock().unwrap().push(prefix.to_string()); + if *self.truncate_without_token.lock().unwrap() { + return Ok((vec!["object-a".to_string()], None, true)); + } + let mut listed = self.listed.lock().unwrap(); if continuation_token.is_none() && !*listed { *listed = true; @@ -2302,6 +2298,39 @@ mod tests { assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2); } + #[tokio::test] + async fn test_recursive_bucket_heal_fails_when_listing_lacks_continuation_token() { + let storage = Arc::new(MockStorage { + truncate_without_token: Mutex::new(true), + ..Default::default() + }); + let request = HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + let err = task + .heal_bucket("bucket-a") + .await + .expect_err("recursive bucket heal must fail on incomplete pagination state"); + + assert!(matches!(err, Error::TaskExecutionFailed { .. })); + assert!(err.to_string().contains("truncated without continuation token")); + assert_eq!( + storage.healed_objects.lock().unwrap().as_slice(), + ["object-a".to_string()], + "the already returned page may be processed, but the task must not report success" + ); + } + #[tokio::test] async fn test_cluster_heal_visits_bucket_objects() { let storage = Arc::new(MockStorage::default()); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 530b0ae74..3e6d427e5 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -461,6 +461,19 @@ fn build_object_heal_request( } } +fn resolve_object_heal_entry(entries: &MetaCacheEntries, resolver: MetadataResolutionParams) -> Option { + if let Some(entry) = entries.resolve(resolver) { + return Some(entry); + } + + entries + .as_ref() + .iter() + .flatten() + .find(|entry| !entry.name.ends_with(SLASH_SEPARATOR)) + .cloned() +} + fn heal_priority_label(priority: HealChannelPriority) -> &'static str { match priority { HealChannelPriority::Low => "low", @@ -2072,44 +2085,49 @@ impl FolderScanner { break; } - let entry_option = match entries.resolve(resolver.clone()){ - Some(entry) => { - Some(entry) + let Some(entry) = resolve_object_heal_entry(&entries, resolver.clone()) else { + continue; + }; + + (self.update_current_path)(&entry.name).await; + + if entry.is_dir() { + continue; } - None => { - let (entry,_) = entries.first_found(); - entry - } - }; + let fivs = match entry.file_info_versions(&bucket) { + Ok(fivs) => fivs, + Err(e) => { + error!( + target: "rustfs::scanner::folder", + event = EVENT_SCANNER_FOLDER_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_FOLDER, + bucket = %bucket, + entry = %entry.name, + state = "file_info_versions_failed", + error = %e, + "Scanner list_path_raw failed to resolve file versions" + ); + send_required_scanner_heal_request( + "object", + &bucket, + Some(&entry.name), + build_object_heal_request( + bucket.clone(), + entry.name.clone(), + None, + self.scan_mode, + HealChannelPriority::High, + ), + ) + .await?; + found_objects = true; + continue; + } + }; - let Some(entry) = entry_option else { - break; - }; - - (self.update_current_path)(&entry.name).await; - - if entry.is_dir() { - continue; - } - - - - - let fivs = match entry.file_info_versions(&bucket) { - Ok(fivs) => fivs, - Err(e) => { - error!( - target: "rustfs::scanner::folder", - event = EVENT_SCANNER_FOLDER_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_FOLDER, - bucket = %bucket, - entry = %entry.name, - state = "file_info_versions_failed", - error = %e, - "Scanner list_path_raw failed to resolve file versions" - ); + for fiv in fivs.versions { send_required_scanner_heal_request( "object", &bucket, @@ -2117,34 +2135,14 @@ impl FolderScanner { build_object_heal_request( bucket.clone(), entry.name.clone(), - None, + fiv.version_id.and_then(|v| if v.is_nil() { None } else { Some(v.to_string()) }), self.scan_mode, HealChannelPriority::High, ), ) .await?; found_objects = true; - continue; } - }; - - for fiv in fivs.versions { - - send_required_scanner_heal_request( - "object", - &bucket, - Some(&entry.name), - build_object_heal_request( - bucket.clone(), - entry.name.clone(), - fiv.version_id.and_then(|v| if v.is_nil() { None } else { Some(v.to_string()) }), - self.scan_mode, - HealChannelPriority::High, - ), - ) - .await?; - found_objects = true; - } } @@ -2421,7 +2419,9 @@ mod tests { use super::*; use crate::{DiskOption, Endpoint, new_disk}; - use rustfs_filemeta::{ReplicateObjectInfo, ReplicationType, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType}; + use rustfs_filemeta::{ + FileInfo, FileMeta, ReplicateObjectInfo, ReplicationType, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, + }; use serial_test::serial; #[cfg(unix)] use std::os::unix::fs::{PermissionsExt, symlink}; @@ -3071,6 +3071,99 @@ mod tests { assert_eq!(request.remove_corrupted, Some(HEAL_DELETE_DANGLING)); } + fn metadata_for_object(bucket: &str, object: &str) -> Vec { + let mut meta = FileMeta::new(); + meta.add_version(FileInfo { + volume: bucket.to_string(), + name: object.to_string(), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }) + .expect("test metadata version should be accepted"); + meta.marshal_msg().expect("test metadata should marshal") + } + + fn test_metadata_resolver(bucket: &str) -> MetadataResolutionParams { + MetadataResolutionParams { + bucket: bucket.to_string(), + dir_quorum: 2, + obj_quorum: 2, + ..Default::default() + } + } + + #[test] + fn test_resolve_object_heal_entry_allows_plain_unresolved_fallback() { + let entries = MetaCacheEntries(vec![Some(MetaCacheEntry { + name: "object".to_string(), + metadata: vec![1, 2, 3], + ..Default::default() + })]); + + let entry = resolve_object_heal_entry(&entries, test_metadata_resolver("bucket")) + .expect("plain object fallback should be eligible for heal"); + + assert_eq!(entry.name, "object"); + } + + #[test] + fn test_resolve_object_heal_entry_skips_unresolved_trailing_slash_fallback() { + let entries = MetaCacheEntries(vec![Some(MetaCacheEntry { + name: "object/".to_string(), + metadata: vec![1, 2, 3], + ..Default::default() + })]); + + assert!( + resolve_object_heal_entry(&entries, test_metadata_resolver("bucket")).is_none(), + "unresolved trailing-slash fallback must not be submitted as an object heal" + ); + } + + #[test] + fn test_resolve_object_heal_entry_uses_plain_fallback_after_trailing_slash() { + let entries = MetaCacheEntries(vec![ + Some(MetaCacheEntry { + name: "object/".to_string(), + metadata: vec![1, 2, 3], + ..Default::default() + }), + Some(MetaCacheEntry { + name: "object".to_string(), + metadata: vec![1, 2, 3], + ..Default::default() + }), + ]); + + let entry = resolve_object_heal_entry(&entries, test_metadata_resolver("bucket")) + .expect("plain object fallback should remain eligible after a trailing-slash candidate"); + + assert_eq!(entry.name, "object"); + } + + #[test] + fn test_resolve_object_heal_entry_preserves_resolved_trailing_slash_object() { + let metadata = metadata_for_object("bucket", "object/"); + let entries = MetaCacheEntries(vec![ + Some(MetaCacheEntry { + name: "object/".to_string(), + metadata: metadata.clone(), + ..Default::default() + }), + Some(MetaCacheEntry { + name: "object/".to_string(), + metadata, + ..Default::default() + }), + ]); + + let entry = resolve_object_heal_entry(&entries, test_metadata_resolver("bucket")) + .expect("resolved trailing-slash object should remain eligible"); + + assert_eq!(entry.name, "object/"); + assert!(entry.is_object_dir()); + } + #[test] fn test_effective_object_heal_scan_mode_keeps_normal_when_bitrot_disabled() { let now = OffsetDateTime::now_utc(); diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index 8ec26d1c9..f76678b90 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -19,7 +19,7 @@ use crate::license::has_valid_license; use crate::server::has_path_prefix; use crate::server::{ CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, HeaderMapCarrier, LICENSE, RUSTFS_ADMIN_PREFIX, - RequestContextLayer, VERSION, + RequestContextLayer, VERSION, liveness_dependency_readiness_report, }; use crate::storage::request_context::RequestContext; use crate::version::build; @@ -597,7 +597,10 @@ async fn health_check(method: Method, uri: Uri) -> Response { } else { HealthProbe::Liveness }; - let readiness_report = collect_dependency_readiness().await; + let readiness_report = match probe { + HealthProbe::Liveness => liveness_dependency_readiness_report(), + HealthProbe::Readiness => collect_dependency_readiness().await, + }; let uptime = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index 79ab62d8f..cd795b62c 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -16,7 +16,7 @@ use super::profile::{TriggerProfileCPU, TriggerProfileMemory}; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::server::{ HEALTH_PREFIX, HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, - collect_dependency_readiness_report as collect_runtime_dependency_readiness_report, + collect_dependency_readiness_report as collect_runtime_dependency_readiness_report, liveness_dependency_readiness_report, }; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; @@ -245,7 +245,10 @@ impl Operation for HealthCheckHandler { } let probe = probe_from_path(req.uri.path()); - let readiness_report = collect_dependency_readiness().await; + let readiness_report = match probe { + HealthProbe::Liveness => liveness_dependency_readiness_report(), + HealthProbe::Readiness => collect_dependency_readiness().await, + }; let response_parts = build_health_response_parts(method.clone(), probe, &readiness_report, "rustfs-endpoint", None, None); @@ -292,6 +295,15 @@ mod tests { assert!(!state.ready); } + #[test] + fn test_liveness_dependency_readiness_report_is_lightweight_ready() { + let report = liveness_dependency_readiness_report(); + assert!(report.readiness.storage_ready); + assert!(report.readiness.iam_ready); + assert!(report.readiness.lock_quorum_ready); + assert!(report.degraded_reasons.is_empty()); + } + #[test] fn test_readiness_state_iam_not_ready() { let state = health_check_state(true, false, true, HealthProbe::Readiness); diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 7e446663e..6ac89e84a 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -23,6 +23,7 @@ use crate::server::{ ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, MINIO_HEALTH_LIVE_PATH, MINIO_HEALTH_READY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, collect_dependency_readiness_report, has_path_prefix, is_admin_path, is_table_catalog_path, + liveness_dependency_readiness_report, }; use crate::storage::apply_cors_headers; use crate::storage::request_context::{ @@ -938,7 +939,10 @@ where .expect("failed to build health busy response"); } - let readiness_report = collect_dependency_readiness_report().await; + let readiness_report = match probe { + HealthProbe::Liveness => liveness_dependency_readiness_report(), + HealthProbe::Readiness => collect_dependency_readiness_report().await, + }; let kms_ready = if probe == HealthProbe::Readiness && health_compat_kms_ready_check_enabled() { Some(health_kms_ready().await) } else { diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index fe0765749..bc3ec4d30 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -61,6 +61,7 @@ pub(crate) use readiness::ReadinessDegradedReason; pub(crate) use readiness::ReadinessGateLayer; pub(crate) use readiness::collect_dependency_readiness; pub(crate) use readiness::collect_dependency_readiness_report; +pub(crate) use readiness::liveness_dependency_readiness_report; pub use readiness::publish_ready_when_runtime_ready; pub(crate) use readiness::snapshot_dependency_readiness_report; diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 581c0e456..9fde5629c 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -442,6 +442,17 @@ fn dependency_readiness_report_from_readiness(readiness: DependencyReadiness) -> } } +pub(crate) fn liveness_dependency_readiness_report() -> DependencyReadinessReport { + DependencyReadinessReport { + readiness: DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: Vec::new(), + } +} + pub async fn collect_dependency_readiness() -> DependencyReadiness { collect_dependency_readiness_report().await.readiness }