From 69345fe059a7b51d5a6578e38ee081869103278e Mon Sep 17 00:00:00 2001 From: weisd Date: Fri, 22 May 2026 14:38:35 +0800 Subject: [PATCH] fix(scanner): preserve background heal compatibility (#3041) --- crates/common/src/heal_channel.rs | 42 +- crates/config/src/constants/scanner.rs | 31 ++ crates/ecstore/src/set_disk.rs | 41 +- crates/heal/src/error.rs | 3 + crates/heal/src/heal/channel.rs | 330 +++++++++++++- crates/heal/src/heal/manager.rs | 589 ++++++++++++++++++++++++- crates/heal/src/heal/task.rs | 276 +++++++++++- crates/scanner/src/scanner.rs | 122 ++++- crates/scanner/src/scanner_folder.rs | 153 ++++++- rustfs/src/admin/handlers/heal.rs | 398 +++++++++++++++-- rustfs/src/main.rs | 8 +- rustfs/src/server/layer.rs | 80 +++- 12 files changed, 1960 insertions(+), 113 deletions(-) diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index 662eda529..f940d29ef 100644 --- a/crates/common/src/heal_channel.rs +++ b/crates/common/src/heal_channel.rs @@ -260,9 +260,17 @@ pub enum HealChannelCommand { response_tx: oneshot::Sender>, }, /// Query heal task status - Query { heal_path: String, client_token: String }, + Query { + heal_path: String, + client_token: String, + response_tx: oneshot::Sender>, + }, /// Cancel heal task - Cancel { heal_path: String }, + Cancel { + heal_path: String, + client_token: String, + response_tx: oneshot::Sender>, + }, } /// Heal request from admin to ahm @@ -407,14 +415,36 @@ pub async fn send_heal_request(request: HealChannelRequest) -> Result<(), String } } +async fn receive_heal_channel_response( + response_rx: oneshot::Receiver>, +) -> Result { + response_rx + .await + .map_err(|e| format!("Failed to receive heal channel response: {e}"))? +} + /// Send heal query request -pub async fn query_heal_status(heal_path: String, client_token: String) -> Result<(), String> { - send_heal_command(HealChannelCommand::Query { heal_path, client_token }).await +pub async fn query_heal_status(heal_path: String, client_token: String) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + send_heal_command(HealChannelCommand::Query { + heal_path, + client_token, + response_tx, + }) + .await?; + receive_heal_channel_response(response_rx).await } /// Send heal cancel request -pub async fn cancel_heal_task(heal_path: String) -> Result<(), String> { - send_heal_command(HealChannelCommand::Cancel { heal_path }).await +pub async fn cancel_heal_task(heal_path: String, client_token: String) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + send_heal_command(HealChannelCommand::Cancel { + heal_path, + client_token, + response_tx, + }) + .await?; + receive_heal_channel_response(response_rx).await } /// Create a new heal request diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index 0603307b6..2afd15a72 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -40,6 +40,37 @@ pub const ENV_SCANNER_SPEED: &str = "RUSTFS_SCANNER_SPEED"; /// Default scanner speed preset. pub const DEFAULT_SCANNER_SPEED: &str = "default"; +/// Environment variable that specifies the periodic bitrot scan cycle in seconds. +/// When set to `0`, `true`, `on`, or `yes`, every scanner cycle runs in deep mode. +/// When set to `false`, `off`, `no`, or `disabled`, periodic deep scans are disabled. +/// - Unit: seconds (u64). +/// - Example: `export RUSTFS_SCANNER_BITROT_CYCLE_SECS=2592000` (30 days) +pub const ENV_SCANNER_BITROT_CYCLE_SECS: &str = "RUSTFS_SCANNER_BITROT_CYCLE_SECS"; + +/// Default bitrot scan cycle used by the scanner. +pub const DEFAULT_SCANNER_BITROT_CYCLE_SECS: u64 = 30 * 24 * 60 * 60; + +/// Environment variable that controls how many object versions trigger scanner alerts. +/// - Example: `export RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS=100` +pub const ENV_SCANNER_ALERT_EXCESS_VERSIONS: &str = "RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS"; + +/// Default object version count that triggers scanner alerts. +pub const DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS: u64 = 100; + +/// Environment variable that controls how many cumulative bytes across object versions trigger scanner alerts. +/// - Example: `export RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE=1099511627776` +pub const ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE: &str = "RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE"; + +/// Default cumulative object version bytes that trigger scanner alerts. +pub const DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE: u64 = 1024 * 1024 * 1024 * 1024; + +/// Environment variable that controls how many subfolders trigger scanner alerts. +/// - Example: `export RUSTFS_SCANNER_ALERT_EXCESS_FOLDERS=50000` +pub const ENV_SCANNER_ALERT_EXCESS_FOLDERS: &str = "RUSTFS_SCANNER_ALERT_EXCESS_FOLDERS"; + +/// Default subfolder count that triggers scanner alerts. +pub const DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS: u64 = 50_000; + /// Environment variable that controls whether the scanner sleeps between operations. /// When `true` (default), the scanner throttles itself. When `false`, it runs at full speed. /// - Example: `export RUSTFS_SCANNER_IDLE_MODE=false` diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 4c5155a26..d1f04f4c0 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -4129,18 +4129,24 @@ async fn disks_with_all_parts( } } - // Build dataErrsByDisk from dataErrsByPart - for (part, disks) in data_errs_by_part.iter() { - for disk_idx in disks.iter() { - if let Some(parts) = data_errs_by_disk.get_mut(disk_idx) - && *part < parts.len() + populate_data_errs_by_disk(&mut data_errs_by_disk, &data_errs_by_part); + + Ok((data_errs_by_disk, data_errs_by_part)) +} + +fn populate_data_errs_by_disk( + data_errs_by_disk: &mut HashMap>, + data_errs_by_part: &HashMap>, +) { + for (part_index, part_errs) in data_errs_by_part { + for (disk_index, part_err) in part_errs.iter().enumerate() { + if let Some(disk_errs) = data_errs_by_disk.get_mut(&disk_index) + && *part_index < disk_errs.len() { - parts[*part] = disks[*disk_idx]; + disk_errs[*part_index] = *part_err; } } } - - Ok((data_errs_by_disk, data_errs_by_part)) } pub fn should_heal_object_on_disk( @@ -5407,6 +5413,25 @@ mod tests { assert!(should_heal); } + #[test] + fn test_populate_data_errs_by_disk_uses_disk_index_not_error_code() { + let mut data_errs_by_disk = HashMap::from([ + (0, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + (1, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + (2, vec![CHECK_PART_UNKNOWN, CHECK_PART_UNKNOWN]), + ]); + let data_errs_by_part = HashMap::from([ + (0, vec![CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]), + (1, vec![CHECK_PART_SUCCESS, CHECK_PART_FILE_CORRUPT, CHECK_PART_SUCCESS]), + ]); + + populate_data_errs_by_disk(&mut data_errs_by_disk, &data_errs_by_part); + + assert_eq!(data_errs_by_disk.get(&0).unwrap(), &vec![CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS]); + assert_eq!(data_errs_by_disk.get(&1).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_FILE_CORRUPT]); + assert_eq!(data_errs_by_disk.get(&2).unwrap(), &vec![CHECK_PART_SUCCESS, CHECK_PART_SUCCESS]); + } + #[tokio::test] async fn test_get_disks_info_preserves_runtime_state_for_suspect_and_offline_disks() { let format = FormatV3::new(1, 3); diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index 21533ae75..4fae5d7f4 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -59,6 +59,9 @@ pub enum Error { #[error("Heal task already exists: {task_id}")] TaskAlreadyExists { task_id: String }, + #[error("Invalid heal client token")] + InvalidClientToken, + #[error("Heal manager is not running")] ManagerNotRunning, diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 3089be598..0d69120b6 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -13,8 +13,8 @@ // limitations under the License. use crate::heal::{ - manager::HealManager, - task::{HealOptions, HealPriority, HealRequest, HealType}, + manager::{HealManager, HealTaskReport}, + task::{HealOptions, HealPriority, HealRequest, HealTaskStatus, HealType}, utils, }; use crate::{Error, Result}; @@ -22,6 +22,8 @@ use rustfs_common::heal_channel::{ HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealChannelResponse, HealScanMode, publish_heal_response, }; +use rustfs_madmin::heal_commands::HealResultItem; +use serde::Serialize; use std::sync::Arc; use tokio::sync::{mpsc, oneshot}; use tracing::{debug, error, info}; @@ -36,6 +38,12 @@ pub struct HealChannelProcessor { response_receiver: mpsc::UnboundedReceiver, } +#[derive(Serialize)] +struct HealTaskStatusPayload { + summary: String, + items: Vec, +} + impl HealChannelProcessor { /// Create new HealChannelProcessor pub fn new(heal_manager: Arc) -> Self { @@ -83,8 +91,16 @@ impl HealChannelProcessor { async fn process_command(&self, command: HealChannelCommand) -> Result<()> { match command { HealChannelCommand::Start { request, response_tx } => self.process_start_request(request, response_tx).await, - HealChannelCommand::Query { heal_path, client_token } => self.process_query_request(heal_path, client_token).await, - HealChannelCommand::Cancel { heal_path } => self.process_cancel_request(heal_path).await, + HealChannelCommand::Query { + heal_path, + client_token, + response_tx, + } => self.process_query_request(heal_path, client_token, response_tx).await, + HealChannelCommand::Cancel { + heal_path, + client_token, + response_tx, + } => self.process_cancel_request(heal_path, client_token, response_tx).await, } } @@ -166,36 +182,114 @@ impl HealChannelProcessor { } /// Process query request - async fn process_query_request(&self, heal_path: String, client_token: String) -> Result<()> { + async fn process_query_request( + &self, + heal_path: String, + client_token: String, + response_tx: oneshot::Sender>, + ) -> Result<()> { info!("Processing heal query request for path: {}", heal_path); - // TODO: Implement query logic based on heal_path and client_token - // For now, return a placeholder response + let (summary, detail, items) = match self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await { + Ok(HealTaskReport { + status: HealTaskStatus::Pending | HealTaskStatus::Running, + result_items, + }) => ("running".to_string(), None, result_items), + Ok(HealTaskReport { + status: HealTaskStatus::Completed, + result_items, + }) => ("finished".to_string(), None, result_items), + Ok(HealTaskReport { + status: HealTaskStatus::Cancelled, + result_items, + }) => ("stopped".to_string(), Some("heal task cancelled".to_string()), result_items), + Ok(HealTaskReport { + status: HealTaskStatus::Timeout, + result_items, + }) => ("stopped".to_string(), Some("heal task timed out".to_string()), result_items), + Ok(HealTaskReport { + status: HealTaskStatus::Failed { error }, + result_items, + }) => ("stopped".to_string(), Some(error), result_items), + Err(crate::Error::TaskNotFound { .. }) => ("finished".to_string(), None, Vec::new()), + Err(crate::Error::InvalidClientToken) => { + let response = HealChannelResponse { + request_id: client_token, + success: false, + data: None, + error: Some("invalid heal client token".to_string()), + }; + let _ = response_tx.send(Ok(response.clone())); + self.publish_response(response); + return Ok(()); + } + Err(err) => { + let error_text = err.to_string(); + let response = HealChannelResponse { + request_id: client_token, + success: false, + data: None, + error: Some(error_text.clone()), + }; + let _ = response_tx.send(Ok(response.clone())); + self.publish_response(response); + return Ok(()); + } + }; + + let data = serde_json::to_vec(&HealTaskStatusPayload { summary, items }) + .map_err(|e| crate::Error::Serialization(format!("failed to serialize heal task status: {e}")))?; + let response = HealChannelResponse { request_id: client_token, success: true, - data: Some(format!("Query result for path: {heal_path}").into_bytes()), - error: None, + data: Some(data), + error: detail, }; + let _ = response_tx.send(Ok(response.clone())); self.publish_response(response); Ok(()) } /// Process cancel request - async fn process_cancel_request(&self, heal_path: String) -> Result<()> { + async fn process_cancel_request( + &self, + heal_path: String, + client_token: String, + response_tx: oneshot::Sender>, + ) -> Result<()> { info!("Processing heal cancel request for path: {}", heal_path); - // TODO: Implement cancel logic based on heal_path - // For now, return a placeholder response - let response = HealChannelResponse { - request_id: heal_path.clone(), - success: true, - data: Some(format!("Cancel request for path: {heal_path}").into_bytes()), - error: None, + let request_id = if client_token.is_empty() { + heal_path.clone() + } else { + client_token.clone() }; + let cancel_result = if client_token.is_empty() { + self.heal_manager.cancel_tasks_for_path(&heal_path).await.map(|_| ()) + } else { + self.heal_manager.cancel_task(&client_token).await + }; + + let response = match cancel_result { + Ok(()) => HealChannelResponse { + request_id, + success: true, + data: Some("stopped".as_bytes().to_vec()), + error: None, + }, + Err(err) => HealChannelResponse { + request_id, + success: false, + data: None, + error: Some(err.to_string()), + }, + }; + + let _ = response_tx.send(Ok(response.clone())); self.publish_response(response); Ok(()) @@ -256,7 +350,9 @@ impl HealChannelProcessor { options.update_parity = true; } - Ok(HealRequest::new(heal_type, options, priority)) + let mut heal_request = HealRequest::new(heal_type, options, priority); + heal_request.id = request.id; + Ok(heal_request) } fn publish_response(&self, response: HealChannelResponse) { @@ -416,6 +512,7 @@ mod tests { }; let heal_request = processor.convert_to_heal_request(channel_request).unwrap(); + assert_eq!(heal_request.id, "test-id"); assert!(matches!(heal_request.heal_type, HealType::Bucket { .. })); assert_eq!(heal_request.priority, HealPriority::Normal); } @@ -691,4 +788,201 @@ mod tests { .expect("processor should surface invalid request through response channel"); assert!(rx.await.expect("oneshot should resolve").is_err()); } + + #[tokio::test] + async fn test_process_query_request_reports_finished_when_task_is_not_active() { + let heal_manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(heal_manager); + let (tx, rx) = oneshot::channel(); + + processor + .process_query_request("bucket".to_string(), "completed-token".to_string(), tx) + .await + .expect("query should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("query response should be returned"); + assert!(response.success); + assert_eq!(response.request_id, "completed-token"); + let payload: serde_json::Value = + serde_json::from_slice(response.data.as_deref().expect("status payload should be present")) + .expect("status payload should be json"); + assert_eq!(payload["summary"], "finished"); + assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0); + } + + #[tokio::test] + async fn test_process_query_request_reports_running_for_queued_task() { + let heal_manager = create_test_heal_manager(); + let request = HealRequest::bucket("bucket".to_string()); + let task_id = request.id.clone(); + assert_eq!( + heal_manager + .submit_heal_request(request) + .await + .expect("request should be accepted"), + HealAdmissionResult::Accepted + ); + + let processor = HealChannelProcessor::new(heal_manager); + let (tx, rx) = oneshot::channel(); + + processor + .process_query_request("bucket".to_string(), task_id.clone(), tx) + .await + .expect("query should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("query response should be returned"); + assert!(response.success); + assert_eq!(response.request_id, task_id); + let payload: serde_json::Value = + serde_json::from_slice(response.data.as_deref().expect("status payload should be present")) + .expect("status payload should be json"); + assert_eq!(payload["summary"], "running"); + assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0); + } + + #[tokio::test] + async fn test_process_query_request_rejects_wrong_token_for_active_path() { + let heal_manager = create_test_heal_manager(); + let request = HealRequest::bucket("bucket".to_string()); + assert_eq!( + heal_manager + .submit_heal_request(request) + .await + .expect("request should be accepted"), + HealAdmissionResult::Accepted + ); + + let processor = HealChannelProcessor::new(heal_manager); + let (tx, rx) = oneshot::channel(); + + processor + .process_query_request("bucket".to_string(), "wrong-token".to_string(), tx) + .await + .expect("query should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("query response should be returned"); + assert!(!response.success); + assert_eq!(response.request_id, "wrong-token"); + assert_eq!(response.error.as_deref(), Some("invalid heal client token")); + } + + #[tokio::test] + async fn test_process_query_request_empty_path_ignores_unrelated_tasks() { + let heal_manager = create_test_heal_manager(); + heal_manager + .submit_heal_request(HealRequest::bucket("bucket".to_string())) + .await + .expect("request should be accepted"); + + let processor = HealChannelProcessor::new(heal_manager); + let (tx, rx) = oneshot::channel(); + + processor + .process_query_request(String::new(), "wrong-token".to_string(), tx) + .await + .expect("query should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("query response should be returned"); + assert!(response.success); + let payload: serde_json::Value = + serde_json::from_slice(response.data.as_deref().expect("status payload should be present")) + .expect("status payload should be json"); + assert_eq!(payload["summary"], "finished"); + assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0); + } + + #[tokio::test] + async fn test_process_cancel_request_cancels_queued_task_by_token() { + let heal_manager = create_test_heal_manager(); + let request = HealRequest::bucket("bucket".to_string()); + let task_id = request.id.clone(); + heal_manager + .submit_heal_request(request) + .await + .expect("request should be accepted"); + + let processor = HealChannelProcessor::new(heal_manager.clone()); + let (tx, rx) = oneshot::channel(); + + processor + .process_cancel_request("bucket".to_string(), task_id.clone(), tx) + .await + .expect("cancel should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("cancel response should be returned"); + assert!(response.success); + assert_eq!(response.request_id, task_id); + assert_eq!(response.data.as_deref(), Some("stopped".as_bytes())); + assert!(matches!( + heal_manager.get_task_status(&response.request_id).await, + Err(crate::Error::TaskNotFound { .. }) + )); + } + + #[tokio::test] + async fn test_process_cancel_request_cancels_queued_task_by_path() { + let heal_manager = create_test_heal_manager(); + let request = HealRequest::bucket("bucket".to_string()); + let task_id = request.id.clone(); + heal_manager + .submit_heal_request(request) + .await + .expect("request should be accepted"); + + let processor = HealChannelProcessor::new(heal_manager.clone()); + let (tx, rx) = oneshot::channel(); + + processor + .process_cancel_request("bucket".to_string(), String::new(), tx) + .await + .expect("cancel should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("cancel response should be returned"); + assert!(response.success); + assert_eq!(response.request_id, "bucket"); + assert_eq!(response.data.as_deref(), Some("stopped".as_bytes())); + assert!(matches!( + heal_manager.get_task_status(&task_id).await, + Err(crate::Error::TaskNotFound { .. }) + )); + } + + #[tokio::test] + async fn test_process_cancel_request_reports_unknown_task() { + let heal_manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(heal_manager); + let (tx, rx) = oneshot::channel(); + + processor + .process_cancel_request("missing".to_string(), "missing-token".to_string(), tx) + .await + .expect("cancel should process"); + + let response = rx + .await + .expect("oneshot should resolve") + .expect("cancel response should be returned"); + assert!(!response.success); + assert_eq!(response.request_id, "missing-token"); + assert!(response.error.unwrap_or_default().contains("Heal task not found")); + } } diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 800c59ee9..6c30d0231 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -23,6 +23,7 @@ use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; use rustfs_ecstore::disk::DiskAPI; use rustfs_ecstore::disk::error::DiskError; use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP; +use rustfs_madmin::heal_commands::HealResultItem; use std::{ collections::{BinaryHeap, HashMap, HashSet}, sync::Arc, @@ -35,6 +36,8 @@ use tokio::{ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; +const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60); + /// Priority queue wrapper for heal requests /// Uses BinaryHeap for priority-based ordering while maintaining FIFO for same-priority items #[derive(Debug)] @@ -88,6 +91,20 @@ enum QueuePushOutcome { Merged, } +#[derive(Debug, Clone)] +struct CompletedHealStatus { + heal_type: HealType, + status: HealTaskStatus, + result_items: Vec, + completed_at: SystemTime, +} + +#[derive(Debug, Clone)] +pub struct HealTaskReport { + pub status: HealTaskStatus, + pub result_items: Vec, +} + impl PriorityHealQueue { fn new() -> Self { Self { @@ -235,6 +252,78 @@ impl PriorityHealQueue { let key = format!("erasure_set:{set_disk_id}"); self.dedup_keys.contains(&key) } + + fn contains_request_id(&self, request_id: &str) -> bool { + self.heap.iter().any(|item| item.request.id == request_id) + } + + fn contains_request_id_matching_path(&self, request_id: &str, heal_path: &str) -> bool { + self.heap + .iter() + .any(|item| item.request.id == request_id && heal_type_matches_path(&item.request.heal_type, heal_path)) + } + + fn contains_matching(&self, mut matches: F) -> bool + where + F: FnMut(&HealRequest) -> bool, + { + self.heap.iter().any(|item| matches(&item.request)) + } + + fn remove_request_id(&mut self, request_id: &str) -> Option { + let mut retained = BinaryHeap::new(); + let mut removed = None; + + while let Some(item) = self.heap.pop() { + if removed.is_none() && item.request.id == request_id { + let key = Self::make_dedup_key(&item.request); + self.dedup_keys.remove(&key); + removed = Some(item.request); + } else { + retained.push(item); + } + } + + self.heap = retained; + removed + } + + fn remove_matching(&mut self, mut should_remove: F) -> usize + where + F: FnMut(&HealRequest) -> bool, + { + let mut retained = BinaryHeap::new(); + let mut removed_count = 0; + + while let Some(item) = self.heap.pop() { + if should_remove(&item.request) { + let key = Self::make_dedup_key(&item.request); + self.dedup_keys.remove(&key); + removed_count += 1; + } else { + retained.push(item); + } + } + + self.heap = retained; + removed_count + } +} + +fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool { + let heal_path = heal_path.trim_matches('/'); + if heal_path.is_empty() { + return false; + } + + match heal_type { + HealType::Object { bucket, object, .. } + | HealType::Metadata { bucket, object } + | HealType::ECDecode { bucket, object, .. } => heal_path == bucket || heal_path == format!("{bucket}/{object}"), + HealType::Bucket { bucket } => heal_path == bucket, + HealType::ErasureSet { set_disk_id, .. } => heal_path == set_disk_id, + HealType::MRF { meta_path } => heal_path == meta_path.trim_matches('/'), + } } fn publish_active_heal_count(active_heals: &HashMap>) { @@ -357,6 +446,8 @@ pub struct HealManager { active_heals: Arc>>>, /// Heal queue (priority-based) heal_queue: Arc>, + /// Recently completed heal statuses retained for status queries. + completed_heals: Arc>>, /// Storage layer interface storage: Arc, /// Cancel token @@ -384,6 +475,7 @@ impl HealManager { state: Arc::new(RwLock::new(HealState::default())), active_heals: Arc::new(Mutex::new(HashMap::new())), heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())), + completed_heals: Arc::new(Mutex::new(HashMap::new())), storage, cancel_token: CancellationToken::new(), statistics: Arc::new(RwLock::new(HealStatistics::new())), @@ -429,6 +521,7 @@ impl HealManager { } active_heals.clear(); publish_active_heal_count(&active_heals); + self.completed_heals.lock().await.clear(); crate::set_heal_queue_length(0); // update state @@ -544,14 +637,139 @@ impl HealManager { /// Get task status pub async fn get_task_status(&self, task_id: &str) -> Result { - let active_heals = self.active_heals.lock().await; - if let Some(task) = active_heals.get(task_id) { - Ok(task.get_status().await) - } else { - Err(Error::TaskNotFound { - task_id: task_id.to_string(), - }) + { + let active_heals = self.active_heals.lock().await; + if let Some(task) = active_heals.get(task_id) { + return Ok(task.get_status().await); + } } + + let queue = self.heal_queue.lock().await; + if queue.contains_request_id(task_id) { + return Ok(HealTaskStatus::Pending); + } + drop(queue); + + let mut completed_heals = self.completed_heals.lock().await; + prune_completed_heal_statuses(&mut completed_heals); + if let Some(completed) = completed_heals.get(task_id) { + return Ok(completed.status.clone()); + } + + Err(Error::TaskNotFound { + task_id: task_id.to_string(), + }) + } + + pub async fn get_task_report_for_path(&self, heal_path: &str, task_id: &str) -> Result { + { + let active_heals = self.active_heals.lock().await; + if let Some(task) = active_heals.get(task_id) + && heal_type_matches_path(&task.heal_type, heal_path) + { + return Ok(HealTaskReport { + status: task.get_status().await, + result_items: task.get_result_items().await, + }); + } + } + + { + let queue = self.heal_queue.lock().await; + if queue.contains_request_id_matching_path(task_id, heal_path) { + return Ok(HealTaskReport { + status: HealTaskStatus::Pending, + result_items: Vec::new(), + }); + } + } + + { + let mut completed_heals = self.completed_heals.lock().await; + prune_completed_heal_statuses(&mut completed_heals); + if let Some(completed) = completed_heals.get(task_id) + && heal_type_matches_path(&completed.heal_type, heal_path) + { + return Ok(HealTaskReport { + status: completed.status.clone(), + result_items: completed.result_items.clone(), + }); + } + } + + if self.path_has_task(heal_path).await { + return Err(Error::InvalidClientToken); + } + + Err(Error::TaskNotFound { + task_id: task_id.to_string(), + }) + } + + /// Get task status for a path-bound client token. + /// + /// If the token is unknown but no task remains for the path, the caller can + /// treat it as an already-finished sequence. If the path still has a live or + /// recently completed task, a different token is invalid for that path. + pub async fn get_task_status_for_path(&self, heal_path: &str, task_id: &str) -> Result { + { + let active_heals = self.active_heals.lock().await; + if let Some(task) = active_heals.get(task_id) + && heal_type_matches_path(&task.heal_type, heal_path) + { + return Ok(task.get_status().await); + } + } + + { + let queue = self.heal_queue.lock().await; + if queue.contains_request_id_matching_path(task_id, heal_path) { + return Ok(HealTaskStatus::Pending); + } + } + + { + let mut completed_heals = self.completed_heals.lock().await; + prune_completed_heal_statuses(&mut completed_heals); + if let Some(completed) = completed_heals.get(task_id) + && heal_type_matches_path(&completed.heal_type, heal_path) + { + return Ok(completed.status.clone()); + } + } + + if self.path_has_task(heal_path).await { + return Err(Error::InvalidClientToken); + } + + Err(Error::TaskNotFound { + task_id: task_id.to_string(), + }) + } + + async fn path_has_task(&self, heal_path: &str) -> bool { + { + let active_heals = self.active_heals.lock().await; + if active_heals + .values() + .any(|task| heal_type_matches_path(&task.heal_type, heal_path)) + { + return true; + } + } + + { + let queue = self.heal_queue.lock().await; + if queue.contains_matching(|request| heal_type_matches_path(&request.heal_type, heal_path)) { + return true; + } + } + + let mut completed_heals = self.completed_heals.lock().await; + prune_completed_heal_statuses(&mut completed_heals); + completed_heals + .values() + .any(|completed| heal_type_matches_path(&completed.heal_type, heal_path)) } /// Get task progress @@ -574,18 +792,68 @@ impl HealManager { /// Cancel task pub async fn cancel_task(&self, task_id: &str) -> Result<()> { - let mut active_heals = self.active_heals.lock().await; - if let Some(task) = active_heals.get(task_id) { - task.cancel().await?; - active_heals.remove(task_id); - publish_active_heal_count(&active_heals); - info!("Cancelled heal task: {}", task_id); - Ok(()) - } else { - Err(Error::TaskNotFound { - task_id: task_id.to_string(), - }) + { + let mut active_heals = self.active_heals.lock().await; + if let Some(task) = active_heals.get(task_id) { + task.cancel().await?; + active_heals.remove(task_id); + publish_active_heal_count(&active_heals); + info!("Cancelled active heal task: {}", task_id); + return Ok(()); + } } + + let mut queue = self.heal_queue.lock().await; + if queue.remove_request_id(task_id).is_some() { + publish_heal_queue_length(&queue); + info!("Cancelled queued heal task: {}", task_id); + return Ok(()); + } + + Err(Error::TaskNotFound { + task_id: task_id.to_string(), + }) + } + + /// Cancel all queued or active tasks matching a heal path. + pub async fn cancel_tasks_for_path(&self, heal_path: &str) -> Result { + let mut cancelled = 0usize; + + { + let mut active_heals = self.active_heals.lock().await; + let task_ids = active_heals + .iter() + .filter_map(|(task_id, task)| heal_type_matches_path(&task.heal_type, heal_path).then_some(task_id.clone())) + .collect::>(); + + for task_id in task_ids { + if let Some(task) = active_heals.get(&task_id) { + task.cancel().await?; + } + active_heals.remove(&task_id); + cancelled += 1; + } + + if cancelled > 0 { + publish_active_heal_count(&active_heals); + } + } + + let mut queue = self.heal_queue.lock().await; + let queued_cancelled = queue.remove_matching(|request| heal_type_matches_path(&request.heal_type, heal_path)); + if queued_cancelled > 0 { + publish_heal_queue_length(&queue); + cancelled += queued_cancelled; + } + + if cancelled == 0 { + return Err(Error::TaskNotFound { + task_id: heal_path.to_string(), + }); + } + + info!("Cancelled {} heal task(s) for path: {}", cancelled, heal_path); + Ok(cancelled) } /// Get statistics @@ -612,6 +880,7 @@ impl HealManager { let config = self.config.clone(); let heal_queue = self.heal_queue.clone(); let active_heals = self.active_heals.clone(); + let completed_heals = self.completed_heals.clone(); let cancel_token = self.cancel_token.clone(); let statistics = self.statistics.clone(); let storage = self.storage.clone(); @@ -628,10 +897,10 @@ impl HealManager { break; } _ = notify.notified(), if event_driven_scheduler_enable => { - Self::process_heal_queue(&heal_queue, &active_heals, &config, &statistics, &storage, ¬ify).await; + Self::process_heal_queue(&heal_queue, &active_heals, &completed_heals, &config, &statistics, &storage, ¬ify).await; } _ = interval.tick() => { - Self::process_heal_queue(&heal_queue, &active_heals, &config, &statistics, &storage, ¬ify).await; + Self::process_heal_queue(&heal_queue, &active_heals, &completed_heals, &config, &statistics, &storage, ¬ify).await; } } } @@ -768,6 +1037,7 @@ impl HealManager { async fn process_heal_queue( heal_queue: &Arc>, active_heals: &Arc>>>, + completed_heals: &Arc>>, config: &Arc>, statistics: &Arc>, storage: &Arc, @@ -827,6 +1097,7 @@ impl HealManager { publish_active_heal_count(&active_heals_guard); update_task_running_metric_for_task(&active_heals_guard, task.as_ref()); let active_heals_clone = active_heals.clone(); + let completed_heals_clone = completed_heals.clone(); let statistics_clone = statistics.clone(); let notify_clone = notify.clone(); let task_type_label_for_spawn = task_type_label.clone(); @@ -851,9 +1122,19 @@ impl HealManager { if let Some(completed_task) = active_heals_guard.remove(&task_id) { publish_active_heal_count(&active_heals_guard); update_task_running_metric_for_task(&active_heals_guard, completed_task.as_ref()); + let completed_status = completed_task.get_status().await; + let completed_status_entry = CompletedHealStatus { + heal_type: completed_task.heal_type.clone(), + status: completed_status.clone(), + result_items: completed_task.get_result_items().await, + completed_at: SystemTime::now(), + }; + let mut completed_heals_guard = completed_heals_clone.lock().await; + prune_completed_heal_statuses(&mut completed_heals_guard); + completed_heals_guard.insert(task_id.clone(), completed_status_entry); // update statistics let mut stats = statistics_clone.write().await; - match completed_task.get_status().await { + match completed_status { HealTaskStatus::Completed => { stats.update_task_completion(true); } @@ -959,6 +1240,20 @@ fn running_erasure_set_counts(active_heals: &HashMap>) -> running } +fn prune_completed_heal_statuses(completed_heals: &mut HashMap) { + let Ok(now) = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) else { + return; + }; + + completed_heals.retain(|_, completed| { + completed + .completed_at + .duration_since(SystemTime::UNIX_EPOCH) + .map(|completed_at| now.saturating_sub(completed_at) <= KEEP_HEAL_TASK_STATUS_DURATION) + .unwrap_or(false) + }); +} + fn can_schedule_request(request: &HealRequest, running_per_set: &HashMap, max_concurrent_per_set: usize) -> bool { match heal_request_set_key(request) { Some(set_key) => running_per_set.get(&set_key).copied().unwrap_or(0) < max_concurrent_per_set, @@ -1497,6 +1792,258 @@ mod tests { ); } + #[tokio::test] + async fn test_get_task_status_reports_pending_for_queued_request() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let request = HealRequest::bucket("bucket".to_string()); + let request_id = request.id.clone(); + + assert_eq!( + manager + .submit_heal_request(request) + .await + .expect("request should be accepted"), + HealAdmissionResult::Accepted + ); + assert_eq!( + manager + .get_task_status(&request_id) + .await + .expect("queued request should have status"), + HealTaskStatus::Pending + ); + } + + #[tokio::test] + async fn test_get_task_status_for_path_rejects_wrong_token_when_path_is_active() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + manager + .submit_heal_request(HealRequest::bucket("bucket".to_string())) + .await + .expect("request should be accepted"); + + assert!(matches!( + manager.get_task_status_for_path("bucket", "wrong-token").await, + Err(Error::InvalidClientToken) + )); + } + + #[tokio::test] + async fn test_get_task_status_for_path_rejects_token_from_other_active_path() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let bucket_request = HealRequest::bucket("bucket".to_string()); + let other_request = HealRequest::bucket("other".to_string()); + let other_request_id = other_request.id.clone(); + + manager + .submit_heal_request(bucket_request) + .await + .expect("bucket request should be accepted"); + manager + .submit_heal_request(other_request) + .await + .expect("other request should be accepted"); + + assert!(matches!( + manager.get_task_status_for_path("bucket", &other_request_id).await, + Err(Error::InvalidClientToken) + )); + } + + #[tokio::test] + async fn test_get_task_status_for_path_does_not_accept_token_from_inactive_path() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let request = HealRequest::bucket("bucket".to_string()); + let request_id = request.id.clone(); + + manager + .submit_heal_request(request) + .await + .expect("request should be accepted"); + + assert!(matches!( + manager.get_task_status_for_path("other", &request_id).await, + Err(Error::TaskNotFound { .. }) + )); + } + + #[tokio::test] + async fn test_get_task_status_for_path_returns_not_found_when_path_is_inactive() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + assert!(matches!( + manager.get_task_status_for_path("bucket", "old-token").await, + Err(Error::TaskNotFound { .. }) + )); + } + + #[tokio::test] + async fn test_get_task_status_for_empty_path_does_not_match_unrelated_tasks() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let request = HealRequest::bucket("bucket".to_string()); + let request_id = request.id.clone(); + + manager + .submit_heal_request(request) + .await + .expect("request should be accepted"); + + assert!(matches!( + manager.get_task_status_for_path("", &request_id).await, + Err(Error::TaskNotFound { .. }) + )); + assert!(matches!( + manager.get_task_status_for_path("", "wrong-token").await, + Err(Error::TaskNotFound { .. }) + )); + } + + #[tokio::test] + async fn test_get_task_status_reads_recent_completed_status() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + manager.completed_heals.lock().await.insert( + "completed-token".to_string(), + CompletedHealStatus { + heal_type: HealType::Bucket { + bucket: "bucket".to_string(), + }, + status: HealTaskStatus::Completed, + result_items: Vec::new(), + completed_at: SystemTime::now(), + }, + ); + + assert_eq!( + manager + .get_task_status_for_path("bucket", "completed-token") + .await + .expect("recent completed task should be queryable"), + HealTaskStatus::Completed + ); + } + + #[tokio::test] + async fn test_get_task_report_for_path_reads_completed_items() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + manager.completed_heals.lock().await.insert( + "completed-token".to_string(), + CompletedHealStatus { + heal_type: HealType::Object { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + }, + status: HealTaskStatus::Completed, + result_items: vec![HealResultItem { + bucket: "bucket".to_string(), + object: "object".to_string(), + object_size: 1024, + ..Default::default() + }], + completed_at: SystemTime::now(), + }, + ); + + let report = manager + .get_task_report_for_path("bucket/object", "completed-token") + .await + .expect("recent completed task report should be queryable"); + + assert_eq!(report.status, HealTaskStatus::Completed); + assert_eq!(report.result_items.len(), 1); + assert_eq!(report.result_items[0].object_size, 1024); + } + + #[tokio::test] + async fn test_get_task_report_for_empty_path_does_not_match_unrelated_tasks() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + manager + .submit_heal_request(HealRequest::bucket("bucket".to_string())) + .await + .expect("request should be accepted"); + + assert!(matches!( + manager.get_task_report_for_path("", "wrong-token").await, + Err(Error::TaskNotFound { .. }) + )); + } + + #[tokio::test] + async fn test_cancel_task_removes_queued_request() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let request = HealRequest::bucket("bucket".to_string()); + let request_id = request.id.clone(); + + manager + .submit_heal_request(request) + .await + .expect("request should be accepted"); + manager + .cancel_task(&request_id) + .await + .expect("queued request should be cancelled"); + + assert!(matches!(manager.get_task_status(&request_id).await, Err(Error::TaskNotFound { .. }))); + } + + #[tokio::test] + async fn test_cancel_tasks_for_path_removes_matching_queued_requests() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new(storage, None); + + let bucket_request = HealRequest::bucket("bucket".to_string()); + let bucket_request_id = bucket_request.id.clone(); + let other_request = HealRequest::bucket("other".to_string()); + let other_request_id = other_request.id.clone(); + + manager + .submit_heal_request(bucket_request) + .await + .expect("bucket request should be accepted"); + manager + .submit_heal_request(other_request) + .await + .expect("other request should be accepted"); + + assert_eq!( + manager + .cancel_tasks_for_path("bucket") + .await + .expect("matching request should be cancelled"), + 1 + ); + assert!(matches!( + manager.get_task_status(&bucket_request_id).await, + Err(Error::TaskNotFound { .. }) + )); + assert_eq!( + manager + .get_task_status(&other_request_id) + .await + .expect("unmatched request should remain queued"), + HealTaskStatus::Pending + ); + } + #[tokio::test] async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index bee286560..322025b81 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -16,6 +16,7 @@ use crate::heal::{ErasureSetHealer, progress::HealProgress, storage::HealStorage use crate::{Error, Result}; use metrics::{counter, histogram}; use rustfs_common::heal_channel::{HealOpts, HealScanMode}; +use rustfs_madmin::heal_commands::HealResultItem; use serde::{Deserialize, Serialize}; use std::{ future::Future, @@ -196,6 +197,8 @@ pub struct HealTask { pub status: Arc>, /// Progress tracking pub progress: Arc>, + /// Result items collected from storage heal calls. + pub result_items: Arc>>, /// Created time pub created_at: SystemTime, /// Queue admission time @@ -220,6 +223,7 @@ impl HealTask { options: request.options, status: Arc::new(RwLock::new(HealTaskStatus::Pending)), progress: Arc::new(RwLock::new(HealProgress::new())), + result_items: Arc::new(RwLock::new(Vec::new())), created_at: request.created_at, enqueued_at: request.enqueued_at, started_at: Arc::new(RwLock::new(None)), @@ -411,6 +415,14 @@ impl HealTask { self.progress.read().await.clone() } + pub async fn get_result_items(&self) -> Vec { + self.result_items.read().await.clone() + } + + async fn record_result_item(&self, result: HealResultItem) { + self.result_items.write().await.push(result); + } + // specific heal implementation method #[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))] async fn heal_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> { @@ -521,6 +533,7 @@ impl HealTask { let mut progress = self.progress.write().await; progress.update_progress(3, 3, object_size, object_size); } + self.record_result_item(result).await; Ok(()) } Err(Error::TaskCancelled) => Err(Error::TaskCancelled), @@ -601,6 +614,7 @@ impl HealTask { let mut progress = self.progress.write().await; progress.update_progress(4, 4, object_size, object_size); } + self.record_result_item(result).await; Ok(()) } Err(Error::TaskCancelled) => Err(Error::TaskCancelled), @@ -659,8 +673,13 @@ impl HealTask { match heal_result { Ok(result) => { info!("Bucket heal completed successfully: {} ({} drives)", bucket, result.after.drives.len()); + self.record_result_item(result).await; - { + if self.options.recursive { + self.heal_bucket_objects(bucket).await?; + } + + if !self.options.recursive { let mut progress = self.progress.write().await; progress.update_progress(3, 3, 0, 0); } @@ -681,6 +700,98 @@ impl HealTask { } } + async fn heal_bucket_objects(&self, bucket: &str) -> Result<()> { + let mut continuation_token: Option = None; + let mut scanned = 0u64; + let mut healed = 0u64; + let mut failed = 0u64; + let mut bytes = 0u64; + + let heal_opts = HealOpts { + recursive: false, + dry_run: self.options.dry_run, + remove: self.options.remove_corrupted, + recreate: self.options.recreate_missing, + scan_mode: self.options.scan_mode, + update_parity: self.options.update_parity, + no_lock: false, + pool: self.options.pool_index, + set: self.options.set_index, + }; + + loop { + self.check_control_flags().await?; + let (objects, next_token, is_truncated) = self + .await_with_control( + self.storage + .list_objects_for_heal_page(bucket, "", continuation_token.as_deref()), + ) + .await?; + + for object in objects { + self.check_control_flags().await?; + scanned += 1; + { + let mut progress = self.progress.write().await; + progress.set_current_object(Some(format!("{bucket}/{object}"))); + progress.update_progress(scanned, healed, failed, bytes); + } + + match self.await_with_control(self.storage.object_exists(bucket, &object)).await { + Ok(false) => { + healed += 1; + } + Ok(true) => match self + .await_with_control(self.storage.heal_object(bucket, &object, None, &heal_opts)) + .await + { + Ok((result, None)) => { + healed += 1; + bytes = bytes.saturating_add(result.object_size as u64); + self.record_result_item(result).await; + } + Ok((_, Some(err))) => { + failed += 1; + warn!("Failed to heal object {}/{}: {}", bucket, object, err); + } + Err(err) => { + failed += 1; + warn!("Failed to heal object {}/{}: {}", bucket, object, err); + } + }, + Err(err) => { + failed += 1; + warn!("Failed to check object {}/{} before heal: {}", bucket, object, err); + } + } + + { + let mut progress = self.progress.write().await; + progress.update_progress(scanned, healed, failed, bytes); + } + } + + if !is_truncated { + break; + } + + continuation_token = next_token; + if continuation_token.is_none() { + warn!("List is truncated but no continuation token was returned for bucket {}", bucket); + break; + } + } + + if failed > 0 { + return Err(Error::TaskExecutionFailed { + message: format!("Failed to heal {failed} object(s) in bucket {bucket}"), + }); + } + + info!("Recursive bucket heal completed for {}: {} scanned, {} healed", bucket, scanned, healed); + Ok(()) + } + async fn heal_metadata(&self, bucket: &str, object: &str) -> Result<()> { info!("Healing metadata: {}/{}", bucket, object); @@ -755,6 +866,7 @@ impl HealTask { let mut progress = self.progress.write().await; progress.update_progress(3, 3, 0, 0); } + self.record_result_item(result).await; Ok(()) } Err(Error::TaskCancelled) => Err(Error::TaskCancelled), @@ -830,6 +942,7 @@ impl HealTask { let mut progress = self.progress.write().await; progress.update_progress(2, 2, 0, 0); } + self.record_result_item(result).await; Ok(()) } Err(Error::TaskCancelled) => Err(Error::TaskCancelled), @@ -923,6 +1036,7 @@ impl HealTask { let mut progress = self.progress.write().await; progress.update_progress(3, 3, object_size, object_size); } + self.record_result_item(result).await; Ok(()) } Err(Error::TaskCancelled) => Err(Error::TaskCancelled), @@ -1072,3 +1186,163 @@ impl std::fmt::Debug for HealTask { .finish() } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::heal::storage::DiskStatus; + use rustfs_ecstore::{ + disk::{DiskStore, endpoint::Endpoint}, + store_api::{BucketInfo, ObjectInfo}, + }; + use rustfs_madmin::heal_commands::HealResultItem; + use std::sync::Mutex; + + #[derive(Default)] + struct MockStorage { + listed: Mutex, + healed_objects: Mutex>, + } + + #[async_trait::async_trait] + impl HealStorageAPI for MockStorage { + async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result> { + Ok(None) + } + + async fn get_object_data(&self, _bucket: &str, _object: &str) -> Result>> { + Ok(None) + } + + async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> Result<()> { + Ok(()) + } + + async fn delete_object(&self, _bucket: &str, _object: &str) -> Result<()> { + Ok(()) + } + + async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> Result { + Ok(true) + } + + async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> Result> { + Ok(Vec::new()) + } + + async fn get_disk_status(&self, _endpoint: &Endpoint) -> Result { + Ok(DiskStatus::Ok) + } + + async fn format_disk(&self, _endpoint: &Endpoint) -> Result<()> { + Ok(()) + } + + async fn get_bucket_info(&self, bucket: &str) -> Result> { + Ok(Some(BucketInfo { + name: bucket.to_string(), + ..Default::default() + })) + } + + async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<()> { + Ok(()) + } + + async fn list_buckets(&self) -> Result> { + Ok(Vec::new()) + } + + async fn object_exists(&self, _bucket: &str, _object: &str) -> Result { + Ok(true) + } + + async fn get_object_size(&self, _bucket: &str, _object: &str) -> Result> { + Ok(None) + } + + async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> Result> { + Ok(None) + } + + async fn heal_object( + &self, + _bucket: &str, + object: &str, + _version_id: Option<&str>, + _opts: &HealOpts, + ) -> Result<(HealResultItem, Option)> { + self.healed_objects.lock().unwrap().push(object.to_string()); + Ok(( + HealResultItem { + object_size: 1, + ..Default::default() + }, + None, + )) + } + + async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result { + Ok(HealResultItem::default()) + } + + async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option)> { + Ok((HealResultItem::default(), None)) + } + + async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result> { + Ok(vec!["object-a".to_string(), "object-b".to_string()]) + } + + async fn list_objects_for_heal_page( + &self, + _bucket: &str, + _prefix: &str, + continuation_token: Option<&str>, + ) -> Result<(Vec, Option, bool)> { + let mut listed = self.listed.lock().unwrap(); + if continuation_token.is_none() && !*listed { + *listed = true; + Ok((vec!["object-a".to_string(), "object-b".to_string()], None, false)) + } else { + Ok((Vec::new(), None, false)) + } + } + + async fn get_disk_for_resume(&self, _set_disk_id: &str) -> Result { + Err(Error::other("not implemented in tests")) + } + } + + #[tokio::test] + async fn test_recursive_bucket_heal_visits_objects() { + let storage = Arc::new(MockStorage::default()); + let request = HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + task.heal_bucket("bucket-a") + .await + .expect("recursive bucket heal should succeed"); + + assert_eq!( + storage.healed_objects.lock().unwrap().as_slice(), + ["object-a".to_string(), "object-b".to_string()] + ); + let progress = task.get_progress().await; + assert_eq!(progress.objects_scanned, 2); + assert_eq!(progress.objects_healed, 2); + let result_items = task.get_result_items().await; + assert_eq!(result_items.len(), 3); + assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2); + } +} diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index bb518037a..167c69cfe 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -15,7 +15,7 @@ use std::sync::Arc; use crate::data_usage_define::{BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH}; -use crate::scanner_folder::data_usage_update_dir_cycles; +use crate::scanner_folder::{data_usage_update_dir_cycles, heal_object_select_prob}; use crate::scanner_io::ScannerIO; use crate::sleeper::SCANNER_SLEEPER; use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError}; @@ -23,7 +23,10 @@ use chrono::{DateTime, Utc}; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::{CurrentCycle, Metric, Metrics, emit_scan_cycle_complete, global_metrics}; use rustfs_config::ScannerSpeed; -use rustfs_config::{DEFAULT_SCANNER_SPEED, ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS}; +use rustfs_config::{ + DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_SPEED, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CYCLE, + ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS, +}; use rustfs_ecstore::StorageAPI as _; use rustfs_ecstore::config::com::{read_config, save_config}; use rustfs_ecstore::disk::RUSTFS_META_BUCKET; @@ -80,7 +83,7 @@ fn initial_scanner_delay() -> Duration { fn initial_scanner_delay_for(start_delay_secs: Option) -> Duration { start_delay_secs .map(|secs| randomized_cycle_delay_for(Duration::from_secs(secs))) - .unwrap_or_else(|| Duration::from_secs(rand::random::() % 5)) + .unwrap_or_else(randomized_cycle_delay) } pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { @@ -111,9 +114,60 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { }); } -fn get_cycle_scan_mode(_current_cycle: u64, _bitrot_start_cycle: u64, _bitrot_start_time: Option>) -> HealScanMode { - // TODO: from config - HealScanMode::Normal +fn bitrot_scan_cycle() -> Option { + let Ok(value) = std::env::var(ENV_SCANNER_BITROT_CYCLE_SECS) else { + return Some(Duration::from_secs(DEFAULT_SCANNER_BITROT_CYCLE_SECS)); + }; + + match value.trim().to_ascii_lowercase().as_str() { + "0" | "true" | "on" | "yes" => Some(Duration::ZERO), + "false" | "off" | "no" | "disabled" => None, + value => value.parse::().ok().map(Duration::from_secs).or_else(|| { + warn!( + env = ENV_SCANNER_BITROT_CYCLE_SECS, + value, + default_secs = DEFAULT_SCANNER_BITROT_CYCLE_SECS, + "Invalid scanner bitrot cycle, using default" + ); + Some(Duration::from_secs(DEFAULT_SCANNER_BITROT_CYCLE_SECS)) + }), + } +} + +fn get_cycle_scan_mode(current_cycle: u64, bitrot_start_cycle: u64, bitrot_start_time: Option>) -> HealScanMode { + let Some(bitrot_cycle) = bitrot_scan_cycle() else { + return HealScanMode::Normal; + }; + + if bitrot_cycle.is_zero() { + return HealScanMode::Deep; + } + + if current_cycle.saturating_sub(bitrot_start_cycle) < heal_object_select_prob() as u64 { + return HealScanMode::Deep; + } + + let Some(bitrot_start_time) = bitrot_start_time else { + return HealScanMode::Deep; + }; + + let elapsed = Utc::now() + .signed_duration_since(bitrot_start_time) + .to_std() + .unwrap_or(Duration::ZERO); + if elapsed >= bitrot_cycle { + HealScanMode::Deep + } else { + HealScanMode::Normal + } +} + +fn retain_recent_cycle_completions(cycle_completed: &mut Vec>) { + let keep = data_usage_update_dir_cycles() as usize; + if cycle_completed.len() > keep { + let drop_count = cycle_completed.len() - keep; + cycle_completed.drain(..drop_count); + } } /// Background healing information @@ -238,9 +292,7 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc info!(duration = ?now.elapsed(), cycles_total=cycle_info.cycle_completed.len(), "Success run data scanner cycle"); - if cycle_info.cycle_completed.len() >= data_usage_update_dir_cycles() as usize { - cycle_info.cycle_completed = cycle_info.cycle_completed.split_off(data_usage_update_dir_cycles() as usize); - } + retain_recent_cycle_completions(&mut cycle_info.cycle_completed); global_metrics().set_cycle(Some(cycle_info.clone())).await; @@ -382,6 +434,16 @@ mod tests { assert!(delay <= Duration::from_secs(132)); } + #[test] + #[serial] + fn test_initial_scanner_delay_uses_cycle_without_explicit_start_delay() { + with_var(ENV_SCANNER_CYCLE, Some("120"), || { + let delay = initial_scanner_delay_for(None); + assert!(delay >= Duration::from_secs(108)); + assert!(delay <= Duration::from_secs(132)); + }); + } + #[test] #[serial] fn test_cycle_interval_prefers_explicit_cycle_override() { @@ -426,4 +488,46 @@ mod tests { assert!(delay >= Duration::from_secs(1), "expected delay >= 1s"); assert!(delay < Duration::from_secs(2), "expected delay < 2s"); } + + #[test] + #[serial] + fn test_get_cycle_scan_mode_runs_deep_until_selection_window_completes() { + with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("3600"), || { + let mode = get_cycle_scan_mode(10, 0, Some(Utc::now())); + assert_eq!(mode, HealScanMode::Deep); + }); + } + + #[test] + #[serial] + fn test_get_cycle_scan_mode_respects_elapsed_bitrot_cycle() { + with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("3600"), || { + let recent = Utc::now() - chrono::Duration::minutes(30); + let old = Utc::now() - chrono::Duration::hours(2); + + assert_eq!(get_cycle_scan_mode(2048, 0, Some(recent)), HealScanMode::Normal); + assert_eq!(get_cycle_scan_mode(2048, 0, Some(old)), HealScanMode::Deep); + }); + } + + #[test] + #[serial] + fn test_get_cycle_scan_mode_can_disable_periodic_deep_scan() { + with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("off"), || { + assert_eq!(get_cycle_scan_mode(1, 0, None), HealScanMode::Normal); + }); + } + + #[test] + fn test_retain_recent_cycle_completions_keeps_last_entries() { + let base = Utc::now(); + let keep = data_usage_update_dir_cycles() as usize; + let mut completed: Vec<_> = (0..keep + 2).map(|i| base + chrono::Duration::seconds(i as i64)).collect(); + + retain_recent_cycle_completions(&mut completed); + + assert_eq!(completed.len(), keep); + assert_eq!(completed.first().copied(), Some(base + chrono::Duration::seconds(2))); + assert_eq!(completed.last().copied(), Some(base + chrono::Duration::seconds((keep + 1) as i64))); + } } diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 7cfa6469a..b4814a8b4 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -69,9 +69,13 @@ const ENV_FAILED_OBJECTS_MAX: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECTS_MAX"; const DEFAULT_FAILED_OBJECT_TTL_SECS: u32 = 86_400; const DEFAULT_FAILED_OBJECTS_MAX: u32 = 10_000; const METRIC_SCANNER_INLINE_HEAL_TOTAL: &str = "rustfs_scanner_inline_heal_total"; +const METRIC_SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL: &str = "rustfs_scanner_excess_object_versions_total"; +const METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL: &str = "rustfs_scanner_excess_object_version_size_total"; +const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders_total"; static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new(); static SCANNER_INLINE_HEAL_METRICS_ONCE: Once = Once::new(); +static SCANNER_ALERT_METRICS_ONCE: Once = Once::new(); pub fn data_usage_update_dir_cycles() -> u32 { rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES) @@ -102,6 +106,50 @@ fn ensure_scanner_inline_heal_metric_registered() { }); } +fn ensure_scanner_alert_metrics_registered() { + SCANNER_ALERT_METRICS_ONCE.call_once(|| { + describe_counter!( + METRIC_SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL, + "Total scanner alerts for objects with too many retained versions." + ); + describe_counter!( + METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL, + "Total scanner alerts for objects whose retained versions exceed the cumulative size threshold." + ); + describe_counter!( + METRIC_SCANNER_EXCESS_FOLDERS_TOTAL, + "Total scanner alerts for folders with too many direct subfolders." + ); + }); +} + +fn scanner_excess_versions_threshold() -> u64 { + rustfs_utils::get_env_u64( + rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSIONS, + rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, + ) +} + +fn scanner_excess_version_size_threshold() -> u64 { + rustfs_utils::get_env_u64( + rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, + rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE, + ) +} + +fn scanner_excess_folders_threshold() -> u64 { + rustfs_utils::get_env_u64( + rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, + rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS, + ) +} + +fn should_alert_excessive_versions(remaining_versions: usize, cumulative_size: i64) -> (bool, bool) { + let too_many_versions = remaining_versions as u64 >= scanner_excess_versions_threshold(); + let too_large_versions = cumulative_size > 0 && cumulative_size as u64 >= scanner_excess_version_size_threshold(); + (too_many_versions, too_large_versions) +} + fn warn_inline_heal_compat_requested() { if !scanner_inline_heal_enabled() { return; @@ -503,7 +551,7 @@ impl ScannerItem { }; let roi = queue_replication_heal_internal(&oi.bucket, oi.clone(), (*replication).clone(), 0).await; - if oi.delete_marker || oi.version_purge_status.is_empty() { + if !Self::should_account_replication_stats(oi) { return; } @@ -545,6 +593,10 @@ impl ScannerItem { } } + fn should_account_replication_stats(oi: &ObjectInfo) -> bool { + !oi.delete_marker && oi.version_purge_status.is_empty() + } + async fn enqueue_heal(&mut self, oi: &ObjectInfo) { let done_heal = Metrics::time(Metric::HealAbandonedObject); debug!( @@ -588,8 +640,38 @@ impl ScannerItem { done_heal(); } - fn alert_excessive_versions(&self, _object_infos_length: usize, _cumulative_size: i64) { - // TODO: Implement alerting for excessive versions + fn alert_excessive_versions(&self, remaining_versions: usize, cumulative_size: i64) { + ensure_scanner_alert_metrics_registered(); + let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size); + if too_many_versions { + counter!( + METRIC_SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL, + "bucket" => self.bucket.clone() + ) + .increment(1); + warn!( + bucket = %self.bucket, + object = %self.object_path(), + versions = remaining_versions, + threshold = scanner_excess_versions_threshold(), + "scanner detected object with excessive retained versions" + ); + } + if too_large_versions { + counter!( + METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL, + "bucket" => self.bucket.clone() + ) + .increment(1); + warn!( + bucket = %self.bucket, + object = %self.object_path(), + versions = remaining_versions, + cumulative_size, + threshold = scanner_excess_version_size_threshold(), + "scanner detected object with excessive retained version size" + ); + } } } @@ -694,6 +776,27 @@ impl FolderScanner { } } + fn alert_excessive_folders(&self, folder: &str, total_folders: usize) { + let threshold = scanner_excess_folders_threshold(); + if total_folders as u64 <= threshold { + return; + } + + ensure_scanner_alert_metrics_registered(); + counter!( + METRIC_SCANNER_EXCESS_FOLDERS_TOTAL, + "root" => self.root.clone() + ) + .increment(1); + warn!( + root = %self.root, + folder, + folders = total_folders, + threshold, + "scanner detected folder with excessive direct subfolders" + ); + } + pub async fn should_heal(&self) -> bool { if self.skip_heal.load(std::sync::atomic::Ordering::Relaxed) { return false; @@ -1011,7 +1114,8 @@ impl FolderScanner { && existing_folders.len() + new_folders.len() >= DATA_SCANNER_COMPACT_AT_FOLDERS) || existing_folders.len() + new_folders.len() >= DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS; - // TODO: Check for excess folders and send events + let total_folders = existing_folders.len() + new_folders.len(); + self.alert_excessive_folders(&folder.name, total_folders); if !into.compacted && should_compact { into.compacted = true; @@ -1557,10 +1661,12 @@ mod tests { use super::*; use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk}; + use rustfs_filemeta::VersionPurgeStatusType; use serial_test::serial; #[cfg(unix)] use std::os::unix::fs::{PermissionsExt, symlink}; use std::sync::atomic::AtomicBool; + use temp_env::with_var; use uuid::Uuid; async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) { @@ -1663,6 +1769,45 @@ mod tests { assert!(!scanner.should_skip_failed("path2")); } + #[test] + fn test_should_account_replication_stats_only_for_live_object_versions() { + let live = ObjectInfo::default(); + assert!(ScannerItem::should_account_replication_stats(&live)); + + let delete_marker = ObjectInfo { + delete_marker: true, + ..Default::default() + }; + assert!(!ScannerItem::should_account_replication_stats(&delete_marker)); + + let purge_version = ObjectInfo { + version_purge_status: VersionPurgeStatusType::Pending, + ..Default::default() + }; + assert!(!ScannerItem::should_account_replication_stats(&purge_version)); + } + + #[test] + #[serial] + fn test_excessive_version_alert_thresholds_use_env() { + with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSIONS, Some("3"), || { + with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, Some("100"), || { + assert_eq!(should_alert_excessive_versions(2, 99), (false, false)); + assert_eq!(should_alert_excessive_versions(3, 99), (true, false)); + assert_eq!(should_alert_excessive_versions(2, 100), (false, true)); + assert_eq!(should_alert_excessive_versions(3, 100), (true, true)); + }); + }); + } + + #[test] + #[serial] + fn test_excessive_folders_threshold_uses_env() { + with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, Some("3"), || { + assert_eq!(scanner_excess_folders_threshold(), 3); + }); + } + #[tokio::test] #[serial] async fn test_record_failed_prunes_to_max_entries() { diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 29add5d80..74e993b91 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -20,7 +20,7 @@ use bytes::Bytes; use http::{HeaderMap, HeaderValue, Uri}; use hyper::{Method, StatusCode}; use matchit::Params; -use rustfs_common::heal_channel::HealOpts; +use rustfs_common::heal_channel::{HealChannelPriority, HealChannelRequest, HealOpts}; use rustfs_config::MAX_HEAL_REQUEST_SIZE; use rustfs_ecstore::bucket::utils::is_valid_object_prefix; use rustfs_ecstore::new_object_layer_fn; @@ -29,10 +29,11 @@ use rustfs_ecstore::store_utils::is_reserved_or_invalid_bucket; use rustfs_policy::policy::action::{Action, AdminAction}; use rustfs_scanner::scanner::{BackgroundHealInfo, read_background_heal_info}; use rustfs_utils::path::path_join; -use s3s::header::CONTENT_TYPE; +use s3s::header::{CONTENT_LENGTH, CONTENT_TYPE}; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; use std::path::PathBuf; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::spawn; use tokio::sync::mpsc; use tracing::{info, warn}; @@ -75,7 +76,7 @@ fn extract_heal_init_params(body: &Bytes, uri: &Uri, params: Params<'_, '_>) -> } } - if (hip.force_start && hip.force_stop) || (!hip.client_token.is_empty() && (hip.force_start || hip.force_stop)) { + if hip.force_start && (hip.force_stop || !hip.client_token.is_empty()) { return Err(s3_error!( InvalidRequest, "invalid combination of clientToken, forceStart, and forceStop parameters" @@ -142,6 +143,34 @@ struct HealResp { api_err: Option, } +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct HealStartSuccess { + client_token: String, + client_address: String, + start_time: String, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct HealTaskStatus { + summary: String, + #[serde(rename = "detail")] + failure_detail: String, + start_time: String, + #[serde(rename = "settings")] + heal_settings: HealOpts, + #[serde(skip_serializing_if = "Vec::is_empty")] + items: Vec, +} + +#[derive(Debug, Deserialize)] +struct HealTaskStatusPayload { + summary: String, + #[serde(default)] + items: Vec, +} + fn map_heal_response(result: Option) -> S3Result<(StatusCode, Vec)> { match result { Some(result) => { @@ -149,12 +178,105 @@ fn map_heal_response(result: Option) -> S3Result<(StatusCode, Vec) return Err(s3_error!(InternalError, "{err}")); } + if result.resp_bytes.is_empty() { + return Err(s3_error!(InternalError, "heal response body is empty")); + } + Ok((StatusCode::OK, result.resp_bytes)) } None => Err(s3_error!(InternalError, "heal channel closed unexpectedly")), } } +fn current_rfc3339_time() -> S3Result { + OffsetDateTime::now_utc() + .format(&Rfc3339) + .map_err(|e| s3_error!(InternalError, "failed to format heal timestamp: {e}")) +} + +fn encode_json(value: &T) -> S3Result> { + serde_json::to_vec(value).map_err(|e| s3_error!(InternalError, "failed to serialize heal response: {e}")) +} + +fn encode_heal_start_success(client_token: String, client_address: String) -> S3Result> { + encode_json(&HealStartSuccess { + client_token, + client_address, + start_time: current_rfc3339_time()?, + }) +} + +fn encode_heal_task_status( + summary: String, + failure_detail: String, + heal_settings: HealOpts, + items: Vec, +) -> S3Result> { + encode_json(&HealTaskStatus { + summary, + failure_detail, + start_time: current_rfc3339_time()?, + heal_settings, + items, + }) +} + +fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { + let mut heal_request = rustfs_common::heal_channel::create_heal_request( + hip.bucket.clone(), + if hip.obj_prefix.is_empty() { + None + } else { + Some(hip.obj_prefix.clone()) + }, + hip.force_start, + Some(HealChannelPriority::Normal), + ); + + heal_request.pool_index = hip.hs.pool; + heal_request.set_index = hip.hs.set; + heal_request.scan_mode = Some(hip.hs.scan_mode); + heal_request.remove_corrupted = Some(hip.hs.remove); + heal_request.recreate_missing = Some(hip.hs.recreate); + heal_request.update_parity = Some(hip.hs.update_parity); + heal_request.recursive = Some(hip.hs.recursive); + heal_request.dry_run = Some(hip.hs.dry_run); + heal_request +} + +fn heal_channel_response_status( + response: &rustfs_common::heal_channel::HealChannelResponse, +) -> (String, Vec) { + let Some(data) = response.data.as_deref() else { + return ("running".to_string(), Vec::new()); + }; + + if let Ok(payload) = serde_json::from_slice::(data) + && !payload.summary.is_empty() + { + return (payload.summary, payload.items); + } + + let summary = std::str::from_utf8(data) + .ok() + .filter(|summary| !summary.is_empty()) + .unwrap_or("running") + .to_string(); + (summary, Vec::new()) +} + +#[cfg(test)] +fn heal_channel_response_summary(response: &rustfs_common::heal_channel::HealChannelResponse) -> String { + heal_channel_response_status(response).0 +} + +#[cfg(test)] +fn heal_channel_response_items( + response: &rustfs_common::heal_channel::HealChannelResponse, +) -> Vec { + heal_channel_response_status(response).1 +} + fn encode_background_heal_status(info: &BackgroundHealInfo) -> S3Result> { serde_json::to_vec(info).map_err(|e| s3_error!(InternalError, "failed to serialize background heal status: {e}")) } @@ -185,6 +307,9 @@ fn map_root_heal_status(heal_err: Option) -> S3Res fn json_response(status: StatusCode, body: Vec) -> S3Response<(StatusCode, Body)> { let mut headers = HeaderMap::new(); headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + if let Ok(value) = HeaderValue::from_str(&body.len().to_string()) { + headers.insert(CONTENT_LENGTH, value); + } S3Response::with_headers((status, Body::from(body)), headers) } @@ -213,6 +338,11 @@ impl Operation for HealHandler { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { warn!("handle HealHandler, req: {:?}, params: {:?}", req, params); validate_heal_admin_request(&req).await?; + let client_address = req + .extensions + .get::>() + .and_then(|opt| opt.map(|addr| addr.0.to_string())) + .unwrap_or_default(); let mut input = req.input; let bytes = match input.store_all_limited(MAX_HEAL_REQUEST_SIZE).await { Ok(b) => b, @@ -236,8 +366,9 @@ impl Operation for HealHandler { .map_err(|e| s3_error!(InternalError, "root heal failed: {e}"))?; map_root_heal_status(heal_err)?; + let body = encode_heal_start_success("root-heal".to_string(), client_address)?; - return Ok(S3Response::new((StatusCode::OK, Body::empty()))); + return Ok(json_response(StatusCode::OK, body)); } validate_heal_request_mode(&hip)?; info!("body: {:?}", hip); @@ -252,11 +383,33 @@ impl Operation for HealHandler { let client_token = hip.client_token.clone(); spawn(async move { match rustfs_common::heal_channel::query_heal_status(heal_path_str, client_token).await { - Ok(_) => { - // TODO: Get actual response from channel + Ok(response) if response.success => { + let (summary, items) = heal_channel_response_status(&response); + let resp_bytes = + encode_heal_task_status(summary, response.error.unwrap_or_default(), HealOpts::default(), items); + match resp_bytes { + Ok(resp_bytes) => { + let _ = tx_clone + .send(HealResp { + resp_bytes, + ..Default::default() + }) + .await; + } + Err(e) => { + let _ = tx_clone + .send(HealResp { + api_err: Some(e.to_string()), + ..Default::default() + }) + .await; + } + } + } + Ok(response) => { let _ = tx_clone .send(HealResp { - resp_bytes: vec![], + api_err: Some(response.error.unwrap_or_else(|| "query heal status failed".to_string())), ..Default::default() }) .await; @@ -275,13 +428,41 @@ impl Operation for HealHandler { // Cancel heal task let tx_clone = tx.clone(); let heal_path_str = heal_path.to_str().unwrap_or_default().to_string(); + let client_token = hip.client_token.clone(); + let client_address = client_address.clone(); + let heal_settings = hip.hs; spawn(async move { - match rustfs_common::heal_channel::cancel_heal_task(heal_path_str).await { - Ok(_) => { - // TODO: Get actual response from channel + match rustfs_common::heal_channel::cancel_heal_task(heal_path_str, client_token.clone()).await { + Ok(response) if response.success => { + let resp_bytes = if client_token.is_empty() { + encode_heal_start_success(response.request_id, client_address) + } else { + let (summary, items) = heal_channel_response_status(&response); + encode_heal_task_status(summary, response.error.unwrap_or_default(), heal_settings, items) + }; + match resp_bytes { + Ok(resp_bytes) => { + let _ = tx_clone + .send(HealResp { + resp_bytes, + ..Default::default() + }) + .await; + } + Err(e) => { + let _ = tx_clone + .send(HealResp { + api_err: Some(e.to_string()), + ..Default::default() + }) + .await; + } + } + } + Ok(response) => { let _ = tx_clone .send(HealResp { - resp_bytes: vec![], + api_err: Some(response.error.unwrap_or_else(|| "cancel heal task failed".to_string())), ..Default::default() }) .await; @@ -299,25 +480,42 @@ impl Operation for HealHandler { } else if hip.client_token.is_empty() { // Use new heal channel mechanism let tx_clone = tx.clone(); + let client_address = client_address.clone(); spawn(async move { // Create heal request through channel - let heal_request = rustfs_common::heal_channel::create_heal_request( - hip.bucket.clone(), - if hip.obj_prefix.is_empty() { - None - } else { - Some(hip.obj_prefix.clone()) - }, - hip.force_start, - Some(rustfs_common::heal_channel::HealChannelPriority::Normal), - ); + let heal_request = build_heal_channel_request(&hip); + let client_token = heal_request.id.clone(); - match rustfs_common::heal_channel::send_heal_request(heal_request).await { - Ok(_) => { - // Success - send empty response for now + match rustfs_common::heal_channel::send_heal_request_with_admission(heal_request).await { + Ok(admission) if admission.is_admitted() => { + let resp_bytes = encode_heal_start_success(client_token, client_address); + match resp_bytes { + Ok(resp_bytes) => { + let _ = tx_clone + .send(HealResp { + resp_bytes, + ..Default::default() + }) + .await; + } + Err(e) => { + let _ = tx_clone + .send(HealResp { + api_err: Some(e.to_string()), + ..Default::default() + }) + .await; + } + } + } + Ok(admission) => { let _ = tx_clone .send(HealResp { - resp_bytes: vec![], + api_err: Some(format!( + "heal request not admitted: admission={}, reason={}", + admission.result_label(), + admission.reason_label() + )), ..Default::default() }) .await; @@ -336,7 +534,7 @@ impl Operation for HealHandler { } let (status, body) = map_heal_response(rx.recv().await)?; - Ok(S3Response::new((status, Body::from(body)))) + Ok(json_response(status, body)) } } @@ -363,8 +561,9 @@ impl Operation for BackgroundHealStatusHandler { mod tests { use super::extract_heal_init_params; use super::{ - HealInitParams, HealResp, encode_background_heal_status, json_response, map_heal_response, map_root_heal_status, - should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target, + HealInitParams, HealResp, build_heal_channel_request, encode_background_heal_status, encode_heal_start_success, + encode_heal_task_status, heal_channel_response_items, heal_channel_response_summary, json_response, map_heal_response, + map_root_heal_status, should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target, }; use bytes::Bytes; use http::StatusCode; @@ -373,8 +572,12 @@ mod tests { use rustfs_common::heal_channel::{HealOpts, HealScanMode}; use rustfs_ecstore::error::StorageError; use rustfs_scanner::scanner::BackgroundHealInfo; - use s3s::{S3ErrorCode, header::CONTENT_TYPE}; + use s3s::{ + S3ErrorCode, + header::{CONTENT_LENGTH, CONTENT_TYPE}, + }; use serde_json::json; + use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::sync::mpsc; use tracing::debug; @@ -442,6 +645,23 @@ mod tests { ); } + #[test] + fn test_extract_heal_init_params_allows_client_token_force_stop() { + let uri: Uri = "/rustfs/admin/v3/heal/test-bucket?clientToken=token&forceStop=true" + .parse() + .expect("uri should parse"); + + let mut router = Router::new(); + router + .insert("/rustfs/admin/v3/heal/{bucket}", ()) + .expect("route should insert"); + let matched = router.at("/rustfs/admin/v3/heal/test-bucket").expect("route should match"); + + let parsed = extract_heal_init_params(&Bytes::new(), &uri, matched.params).expect("client-token stop should be accepted"); + assert_eq!(parsed.client_token, "token"); + assert!(parsed.force_stop); + } + #[test] fn test_extract_heal_init_params_allows_root_heal_target() { let uri: Uri = "/rustfs/admin/v3/heal/".parse().expect("uri should parse"); @@ -566,6 +786,17 @@ mod tests { assert!(result.is_err()); assert_eq!(result.unwrap_err().code(), &S3ErrorCode::InternalError); + let (tx, mut rx) = mpsc::channel(1); + tx.send(HealResp { + resp_bytes: vec![], + api_err: None, + }) + .await + .unwrap(); + let result = map_heal_response(rx.recv().await); + assert!(result.is_err()); + assert_eq!(result.unwrap_err().code(), &S3ErrorCode::InternalError); + let (tx, mut rx) = mpsc::channel(1); let _ = tx .send(HealResp { @@ -594,10 +825,119 @@ mod tests { assert!(json["bitrotStartTime"].is_null()); } + #[test] + fn test_encode_heal_start_success_uses_client_wire_shape() { + let encoded = encode_heal_start_success("token-1".to_string(), "127.0.0.1:9000".to_string()) + .expect("start response should serialize"); + let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize"); + + assert_eq!(json["clientToken"], "token-1"); + assert_eq!(json["clientAddress"], "127.0.0.1:9000"); + let start_time = json["startTime"].as_str().expect("startTime should be a string"); + OffsetDateTime::parse(start_time, &Rfc3339).expect("startTime should be RFC3339"); + } + + #[test] + fn test_encode_heal_task_status_uses_client_wire_shape() { + let encoded = + encode_heal_task_status("Heal status query accepted".to_string(), String::new(), HealOpts::default(), Vec::new()) + .expect("status response should serialize"); + let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize"); + + assert_eq!(json["summary"], "Heal status query accepted"); + assert_eq!(json["detail"], ""); + assert!(json["items"].is_null()); + assert!(json["settings"].is_object()); + let start_time = json["startTime"].as_str().expect("startTime should be a string"); + OffsetDateTime::parse(start_time, &Rfc3339).expect("startTime should be RFC3339"); + } + + #[test] + fn test_build_heal_channel_request_preserves_client_options() { + let hip = HealInitParams { + bucket: "bucket-a".to_string(), + obj_prefix: "prefix-a".to_string(), + force_start: true, + hs: HealOpts { + recursive: true, + dry_run: true, + remove: true, + recreate: false, + scan_mode: HealScanMode::Deep, + update_parity: false, + no_lock: true, + pool: Some(1), + set: Some(2), + }, + ..Default::default() + }; + + let request = build_heal_channel_request(&hip); + + assert_eq!(request.bucket, "bucket-a"); + assert_eq!(request.object_prefix.as_deref(), Some("prefix-a")); + assert!(request.force_start); + assert_eq!(request.pool_index, Some(1)); + assert_eq!(request.set_index, Some(2)); + assert_eq!(request.scan_mode, Some(HealScanMode::Deep)); + assert_eq!(request.remove_corrupted, Some(true)); + assert_eq!(request.recreate_missing, Some(false)); + assert_eq!(request.update_parity, Some(false)); + assert_eq!(request.recursive, Some(true)); + assert_eq!(request.dry_run, Some(true)); + } + + #[test] + fn test_heal_channel_response_summary_defaults_to_running() { + let response = rustfs_common::heal_channel::create_heal_response("token".to_string(), true, None, None); + assert_eq!(heal_channel_response_summary(&response), "running"); + + let response = + rustfs_common::heal_channel::create_heal_response("token".to_string(), true, Some(b"finished".to_vec()), None); + assert_eq!(heal_channel_response_summary(&response), "finished"); + } + + #[test] + fn test_heal_channel_response_status_preserves_items() { + let payload = serde_json::json!({ + "summary": "finished", + "items": [{ + "resultId": 0, + "type": "object", + "bucket": "bucket-a", + "object": "object-a", + "versionId": "", + "detail": "", + "parityBlocks": 2, + "dataBlocks": 2, + "diskCount": 4, + "setCount": 1, + "before": { "drives": [] }, + "after": { "drives": [] }, + "objectSize": 1024 + }] + }); + let response = rustfs_common::heal_channel::create_heal_response( + "token".to_string(), + true, + Some(serde_json::to_vec(&payload).expect("payload should serialize")), + None, + ); + + assert_eq!(heal_channel_response_summary(&response), "finished"); + let items = heal_channel_response_items(&response); + assert_eq!(items.len(), 1); + assert_eq!(items[0].bucket, "bucket-a"); + assert_eq!(items[0].object, "object-a"); + assert_eq!(items[0].object_size, 1024); + } + #[test] fn test_json_response_sets_application_json_content_type() { let response = json_response(StatusCode::OK, b"{}".to_vec()); let content_type = response.headers.get(CONTENT_TYPE).and_then(|value| value.to_str().ok()); assert_eq!(content_type, Some("application/json"),); + let content_length = response.headers.get(CONTENT_LENGTH).and_then(|value| value.to_str().ok()); + assert_eq!(content_length, Some("2"),); } } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 3e188e1e2..c306fa6ea 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -572,10 +572,6 @@ async fn run(config: rustfs::config::Config) -> Result<()> { init_heal_manager(heal_storage, None).await?; } - if enable_scanner { - init_data_scanner(ctx.clone(), store.clone()).await; - } - if !enable_heal && !enable_scanner { info!(target: "rustfs::main::run","Both scanner and heal are disabled, skipping AHM service initialization"); } @@ -610,6 +606,10 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // Publish ready only after all critical bootstrap metadata is in place state_manager.update(ServiceState::Ready); + if enable_scanner { + init_data_scanner(ctx.clone(), store.clone()).await; + } + // Perform hibernation for 1 second tokio::time::sleep(SHUTDOWN_TIMEOUT).await; // listen to the shutdown signal diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index ebac6d33a..6fa1791ee 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -275,7 +275,7 @@ fn should_force_zero_content_length_for_empty_body_route(req: &HttpRequest return false; } - if is_empty_body_admin_path(req.method(), req.uri().path()) { + if is_empty_body_admin_path(req.method(), req.uri()) { return true; } @@ -286,7 +286,8 @@ fn should_force_zero_content_length_for_empty_body_route(req: &HttpRequest is_empty_body_s3_path(req.method(), req.uri()) } -fn is_empty_body_admin_path(method: &Method, path: &str) -> bool { +fn is_empty_body_admin_path(method: &Method, uri: &http::Uri) -> bool { + let path = uri.path(); match *method { Method::PUT => matches!( path, @@ -295,21 +296,40 @@ fn is_empty_body_admin_path(method: &Method, path: &str) -> bool { | "/rustfs/admin/v3/set-user-status" | "/rustfs/admin/v3/set-group-status" ), - Method::POST => matches!( - path, - "/minio/admin/v3/rebalance/start" - | "/minio/admin/v3/rebalance/stop" - | "/minio/admin/v3/pools/decommission" - | "/minio/admin/v3/pools/cancel" - | "/rustfs/admin/v3/rebalance/start" - | "/rustfs/admin/v3/rebalance/stop" - | "/rustfs/admin/v3/pools/decommission" - | "/rustfs/admin/v3/pools/cancel" - ), + Method::POST => { + matches!( + path, + "/minio/admin/v3/rebalance/start" + | "/minio/admin/v3/rebalance/stop" + | "/minio/admin/v3/pools/decommission" + | "/minio/admin/v3/pools/cancel" + | "/rustfs/admin/v3/rebalance/start" + | "/rustfs/admin/v3/rebalance/stop" + | "/rustfs/admin/v3/pools/decommission" + | "/rustfs/admin/v3/pools/cancel" + ) || is_heal_status_query(path, uri.query()) + } _ => false, } } +fn is_heal_status_query(path: &str, query: Option<&str>) -> bool { + let Some(query) = query else { + return false; + }; + + if !path.find("/v3/heal/").is_some_and(|index| path[..index].ends_with("/admin")) { + return false; + } + + query.split('&').any(|param| { + param + .split_once('=') + .map(|(key, value)| key == "clientToken" && !value.is_empty()) + .unwrap_or(false) + }) +} + fn is_empty_body_s3_path(method: &Method, uri: &http::Uri) -> bool { *method == Method::DELETE && ConditionalCorsLayer::is_s3_path(uri.path()) } @@ -1320,6 +1340,40 @@ mod tests { assert!(headers.get(http::header::TRANSFER_ENCODING).is_none()); } + #[test] + fn admin_heal_status_query_without_content_length_is_normalized() { + let paths = [ + format!("{ADMIN_PREFIX}/v3/heal/?clientToken=root-heal"), + format!("{ADMIN_PREFIX}/v3/heal/bucket?clientToken=bucket-heal"), + ]; + + for path in paths { + let request = Request::builder() + .method(Method::POST) + .uri(path.clone()) + .header(http::header::TRANSFER_ENCODING, "chunked") + .body(()) + .expect("request"); + + assert!( + should_force_zero_content_length_for_empty_body_route(&request), + "{path} should force Content-Length: 0" + ); + } + } + + #[test] + fn admin_heal_start_without_status_token_is_not_normalized() { + let request = Request::builder() + .method(Method::POST) + .uri(format!("{ADMIN_PREFIX}/v3/heal/")) + .header(http::header::TRANSFER_ENCODING, "chunked") + .body(()) + .expect("request"); + + assert!(!should_force_zero_content_length_for_empty_body_route(&request)); + } + #[test] fn s3_delete_object_version_without_content_length_is_normalized() { let request = Request::builder()