fix: stabilize head metadata responses and heal tests (#1732)

Signed-off-by: LoganZ2 <103290230+LoganZ2@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
LoganZ2
2026-02-10 09:44:14 +08:00
committed by GitHub
parent 682b5bbb2f
commit ccf3b29df5
7 changed files with 142 additions and 43 deletions
+1
View File
@@ -49,3 +49,4 @@ serial_test = { workspace = true }
tracing-subscriber = { workspace = true }
tempfile = { workspace = true }
walkdir = { workspace = true }
http = { workspace = true }
+71 -31
View File
@@ -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<PathBuf>, Arc<ECStore>, Arc<ECStoreHealStorage>) {
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)]
@@ -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");