diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index 19d6f0f49..0f52b8334 100644 --- a/crates/common/src/heal_channel.rs +++ b/crates/common/src/heal_channel.rs @@ -18,7 +18,7 @@ use std::{ fmt::{self, Display}, sync::OnceLock, }; -use tokio::sync::{broadcast, mpsc}; +use tokio::sync::mpsc; use uuid::Uuid; pub const HEAL_DELETE_DANGLING: bool = true; @@ -290,11 +290,6 @@ pub type HealChannelReceiver = mpsc::UnboundedReceiver; /// Global heal channel sender static GLOBAL_HEAL_CHANNEL_SENDER: OnceLock = OnceLock::new(); -type HealResponseSender = broadcast::Sender; - -/// Global heal response broadcaster -static GLOBAL_HEAL_RESPONSE_SENDER: OnceLock = OnceLock::new(); - /// Initialize global heal channel pub fn init_heal_channel() -> HealChannelReceiver { let (tx, rx) = mpsc::unbounded_channel(); @@ -321,23 +316,6 @@ pub async fn send_heal_command(command: HealChannelCommand) -> Result<(), String } } -fn heal_response_sender() -> &'static HealResponseSender { - GLOBAL_HEAL_RESPONSE_SENDER.get_or_init(|| { - let (tx, _rx) = broadcast::channel(1024); - tx - }) -} - -/// Publish a heal response to subscribers. -pub fn publish_heal_response(response: HealChannelResponse) -> Result<(), broadcast::error::SendError> { - heal_response_sender().send(response).map(|_| ()) -} - -/// Subscribe to heal responses. -pub fn subscribe_heal_responses() -> broadcast::Receiver { - heal_response_sender().subscribe() -} - /// Send heal start request pub async fn send_heal_request(request: HealChannelRequest) -> Result<(), String> { send_heal_command(HealChannelCommand::Start(request)).await @@ -536,20 +514,3 @@ pub async fn send_heal_disk(set_disk_id: String, priority: Option 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(); @@ -329,25 +340,11 @@ mod serial_tests { let (_result, error) = heal_storage.heal_format(false).await.expect("Failed to heal format"); assert!(error.is_none(), "Heal format returned error: {error:?}"); - // ─── 2️⃣ wait for format.json to be restored with polling + timeout ─────── - // The minimal scanner interval is clamped to 10s in manager.rs, so we set timeout to 20s - let timeout_duration = Duration::from_secs(20); - let poll_interval = Duration::from_millis(200); + // Wait for task completion + let restored = wait_for_path_exists(&format_path, Duration::from_secs(20)).await; - let result = tokio::time::timeout(timeout_duration, async { - loop { - if format_path.exists() { - break; - } - tokio::time::sleep(poll_interval).await; - } - }) - .await; - - assert!(result.is_ok(), "format.json was not restored within timeout period"); - - // ─── 3️⃣ verify format.json is restored ─────── - assert!(format_path.exists(), "format.json does not exist on disk after heal"); + // ─── 2️⃣ verify format.json is restored ─────── + assert!(restored, "format.json does not exist on disk after heal"); info!("Heal format basic test passed"); } @@ -365,51 +362,37 @@ 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 (_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 + let restored = wait_for_path_exists(&format_path, Duration::from_secs(20)).await; + + // ─── 2️⃣ verify format.json is restored ─────── + assert!(restored, "format.json does not exist on disk after heal"); + // ─── 3 verify each part file is restored ─────── + 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 heal_manager = HealManager::new(heal_storage.clone(), Some(cfg)); - heal_manager.start().await.unwrap(); - - // ─── 2️⃣ wait for format.json and part file to be restored with polling + timeout ─────── - // The minimal scanner interval is clamped to 10s in manager.rs, so we set timeout to 20s - let timeout_duration = Duration::from_secs(20); - let poll_interval = Duration::from_millis(200); - - let result = tokio::time::timeout(timeout_duration, async { - loop { - if format_path.exists() && target_part.exists() { - break; - } - tokio::time::sleep(poll_interval).await; - } - }) - .await; - - assert!(result.is_ok(), "format.json or part file was not restored within timeout period"); - - // ─── 3️⃣ verify format.json is restored ─────── - assert!(format_path.exists(), "format.json does not exist on disk after heal"); - // ─── 4️⃣ verify each part file is restored ─────── - assert!(target_part.exists()); + 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:?}"); // Verify object metadata is accessible let obj_info = ecstore