diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index bcb730c2d..8086c3229 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -228,15 +228,6 @@ pub const DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS: usize = 4; /// Default object interval for cooperative scanner yields. pub const DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS: u64 = 128; -/// Compatibility flag kept for Patch 3 rollback windows. -/// -/// Inline scanner heal execution has been removed in favor of heal-candidate enqueue. -/// When this flag is enabled, RustFS logs a warning and continues to use enqueue-based heal. -pub const ENV_SCANNER_INLINE_HEAL_ENABLE: &str = "RUSTFS_SCANNER_INLINE_HEAL_ENABLE"; - -/// Default inline scanner heal compatibility mode. -pub const DEFAULT_SCANNER_INLINE_HEAL_ENABLE: bool = false; - /// Scanner speed preset controlling throttling behavior. /// /// Each preset defines three parameters: diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 8f4ab9e8a..edfcf2613 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -759,7 +759,7 @@ impl HealChannelProcessor { #[cfg(test)] mod tests { - use super::super::{DiskStore, Endpoint}; + use super::super::DiskStore; use super::*; use crate::heal::manager::HealConfig; use crate::heal::storage::{HealObjectInfo, HealStorageAPI}; @@ -776,45 +776,18 @@ mod tests { async fn get_object_meta(&self, _bucket: &str, _object: &str) -> crate::Result> { Ok(None) } - async fn get_object_data(&self, _bucket: &str, _object: &str) -> crate::Result>> { - Ok(None) - } - async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> crate::Result<()> { - Ok(()) - } - async fn delete_object(&self, _bucket: &str, _object: &str) -> crate::Result<()> { - Ok(()) - } - async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> crate::Result { - Ok(true) - } async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> crate::Result> { Ok(vec![]) } - async fn get_disk_status(&self, _endpoint: &Endpoint) -> crate::Result { - Ok(crate::heal::storage::DiskStatus::Ok) - } - async fn format_disk(&self, _endpoint: &Endpoint) -> crate::Result<()> { - Ok(()) - } async fn get_bucket_info(&self, _bucket: &str) -> crate::Result> { Ok(None) } - async fn heal_bucket_metadata(&self, _bucket: &str) -> crate::Result<()> { - Ok(()) - } async fn list_buckets(&self) -> crate::Result> { Ok(vec![]) } async fn object_exists(&self, _bucket: &str, _object: &str) -> crate::Result { Ok(false) } - async fn get_object_size(&self, _bucket: &str, _object: &str) -> crate::Result> { - Ok(None) - } - async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> crate::Result> { - Ok(None) - } async fn heal_object( &self, _bucket: &str, @@ -837,13 +810,6 @@ mod tests { ) -> crate::Result<(rustfs_madmin::heal_commands::HealResultItem, Option)> { Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None)) } - async fn list_objects_for_heal( - &self, - _bucket: &str, - _prefix: &str, - ) -> crate::Result> { - Ok(vec![]) - } async fn list_objects_for_heal_page( &self, _bucket: &str, diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 04075837c..920fffcf5 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -1267,7 +1267,7 @@ mod resume_loop_tests { CheckpointManager, RESUME_CHECKPOINT_FILE, ReplacementTargetIdentity, ResumeDeleteFailure, ResumeManager, ResumeUtils, compose_key, }; - use crate::heal::storage::{DiskStatus, HealLifecycleExpiryContext, HealListItem, HealObjectInfo, HealStorageAPI}; + use crate::heal::storage::{HealLifecycleExpiryContext, HealListItem, HealObjectInfo, HealStorageAPI}; use crate::heal::storage_api::status::BucketInfo; use crate::heal::{ BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, RUSTFS_META_BUCKET, new_disk, @@ -1448,36 +1448,15 @@ mod resume_loop_tests { async fn get_object_meta(&self, _b: &str, _o: &str) -> Result> { Ok(None) } - async fn get_object_data(&self, _b: &str, _o: &str) -> Result>> { - Ok(None) - } - async fn put_object_data(&self, _b: &str, _o: &str, _d: &[u8]) -> Result<()> { - Ok(()) - } - async fn delete_object(&self, _b: &str, _o: &str) -> Result<()> { - Ok(()) - } - async fn verify_object_integrity(&self, _b: &str, _o: &str) -> Result { - Ok(true) - } async fn ec_decode_rebuild(&self, _b: &str, _o: &str) -> Result> { Ok(Vec::new()) } - async fn get_disk_status(&self, _e: &Endpoint) -> Result { - Ok(DiskStatus::Ok) - } - async fn format_disk(&self, _e: &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, _b: &str) -> Result<()> { - Ok(()) - } async fn list_buckets(&self) -> Result> { Ok(Vec::new()) } @@ -1485,12 +1464,6 @@ mod resume_loop_tests { // Must never be consulted: the resume loop always goes through heal_object. panic!("object_exists must not be called by the resume heal loop"); } - async fn get_object_size(&self, _b: &str, _o: &str) -> Result> { - Ok(None) - } - async fn get_object_checksum(&self, _b: &str, _o: &str) -> Result> { - Ok(None) - } async fn load_heal_lifecycle_expiry_context(&self, _bucket: &str) -> Result> { Ok((!self.lifecycle_expired.lock().unwrap().is_empty()).then(HealLifecycleExpiryContext::test)) } @@ -1556,9 +1529,6 @@ mod resume_loop_tests { ReplacementCommitEvidence::Error(message) => Err(Error::other(message)), } } - async fn list_objects_for_heal(&self, _b: &str, _p: &str) -> Result> { - Ok(Vec::new()) - } async fn list_objects_for_heal_page( &self, _bucket: &str, diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 3a243c88e..4c3c622c7 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -3882,42 +3882,14 @@ mod tests { 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(crate::heal::storage::DiskStatus::Ok) - } - - async fn format_disk(&self, _endpoint: &Endpoint) -> Result<()> { - Ok(()) - } - async fn get_bucket_info(&self, _bucket: &str) -> Result> { Ok(None) } - async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<()> { - Ok(()) - } - async fn list_buckets(&self) -> Result> { if let Some(hook) = manager_recovery_test_hook() { *hook.listed.lock().expect("manager recovery listed lock should not poison") = true; @@ -3929,14 +3901,6 @@ mod tests { Ok(bucket == "retry-transition") } - 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, @@ -3998,10 +3962,6 @@ mod tests { Ok((HealResultItem::default(), None)) } - async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result> { - Ok(Vec::new()) - } - async fn list_objects_for_heal_page( &self, _bucket: &str, diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index fbe51f050..63ca002e4 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -27,7 +27,7 @@ use super::storage_api::storage::{ BucketInfo, BucketOperations, DiskSetSelector, HealOperations as _, ListOperations as _, ObjectIO as _, ObjectOperations as _, StorageAdminApi, }; -use super::{DiskStore, ECStore, Endpoint, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity}; +use super::{DiskStore, ECStore, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity}; pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader}; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] @@ -65,7 +65,6 @@ const LOG_COMPONENT_HEAL: &str = "heal"; const LOG_SUBSYSTEM_STORAGE: &str = "storage"; const EVENT_HEAL_STORAGE_OBJECT_IO: &str = "heal_storage_object_io"; const EVENT_HEAL_STORAGE_OBJECT_READ_LIMIT: &str = "heal_storage_object_read_limit"; -const EVENT_HEAL_STORAGE_OBJECT_VERIFY: &str = "heal_storage_object_verify"; const EVENT_HEAL_STORAGE_ADMIN_OP: &str = "heal_storage_admin_op"; const EVENT_HEAL_STORAGE_REPAIR_OP: &str = "heal_storage_repair_op"; @@ -312,56 +311,23 @@ pub struct HealListItem { pub is_delete_marker: bool, } -/// Disk status for heal operations -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum DiskStatus { - /// Ok - Ok, - /// Offline - Offline, - /// Corrupt - Corrupt, - /// Missing - Missing, - /// Permission denied - PermissionDenied, - /// Faulty - Faulty, - /// Root mount - RootMount, - /// Unknown - Unknown, - /// Unformatted - Unformatted, -} - /// Heal storage layer interface #[async_trait] pub trait HealStorageAPI: Send + Sync { /// Get object meta + /// + /// Reserved for HS-01 MRF wiring (rustfs/backlog#1865): MRF intents + /// currently execute through `heal_object`; keep this entry point for the + /// metadata-corruption variant that must inspect metadata first. async fn get_object_meta(&self, bucket: &str, object: &str) -> Result>; - /// Get object data - async fn get_object_data(&self, bucket: &str, object: &str) -> Result>>; - - /// Put object data - async fn put_object_data(&self, bucket: &str, object: &str, data: &[u8]) -> Result<()>; - - /// Delete object - async fn delete_object(&self, bucket: &str, object: &str) -> Result<()>; - - /// Check object integrity - async fn verify_object_integrity(&self, bucket: &str, object: &str) -> Result; - /// EC decode rebuild + /// + /// Reserved for HS-01 MRF wiring (rustfs/backlog#1865): urgent ECDecode + /// requests currently execute through `heal_object`; keep the explicit + /// rebuild-and-read path for the decode-failure fast variant. async fn ec_decode_rebuild(&self, bucket: &str, object: &str) -> Result>; - /// Get disk status - async fn get_disk_status(&self, endpoint: &Endpoint) -> Result; - - /// Format disk - async fn format_disk(&self, endpoint: &Endpoint) -> Result<()>; - /// Get bucket info async fn get_bucket_info(&self, bucket: &str) -> Result>; @@ -387,21 +353,12 @@ pub trait HealStorageAPI: Send + Sync { Ok(false) } - /// Fix bucket metadata - async fn heal_bucket_metadata(&self, bucket: &str) -> Result<()>; - /// Get all buckets async fn list_buckets(&self) -> Result>; /// Check object exists async fn object_exists(&self, bucket: &str, object: &str) -> Result; - /// Get object size - async fn get_object_size(&self, bucket: &str, object: &str) -> Result>; - - /// Get object checksum - async fn get_object_checksum(&self, bucket: &str, object: &str) -> Result>; - /// Heal object using ecstore async fn heal_object( &self, @@ -453,12 +410,6 @@ pub trait HealStorageAPI: Send + Sync { Ok(false) } - /// List object versions for healing (returns all versions, may use significant memory for large buckets) - /// - /// WARNING: This method loads all object versions into memory at once. For buckets with many - /// objects/versions, consider using `list_objects_for_heal_page` instead to process versions in pages. - async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result>; - /// List object versions for healing with pagination (returns one page and continuation token) /// Returns (versions, next_continuation_token, is_truncated). The continuation token is an /// opaque composite `(marker, version_marker)` value — see `encode_heal_token`/`decode_heal_token`. @@ -527,89 +478,11 @@ impl ECStoreHealStorage { pub fn new(ecstore: Arc) -> Self { Self { ecstore } } -} - -fn is_transient_object_exists_message(message: &str) -> bool { - let message = message.to_ascii_lowercase(); - - [ - "failed to acquire read lock", - "lock acquisition failed", - "lock acquisition timeout", - "quorum not reached", - "deadline has elapsed", - "timed out", - "network error", - "transport error", - "connection refused", - ] - .iter() - .any(|pattern| message.contains(pattern)) -} - -fn is_transient_object_exists_error(err: &StorageError) -> bool { - if err.is_quorum_error() { - return true; - } - - match err { - StorageError::Lock(lock_err) => lock_err.is_retryable() || is_transient_object_exists_message(&lock_err.to_string()), - StorageError::Io(io_err) => is_transient_object_exists_message(&io_err.to_string()), - StorageError::SlowDown | StorageError::OperationCanceled => true, - _ => false, - } -} - -#[async_trait] -impl HealStorageAPI for ECStoreHealStorage { - async fn get_object_meta(&self, bucket: &str, object: &str) -> Result> { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "get_object_meta", - bucket, - object, - "Heal storage request started" - ); - - match self.ecstore.get_object_info(bucket, object, &Default::default()).await { - Ok(info) => Ok(Some(info)), - Err(e) => { - // Map ObjectNotFound to None to align with Option return type - if matches!(e, StorageError::ObjectNotFound(_, _)) { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "get_object_meta", - bucket, - object, - result = "not_found", - "Heal storage object metadata missing" - ); - Ok(None) - } else { - error!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "get_object_meta", - bucket, - object, - result = "failed", - error = %e, - "Heal storage request failed" - ); - Err(Error::other(e)) - } - } - } - } + /// Read back an object's bytes, capped to bound memory. + /// + /// Private support for the reserved `ec_decode_rebuild` (HS-01); not part + /// of the storage trait surface. async fn get_object_data(&self, bucket: &str, object: &str) -> Result>> { debug!( target: "rustfs::heal::storage", @@ -695,196 +568,85 @@ impl HealStorageAPI for ECStoreHealStorage { } Ok(Some(buf)) } +} - async fn put_object_data(&self, bucket: &str, object: &str, data: &[u8]) -> Result<()> { +fn is_transient_object_exists_message(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + + [ + "failed to acquire read lock", + "lock acquisition failed", + "lock acquisition timeout", + "quorum not reached", + "deadline has elapsed", + "timed out", + "network error", + "transport error", + "connection refused", + ] + .iter() + .any(|pattern| message.contains(pattern)) +} + +fn is_transient_object_exists_error(err: &StorageError) -> bool { + if err.is_quorum_error() { + return true; + } + + match err { + StorageError::Lock(lock_err) => lock_err.is_retryable() || is_transient_object_exists_message(&lock_err.to_string()), + StorageError::Io(io_err) => is_transient_object_exists_message(&io_err.to_string()), + StorageError::SlowDown | StorageError::OperationCanceled => true, + _ => false, + } +} + +#[async_trait] +impl HealStorageAPI for ECStoreHealStorage { + async fn get_object_meta(&self, bucket: &str, object: &str) -> Result> { debug!( target: "rustfs::heal::storage", event = EVENT_HEAL_STORAGE_OBJECT_IO, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "put_object_data", - bucket, - object, - bytes = data.len(), - "Heal storage request started" - ); - - let mut reader = HealPutObjReader::from_vec(data.to_vec()); - match (*self.ecstore) - .put_object(bucket, object, &mut reader, &Default::default()) - .await - { - Ok(_) => { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "put_object_data", - bucket, - object, - result = "ok", - "Heal storage object write completed" - ); - Ok(()) - } - Err(e) => { - error!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "put_object_data", - bucket, - object, - result = "failed", - error = %e, - "Heal storage request failed" - ); - Err(Error::other(e)) - } - } - } - - async fn delete_object(&self, bucket: &str, object: &str) -> Result<()> { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "delete_object", + operation = "get_object_meta", bucket, object, "Heal storage request started" ); - match self.ecstore.delete_object(bucket, object, Default::default()).await { - Ok(_) => { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "delete_object", - bucket, - object, - result = "ok", - "Heal storage object delete completed" - ); - Ok(()) - } + match self.ecstore.get_object_info(bucket, object, &Default::default()).await { + Ok(info) => Ok(Some(info)), Err(e) => { - error!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "delete_object", - bucket, - object, - result = "failed", - error = %e, - "Heal storage request failed" - ); - Err(Error::other(e)) - } - } - } - - async fn verify_object_integrity(&self, bucket: &str, object: &str) -> Result { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_VERIFY, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - bucket, - object, - state = "started", - "Heal storage object verification started" - ); - - // Check object metadata first - match self.get_object_meta(bucket, object).await? { - Some(obj_info) => { - if obj_info.size < 0 { - warn!( + // Map ObjectNotFound to None to align with Option return type + if matches!(e, StorageError::ObjectNotFound(_, _)) { + debug!( target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_VERIFY, + event = EVENT_HEAL_STORAGE_OBJECT_IO, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "get_object_meta", bucket, object, - state = "invalid_size", - "Heal storage object verification failed" + result = "not_found", + "Heal storage object metadata missing" ); - return Ok(false); + Ok(None) + } else { + error!( + target: "rustfs::heal::storage", + event = EVENT_HEAL_STORAGE_OBJECT_IO, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_STORAGE, + operation = "get_object_meta", + bucket, + object, + result = "failed", + error = %e, + "Heal storage request failed" + ); + Err(Error::other(e)) } - - // Stream-read the object to a sink to avoid loading into memory - match (*self.ecstore) - .get_object_reader(bucket, object, None, Default::default(), &Default::default()) - .await - { - Ok(reader) => { - let mut stream = reader.stream; - match tokio::io::copy(&mut stream, &mut tokio::io::sink()).await { - Ok(_) => { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_VERIFY, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - bucket, - object, - state = "ok", - "Heal storage object verified" - ); - Ok(true) - } - Err(e) => { - warn!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_VERIFY, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - bucket, - object, - state = "stream_read_failed", - error = %e, - "Heal storage object verification failed" - ); - Ok(false) - } - } - } - Err(e) => { - warn!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_VERIFY, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - bucket, - object, - state = "reader_open_failed", - error = %e, - "Heal storage object verification failed" - ); - Ok(false) - } - } - } - None => { - warn!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_VERIFY, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - bucket, - object, - state = "metadata_missing", - "Heal storage object verification failed" - ); - Ok(false) } } } @@ -976,81 +738,6 @@ impl HealStorageAPI for ECStoreHealStorage { } } - async fn get_disk_status(&self, endpoint: &Endpoint) -> Result { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "get_disk_status", - endpoint = ?endpoint, - state = "started", - "Heal storage admin operation started" - ); - - // TODO: implement disk status check using ecstore - // For now, return Ok status - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "get_disk_status", - endpoint = ?endpoint, - result = "ok", - disk_status = "ok", - "Heal storage disk status resolved" - ); - Ok(DiskStatus::Ok) - } - - async fn format_disk(&self, endpoint: &Endpoint) -> Result<()> { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "format_disk", - endpoint = ?endpoint, - state = "started", - "Heal storage admin operation started" - ); - - // Use ecstore's heal_format - match self.heal_format(false).await { - Ok((_, error)) => { - if error.is_some() { - return Err(Error::other(format!("Format failed: {error:?}"))); - } - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "format_disk", - endpoint = ?endpoint, - result = "ok", - "Heal storage disk format completed" - ); - Ok(()) - } - Err(e) => { - error!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "format_disk", - endpoint = ?endpoint, - result = "failed", - error = %e, - "Heal storage admin operation failed" - ); - Err(e) - } - } - } - async fn get_bucket_info(&self, bucket: &str) -> Result> { debug!( target: "rustfs::heal::storage", @@ -1161,61 +848,6 @@ impl HealStorageAPI for ECStoreHealStorage { } } - async fn heal_bucket_metadata(&self, bucket: &str) -> Result<()> { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_REPAIR_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "heal_bucket_metadata", - bucket, - state = "started", - "Heal storage repair started" - ); - - let heal_opts = HealOpts { - recursive: true, - dry_run: false, - remove: false, - recreate: false, - scan_mode: HealScanMode::Normal, - update_parity: false, - no_lock: false, - pool: None, - set: None, - }; - - match self.heal_bucket(bucket, &heal_opts).await { - Ok(_) => { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_REPAIR_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "heal_bucket_metadata", - bucket, - result = "ok", - "Heal storage bucket metadata repaired" - ); - Ok(()) - } - Err(e) => { - error!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_REPAIR_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "heal_bucket_metadata", - bucket, - result = "failed", - error = %e, - "Heal storage repair failed" - ); - Err(e) - } - } - } - async fn list_buckets(&self) -> Result> { debug!( target: "rustfs::heal::storage", @@ -1315,48 +947,6 @@ impl HealStorageAPI for ECStoreHealStorage { } } - async fn get_object_size(&self, bucket: &str, object: &str) -> Result> { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "get_object_size", - bucket, - object, - "Heal storage request started" - ); - - match self.get_object_meta(bucket, object).await { - Ok(Some(obj_info)) => Ok(Some(obj_info.size as u64)), - Ok(None) => Ok(None), - Err(e) => Err(e), - } - } - - async fn get_object_checksum(&self, bucket: &str, object: &str) -> Result> { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_OBJECT_IO, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "get_object_checksum", - bucket, - object, - "Heal storage request started" - ); - - match self.get_object_meta(bucket, object).await { - Ok(Some(obj_info)) => { - // Convert checksum bytes to hex string - let checksum = obj_info.checksum.iter().map(|b| format!("{b:02x}")).collect::(); - Ok(Some(checksum)) - } - Ok(None) => Ok(None), - Err(e) => Err(e), - } - } - async fn heal_object( &self, bucket: &str, @@ -1547,65 +1137,6 @@ impl HealStorageAPI for ECStoreHealStorage { .map_err(Error::Storage) } - async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result> { - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "list_objects_for_heal", - bucket, - prefix, - state = "started", - "Heal storage admin operation started" - ); - warn!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "list_objects_for_heal", - bucket, - prefix, - state = "memory_heavy", - "Heal storage version listing loads all versions into memory (footprint is per-version, not per-object)" - ); - - let mut all_objects: Vec = Vec::new(); - let mut continuation_token: Option = None; - - loop { - let (page_objects, next_token, is_truncated) = self - .list_objects_for_heal_page(bucket, prefix, continuation_token.as_deref(), false) - .await?; - - all_objects.extend(page_objects); - - if !is_truncated { - break; - } - - continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?; - if continuation_token.is_none() { - break; - } - } - - debug!( - target: "rustfs::heal::storage", - event = EVENT_HEAL_STORAGE_ADMIN_OP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_STORAGE, - operation = "list_objects_for_heal", - bucket, - prefix, - object_count = all_objects.len(), - result = "ok", - "Heal storage object listing completed" - ); - Ok(all_objects) - } - async fn list_objects_for_heal_page( &self, bucket: &str, diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 2c0626d8f..7b1d33348 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -2822,7 +2822,7 @@ impl std::fmt::Debug for HealTask { mod tests { use super::super::{DiskOption, DiskStore, Endpoint, HealDiskExt as _, new_disk}; use super::*; - use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo}; + use crate::heal::storage::{HealListItem, HealObjectInfo}; use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, TraceSubscription, TraceVal, subscribe_trace_events}; use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos}; use std::collections::{HashMap, VecDeque}; @@ -3354,7 +3354,6 @@ mod tests { object_exists_by_name: Mutex>, heal_object_outcome: Mutex>, heal_object_outcomes: Mutex>>, - deleted_objects: Mutex>, format_no_heal_required: Mutex, global_format_calls: Mutex, replacement_format_calls: Mutex)>>, @@ -3547,35 +3546,10 @@ mod tests { 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<()> { - self.deleted_objects.lock().unwrap().push(object.to_string()); - 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(), @@ -3590,10 +3564,6 @@ mod tests { Ok(*self.usage_baseline.lock().unwrap()) } - async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<()> { - Ok(()) - } - async fn list_buckets(&self) -> Result> { let buckets = self .listed_buckets @@ -3621,14 +3591,6 @@ mod tests { Ok(self.object_exists.lock().unwrap().unwrap_or(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, @@ -3764,10 +3726,6 @@ mod tests { Ok(*self.replacement_targets_ready.lock().unwrap()) } - async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result> { - Ok(vec![heal_item("object-a"), heal_item("object-b")]) - } - async fn list_objects_for_heal_page( &self, bucket: &str, @@ -4858,7 +4816,7 @@ mod tests { } #[tokio::test] - async fn test_heal_failure_with_remove_corrupted_does_not_delete_object() { + async fn test_heal_failure_with_remove_corrupted_propagates_remove_flag() { let storage = Arc::new(MockStorage { object_exists: Mutex::new(Some(true)), heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::OkWithOtherError( @@ -4884,7 +4842,6 @@ mod tests { let err = task.execute().await.expect_err("heal failure should still be reported"); assert!(matches!(err, Error::TaskExecutionFailed { .. })); - assert!(storage.deleted_objects.lock().unwrap().is_empty()); assert!(storage.object_heal_opts.lock().unwrap()[0].remove); } diff --git a/crates/heal/src/lib.rs b/crates/heal/src/lib.rs index f1ec4cebe..29444d0f5 100644 --- a/crates/heal/src/lib.rs +++ b/crates/heal/src/lib.rs @@ -352,8 +352,8 @@ pub(crate) fn set_heal_queue_length(count: usize) { mod tests { use super::{ Error, HEAL_RUNTIME_INIT_TEST_HOOK, HealRuntimeInitTestHook, get_heal_channel_processor, get_heal_manager, - heal::DiskStore, heal::Endpoint, heal::manager::HealConfig, heal::storage::DiskStatus, heal::storage::HealListItem, - heal::storage::HealObjectInfo, heal::storage::HealStorageAPI, init_heal_manager, run_owned_initialization, + heal::DiskStore, heal::manager::HealConfig, heal::storage::HealListItem, heal::storage::HealObjectInfo, + heal::storage::HealStorageAPI, init_heal_manager, run_owned_initialization, }; use crate::heal::storage_api::status::BucketInfo; use rustfs_common::heal_channel::HealOpts; @@ -370,42 +370,14 @@ mod tests { Ok(None) } - async fn get_object_data(&self, _bucket: &str, _object: &str) -> Result>, Error> { - Ok(None) - } - - async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> Result<(), Error> { - Ok(()) - } - - async fn delete_object(&self, _bucket: &str, _object: &str) -> Result<(), Error> { - 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, Error> { Ok(Vec::new()) } - async fn get_disk_status(&self, _endpoint: &Endpoint) -> Result { - Ok(DiskStatus::Ok) - } - - async fn format_disk(&self, _endpoint: &Endpoint) -> Result<(), Error> { - Ok(()) - } - async fn get_bucket_info(&self, _bucket: &str) -> Result, Error> { Ok(None) } - async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<(), Error> { - Ok(()) - } - async fn list_buckets(&self) -> Result, Error> { Ok(Vec::new()) } @@ -414,14 +386,6 @@ mod tests { Ok(false) } - async fn get_object_size(&self, _bucket: &str, _object: &str) -> Result, Error> { - Ok(None) - } - - async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> Result, Error> { - Ok(None) - } - async fn heal_object( &self, _bucket: &str, @@ -440,10 +404,6 @@ mod tests { Ok((HealResultItem::default(), None)) } - async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result, Error> { - Ok(Vec::new()) - } - async fn list_objects_for_heal_page( &self, _bucket: &str, diff --git a/crates/heal/tests/heal_bug_fixes_test.rs b/crates/heal/tests/heal_bug_fixes_test.rs index dc5addef2..1bdb00348 100644 --- a/crates/heal/tests/heal_bug_fixes_test.rs +++ b/crates/heal/tests/heal_bug_fixes_test.rs @@ -184,45 +184,18 @@ fn test_heal_task_status_atomic_update() { async fn get_object_meta(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { Ok(None) } - async fn get_object_data(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result>> { - Ok(None) - } - async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> rustfs_heal::Result<()> { - Ok(()) - } - async fn delete_object(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<()> { - Ok(()) - } - async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result { - Ok(true) - } async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { Ok(vec![]) } - async fn get_disk_status(&self, _endpoint: &Endpoint) -> rustfs_heal::Result { - Ok(rustfs_heal::heal::storage::DiskStatus::Ok) - } - async fn format_disk(&self, _endpoint: &Endpoint) -> rustfs_heal::Result<()> { - Ok(()) - } async fn get_bucket_info(&self, _bucket: &str) -> rustfs_heal::Result> { Ok(None) } - async fn heal_bucket_metadata(&self, _bucket: &str) -> rustfs_heal::Result<()> { - Ok(()) - } async fn list_buckets(&self) -> rustfs_heal::Result> { Ok(vec![]) } async fn object_exists(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result { Ok(false) } - async fn get_object_size(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { - Ok(None) - } - async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { - Ok(None) - } async fn heal_object( &self, _bucket: &str, @@ -245,9 +218,6 @@ fn test_heal_task_status_atomic_update() { ) -> rustfs_heal::Result<(rustfs_madmin::heal_commands::HealResultItem, Option)> { Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None)) } - async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result> { - Ok(vec![]) - } async fn list_objects_for_heal_page( &self, _bucket: &str, @@ -289,7 +259,7 @@ fn test_heal_task_status_atomic_update() { #[tokio::test] async fn test_heal_task_transient_object_exists_skip_avoids_recreate() { - use rustfs_heal::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI}; + use rustfs_heal::heal::storage::{HealListItem, HealObjectInfo, HealStorageAPI}; use rustfs_heal::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType}; use std::sync::{ Arc, @@ -307,42 +277,14 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() { Ok(None) } - async fn get_object_data(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result>> { - Ok(None) - } - - async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> rustfs_heal::Result<()> { - Ok(()) - } - - async fn delete_object(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<()> { - Ok(()) - } - - async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result { - Ok(true) - } - async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { Ok(Vec::new()) } - async fn get_disk_status(&self, _endpoint: &Endpoint) -> rustfs_heal::Result { - Ok(DiskStatus::Ok) - } - - async fn format_disk(&self, _endpoint: &Endpoint) -> rustfs_heal::Result<()> { - Ok(()) - } - async fn get_bucket_info(&self, _bucket: &str) -> rustfs_heal::Result> { Ok(None) } - async fn heal_bucket_metadata(&self, _bucket: &str) -> rustfs_heal::Result<()> { - Ok(()) - } - async fn list_buckets(&self) -> rustfs_heal::Result> { Ok(Vec::new()) } @@ -354,14 +296,6 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() { )) } - async fn get_object_size(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { - Ok(None) - } - - async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { - Ok(None) - } - async fn heal_object( &self, _bucket: &str, @@ -388,10 +322,6 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() { Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None)) } - async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result> { - Ok(Vec::new()) - } - async fn list_objects_for_heal_page( &self, _bucket: &str, diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index c573065f7..c15ed6c41 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -71,7 +71,6 @@ const EVENT_SCANNER_METADATA_CORRUPT: &str = "scanner_metadata_corrupt"; const EVENT_SCANNER_LIFECYCLE_ACTION: &str = "scanner_lifecycle_action"; const EVENT_SCANNER_HEAL_ADMISSION: &str = "scanner_heal_admission"; const EVENT_SCANNER_ALERT_STATE: &str = "scanner_alert_state"; -const EVENT_SCANNER_COMPAT_STATE: &str = "scanner_compat_state"; const DATA_USAGE_UPDATE_DIR_CYCLES: u32 = 16; const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500; @@ -92,7 +91,6 @@ 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 DEFAULT_SCANNER_DEEP_VERIFY_COOLDOWN_SECS: u64 = 60; -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"; @@ -196,8 +194,6 @@ fn emit_scanner_alert_event(event_name: &str, bucket: &str, object: &str, size: } const MAX_PENDING_SCANNER_HEALS_PER_BUCKET: usize = 10_000; -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(); #[cfg(test)] @@ -251,27 +247,6 @@ fn effective_object_heal_scan_mode(heal_bitrot: bool, mod_time: Option bool { - scanner_inline_heal_enabled_from_value(std::env::var(rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE).ok().as_deref()) -} - -fn scanner_inline_heal_enabled_from_value(value: Option<&str>) -> bool { - match value { - Some(value) => matches!(value.trim().to_ascii_lowercase().as_str(), "1" | "true" | "on" | "yes"), - None => rustfs_config::DEFAULT_SCANNER_INLINE_HEAL_ENABLE, - } -} - -fn ensure_scanner_inline_heal_metric_registered() { - SCANNER_INLINE_HEAL_METRICS_ONCE.call_once(|| { - describe_counter!( - METRIC_SCANNER_INLINE_HEAL_TOTAL, - "Total number of inline heal operations executed directly by scanner." - ); - counter!(METRIC_SCANNER_INLINE_HEAL_TOTAL).increment(0); - }); -} - fn ensure_scanner_alert_metrics_registered() { SCANNER_ALERT_METRICS_ONCE.call_once(|| { describe_counter!( @@ -505,24 +480,6 @@ fn should_alert_excessive_versions(remaining_versions: usize, cumulative_size: i (too_many_versions, too_large_versions) } -fn warn_inline_heal_compat_requested() { - if !scanner_inline_heal_enabled() { - return; - } - - SCANNER_INLINE_HEAL_WARN_ONCE.call_once(|| { - warn!( - target: "rustfs::scanner::folder", - event = EVENT_SCANNER_COMPAT_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_HEAL, - env = rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE, - state = "inline_heal_rollback_unsupported", - "Scanner inline-heal rollback is unsupported; using async heal admission" - ); - }); -} - fn non_negative_i64_to_u64(value: i64) -> u64 { value.max(0) as u64 } @@ -1282,7 +1239,6 @@ impl ScannerItem { async fn heal_actions(&mut self, oi: &ObjectInfo, actual_size: i64, size_summary: &mut SizeSummary) -> i64 { if self.heal_enabled { - warn_inline_heal_compat_requested(); self.enqueue_heal(oi).await; } @@ -3231,8 +3187,6 @@ pub async fn scan_data_folder( ) -> Result { use crate::data_usage_define::DATA_USAGE_ROOT; - ensure_scanner_inline_heal_metric_registered(); - // Check that we're not trying to scan the root if cache.info.name.is_empty() || cache.info.name == DATA_USAGE_ROOT { return Err(ScannerError::Other("internal error: root scan attempted".to_string())); @@ -4325,19 +4279,6 @@ mod tests { assert!(!scanner.new_cache.info.failed_objects.contains_key("expired")); } - #[test] - fn test_scanner_inline_heal_enabled_defaults_to_false() { - assert!(!scanner_inline_heal_enabled_from_value(None)); - } - - #[test] - fn test_scanner_inline_heal_enabled_reads_env_override() { - assert!(scanner_inline_heal_enabled_from_value(Some("true"))); - assert!(scanner_inline_heal_enabled_from_value(Some("YES"))); - assert!(scanner_inline_heal_enabled_from_value(Some("1"))); - assert!(!scanner_inline_heal_enabled_from_value(Some("false"))); - } - #[test] fn test_build_object_heal_request_omits_nil_version_id() { let request = build_object_heal_request( diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 3d98e966a..9ae59353b 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -14,10 +14,9 @@ use crate::admin::auth::{authenticate_request, validate_admin_request}; use crate::admin::router::{AdminOperation, Operation, S3Router}; -use crate::admin::runtime_sources::{app_context_from_req, object_store_from_extensions}; +use crate::admin::runtime_sources::app_context_from_req; use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket; use crate::admin::storage_api::bucket::utils::is_valid_object_prefix; -use crate::admin::storage_api::contract::heal::HealOperations as _; use crate::server::ADMIN_PREFIX; use crate::server::RemoteAddr; use crate::storage::rpc::node_service::heal::{ @@ -1219,41 +1218,6 @@ fn validate_heal_request_mode(hip: &HealInitParams) -> S3Result<()> { Ok(()) } -fn should_handle_root_heal_directly(_hip: &HealInitParams) -> bool { - false -} - -fn map_root_heal_status(heal_err: Option) -> S3Result<()> { - match heal_err { - None => Ok(()), - Some(crate::admin::storage_api::error::StorageError::NoHealRequired) => { - info!( - event = EVENT_ADMIN_RESPONSE_EMITTED, - component = LOG_COMPONENT_ADMIN_API, - subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, - operation = "root_heal", - result = "success", - state = "no_heal_required", - "admin response emitted" - ); - Ok(()) - } - Some(err) => { - warn!( - event = EVENT_ADMIN_REQUEST_FAILED, - component = LOG_COMPONENT_ADMIN_API, - subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, - operation = "root_heal", - result = "failed", - reason = "root_heal_failed", - error = %err, - "admin request failed" - ); - Err(s3_error!(InternalError, "root heal failed: {err}")) - } - } -} - fn json_response(status: StatusCode, body: Vec) -> S3Response<(StatusCode, Body)> { let mut headers = HeaderMap::new(); headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); @@ -1358,50 +1322,6 @@ impl Operation for HealHandler { } }; let hip = extract_heal_init_params(&bytes, &req.uri, params)?; - // The heal channel currently models bucket/object work. Root heal reuses the - // existing format-heal path directly so `/v3/heal/` is accepted intentionally. - if should_handle_root_heal_directly(&hip) { - let Some(store) = object_store_from_extensions(&req.extensions) else { - warn!( - event = EVENT_ADMIN_REQUEST_FAILED, - component = LOG_COMPONENT_ADMIN_API, - subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, - operation = "root_heal", - result = "failed", - reason = "server_not_initialized", - "admin request failed" - ); - return Err(s3_error!(InternalError, "server not initialized")); - }; - - let (_, heal_err) = store.heal_format(hip.hs.dry_run).await.map_err(|e| { - warn!( - event = EVENT_ADMIN_REQUEST_FAILED, - component = LOG_COMPONENT_ADMIN_API, - subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, - operation = "root_heal", - result = "failed", - reason = "heal_format_failed", - error = %e, - "admin request failed" - ); - 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)?; - info!( - event = EVENT_ADMIN_RESPONSE_EMITTED, - component = LOG_COMPONENT_ADMIN_API, - subsystem = LOG_SUBSYSTEM_HEAL_ADMIN, - operation = "root_heal", - result = "success", - state = "started", - "admin response emitted" - ); - - return Ok(json_response(StatusCode::OK, body)); - } validate_heal_request_mode(&hip)?; let response_operation = if hip.force_stop { "cancel_heal" @@ -1614,11 +1534,9 @@ mod tests { build_replacement_recovery_status_response, encode_background_heal_status, encode_heal_control_path, encode_heal_start_success, encode_heal_task_status, execute_after_heal_control_capability, heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary, heal_control_response_id, json_response, - map_heal_response, map_root_heal_status, merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status, - query_peer_replacement_recovery_status, reject_heal_admission, should_handle_root_heal_directly, - validate_heal_request_mode, validate_heal_target, + map_heal_response, merge_peer_heal_statuses, peer_topology_complete, query_peer_heal_status, + query_peer_replacement_recovery_status, reject_heal_admission, validate_heal_request_mode, validate_heal_target, }; - use crate::admin::storage_api::error::StorageError; use crate::storage::rpc::node_service::heal::{ NodeHealProgress, NodeHealStatusSnapshot, NodeReplacementRecoveryStatusSnapshot, encode_node_replacement_recovery_status, }; @@ -2086,48 +2004,63 @@ mod tests { } #[test] - fn test_should_handle_root_heal_directly_is_disabled_for_root_start_modes() { - assert!(!should_handle_root_heal_directly(&HealInitParams::default())); - assert!(!should_handle_root_heal_directly(&HealInitParams { - force_start: true, - ..Default::default() - })); - } - - #[test] - fn test_should_handle_root_heal_directly_skips_query_cancel_and_bucket_targets() { - assert!(!should_handle_root_heal_directly(&HealInitParams { - client_token: "heal-token".to_string(), - ..Default::default() - })); - assert!(!should_handle_root_heal_directly(&HealInitParams { - force_stop: true, - ..Default::default() - })); - assert!(!should_handle_root_heal_directly(&HealInitParams { - bucket: "bucket".to_string(), - ..Default::default() - })); - assert!(!should_handle_root_heal_directly(&HealInitParams { - hs: HealOpts { - pool: Some(1), - set: Some(2), + fn test_root_heal_shapes_route_through_cluster_coordination() { + // Root heal has no direct local store path: every start shape is either + // rejected by validate_heal_request_mode or submitted to the cluster + // heal channel as an Admin-sourced request (see HealHandler::call). + let accepted_root_starts = [ + HealInitParams { + hs: HealOpts { + recursive: true, + ..Default::default() + }, ..Default::default() }, - ..Default::default() - })); - } + HealInitParams { + force_start: true, + hs: HealOpts { + recursive: true, + ..Default::default() + }, + ..Default::default() + }, + HealInitParams { + hs: HealOpts { + pool: Some(1), + set: Some(2), + ..Default::default() + }, + ..Default::default() + }, + ]; + for hip in accepted_root_starts { + validate_heal_request_mode(&hip).expect("accepted root heal start must reach cluster coordination"); + let request = build_heal_channel_request(&hip); + assert_eq!(request.bucket, "", "root heal must stay cluster-scoped"); + assert_eq!(request.source, HealRequestSource::Admin); + assert!(!request.id.is_empty(), "cluster heal requests carry a dedup id"); + } - #[test] - fn test_map_root_heal_status_allows_no_heal_required() { - map_root_heal_status(Some(StorageError::NoHealRequired)).expect("NoHealRequired should stay non-fatal"); - } - - #[test] - fn test_map_root_heal_status_rejects_fatal_errors() { - let err = map_root_heal_status(Some(StorageError::Unexpected)).expect_err("fatal status must fail"); - assert_eq!(err.code(), &S3ErrorCode::InternalError); - assert!(err.to_string().contains("root heal failed: Unexpected error")); + // Shapes that cannot start a tracked heal (plain start, bare force_start + // without recursive, bare pool) are rejected instead of falling back to + // a direct local path. + for hip in [ + HealInitParams::default(), + HealInitParams { + force_start: true, + ..Default::default() + }, + HealInitParams { + hs: HealOpts { + pool: Some(1), + ..Default::default() + }, + ..Default::default() + }, + ] { + let err = validate_heal_request_mode(&hip).expect_err("unscoped root heal start must be rejected"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } } #[test] diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index eb38e9640..318718b49 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -909,10 +909,6 @@ pub(crate) mod contract { }; } - pub(crate) mod heal { - pub(crate) use super::super::storage_contracts::HealOperations; - } - pub(crate) mod list { pub(crate) use super::super::storage_contracts::ListOperations; } diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 352bfbd58..2503e25c7 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -2206,7 +2206,7 @@ mod tests { previous_scanner_activity_response, remove_heal_control_replay, scanner_activity_response, stop_rebalance_response, }; use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; - use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo, HealEndpoint}; + use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo}; use crate::storage::storage_api::set_tonic_canonical_body_digest; use crate::storage::storage_api::{ Endpoint, @@ -2334,42 +2334,14 @@ mod tests { Ok(None) } - async fn get_object_data(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result>> { - Ok(None) - } - - async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> rustfs_heal::Result<()> { - Ok(()) - } - - async fn delete_object(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<()> { - Ok(()) - } - - async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result { - Ok(true) - } - async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { Ok(Vec::new()) } - async fn get_disk_status(&self, _endpoint: &HealEndpoint) -> rustfs_heal::Result { - Ok(rustfs_heal::heal::storage::DiskStatus::Ok) - } - - async fn format_disk(&self, _endpoint: &HealEndpoint) -> rustfs_heal::Result<()> { - Ok(()) - } - async fn get_bucket_info(&self, _bucket: &str) -> rustfs_heal::Result> { Ok(None) } - async fn heal_bucket_metadata(&self, _bucket: &str) -> rustfs_heal::Result<()> { - Ok(()) - } - async fn list_buckets(&self) -> rustfs_heal::Result> { Ok(Vec::new()) } @@ -2378,14 +2350,6 @@ mod tests { Ok(false) } - async fn get_object_size(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { - Ok(None) - } - - async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { - Ok(None) - } - async fn heal_object( &self, _bucket: &str, @@ -2411,14 +2375,6 @@ mod tests { Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None)) } - async fn list_objects_for_heal( - &self, - _bucket: &str, - _prefix: &str, - ) -> rustfs_heal::Result> { - Ok(Vec::new()) - } - async fn list_objects_for_heal_page( &self, _bucket: &str, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 534269546..ef9a92414 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -252,8 +252,6 @@ pub(crate) mod rpc_consumer { }; pub(crate) type StorageResult = super::super::Result; - #[cfg(test)] - pub(crate) type HealEndpoint = super::super::ecstore_disk::endpoint::Endpoint; #[cfg(test)] pub(crate) type HealBucketInfo = super::super::contract::bucket::BucketInfo;