From ccf3b29df5261473122468cfc735026f9502d2e9 Mon Sep 17 00:00:00 2001 From: LoganZ2 <103290230+LoganZ2@users.noreply.github.com> Date: Tue, 10 Feb 2026 09:44:14 +0800 Subject: [PATCH] fix: stabilize head metadata responses and heal tests (#1732) Signed-off-by: LoganZ2 <103290230+LoganZ2@users.noreply.github.com> Co-authored-by: houseme Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- Cargo.lock | 1 + crates/heal/Cargo.toml | 1 + crates/heal/tests/heal_integration_test.rs | 102 ++++++++++++------ .../tests/unit/config_tests.rs | 27 ++++- rustfs/src/storage/concurrency.rs | 15 ++- rustfs/src/storage/ecfs.rs | 24 +++++ rustfs/src/storage/options.rs | 15 ++- 7 files changed, 142 insertions(+), 43 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e6d97d1fd..58743d0f3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7148,6 +7148,7 @@ dependencies = [ "anyhow", "async-trait", "futures", + "http 1.4.0", "rustfs-common", "rustfs-config", "rustfs-ecstore", diff --git a/crates/heal/Cargo.toml b/crates/heal/Cargo.toml index a44091e24..423deead3 100644 --- a/crates/heal/Cargo.toml +++ b/crates/heal/Cargo.toml @@ -49,3 +49,4 @@ serial_test = { workspace = true } tracing-subscriber = { workspace = true } tempfile = { workspace = true } walkdir = { workspace = true } +http = { workspace = true } diff --git a/crates/heal/tests/heal_integration_test.rs b/crates/heal/tests/heal_integration_test.rs index 9e9df64e9..bf8e117a6 100644 --- a/crates/heal/tests/heal_integration_test.rs +++ b/crates/heal/tests/heal_integration_test.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use http::HeaderMap; use rustfs_common::heal_channel::{HealOpts, HealScanMode}; use rustfs_ecstore::{ disk::endpoint::Endpoint, @@ -26,7 +27,7 @@ use rustfs_heal::heal::{ }; use serial_test::serial; use std::{ - path::PathBuf, + path::{Path, PathBuf}, sync::{Arc, Once, OnceLock}, time::Duration, }; @@ -48,6 +49,17 @@ pub fn init_tracing() { }); } +async fn wait_for_path_exists(path: &Path, timeout: Duration) -> bool { + let deadline = std::time::Instant::now() + timeout; + while std::time::Instant::now() < deadline { + if path.exists() { + return true; + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + path.exists() +} + /// Test helper: Create test environment with ECStore async fn setup_test_env() -> (Vec, Arc, Arc) { init_tracing(); @@ -229,6 +241,19 @@ mod serial_tests { // ─── 2️⃣ verify each part file is restored ─────── assert!(target_part.exists()); + // ─── 3️⃣ verify object data integrity by actually reading it ─────── + let mut reader = ecstore + .get_object_reader(bucket_name, object_name, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("Failed to get object reader after heal"); + + let mut downloaded_data = Vec::new(); + tokio::io::copy(&mut reader, &mut downloaded_data) + .await + .expect("Failed to read healed object data"); + + assert_eq!(downloaded_data, test_data, "Healed object data does not match original"); + info!("Heal object basic test passed"); } @@ -312,19 +337,14 @@ mod serial_tests { assert!(!format_path.exists(), "format.json still exists after deletion"); println!("✅ Deleted format.json on disk: {format_path:?}"); - // Create heal manager with faster interval - let cfg = HealConfig { - heal_interval: Duration::from_secs(2), - ..Default::default() - }; - let heal_manager = HealManager::new(heal_storage.clone(), Some(cfg)); - heal_manager.start().await.unwrap(); + let (_result, error) = heal_storage.heal_format(false).await.expect("Failed to heal format"); + assert!(error.is_none(), "Heal format returned error: {error:?}"); // Wait for task completion - tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + let restored = wait_for_path_exists(&format_path, Duration::from_secs(20)).await; // ─── 2️⃣ verify format.json is restored ─────── - assert!(format_path.exists(), "format.json does not exist on disk after heal"); + assert!(restored, "format.json does not exist on disk after heal"); info!("Heal format basic test passed"); } @@ -342,39 +362,59 @@ mod serial_tests { create_test_bucket(&ecstore, bucket_name).await; upload_test_object(&ecstore, bucket_name, object_name, test_data).await; - let obj_dir = disk_paths[0].join(bucket_name).join(object_name); - let target_part = WalkDir::new(&obj_dir) - .min_depth(2) - .max_depth(2) - .into_iter() - .filter_map(Result::ok) - .find(|e| e.file_type().is_file() && e.file_name().to_str().map(|n| n.starts_with("part.")).unwrap_or(false)) - .map(|e| e.into_path()) - .expect("Failed to locate part file to delete"); - // ─── 1️⃣ delete format.json on one disk ────────────── let format_path = disk_paths[0].join(".rustfs.sys").join("format.json"); std::fs::remove_dir_all(&disk_paths[0]).expect("failed to delete all contents under disk_paths[0]"); std::fs::create_dir_all(&disk_paths[0]).expect("failed to recreate disk_paths[0] directory"); println!("✅ Deleted format.json on disk: {:?}", disk_paths[0]); - // Create heal manager with faster interval - let cfg = HealConfig { - heal_interval: Duration::from_secs(1), - ..Default::default() - }; - let heal_manager = HealManager::new(heal_storage.clone(), Some(cfg)); - heal_manager.start().await.unwrap(); + let (_result, error) = heal_storage.heal_format(false).await.expect("Failed to heal format"); + assert!(error.is_none(), "Heal format returned error: {error:?}"); // Wait for task completion - tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + let restored = wait_for_path_exists(&format_path, Duration::from_secs(20)).await; // ─── 2️⃣ verify format.json is restored ─────── - assert!(format_path.exists(), "format.json does not exist on disk after heal"); + assert!(restored, "format.json does not exist on disk after heal"); // ─── 3 verify each part file is restored ─────── - assert!(target_part.exists()); + let heal_opts = HealOpts { + recursive: false, + dry_run: false, + remove: false, + recreate: true, + scan_mode: HealScanMode::Normal, + update_parity: true, + no_lock: false, + pool: None, + set: None, + }; + let (_result, error) = heal_storage + .heal_object(bucket_name, object_name, None, &heal_opts) + .await + .expect("Failed to heal object"); + assert!(error.is_none(), "Heal object returned error: {error:?}"); - info!("Heal format basic test passed"); + // Verify object metadata is accessible + let obj_info = ecstore + .get_object_info(bucket_name, object_name, &ObjectOptions::default()) + .await + .expect("Expected object to be readable after heal"); + assert_eq!(obj_info.size as usize, test_data.len()); + + // Actually read the object data to verify integrity + let mut reader = ecstore + .get_object_reader(bucket_name, object_name, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("Failed to get object reader"); + + let mut downloaded_data = Vec::new(); + tokio::io::copy(&mut reader, &mut downloaded_data) + .await + .expect("Failed to read object data"); + + assert_eq!(downloaded_data, test_data, "Healed object data does not match original"); + + info!("Heal format with data test passed"); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] diff --git a/crates/trusted-proxies/tests/unit/config_tests.rs b/crates/trusted-proxies/tests/unit/config_tests.rs index 695e2a9bc..ba05f79e4 100644 --- a/crates/trusted-proxies/tests/unit/config_tests.rs +++ b/crates/trusted-proxies/tests/unit/config_tests.rs @@ -12,16 +12,32 @@ // See the License for the specific language governing permissions and // limitations under the License. -use rustfs_config::{DEFAULT_TRUSTED_PROXY_PROXIES, ENV_TRUSTED_PROXY_PROXIES}; +use rustfs_config::{ + DEFAULT_TRUSTED_PROXY_PROXIES, ENV_TRUSTED_PROXY_ENABLE_RFC7239, ENV_TRUSTED_PROXY_MAX_HOPS, ENV_TRUSTED_PROXY_PROXIES, + ENV_TRUSTED_PROXY_VALIDATION_MODE, +}; use rustfs_trusted_proxies::{ConfigLoader, TrustedProxy, TrustedProxyConfig, ValidationMode}; use std::net::IpAddr; +use std::sync::Mutex; + +static ENV_MUTEX: Mutex<()> = Mutex::new(()); #[test] #[allow(unsafe_code)] fn test_config_loader_default() { + let _guard = ENV_MUTEX.lock().unwrap(); unsafe { std::env::remove_var(ENV_TRUSTED_PROXY_PROXIES); } + unsafe { + std::env::remove_var(ENV_TRUSTED_PROXY_VALIDATION_MODE); + } + unsafe { + std::env::remove_var(ENV_TRUSTED_PROXY_MAX_HOPS); + } + unsafe { + std::env::remove_var(ENV_TRUSTED_PROXY_ENABLE_RFC7239); + } let config = ConfigLoader::from_env_or_default(); assert_eq!(config.server_addr.port(), 9000); assert!(!config.proxy.proxies.is_empty()); @@ -33,14 +49,15 @@ fn test_config_loader_default() { #[test] #[allow(unsafe_code)] fn test_config_loader_env_vars() { + let _guard = ENV_MUTEX.lock().unwrap(); unsafe { std::env::set_var(ENV_TRUSTED_PROXY_PROXIES, "192.168.1.0/24,10.0.0.0/8"); } unsafe { - std::env::set_var("RUSTFS_TRUSTED_PROXY_VALIDATION_MODE", "strict"); + std::env::set_var(ENV_TRUSTED_PROXY_VALIDATION_MODE, "strict"); } unsafe { - std::env::set_var("RUSTFS_TRUSTED_PROXY_MAX_HOPS", "5"); + std::env::set_var(ENV_TRUSTED_PROXY_MAX_HOPS, "5"); } let config = ConfigLoader::from_env(); @@ -54,10 +71,10 @@ fn test_config_loader_env_vars() { std::env::remove_var(ENV_TRUSTED_PROXY_PROXIES); } unsafe { - std::env::remove_var("RUSTFS_TRUSTED_PROXY_VALIDATION_MODE"); + std::env::remove_var(ENV_TRUSTED_PROXY_VALIDATION_MODE); } unsafe { - std::env::remove_var("RUSTFS_TRUSTED_PROXY_MAX_HOPS"); + std::env::remove_var(ENV_TRUSTED_PROXY_MAX_HOPS); } unsafe { std::env::remove_var("SERVER_PORT"); diff --git a/rustfs/src/storage/concurrency.rs b/rustfs/src/storage/concurrency.rs index 33b53f816..0216e9a00 100644 --- a/rustfs/src/storage/concurrency.rs +++ b/rustfs/src/storage/concurrency.rs @@ -426,8 +426,19 @@ impl Drop for GetObjectGuard { /// This ensures accurate tracking even in error/panic scenarios, as Drop /// is called during stack unwinding (unless explicitly forgotten). fn drop(&mut self) { - // Decrement concurrent request counter - ACTIVE_GET_REQUESTS.fetch_sub(1, Ordering::Relaxed); + // Decrement concurrent request counter without underflow. + // If the counter is already 0, `checked_sub` returns None and `fetch_update` + // yields Err(previous). We treat that as a lifecycle bug (e.g., unmatched drop) + // and surface it in debug builds instead of silently discarding it. + if let Err(previous) = + ACTIVE_GET_REQUESTS.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)) + { + debug_assert_eq!( + previous, 0, + "ACTIVE_GET_REQUESTS underflow attempt in GetObjectGuard::drop; previous value = {}", + previous + ); + } // Record Prometheus metrics for monitoring and alerting #[cfg(feature = "metrics")] diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index f6e7fe531..0d68e0942 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -3428,6 +3428,30 @@ impl S3 for FS { // takes precedence for these read operations. let mut response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await; + if let Some(content_disposition) = metadata_map.get("content-disposition") { + // Only set Content-Disposition from metadata if it has not already been set + if response.headers.get(http::header::CONTENT_DISPOSITION).is_none() { + match HeaderValue::from_str(content_disposition) { + Ok(header_value) => { + response.headers.insert(http::header::CONTENT_DISPOSITION, header_value); + } + Err(err) => { + warn!( + "Failed to parse content-disposition metadata value `{}` into HeaderValue: {}", + content_disposition, err + ); + } + } + } + } + + if !response.headers.contains_key(http::header::CONTENT_TYPE) + && let Some(content_type) = metadata_map.get("content-type") + && let Ok(header_value) = HeaderValue::from_str(content_type) + { + response.headers.insert(http::header::CONTENT_TYPE, header_value); + } + // Add x-amz-tagging-count header if object has tags // Per S3 API spec, this header should be present in HEAD object response when tags exist if tag_count > 0 { diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index 2372bc70d..7ec0e26f3 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -390,12 +390,16 @@ pub fn extract_metadata_from_mime_with_object_name( } pub(crate) fn filter_object_metadata(metadata: &HashMap) -> Option> { - // Standard HTTP headers that should NOT be returned in the Metadata field - // These are returned as separate response headers, not user metadata + // HTTP headers that should NOT be returned in the Metadata field. + // These headers are returned as separate response headers, not user metadata. + // + // Note: content-type and content-disposition are intentionally NOT excluded here. + // They remain in the filtered metadata so they can continue to be exposed via + // x-amz-meta-* style user metadata for backward compatibility, while the HEAD + // implementation also mirrors their values into the standard Content-Type and + // Content-Disposition response headers where appropriate. const EXCLUDED_HEADERS: &[&str] = &[ - "content-type", "content-encoding", - "content-disposition", "content-language", "cache-control", "expires", @@ -432,7 +436,8 @@ pub(crate) fn filter_object_metadata(metadata: &HashMap) -> Opti continue; } - // Skip standard HTTP headers (they are returned as separate headers, not metadata) + // Skip excluded HTTP headers (they are returned as separate headers, not metadata) + // Note: content-type and content-disposition are NOT excluded and will be treated as user metadata if EXCLUDED_HEADERS.contains(&lower_key.as_str()) { continue; }