From 2aa0148454857e5e26d1bb1cb874923b348e52fb Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 11 Aug 2026 10:55:22 +0800 Subject: [PATCH] fix: report stalled object traffic as unready (#5936) * fix: report stalled object traffic as unready * fix: track fully received PUT storage progress --- crates/config/README.md | 8 + crates/config/src/constants/health.rs | 13 + rustfs/src/admin/console.rs | 53 ++- rustfs/src/admin/handlers/health.rs | 4 +- rustfs/src/app/context/global.rs | 13 + rustfs/src/app/gating_test_env.rs | 28 ++ rustfs/src/app/mod.rs | 1 + rustfs/src/app/object_traffic_health.rs | 335 +++++++++++++++++++ rustfs/src/app/object_usecase.rs | 416 ++++++++++++++++++++++-- rustfs/src/server/health.rs | 147 ++++++++- rustfs/src/server/http.rs | 4 +- rustfs/src/server/layer.rs | 153 ++++++++- rustfs/src/server/readiness.rs | 9 + rustfs/src/storage/options.rs | 40 +++ 14 files changed, 1164 insertions(+), 60 deletions(-) create mode 100644 rustfs/src/app/object_traffic_health.rs diff --git a/crates/config/README.md b/crates/config/README.md index 92141d4e8..0c703b57c 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -97,6 +97,14 @@ Current guidance: - enables minimal payload mode for GET health responses (`status`, `ready` only). - `RUSTFS_HEALTH_READINESS_CACHE_TTL_MS` - TTL for readiness cache evaluation. +- `RUSTFS_HEALTH_OBJECT_PROGRESS_ENABLE` + - withdraws readiness when bounded object read/write stages stop completing while requests remain active. + - default is `true`. +- `RUSTFS_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS` + - maximum time without completion in a bounded object stage before readiness is withdrawn. + - default is `30000`; `0` uses the default. + - the effective value is at least 5 seconds longer than `RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT`. + - this readiness SLO is independent of disk read/write failure deadlines and may withdraw traffic before those deadlines expire. - `RUSTFS_HEALTH_COMPAT_BUSY_CHECK_ENABLE` - enables busy protection behavior for health probes. - default is `false`. diff --git a/crates/config/src/constants/health.rs b/crates/config/src/constants/health.rs index 1a83fe326..89be0c036 100644 --- a/crates/config/src/constants/health.rs +++ b/crates/config/src/constants/health.rs @@ -22,6 +22,19 @@ pub const DEFAULT_HEALTH_ENDPOINT_ENABLE: bool = true; pub const ENV_HEALTH_READINESS_CACHE_TTL_MS: &str = "RUSTFS_HEALTH_READINESS_CACHE_TTL_MS"; pub const DEFAULT_HEALTH_READINESS_CACHE_TTL_MS: u64 = 1000; +/// Enable readiness withdrawal when bounded object read/write stages stop +/// completing while requests remain active. +pub const ENV_HEALTH_OBJECT_PROGRESS_ENABLE: &str = "RUSTFS_HEALTH_OBJECT_PROGRESS_ENABLE"; +pub const DEFAULT_HEALTH_OBJECT_PROGRESS_ENABLE: bool = true; + +/// Requested time without completion in a bounded object stage before local +/// readiness is withdrawn (milliseconds). A value of `0` uses the default; +/// runtime adds a safety floor based on the object-lock acquisition timeout. +pub const ENV_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS: &str = "RUSTFS_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS"; +pub const DEFAULT_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS: u64 = 30_000; +/// Additional time beyond the configured object-lock acquisition deadline. +pub const HEALTH_OBJECT_PROGRESS_LOCK_MARGIN_MS: u64 = 5_000; + /// Timeout for cluster health readiness collectors (milliseconds). /// This bounds expensive storage and lock quorum checks used by cluster probes. pub const ENV_HEALTH_CLUSTER_TIMEOUT_MS: &str = "RUSTFS_HEALTH_CLUSTER_TIMEOUT_MS"; diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index 50de63b10..acf820e6b 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -28,7 +28,7 @@ use crate::server::{ }; use crate::version::build; use axum::{ - Json, Router, + Extension, Json, Router, body::Body, extract::Request, middleware, @@ -632,13 +632,22 @@ fn setup_console_middleware_stack( /// # Returns: /// - A `Response` containing the health check result. #[instrument] -async fn health_check(method: Method, uri: Uri) -> Response { +async fn health_check( + method: Method, + uri: Uri, + server_ctx: Option>>, +) -> Response { let probe = if uri.path().strip_prefix(CONSOLE_PREFIX) == Some(HEALTH_READY_PATH) { HealthProbe::Readiness } else { HealthProbe::Liveness }; - let readiness_report = collect_probe_readiness(probe).await; + let app_context = match server_ctx { + Some(Extension(server_ctx)) => server_ctx.installed_app_context(), + None => crate::runtime_sources::current_app_context(), + }; + let object_traffic_health = app_context.map(|context| context.object_traffic_health()); + let readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await; let uptime = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -919,6 +928,44 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn console_readiness_uses_the_request_server_object_progress() { + temp_env::async_with_vars([(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false"))], async { + let object_traffic_health = + Arc::new(crate::app::object_traffic_health::ObjectTrafficHealth::enabled_for_test(Duration::ZERO)); + let stalled = object_traffic_health + .track_write_storage() + .expect("write tracking must be enabled"); + let app_context = + crate::app::gating_test_env::app_context_with_object_traffic_health(Arc::clone(&object_traffic_health)).await; + let server_ctx = crate::runtime_sources::ServerContextSlot::new(); + assert!(server_ctx.install(app_context)); + + let response = health_check( + Method::GET, + format!("{CONSOLE_PREFIX}{HEALTH_READY_PATH}") + .parse() + .expect("console readiness URI"), + Some(Extension(server_ctx)), + ) + .await; + + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + let body = response + .into_body() + .collect() + .await + .expect("console readiness body") + .to_bytes(); + let payload: serde_json::Value = serde_json::from_slice(&body).expect("console readiness JSON"); + assert_eq!(payload["ready"], false); + assert_eq!(payload["degradedReasons"], serde_json::json!(["object_write_stalled"])); + drop(stalled); + }) + .await; + } + // setup_console_middleware_stack reads ENV_HEALTH_ENDPOINT_ENABLE (see above). #[tokio::test] #[serial] diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index b44e2922d..8efd5ba51 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -14,6 +14,7 @@ use super::profile::{TriggerProfileCPU, TriggerProfileMemory}; use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::runtime_sources::app_context_from_req; use crate::server::{ HEALTH_PREFIX, HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, build_health_response_parts, collect_probe_readiness, probe_from_path, @@ -51,6 +52,7 @@ pub struct HealthCheckHandler {} #[async_trait::async_trait] impl Operation for HealthCheckHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let object_traffic_health = app_context_from_req(&req).map(|context| context.object_traffic_health()); // Extract the original HTTP Method (encapsulated by s3s into S3Request) let method = req.method; @@ -66,7 +68,7 @@ impl Operation for HealthCheckHandler { } let probe = probe_from_path(req.uri.path()); - let readiness_report = collect_probe_readiness(probe).await; + let readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await; let response_parts = build_health_response_parts(method.clone(), probe, readiness_report.as_ref(), "rustfs-endpoint", None, None); diff --git a/rustfs/src/app/context/global.rs b/rustfs/src/app/context/global.rs index 84118caba..6a0fb763f 100644 --- a/rustfs/src/app/context/global.rs +++ b/rustfs/src/app/context/global.rs @@ -34,6 +34,7 @@ use super::interfaces::{ ScannerMetricsInterface, ServerConfigInterface, StorageClassInterface, TierConfigInterface, TransitionStateInterface, }; use crate::app::object_data_cache::ObjectDataCacheAdapter; +use crate::app::object_traffic_health::ObjectTrafficHealth; use rustfs_iam::{federation::FederatedIdentityService, store::object::ObjectStore, sys::IamSys}; use rustfs_kms::KmsServiceManager; use std::sync::{Arc, OnceLock}; @@ -74,6 +75,7 @@ pub struct AppContext { storage_class: Arc, buffer_config: Arc, object_data_cache: Arc, + object_traffic_health: Arc, } impl AppContext { @@ -122,6 +124,7 @@ impl AppContext { storage_class: default_storage_class_interface(), buffer_config: default_buffer_config_interface(), object_data_cache, + object_traffic_health: Arc::new(ObjectTrafficHealth::from_env()), } } @@ -137,6 +140,10 @@ impl AppContext { self.object_store.clone() } + pub(crate) fn object_traffic_health(&self) -> Arc { + Arc::clone(&self.object_traffic_health) + } + pub fn iam(&self) -> Arc { self.iam.clone() } @@ -342,9 +349,15 @@ impl AppContext { storage_class: interfaces.storage_class, buffer_config: interfaces.buffer_config, object_data_cache: ObjectDataCacheAdapter::disabled_arc(), + object_traffic_health: Arc::new(ObjectTrafficHealth::from_env()), } } + pub(crate) fn with_test_object_traffic_health(mut self, object_traffic_health: Arc) -> Self { + self.object_traffic_health = object_traffic_health; + self + } + pub(crate) fn with_test_runtime_config_interfaces( mut self, server_config: Arc, diff --git a/rustfs/src/app/gating_test_env.rs b/rustfs/src/app/gating_test_env.rs index 06b8ce007..e25a4471e 100644 --- a/rustfs/src/app/gating_test_env.rs +++ b/rustfs/src/app/gating_test_env.rs @@ -25,6 +25,7 @@ use super::storage_api::test::bucket::metadata_sys; use super::storage_api::test::contract::bucket::{BucketOperations, BucketOptions}; use super::storage_api::test::{ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints}; +use super::{context::AppContext, object_traffic_health::ObjectTrafficHealth}; use std::path::PathBuf; use std::sync::{Arc, OnceLock}; use tempfile::TempDir; @@ -32,6 +33,7 @@ use tokio::fs; use tokio_util::sync::CancellationToken; static SHARED_GATING_ENV: OnceLock<(Vec, Arc, TempDir)> = OnceLock::new(); +static SHARED_GATING_INIT: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); /// Return a shared 4-disk `ECStore` with bucket metadata initialized. /// @@ -42,6 +44,10 @@ pub(crate) async fn shared_gating_ecstore() -> Arc { if let Some((_paths, store, _)) = SHARED_GATING_ENV.get() { return store.clone(); } + let _init_guard = SHARED_GATING_INIT.lock().await; + if let Some((_paths, store, _)) = SHARED_GATING_ENV.get() { + return store.clone(); + } let temp_dir = TempDir::new().expect("create temp dir for gating test env"); let temp_path = temp_dir.path().to_path_buf(); @@ -101,6 +107,28 @@ pub(crate) async fn shared_gating_ecstore() -> Arc { ecstore } +pub(crate) async fn shared_gating_ambient() -> Arc { + let store = shared_gating_ecstore().await; + if let Some(ambient) = crate::runtime_sources::current_app_context() { + return ambient; + } + + let _init_guard = SHARED_GATING_INIT.lock().await; + if crate::runtime_sources::current_app_context().is_none() { + super::runtime_sources::install_test_app_context(Arc::clone(&store)).await; + } + crate::runtime_sources::current_app_context().expect("object traffic test context must be installed") +} + +pub(crate) fn app_context_from_current_environment(ambient: &AppContext) -> AppContext { + AppContext::new(ambient.object_store(), ambient.iam(), ambient.kms()) +} + +pub(crate) async fn app_context_with_object_traffic_health(object_traffic_health: Arc) -> Arc { + let ambient = shared_gating_ambient().await; + Arc::new(app_context_from_current_environment(&ambient).with_test_object_traffic_health(object_traffic_health)) +} + /// Like [`shared_gating_ecstore`], but also returns the backing disk paths so /// tests can remove on-disk shards and simulate the object data vanishing /// mid-stream. diff --git a/rustfs/src/app/mod.rs b/rustfs/src/app/mod.rs index 7d3e05ab7..107ae3e0c 100644 --- a/rustfs/src/app/mod.rs +++ b/rustfs/src/app/mod.rs @@ -21,6 +21,7 @@ pub mod context; pub(crate) mod metadata_route; pub mod multipart_usecase; pub(crate) mod object_data_cache; +pub(crate) mod object_traffic_health; pub mod object_usecase; pub(crate) mod runtime_sources; mod select_object; diff --git a/rustfs/src/app/object_traffic_health.rs b/rustfs/src/app/object_traffic_health.rs new file mode 100644 index 000000000..afa39c923 --- /dev/null +++ b/rustfs/src/app/object_traffic_health.rs @@ -0,0 +1,335 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct ObjectTrafficSnapshot { + pub(crate) read_stalled: bool, + pub(crate) write_stalled: bool, +} + +/// Detects bounded object stages that stop returning from foreground requests. +/// Both success and error returns are progress: dependency correctness remains +/// the responsibility of the existing readiness checks. +#[derive(Debug)] +pub(crate) struct ObjectTrafficHealth { + started_at: Instant, + stall_after_ms: u64, + enabled: bool, + read_metadata: OperationProgress, + read_storage: OperationProgress, + write_storage: OperationProgress, +} + +impl ObjectTrafficHealth { + pub(crate) fn from_env() -> Self { + let enabled = rustfs_utils::get_env_bool( + rustfs_config::ENV_HEALTH_OBJECT_PROGRESS_ENABLE, + rustfs_config::DEFAULT_HEALTH_OBJECT_PROGRESS_ENABLE, + ); + let configured_timeout_ms = rustfs_utils::get_env_u64( + rustfs_config::ENV_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS, + rustfs_config::DEFAULT_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS, + ); + let requested_timeout_ms = if configured_timeout_ms == 0 { + rustfs_config::DEFAULT_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS + } else { + configured_timeout_ms + }; + let minimum_timeout_ms = duration_ms_saturating(crate::storage::get_lock_acquire_timeout()) + .saturating_add(rustfs_config::HEALTH_OBJECT_PROGRESS_LOCK_MARGIN_MS); + let stall_after_ms = requested_timeout_ms.max(minimum_timeout_ms); + + Self::new(enabled, stall_after_ms) + } + + fn new(enabled: bool, stall_after_ms: u64) -> Self { + Self { + started_at: Instant::now(), + stall_after_ms, + enabled, + read_metadata: OperationProgress::default(), + read_storage: OperationProgress::default(), + write_storage: OperationProgress::default(), + } + } + + pub(crate) fn track_read_metadata(&self) -> Option> { + self.track(&self.read_metadata) + } + + pub(crate) fn track_read_storage(&self) -> Option> { + self.track(&self.read_storage) + } + + pub(crate) fn track_write_storage(&self) -> Option> { + self.track(&self.write_storage) + } + + pub(crate) fn snapshot(&self) -> ObjectTrafficSnapshot { + if !self.enabled { + return ObjectTrafficSnapshot::default(); + } + + let now_ms = self.now_ms(); + ObjectTrafficSnapshot { + read_stalled: self.read_metadata.is_stalled_at(now_ms, self.stall_after_ms) + || self.read_storage.is_stalled_at(now_ms, self.stall_after_ms), + write_stalled: self.write_storage.is_stalled_at(now_ms, self.stall_after_ms), + } + } + + fn track<'a>(&'a self, progress: &'a OperationProgress) -> Option> { + if !self.enabled || !progress.begin_at(self.now_ms()) { + return None; + } + Some(ObjectTrafficProgressGuard { health: self, progress }) + } + + fn now_ms(&self) -> u64 { + duration_ms_saturating(self.started_at.elapsed()) + } + + #[cfg(test)] + pub(crate) fn enabled_for_test(stall_after: Duration) -> Self { + Self::new(true, duration_ms_saturating(stall_after)) + } + + #[cfg(test)] + pub(crate) fn read_storage_stalled_for_test(&self) -> bool { + self.read_storage.is_stalled_at(self.now_ms(), self.stall_after_ms) + } +} + +#[derive(Debug, Default)] +struct OperationProgress { + active: AtomicU64, + last_progress_ms: AtomicU64, +} + +impl OperationProgress { + fn begin_at(&self, now_ms: u64) -> bool { + let mut active = self.active.load(Ordering::Relaxed); + loop { + let Some(next) = active.checked_add(1) else { + return false; + }; + if active == 0 { + self.last_progress_ms.fetch_max(now_ms, Ordering::Relaxed); + } + match self + .active + .compare_exchange_weak(active, next, Ordering::Release, Ordering::Relaxed) + { + Ok(_) => return true, + Err(observed) => active = observed, + } + } + } + + fn complete_at(&self, now_ms: u64) { + self.last_progress_ms.fetch_max(now_ms, Ordering::Relaxed); + let previous = self.active.fetch_sub(1, Ordering::Release); + debug_assert!(previous > 0, "object traffic progress guard underflow"); + } + + fn is_stalled_at(&self, now_ms: u64, stall_after_ms: u64) -> bool { + self.active.load(Ordering::Acquire) > 0 + && now_ms.saturating_sub(self.last_progress_ms.load(Ordering::Relaxed)) >= stall_after_ms + } +} + +#[must_use = "dropping the guard records operation completion"] +pub(crate) struct ObjectTrafficProgressGuard<'a> { + health: &'a ObjectTrafficHealth, + progress: &'a OperationProgress, +} + +impl Drop for ObjectTrafficProgressGuard<'_> { + fn drop(&mut self) { + self.progress.complete_at(self.health.now_ms()); + } +} + +fn duration_ms_saturating(duration: Duration) -> u64 { + duration + .as_secs() + .saturating_mul(1_000) + .saturating_add(u64::from(duration.subsec_millis())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_active_operation_stalls_at_the_exact_boundary() { + let progress = OperationProgress::default(); + assert!(progress.begin_at(10)); + + assert!(!progress.is_stalled_at(39, 30)); + assert!(progress.is_stalled_at(40, 30)); + } + + #[test] + fn later_arrivals_do_not_hide_an_existing_stall() { + let progress = OperationProgress::default(); + assert!(progress.begin_at(10)); + assert!(progress.begin_at(35)); + + assert!(progress.is_stalled_at(40, 30)); + } + + #[test] + fn a_completion_resets_progress_until_the_remaining_operation_stalls() { + let progress = OperationProgress::default(); + assert!(progress.begin_at(10)); + assert!(progress.begin_at(20)); + + progress.complete_at(35); + + assert!(!progress.is_stalled_at(64, 30)); + assert!(progress.is_stalled_at(65, 30)); + progress.complete_at(65); + assert!(!progress.is_stalled_at(u64::MAX, 30)); + } + + #[test] + fn a_stale_begin_timestamp_cannot_overwrite_newer_progress() { + let progress = OperationProgress::default(); + assert!(progress.begin_at(10)); + progress.complete_at(100); + + assert!(progress.begin_at(10)); + assert!(!progress.is_stalled_at(129, 30)); + assert!(progress.is_stalled_at(130, 30)); + } + + #[test] + #[serial_test::serial] + fn environment_configuration_is_sanitized() { + temp_env::with_vars( + [ + (rustfs_config::ENV_HEALTH_OBJECT_PROGRESS_ENABLE, Some("false")), + (rustfs_config::ENV_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS, Some("1")), + ], + || { + let minimum_timeout_ms = duration_ms_saturating(crate::storage::get_lock_acquire_timeout()) + .saturating_add(rustfs_config::HEALTH_OBJECT_PROGRESS_LOCK_MARGIN_MS); + let health = ObjectTrafficHealth::from_env(); + assert!(!health.enabled); + assert_eq!(health.stall_after_ms, minimum_timeout_ms); + }, + ); + + temp_env::with_vars([(rustfs_config::ENV_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS, Some("0"))], || { + let minimum_timeout_ms = duration_ms_saturating(crate::storage::get_lock_acquire_timeout()) + .saturating_add(rustfs_config::HEALTH_OBJECT_PROGRESS_LOCK_MARGIN_MS); + let health = ObjectTrafficHealth::from_env(); + assert_eq!( + health.stall_after_ms, + rustfs_config::DEFAULT_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS.max(minimum_timeout_ms) + ); + }); + } + + #[test] + fn read_and_write_progress_are_independent() { + let health = ObjectTrafficHealth::enabled_for_test(Duration::ZERO); + let read = health.track_read_storage().expect("read tracking must be enabled"); + + assert_eq!( + health.snapshot(), + ObjectTrafficSnapshot { + read_stalled: true, + write_stalled: false, + } + ); + + drop(read); + let write = health.track_write_storage().expect("write tracking must be enabled"); + assert_eq!( + health.snapshot(), + ObjectTrafficSnapshot { + read_stalled: false, + write_stalled: true, + } + ); + drop(write); + assert_eq!(health.snapshot(), ObjectTrafficSnapshot::default()); + } + + #[test] + fn disabled_tracking_never_withdraws_readiness() { + let health = ObjectTrafficHealth::new(false, 0); + + assert!(health.track_read_metadata().is_none()); + assert!(health.track_read_storage().is_none()); + assert!(health.track_write_storage().is_none()); + assert_eq!(health.snapshot(), ObjectTrafficSnapshot::default()); + } + + #[test] + fn metadata_completions_do_not_hide_a_storage_stall() { + let health = ObjectTrafficHealth::enabled_for_test(Duration::ZERO); + let storage = health.track_read_storage().expect("read storage tracking must be enabled"); + let metadata = health.track_read_metadata().expect("read metadata tracking must be enabled"); + drop(metadata); + + assert!(health.snapshot().read_stalled); + drop(storage); + } + + #[tokio::test] + async fn aborting_a_tracked_future_clears_the_active_operation() { + let health = std::sync::Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO)); + let task_health = std::sync::Arc::clone(&health); + let task = tokio::spawn(async move { + let _progress = task_health.track_read_storage().expect("read tracking must be enabled"); + std::future::pending::<()>().await; + }); + + if tokio::time::timeout(Duration::from_secs(2), async { + while !health.snapshot().read_stalled { + tokio::task::yield_now().await; + } + }) + .await + .is_err() + { + task.abort(); + let _ = task.await; + panic!("tracked future did not publish an active operation"); + } + task.abort(); + assert!(task.await.expect_err("tracked task must be cancelled").is_cancelled()); + assert!(!health.snapshot().read_stalled); + } + + #[tokio::test] + #[serial_test::serial] + async fn app_context_honors_the_disabled_progress_environment() { + let ambient = crate::app::gating_test_env::shared_gating_ambient().await; + temp_env::async_with_vars([(rustfs_config::ENV_HEALTH_OBJECT_PROGRESS_ENABLE, Some("false"))], async { + let context = crate::app::gating_test_env::app_context_from_current_environment(&ambient); + assert!(context.object_traffic_health().track_read_storage().is_none()); + }) + .await; + + let installed = crate::app::runtime_sources::current_app_context().expect("test AppContext must remain installed"); + assert!(std::sync::Arc::ptr_eq(&ambient, &installed)); + } +} diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index bca116cab..9995dee53 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -222,6 +222,7 @@ use crate::app::object_data_cache::{ }; #[cfg(test)] use crate::app::object_data_cache::{ColdFillRole, ColdFillWaitOutcome, scope_cold_fill_disk_permit_owner_for_test}; +use crate::app::object_traffic_health::ObjectTrafficHealth; type S3StdError = Box; @@ -2951,6 +2952,8 @@ fn normalize_delete_objects_version_id( #[cfg(test)] type DeleteSnapshotTestHook = (String, Arc, Arc); +#[cfg(test)] +type PutPostStoreTestHook = (String, Arc, Arc); #[cfg(test)] static DELETE_SNAPSHOT_TEST_HOOK: OnceLock>> = OnceLock::new(); @@ -2958,6 +2961,8 @@ static DELETE_SNAPSHOT_TEST_HOOK: OnceLock> static DELETE_SOURCE_TEST_HOOK: OnceLock>> = OnceLock::new(); #[cfg(test)] static DELETE_OBJECTS_AUTH_TEST_HOOK: OnceLock>> = OnceLock::new(); +#[cfg(test)] +static PUT_POST_STORE_TEST_HOOK: OnceLock>> = OnceLock::new(); #[cfg(test)] pub(crate) fn install_delete_snapshot_test_hook( @@ -3052,6 +3057,33 @@ async fn wait_for_delete_objects_auth_test_hook(bucket: &str) { } } +#[cfg(test)] +fn install_put_post_store_test_hook(bucket: String, entered: Arc, resume: Arc) { + *PUT_POST_STORE_TEST_HOOK + .get_or_init(|| Mutex::new(None)) + .lock() + .expect("PUT post-store test hook lock should not be poisoned") = Some((bucket, entered, resume)); +} + +#[cfg(test)] +async fn wait_for_put_post_store_test_hook(bucket: &str) { + let hook = { + let mut slot = PUT_POST_STORE_TEST_HOOK + .get_or_init(|| Mutex::new(None)) + .lock() + .expect("PUT post-store test hook lock should not be poisoned"); + if slot.as_ref().is_some_and(|(expected_bucket, _, _)| expected_bucket == bucket) { + slot.take() + } else { + None + } + }; + if let Some((_bucket, entered, resume)) = hook { + entered.wait().await; + resume.wait().await; + } +} + fn build_put_object_expiration_header(event: &lifecycle::Event) -> Option { if !event.action.delete() { return None; @@ -3895,6 +3927,14 @@ pub struct DefaultObjectUsecase { get_object_timeout_policy: Option, } +async fn track_object_read_setup(health: Option<&ObjectTrafficHealth>, future: F) -> F::Output +where + F: std::future::Future, +{ + let _progress = health.and_then(ObjectTrafficHealth::track_read_storage); + future.await +} + impl DefaultObjectUsecase { fn should_use_large_put_concurrency_tuning(size: i64) -> bool { size >= DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES @@ -3951,6 +3991,13 @@ impl DefaultObjectUsecase { current_object_data_cache_for_context(self.context.as_deref()) } + fn object_traffic_health(&self) -> Option> { + self.context + .as_ref() + .map(|context| context.object_traffic_health()) + .or_else(|| current_app_context().map(|context| context.object_traffic_health())) + } + fn base_buffer_size(&self) -> usize { self.context .clone() @@ -4351,6 +4398,7 @@ impl DefaultObjectUsecase { rs: Option, opts: &ObjectOptions, part_number: Option, + object_traffic_health: Option>, ) -> S3Result { let read_start = std::time::Instant::now(); let read_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then_some(read_start); @@ -4366,10 +4414,12 @@ impl DefaultObjectUsecase { key, ) .await?; - let reader = store - .get_object_reader(bucket, key, rs.clone(), req.headers.clone(), opts) - .await - .map_err(map_get_object_reader_error)?; + let reader = track_object_read_setup( + object_traffic_health.as_deref(), + store.get_object_reader(bucket, key, rs.clone(), req.headers.clone(), opts), + ) + .await + .map_err(map_get_object_reader_error)?; let read_setup = Self::finish_get_object_read(req, manager, bucket, key, rs, part_number, read_start, reader, true).await?; return Ok(GetObjectPreparedRead { io_planning, read_setup }); @@ -4390,10 +4440,12 @@ impl DefaultObjectUsecase { .await?, ); let mut prepared = Some( - store - .prepare_get_object_reader(bucket, key, rs.clone(), HeaderMap::new(), opts) - .await - .map_err(map_get_object_reader_error)?, + track_object_read_setup( + object_traffic_health.as_deref(), + store.prepare_get_object_reader(bucket, key, rs.clone(), HeaderMap::new(), opts), + ) + .await + .map_err(map_get_object_reader_error)?, ); let mut cache_fill_allowed = true; let mut legacy_hook_missed = false; @@ -4494,6 +4546,7 @@ impl DefaultObjectUsecase { let headers = &req.headers; let store = &store; let range = &rs; + let object_traffic_health = &object_traffic_health; move |producer| { let adapter = Arc::clone(adapter); let engine_plan = engine_plan.clone(); @@ -4503,6 +4556,7 @@ impl DefaultObjectUsecase { let bucket = bucket.to_owned(); let key = key.to_owned(); let opts = opts.clone(); + let object_traffic_health = object_traffic_health.as_ref().map(Arc::clone); async move { let producer_deadline = producer.deadline(); let cancellation = producer.cancellation_token(); @@ -4548,7 +4602,10 @@ impl DefaultObjectUsecase { } }; - let prepare = store.prepare_get_object_reader(&bucket, &key, range.clone(), HeaderMap::new(), &opts); + let prepare = track_object_read_setup( + object_traffic_health.as_deref(), + store.prepare_get_object_reader(&bucket, &key, range.clone(), HeaderMap::new(), &opts), + ); let prepared = match match await_cold_fill_startup(prepare, &cancellation, producer_deadline).await { Ok(result) => result, Err(ColdFillStartupWaitError::Cancelled) => { @@ -4602,7 +4659,8 @@ impl DefaultObjectUsecase { || { #[cfg(test)] record_cold_fill_reader_open_for_test(&reader_open_plan); - prepared.with_headers(h).into_reader() + let open_reader = prepared.with_headers(h).into_reader(); + async move { track_object_read_setup(object_traffic_health.as_deref(), open_reader).await } }, ColdFillProducerExecution { expected, @@ -4647,11 +4705,12 @@ impl DefaultObjectUsecase { let io_planning = metadata_admission .take() .ok_or_else(|| s3_error!(InternalError, "prepared metadata admission is unavailable"))?; - let reader = prepared - .with_headers(req.headers.clone()) - .into_reader() - .await - .map_err(map_get_object_reader_error)?; + let reader = track_object_read_setup( + object_traffic_health.as_deref(), + prepared.with_headers(req.headers.clone()).into_reader(), + ) + .await + .map_err(map_get_object_reader_error)?; (io_planning, reader) } else { let io_planning = Self::acquire_get_object_io_planning( @@ -4665,19 +4724,25 @@ impl DefaultObjectUsecase { ) .await?; let reader = if legacy_hook_missed { - store - .prepare_get_object_reader(bucket, key, rs.clone(), HeaderMap::new(), opts) - .await - .map_err(map_get_object_reader_error)? - .with_headers(req.headers.clone()) - .into_reader() - .await - .map_err(map_get_object_reader_error)? + let prepared = track_object_read_setup( + object_traffic_health.as_deref(), + store.prepare_get_object_reader(bucket, key, rs.clone(), HeaderMap::new(), opts), + ) + .await + .map_err(map_get_object_reader_error)?; + track_object_read_setup( + object_traffic_health.as_deref(), + prepared.with_headers(req.headers.clone()).into_reader(), + ) + .await + .map_err(map_get_object_reader_error)? } else { - store - .get_object_reader(bucket, key, rs.clone(), req.headers.clone(), opts) - .await - .map_err(map_get_object_reader_error)? + track_object_read_setup( + object_traffic_health.as_deref(), + store.get_object_reader(bucket, key, rs.clone(), req.headers.clone(), opts), + ) + .await + .map_err(map_get_object_reader_error)? }; (io_planning, reader) }; @@ -5485,8 +5550,8 @@ impl DefaultObjectUsecase { debug!("Zero-copy write enabled for {} byte object (bucket={}, key={})", size, bucket, key); } - let use_small_eager_put_path = - should_use_small_eager_put_path(size, &req.headers, server_side_encryption_requested, should_compress, false); + let use_empty_or_small_eager_put_path = size == 0 + || should_use_small_eager_put_path(size, &req.headers, server_side_encryption_requested, should_compress, false); let zero_copy_eager_put_path_status = zero_copy_eager_put_path_status(size, &req.headers, server_side_encryption_requested, should_compress, false); let use_zero_copy_eager_put_path = zero_copy_eager_put_path_status == PUT_EAGER_STATUS_ELIGIBLE; @@ -5498,7 +5563,7 @@ impl DefaultObjectUsecase { "stream_compressed" } else if use_zero_copy_eager_put_path { "zero_copy_eager" - } else if use_small_eager_put_path { + } else if use_empty_or_small_eager_put_path { "small_eager" } else { "streaming" @@ -5712,7 +5777,7 @@ impl DefaultObjectUsecase { let eager_body = read_zero_copy_put_body_exact(body, actual_size as usize).await?; rustfs_io_metrics::record_zero_copy_write(actual_size as usize, zero_copy_start.elapsed().as_secs_f64() * 1000.0); HashReader::from_stream(eager_body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)? - } else if use_small_eager_put_path { + } else if use_empty_or_small_eager_put_path { if (actual_size as usize) <= POOL_BYPASS_MAX_SIZE { // Bypass BytesPool for very small objects to avoid Small-tier // Mutex contention under high concurrency. Direct allocation @@ -5866,6 +5931,14 @@ impl DefaultObjectUsecase { } }); + let object_traffic_health = if use_zero_copy_eager_put_path || use_empty_or_small_eager_put_path { + self.object_traffic_health() + } else { + None + }; + let object_traffic_progress = object_traffic_health + .as_deref() + .and_then(ObjectTrafficHealth::track_write_storage); let (obj_info, backfilled_old_current_size) = match store .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts) .await @@ -5912,6 +5985,9 @@ impl DefaultObjectUsecase { return result; } }; + drop(object_traffic_progress); + #[cfg(test)] + wait_for_put_post_store_test_hook(&bucket).await; maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await; @@ -6352,6 +6428,10 @@ impl DefaultObjectUsecase { // naming nonexistent buckets fail before the versioning lookup in // get_opts. The store comes from the request-bound server context // (backlog#1052 S6), not the process-global handle. + let object_traffic_health = self.object_traffic_health(); + let object_metadata_progress = object_traffic_health + .as_deref() + .and_then(ObjectTrafficHealth::track_read_metadata); let store_lookup_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); let Some(store) = self.object_store() else { lifecycle.finish_err(); @@ -6392,6 +6472,7 @@ impl DefaultObjectUsecase { rs, opts, } = request_context; + drop(object_metadata_progress); let manager = get_concurrency_manager(); @@ -6407,6 +6488,7 @@ impl DefaultObjectUsecase { rs, &opts, part_number, + object_traffic_health, ) .await { @@ -11142,6 +11224,278 @@ mod tests { (store, context) } + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn object_progress_tracks_real_get_and_small_put_lock_waits() { + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + + let object_traffic_health = Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO)); + let context = temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_DATA_CACHE_ENABLE, Some("false"))], async { + crate::app::gating_test_env::app_context_with_object_traffic_health(Arc::clone(&object_traffic_health)).await + }) + .await; + let store = context.object_store(); + let bucket = format!("object-progress-{}", Uuid::new_v4()); + let object = "object.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("object progress bucket must be created"); + put_real_cold_fill_object(&store, &bucket, object, b"initial").await; + + let metadata_entered = Arc::new(tokio::sync::Barrier::new(2)); + let metadata_resume = Arc::new(tokio::sync::Barrier::new(2)); + crate::storage::options::install_versioning_config_test_hook( + bucket.clone(), + Arc::clone(&metadata_entered), + Arc::clone(&metadata_resume), + ); + let metadata_input = GetObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .build() + .expect("metadata GET input must build"); + let metadata_usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context))); + let metadata_get = tokio::spawn(async move { + metadata_usecase + .execute_get_object(build_request(metadata_input, Method::GET)) + .await + }); + tokio::time::timeout(Duration::from_secs(2), metadata_entered.wait()) + .await + .expect("GET must enter the bucket metadata stage"); + assert!(object_traffic_health.snapshot().read_stalled); + assert!(!metadata_get.is_finished(), "GET must still be waiting in bucket metadata"); + metadata_resume.wait().await; + let metadata_response = tokio::time::timeout(Duration::from_secs(10), metadata_get) + .await + .expect("metadata GET must finish after release") + .expect("metadata GET task must join") + .expect("metadata GET must succeed after release"); + assert!(!object_traffic_health.snapshot().read_stalled); + drop(metadata_response); + + let read_lock = store + .new_ns_lock(&bucket, object) + .await + .expect("read test namespace lock must be created") + .get_write_lock(Duration::from_secs(5)) + .await + .expect("read test namespace lock must be held"); + let get_input = GetObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .build() + .expect("GET input must build"); + let get_usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context))); + let get = tokio::spawn(async move { get_usecase.execute_get_object(build_request(get_input, Method::GET)).await }); + tokio::time::timeout(Duration::from_secs(2), async { + while !object_traffic_health.read_storage_stalled_for_test() { + tokio::task::yield_now().await; + } + }) + .await + .expect("blocked GET must publish a storage stall"); + assert!(!get.is_finished(), "GET must still be waiting for the held namespace lock"); + drop(read_lock); + let get_response = tokio::time::timeout(Duration::from_secs(10), get) + .await + .expect("GET must finish after releasing the lock") + .expect("GET task must join") + .expect("GET must succeed after releasing the lock"); + assert!(!object_traffic_health.snapshot().read_stalled); + drop(get_response); + + let write_lock = store + .new_ns_lock(&bucket, object) + .await + .expect("write test namespace lock must be created") + .get_write_lock(Duration::from_secs(5)) + .await + .expect("write test namespace lock must be held"); + let post_store_entered = Arc::new(tokio::sync::Barrier::new(2)); + let post_store_resume = Arc::new(tokio::sync::Barrier::new(2)); + install_put_post_store_test_hook(bucket.clone(), Arc::clone(&post_store_entered), Arc::clone(&post_store_resume)); + let payload = Bytes::from_static(b"replacement"); + let put_input = PutObjectInput::builder() + .bucket(bucket) + .key(object.to_string()) + .body(Some(StreamingBlob::from(s3s::Body::from(payload.clone())))) + .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64"))) + .build() + .expect("PUT input must build"); + let put_usecase = DefaultObjectUsecase::with_context(Some(context)); + let put = tokio::spawn(async move { + put_usecase + .execute_put_object(&FS::new(), build_request(put_input, Method::PUT)) + .await + }); + tokio::time::timeout(Duration::from_secs(2), async { + while !object_traffic_health.snapshot().write_stalled { + tokio::task::yield_now().await; + } + }) + .await + .expect("blocked small PUT must publish a storage stall"); + assert!(!put.is_finished(), "PUT must still be waiting for the held namespace lock"); + drop(write_lock); + tokio::time::timeout(Duration::from_secs(10), post_store_entered.wait()) + .await + .expect("PUT must reach the first post-store hook"); + assert!(!object_traffic_health.snapshot().write_stalled); + assert!(!put.is_finished(), "PUT must remain blocked after the store guard has ended"); + post_store_resume.wait().await; + tokio::time::timeout(Duration::from_secs(10), put) + .await + .expect("PUT must finish after releasing the lock") + .expect("PUT task must join") + .expect("PUT must succeed after releasing the lock"); + let recovered = object_traffic_health.snapshot(); + assert!(!recovered.read_stalled); + assert!(!recovered.write_stalled); + } + + #[tokio::test] + async fn object_progress_tracks_zero_byte_and_zero_copy_put_lock_waits() { + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + + let object_traffic_health = Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO)); + let context = + crate::app::gating_test_env::app_context_with_object_traffic_health(Arc::clone(&object_traffic_health)).await; + let store = context.object_store(); + let bucket = format!("progress-buffered-{}", Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("buffered PUT progress bucket must be created"); + + let extra_body_object = "zero-byte-extra.bin"; + let extra_body_input = PutObjectInput::builder() + .bucket(bucket.clone()) + .key(extra_body_object.to_string()) + .body(Some(StreamingBlob::from(s3s::Body::from(Bytes::from_static(b"x"))))) + .content_length(Some(88)) + .build() + .expect("zero-byte extra-body PUT input must build"); + let extra_body_usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context))); + let mut extra_body_request = build_request(extra_body_input, Method::PUT); + extra_body_request.headers = streaming_headers(Some("0")); + let extra_body_err = extra_body_usecase + .execute_put_object(&FS::new(), extra_body_request) + .await + .expect_err("decoded zero-byte PUT with body data must fail"); + assert_eq!(extra_body_err.code(), &S3ErrorCode::UnexpectedContent); + assert!(!object_traffic_health.snapshot().write_stalled); + let lookup_err = store + .get_object_info(&bucket, extra_body_object, &ObjectOptions::default()) + .await + .expect_err("rejected zero-byte PUT must not create an object"); + assert!(is_err_object_not_found(&lookup_err)); + + let zero_object = "zero-byte.bin"; + let zero_write_lock = store + .new_ns_lock(&bucket, zero_object) + .await + .expect("zero-byte PUT namespace lock must be created") + .get_write_lock(Duration::from_secs(30)) + .await + .expect("zero-byte PUT namespace lock must be held"); + let (body_polled_tx, body_polled_rx) = tokio::sync::oneshot::channel(); + let (body_release_tx, body_release_rx) = tokio::sync::oneshot::channel(); + let pending_zero_body = StreamingBlob::wrap(futures::stream::once(async move { + body_polled_tx.send(()).expect("zero-byte body poll signal must be received"); + body_release_rx.await.expect("zero-byte body EOF must be released"); + Ok::(Bytes::new()) + })); + let zero_input = PutObjectInput::builder() + .bucket(bucket.clone()) + .key(zero_object.to_string()) + .body(Some(pending_zero_body)) + .content_length(Some(87)) + .build() + .expect("zero-byte PUT input must build"); + let zero_usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context))); + let mut zero_request = build_request(zero_input, Method::PUT); + zero_request.headers = streaming_headers(Some("0")); + let zero_put = tokio::spawn(async move { zero_usecase.execute_put_object(&FS::new(), zero_request).await }); + + tokio::time::timeout(Duration::from_secs(30), body_polled_rx) + .await + .expect("zero-byte PUT body must be polled for EOF") + .expect("zero-byte PUT body poll signal must be sent"); + assert!(!object_traffic_health.snapshot().write_stalled); + assert!(!zero_put.is_finished(), "zero-byte PUT must still be waiting for request EOF"); + + body_release_tx.send(()).expect("zero-byte PUT body EOF must be released"); + tokio::time::timeout(Duration::from_secs(30), async { + while !object_traffic_health.snapshot().write_stalled { + tokio::task::yield_now().await; + } + }) + .await + .expect("fully received zero-byte PUT must publish a storage stall"); + assert!(!zero_put.is_finished(), "zero-byte PUT must still be waiting for the held namespace lock"); + + drop(zero_write_lock); + tokio::time::timeout(Duration::from_secs(30), zero_put) + .await + .expect("zero-byte PUT must finish after releasing the lock") + .expect("zero-byte PUT task must join") + .expect("zero-byte PUT must succeed after releasing the lock"); + assert!(!object_traffic_health.snapshot().write_stalled); + + let zero_copy_object = "zero-copy-eager.jpg"; + let zero_copy_payload = Bytes::from(vec![b'z'; 1024 * 1024 + 1]); + let zero_copy_size = i64::try_from(zero_copy_payload.len()).expect("zero-copy payload length must fit i64"); + let zero_copy_headers = HeaderMap::new(); + assert!(!is_disk_compressible(&zero_copy_headers, zero_copy_object)); + assert_eq!( + zero_copy_eager_put_path_status(zero_copy_size, &zero_copy_headers, false, false, false), + PUT_EAGER_STATUS_ELIGIBLE, + "test payload must exercise the production zero-copy eager path", + ); + let zero_copy_write_lock = store + .new_ns_lock(&bucket, zero_copy_object) + .await + .expect("zero-copy PUT namespace lock must be created") + .get_write_lock(Duration::from_secs(30)) + .await + .expect("zero-copy PUT namespace lock must be held"); + let zero_copy_input = PutObjectInput::builder() + .bucket(bucket) + .key(zero_copy_object.to_string()) + .body(Some(StreamingBlob::from(s3s::Body::from(zero_copy_payload)))) + .content_length(Some(zero_copy_size)) + .build() + .expect("zero-copy PUT input must build"); + let zero_copy_usecase = DefaultObjectUsecase::with_context(Some(context)); + let zero_copy_put = tokio::spawn(async move { + zero_copy_usecase + .execute_put_object(&FS::new(), build_request(zero_copy_input, Method::PUT)) + .await + }); + + tokio::time::timeout(Duration::from_secs(30), async { + while !object_traffic_health.snapshot().write_stalled { + tokio::task::yield_now().await; + } + }) + .await + .expect("blocked zero-copy eager PUT must publish a storage stall"); + assert!( + !zero_copy_put.is_finished(), + "zero-copy PUT must still be waiting for the held namespace lock" + ); + + drop(zero_copy_write_lock); + tokio::time::timeout(Duration::from_secs(30), zero_copy_put) + .await + .expect("zero-copy PUT must finish after releasing the lock") + .expect("zero-copy PUT task must join") + .expect("zero-copy PUT must succeed after releasing the lock"); + assert!(!object_traffic_health.snapshot().write_stalled); + } + async fn put_real_cold_fill_object(store: &Arc, bucket: &str, object: &str, body: &[u8]) -> ObjectInfo { let mut reader = PutObjReader::from_vec(body.to_vec()); store diff --git a/rustfs/src/server/health.rs b/rustfs/src/server/health.rs index 5cac23e3f..63de3ccd0 100644 --- a/rustfs/src/server/health.rs +++ b/rustfs/src/server/health.rs @@ -12,11 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::readiness::{DependencyReadinessReport, ReadinessDegradedReason}; +use super::readiness::{DependencyReadinessReport, ReadinessDegradedReason, record_readiness_overlay_reason}; use super::{ HEALTH_READY_PATH, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_READY_PATH, collect_cluster_read_health_report, collect_cluster_write_health_report, collect_node_readiness_report, }; +use crate::app::object_traffic_health::{ObjectTrafficHealth, ObjectTrafficSnapshot}; use http::{Method, StatusCode}; use rustfs_kms::ProbeStatus; use rustfs_kms::probe::{DEFAULT_PROBE_INTERVAL, ENV_KMS_PROBE_INTERVAL_SECS, MIN_PROBE_INTERVAL}; @@ -70,11 +71,33 @@ pub(crate) struct HealthPayloadContext<'a> { pub(crate) include_dependency_details: bool, } -pub(crate) async fn collect_probe_readiness(probe: HealthProbe) -> Option { - match readiness_source_for_probe(probe)? { - HealthReadinessSource::Node => Some(collect_node_readiness_report().await), - HealthReadinessSource::ClusterWrite => Some(collect_cluster_write_health_report().await), - HealthReadinessSource::ClusterRead => Some(collect_cluster_read_health_report().await), +pub(crate) async fn collect_probe_readiness( + probe: HealthProbe, + object_traffic_health: Option<&ObjectTrafficHealth>, +) -> Option { + let mut report = match readiness_source_for_probe(probe)? { + HealthReadinessSource::Node => collect_node_readiness_report().await, + HealthReadinessSource::ClusterWrite => collect_cluster_write_health_report().await, + HealthReadinessSource::ClusterRead => collect_cluster_read_health_report().await, + }; + if probe == HealthProbe::Readiness + && let Some(object_traffic_health) = object_traffic_health + { + apply_object_traffic_snapshot(&mut report, object_traffic_health.snapshot()); + } + Some(report) +} + +fn apply_object_traffic_snapshot(report: &mut DependencyReadinessReport, snapshot: ObjectTrafficSnapshot) { + if snapshot.read_stalled { + let reason = ReadinessDegradedReason::ObjectReadStalled; + report.degraded_reasons.push(reason); + record_readiness_overlay_reason(reason); + } + if snapshot.write_stalled { + let reason = ReadinessDegradedReason::ObjectWriteStalled; + report.degraded_reasons.push(reason); + record_readiness_overlay_reason(reason); } } @@ -300,13 +323,19 @@ pub(crate) fn build_health_response_parts( ), }; - if probe == HealthProbe::Readiness && matches!(kms_ready, Some(false)) { + let object_traffic_stalled = degraded_reasons.iter().any(|reason| { + matches!( + reason, + ReadinessDegradedReason::ObjectReadStalled | ReadinessDegradedReason::ObjectWriteStalled + ) + }); + if probe == HealthProbe::Readiness && (object_traffic_stalled || matches!(kms_ready, Some(false))) { health = HealthCheckState { status_code: StatusCode::SERVICE_UNAVAILABLE, status: "degraded", ready: false, }; - if !degraded_reasons.contains(&ReadinessDegradedReason::KmsNotReady) { + if matches!(kms_ready, Some(false)) && !degraded_reasons.contains(&ReadinessDegradedReason::KmsNotReady) { degraded_reasons.push(ReadinessDegradedReason::KmsNotReady); } } @@ -365,6 +394,8 @@ pub(crate) fn build_health_payload(ctx: HealthPayloadContext<'_>) -> Value { mod tests { use super::super::readiness::DependencyReadiness; use super::*; + use metrics_util::MetricKind; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use rustfs_kms::{ProbeFailureKind, ProbeResult}; use serial_test::serial; use temp_env::with_var; @@ -394,6 +425,106 @@ mod tests { } } + #[tokio::test] + async fn readiness_collects_object_stalls_and_recovers_on_completion() { + let object_traffic_health = ObjectTrafficHealth::enabled_for_test(Duration::ZERO); + let read = object_traffic_health + .track_read_storage() + .expect("read tracking must be enabled"); + let write = object_traffic_health + .track_write_storage() + .expect("write tracking must be enabled"); + + let stalled = collect_probe_readiness(HealthProbe::Readiness, Some(&object_traffic_health)) + .await + .expect("readiness must have a dependency report"); + assert!(stalled.degraded_reasons.contains(&ReadinessDegradedReason::ObjectReadStalled)); + assert!( + stalled + .degraded_reasons + .contains(&ReadinessDegradedReason::ObjectWriteStalled) + ); + + drop(read); + drop(write); + let recovered = collect_probe_readiness(HealthProbe::Readiness, Some(&object_traffic_health)) + .await + .expect("readiness must have a dependency report"); + assert!( + !recovered + .degraded_reasons + .contains(&ReadinessDegradedReason::ObjectReadStalled) + ); + assert!( + !recovered + .degraded_reasons + .contains(&ReadinessDegradedReason::ObjectWriteStalled) + ); + } + + #[test] + #[serial] + fn an_object_stall_degrades_readiness_without_changing_dependency_details() { + with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false"), || { + let mut report = ready_report(); + report.degraded_reasons.push(ReadinessDegradedReason::ObjectReadStalled); + + let parts = + build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&report), "rustfs-endpoint", None, None); + + assert_eq!(parts.status_code, StatusCode::SERVICE_UNAVAILABLE); + let payload = parts.payload.expect("GET should include payload"); + assert_eq!(payload["ready"], false); + assert_eq!(payload["details"]["storage"]["ready"], true); + assert_eq!(payload["degradedReasons"], json!(["object_read_stalled"])); + }); + } + + #[test] + fn object_stalls_do_not_change_liveness() { + let mut report = ready_report(); + report.degraded_reasons.push(ReadinessDegradedReason::ObjectWriteStalled); + + let parts = + build_health_response_parts(Method::HEAD, HealthProbe::Liveness, Some(&report), "rustfs-endpoint", None, None); + + assert_eq!(parts.status_code, StatusCode::OK); + assert!(parts.payload.is_none()); + } + + #[test] + fn object_stall_overlay_records_the_final_readiness_metrics() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + let mut report = ready_report(); + apply_object_traffic_snapshot( + &mut report, + ObjectTrafficSnapshot { + read_stalled: true, + write_stalled: false, + }, + ); + }); + + let entries = snapshotter.snapshot().into_vec(); + let ready = entries.iter().find_map(|(composite, _, _, value)| { + (composite.kind() == MetricKind::Gauge && composite.key().name() == "rustfs_runtime_readiness_ready").then_some(value) + }); + assert!(matches!(ready, Some(DebugValue::Gauge(value)) if value.into_inner() == 0.0)); + + let degraded = entries.iter().find_map(|(composite, _, _, value)| { + (composite.kind() == MetricKind::Counter + && composite.key().name() == "rustfs_runtime_readiness_degraded_total" + && composite + .key() + .labels() + .any(|label| label.key() == "reason" && label.value() == "object_read_stalled")) + .then_some(value) + }); + assert!(matches!(degraded, Some(DebugValue::Counter(1)))); + } + #[tokio::test(start_paused = true)] async fn a_fresh_successful_round_keeps_the_service_ready() { let round_at = Instant::now(); diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 969263f38..c45c25ab4 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -1604,7 +1604,7 @@ fn process_connection( .option_layer(if is_console { Some(RedirectLayer) } else { None }) .layer(BodylessStatusFixLayer) .layer(HeadRequestBodyFixLayer) - .layer(PublicHealthEndpointLayer) + .layer(PublicHealthEndpointLayer::new(Arc::clone(&server_ctx))) .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer)) .layer(DoubleSlashListBucketsCompatLayer) .service(service) @@ -1701,7 +1701,7 @@ fn process_connection( .option_layer(if is_console { Some(RedirectLayer) } else { None }) .layer(BodylessStatusFixLayer) .layer(HeadRequestBodyFixLayer) - .layer(PublicHealthEndpointLayer) + .layer(PublicHealthEndpointLayer::new(Arc::clone(&server_ctx))) .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer)) .layer(DoubleSlashListBucketsCompatLayer) .service(service) diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index c6c889e53..8902b7487 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -14,6 +14,7 @@ use super::runtime_sources; use crate::admin::console::is_console_path; +use crate::app::object_traffic_health::ObjectTrafficHealth; use crate::error::ApiError; use crate::server::RemoteAddr; use crate::server::cors; @@ -1238,19 +1239,31 @@ where } #[derive(Clone)] -pub struct PublicHealthEndpointLayer; +pub struct PublicHealthEndpointLayer { + server_ctx: Arc, +} + +impl PublicHealthEndpointLayer { + pub fn new(server_ctx: Arc) -> Self { + Self { server_ctx } + } +} impl Layer for PublicHealthEndpointLayer { type Service = PublicHealthEndpointService; fn layer(&self, inner: S) -> Self::Service { - PublicHealthEndpointService { inner } + PublicHealthEndpointService { + inner, + server_ctx: Arc::clone(&self.server_ctx), + } } } #[derive(Clone)] pub struct PublicHealthEndpointService { inner: S, + server_ctx: Arc, } fn health_endpoint_enabled() -> bool { @@ -1318,6 +1331,7 @@ async fn health_kms_ready() -> bool { async fn build_public_health_http_response( method: Method, path: String, + object_traffic_health: Option>, ) -> Response> where RestBody: From, @@ -1342,7 +1356,7 @@ where .expect("failed to build health busy response"); } - let readiness_report = collect_probe_readiness(probe).await; + let readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await; let kms_ready = if probe == HealthProbe::Readiness && health_compat_kms_ready_check_enabled() { Some(health_kms_ready().await) } else { @@ -1388,7 +1402,11 @@ where if is_public_health_endpoint_request(method, path) { let method = method.clone(); let path = path.to_owned(); - return Box::pin(async move { Ok(build_public_health_http_response(method, path).await) }); + let object_traffic_health = self + .server_ctx + .installed_app_context() + .map(|context| context.object_traffic_health()); + return Box::pin(async move { Ok(build_public_health_http_response(method, path, object_traffic_health).await) }); } let mut inner = self.inner.clone(); @@ -2185,6 +2203,17 @@ mod tests { use temp_env::{async_with_vars, with_var}; use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt}; + fn public_health_layer() -> PublicHealthEndpointLayer { + PublicHealthEndpointLayer::new(crate::runtime_sources::ServerContextSlot::new()) + } + + async fn public_health_layer_with_tracker(object_traffic_health: Arc) -> PublicHealthEndpointLayer { + let app_context = crate::app::gating_test_env::app_context_with_object_traffic_health(object_traffic_health).await; + let server_ctx = crate::runtime_sources::ServerContextSlot::new(); + assert!(server_ctx.install(app_context)); + PublicHealthEndpointLayer::new(server_ctx) + } + #[derive(Clone, Debug)] struct CaptureService; @@ -2651,7 +2680,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -2846,7 +2875,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -2874,7 +2903,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -2899,7 +2928,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -2924,7 +2953,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -2946,13 +2975,107 @@ mod tests { .await; } + #[tokio::test] + #[serial] + async fn public_readiness_aliases_use_the_installed_object_progress() { + async_with_vars( + [ + (rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true")), + (rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false")), + ], + async { + let object_traffic_health = Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO)); + let stalled = object_traffic_health + .track_read_storage() + .expect("read tracking must be enabled"); + let inner = CountingHybridService::default(); + let calls = inner.calls(); + let mut service = public_health_layer_with_tracker(Arc::clone(&object_traffic_health)) + .await + .layer(inner); + + let response = service + .call( + Request::builder() + .method(Method::GET) + .uri(HEALTH_READY_PATH) + .body(Full::::from(Bytes::new())) + .expect("canonical readiness request"), + ) + .await + .expect("canonical readiness response"); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + let body = BodyExt::collect(response.into_body()) + .await + .expect("readiness body") + .to_bytes(); + let payload: serde_json::Value = serde_json::from_slice(&body).expect("readiness JSON"); + assert_eq!(payload["ready"], false); + assert_eq!(payload["degradedReasons"], serde_json::json!(["object_read_stalled"])); + + let response = service + .call( + Request::builder() + .method(Method::HEAD) + .uri(MINIO_HEALTH_READY_PATH) + .body(Full::::from(Bytes::new())) + .expect("MinIO readiness request"), + ) + .await + .expect("MinIO readiness response"); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert!( + BodyExt::collect(response.into_body()) + .await + .expect("HEAD body") + .to_bytes() + .is_empty() + ); + + let response = service + .call( + Request::builder() + .method(Method::GET) + .uri(HEALTH_COMPAT_LIVE_PATH) + .body(Full::::from(Bytes::new())) + .expect("liveness request"), + ) + .await + .expect("liveness response"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(calls.load(Ordering::SeqCst), 0); + + drop(stalled); + let response = service + .call( + Request::builder() + .method(Method::HEAD) + .uri(HEALTH_READY_PATH) + .body(Full::::from(Bytes::new())) + .expect("recovered readiness request"), + ) + .await + .expect("recovered readiness response"); + assert_eq!(response.status(), StatusCode::OK); + assert!( + BodyExt::collect(response.into_body()) + .await + .expect("HEAD body") + .to_bytes() + .is_empty() + ); + }, + ) + .await; + } + #[tokio::test] #[serial] async fn public_health_endpoint_layer_handles_minio_health_cluster_before_inner_service() { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -2977,7 +3100,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -3002,7 +3125,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("false"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -3027,7 +3150,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("false"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -3069,7 +3192,7 @@ mod tests { async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("false"))], async { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( @@ -3092,7 +3215,7 @@ mod tests { async fn public_health_endpoint_layer_forwards_non_health_requests() { let inner = CountingHybridService::default(); let calls = inner.calls(); - let mut service = PublicHealthEndpointLayer.layer(inner); + let mut service = public_health_layer().layer(inner); let response = service .call( diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index a4064081d..aa2697f67 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -88,6 +88,8 @@ pub enum ReadinessDegradedReason { IamNotReady, LockQuorumUnavailable, KmsNotReady, + ObjectReadStalled, + ObjectWriteStalled, ClusterHealthTimeout, PeerHealthUnavailable, StorageAndIamUnavailable, @@ -103,6 +105,8 @@ impl ReadinessDegradedReason { ReadinessDegradedReason::IamNotReady => "iam_not_ready", ReadinessDegradedReason::LockQuorumUnavailable => "lock_quorum_unavailable", ReadinessDegradedReason::KmsNotReady => "kms_not_ready", + ReadinessDegradedReason::ObjectReadStalled => "object_read_stalled", + ReadinessDegradedReason::ObjectWriteStalled => "object_write_stalled", ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout", ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable", ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable", @@ -714,6 +718,11 @@ fn record_readiness_report(report: &DependencyReadinessReport) { } } +pub(crate) fn record_readiness_overlay_reason(reason: ReadinessDegradedReason) { + gauge!(METRIC_RUNTIME_READINESS_READY).set(0.0); + counter!(METRIC_RUNTIME_READINESS_DEGRADED_TOTAL, "reason" => reason.as_str()).increment(1); +} + fn dependency_readiness_report_from_readiness(readiness: DependencyReadiness) -> DependencyReadinessReport { DependencyReadinessReport { degraded_reasons: degraded_reasons(readiness), diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index 39825963b..64cb40c2d 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -59,11 +59,51 @@ use s3s::dto::VersioningConfiguration; #[cfg(test)] pub(crate) static VERSIONING_CONFIG_LOOKUPS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); +#[cfg(test)] +type VersioningConfigTestHook = (String, std::sync::Arc, std::sync::Arc); + +#[cfg(test)] +static VERSIONING_CONFIG_TEST_HOOK: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + +#[cfg(test)] +pub(crate) fn install_versioning_config_test_hook( + bucket: String, + entered: std::sync::Arc, + resume: std::sync::Arc, +) { + *VERSIONING_CONFIG_TEST_HOOK + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("versioning config test hook lock should not be poisoned") = Some((bucket, entered, resume)); +} + +#[cfg(test)] +async fn wait_for_versioning_config_test_hook(bucket: &str) { + let hook = { + let mut slot = VERSIONING_CONFIG_TEST_HOOK + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("versioning config test hook lock should not be poisoned"); + if slot.as_ref().is_some_and(|(expected_bucket, _, _)| expected_bucket == bucket) { + slot.take() + } else { + None + } + }; + if let Some((_bucket, entered, resume)) = hook { + entered.wait().await; + resume.wait().await; + } +} + /// Fetch the bucket's versioning configuration once so callers can derive /// enabled/suspended state without repeated metadata-sys lookups per request. pub(crate) async fn bucket_versioning_config(bucket: &str) -> VersioningConfiguration { #[cfg(test)] VERSIONING_CONFIG_LOOKUPS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + #[cfg(test)] + wait_for_versioning_config_test_hook(bucket).await; match BucketVersioningSys::get(bucket).await { Ok(cfg) => cfg, Err(err) => {