diff --git a/crates/ecstore/src/erasure_coding/heal.rs b/crates/ecstore/src/erasure_coding/heal.rs index 797d95bc4..4d5874b58 100644 --- a/crates/ecstore/src/erasure_coding/heal.rs +++ b/crates/ecstore/src/erasure_coding/heal.rs @@ -19,7 +19,7 @@ use crate::erasure_coding::decode::ParallelReader; use crate::erasure_coding::encode::MultiWriter; use bytes::Bytes; use tokio::io::AsyncRead; -use tracing::info; +use tracing::{info, warn}; impl super::Erasure { pub async fn heal( @@ -60,10 +60,14 @@ impl super::Erasure { // We need at least data_shards available shards (data + parity combined) let available_shards = errs.iter().filter(|e| e.is_none()).count(); if available_shards < self.data_shards { - return Err(Error::other(format!( - "can not reconstruct data: not enough available shards (need {}, have {}) {errs:?}", - self.data_shards, available_shards - ))); + warn!( + required_data_shards = self.data_shards, + available_shards, + total_shards = errs.len(), + errors = ?errs, + "Erasure heal read quorum unavailable" + ); + return Err(Error::ErasureReadQuorum); } if self.parity_shards > 0 { @@ -193,4 +197,36 @@ mod tests { .expect("inline writer should retain data"); assert_eq!(healed, encoded[missing_data].to_vec()); } + + #[tokio::test] + async fn heal_returns_read_quorum_when_available_shards_are_insufficient() { + let erasure = Erasure::new(3, 2, 64); + let data = b"heal should fail before decode when too few shards are readable"; + let encoded = erasure.encode_data(data).expect("encode should succeed"); + + let readers = encoded + .iter() + .enumerate() + .map(|(index, shard)| { + if index < 2 { + Some(BitrotReader::new( + Cursor::new(shard.to_vec()), + erasure.shard_size(), + HashAlgorithm::None, + false, + )) + } else { + None + } + }) + .collect::>(); + let mut writers = (0..erasure.total_shard_count()).map(|_| None).collect::>(); + + let err = erasure + .heal(&mut writers, readers, data.len(), &[]) + .await + .expect_err("heal should fail when available shards are below data shards"); + + assert!(matches!(err, Error::ErasureReadQuorum)); + } } diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 3305ec513..236cde17d 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -4763,7 +4763,9 @@ impl rustfs_storage_api::HealOperations for SetDisks { let disks = self.disks.read().await; let disks = disks.clone(); - let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false).await?; + let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false) + .await + .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; if DiskError::is_all_not_found(&errs) { debug!( event = EVENT_SET_DISK_HEAL, @@ -4791,14 +4793,20 @@ impl rustfs_storage_api::HealOperations for SetDisks { // Pass no_lock=true since we already obtained write lock (or are already called with no_lock=true) let mut inner_opts = *opts; inner_opts.no_lock = true; - let (result, err) = self.heal_object(bucket, object, version_id, &inner_opts).await?; + let (result, err) = self + .heal_object(bucket, object, version_id, &inner_opts) + .await + .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; if let Some(err) = err.as_ref() { match err { &DiskError::FileCorrupt if opts.scan_mode != HealScanMode::Deep => { // Instead of returning an error when a bitrot error is detected // during a normal heal scan, heal again with bitrot flag enabled. inner_opts.scan_mode = HealScanMode::Deep; - let (result, err) = self.heal_object(bucket, object, version_id, &inner_opts).await?; + let (result, err) = self + .heal_object(bucket, object, version_id, &inner_opts) + .await + .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; return Ok((result, err.map(|e| e.into()))); } _ => {} diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index 67ce3a155..ca3c3830f 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/heal.rs @@ -215,13 +215,14 @@ impl SetDisks { let required_data = total_disks.saturating_sub(latest_meta.erasure.parity_blocks); error!( - "Data corruption detected for {}/{}: Insufficient healthy shards. Need at least {} data shards, but found only {} healthy disks. (Missing/Corrupt: {}, Parity: {})", bucket, object, - required_data, - healthy_count, - disks_to_heal_count, - latest_meta.erasure.parity_blocks + version_id, + required_data_shards = required_data, + healthy_shards = healthy_count, + missing_or_corrupt_shards = disks_to_heal_count, + parity_shards = latest_meta.erasure.parity_blocks, + "Heal object cannot reconstruct with available shards" ); // Allow for dangling deletes, on versions that have DataDir missing etc. @@ -253,16 +254,23 @@ impl SetDisks { Ok((self.default_heal_result(m, &t_errs, bucket, object, version_id).await, Some(derr))) } Err(err) => { - // t_errs = vec![Some(err.clone()]; errs.len()); + error!( + bucket, + object, + version_id, + error = %err, + "Heal object dangling cleanup could not prove object deletion" + ); + let quorum_err = DiskError::ErasureReadQuorum; let mut t_errs = Vec::with_capacity(errs.len()); for _ in 0..errs.len() { - t_errs.push(Some(err.clone())); + t_errs.push(Some(quorum_err.clone())); } Ok(( self.default_heal_result(FileInfo::default(), &t_errs, bucket, object, version_id) .await, - Some(err), + Some(quorum_err), )) } }; diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index a12497500..f8cfd5bfa 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -21,7 +21,7 @@ use crate::heal::{ use crate::{Error, Result}; use rustfs_common::heal_channel::{ HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealChannelResponse, - HealScanMode, publish_heal_response, + HealRequestSource, HealScanMode, publish_heal_response, }; use rustfs_madmin::heal_commands::HealResultItem; use serde::Serialize; @@ -460,11 +460,16 @@ impl HealChannelProcessor { HealChannelPriority::Critical => HealPriority::Urgent, }; + let recreate_missing = request.recreate_missing.unwrap_or(match request.source { + HealRequestSource::Scanner => false, + HealRequestSource::Admin | HealRequestSource::AutoHeal | HealRequestSource::Internal => true, + }); + // Build HealOptions with all available fields let options = HealOptions { scan_mode: request.scan_mode.unwrap_or(HealScanMode::Normal), remove_corrupted: request.remove_corrupted.unwrap_or(false), - recreate_missing: request.recreate_missing.unwrap_or(true), + recreate_missing, update_parity: request.update_parity.unwrap_or(true), recursive, dry_run: request.dry_run.unwrap_or(false), @@ -725,6 +730,109 @@ mod tests { assert!(heal_request.options.recreate_missing); } + #[tokio::test] + async fn test_convert_to_heal_request_scanner_defaults_recreate_missing_false() { + let heal_manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(heal_manager); + + let channel_request = HealChannelRequest { + id: "test-id".to_string(), + bucket: "test-bucket".to_string(), + object_prefix: Some("test-object".to_string()), + object_version_id: None, + disk: None, + priority: HealChannelPriority::Low, + scan_mode: None, + remove_corrupted: None, + recreate_missing: None, + update_parity: None, + recursive: Some(false), + dry_run: None, + timeout_seconds: None, + pool_index: None, + set_index: None, + force_start: false, + source: HealRequestSource::Scanner, + }; + + let heal_request = processor.convert_to_heal_request(channel_request).unwrap(); + + assert!(!heal_request.options.recreate_missing); + } + + #[tokio::test] + async fn test_convert_to_heal_request_non_scanner_defaults_recreate_missing_true() { + let heal_manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(heal_manager); + + for source in [ + HealRequestSource::Admin, + HealRequestSource::AutoHeal, + HealRequestSource::Internal, + ] { + let channel_request = HealChannelRequest { + id: format!("test-id-{source:?}"), + bucket: "test-bucket".to_string(), + object_prefix: Some("test-object".to_string()), + object_version_id: None, + disk: None, + priority: HealChannelPriority::Normal, + scan_mode: None, + remove_corrupted: None, + recreate_missing: None, + update_parity: None, + recursive: Some(false), + dry_run: None, + timeout_seconds: None, + pool_index: None, + set_index: None, + force_start: false, + source, + }; + + let heal_request = processor.convert_to_heal_request(channel_request).unwrap(); + + assert!(heal_request.options.recreate_missing); + } + } + + #[tokio::test] + async fn test_convert_to_heal_request_keeps_explicit_recreate_missing() { + let heal_manager = create_test_heal_manager(); + let processor = HealChannelProcessor::new(heal_manager); + + for (source, recreate_missing) in [ + (HealRequestSource::Scanner, true), + (HealRequestSource::Admin, false), + (HealRequestSource::AutoHeal, false), + (HealRequestSource::Internal, false), + ] { + let channel_request = HealChannelRequest { + id: format!("test-id-{source:?}-{recreate_missing}"), + bucket: "test-bucket".to_string(), + object_prefix: Some("test-object".to_string()), + object_version_id: None, + disk: None, + priority: HealChannelPriority::Normal, + scan_mode: None, + remove_corrupted: None, + recreate_missing: Some(recreate_missing), + update_parity: None, + recursive: Some(false), + dry_run: None, + timeout_seconds: None, + pool_index: None, + set_index: None, + force_start: false, + source, + }; + + let heal_request = processor.convert_to_heal_request(channel_request).unwrap(); + + assert_eq!(heal_request.options.recreate_missing, recreate_missing); + } + } + #[tokio::test] async fn test_convert_to_heal_request_prefix_when_recursive() { let heal_manager = create_test_heal_manager(); diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 250a0d92d..322aa46f2 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -33,7 +33,7 @@ use tokio::{ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; -use super::{DiskError, GLOBAL_LOCAL_DISK_MAP, HealDiskExt as _}; +use super::{DiskError, EcstoreError, GLOBAL_LOCAL_DISK_MAP, HealDiskExt as _}; const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60); const LOG_COMPONENT_HEAL: &str = "heal"; @@ -557,17 +557,31 @@ fn is_recoverable_heal_error(err: &Error, error: &str) -> bool { match err { Error::TaskCancelled => false, Error::TaskTimeout | Error::TransientSkip { .. } => true, - Error::TaskExecutionFailed { .. } - | Error::Storage(_) - | Error::Disk(_) - | Error::Io(_) - | Error::IO(_) - | Error::Anyhow(_) - | Error::Other(_) => is_recoverable_heal_error_message(error), + Error::Storage(err) => is_recoverable_storage_heal_error(err) || is_recoverable_heal_error_message(error), + Error::Disk(err) => is_recoverable_disk_heal_error(err) || is_recoverable_heal_error_message(error), + Error::TaskExecutionFailed { .. } | Error::Io(_) | Error::IO(_) | Error::Anyhow(_) | Error::Other(_) => { + is_recoverable_heal_error_message(error) + } _ => false, } } +fn is_recoverable_storage_heal_error(err: &EcstoreError) -> bool { + err.is_quorum_error() || matches!(err, EcstoreError::SlowDown | EcstoreError::OperationCanceled | EcstoreError::Lock(_)) +} + +fn is_recoverable_disk_heal_error(err: &DiskError) -> bool { + matches!( + err, + DiskError::ErasureReadQuorum + | DiskError::ErasureWriteQuorum + | DiskError::Timeout + | DiskError::SourceStalled + | DiskError::FaultyRemoteDisk + | DiskError::FaultyDisk + ) +} + fn is_recoverable_heal_error_message(error: &str) -> bool { let error = error.to_ascii_lowercase(); [ @@ -3031,6 +3045,33 @@ mod tests { assert!(retry_error.contains("Lock acquisition timeout")); } + #[test] + fn test_retry_request_for_typed_read_quorum_error() { + let storage: Arc = Arc::new(MockStorage); + let task = HealTask::from_request(HealRequest::object("bucket".to_string(), "object".to_string(), None), storage); + let result = Err(Error::Storage(EcstoreError::InsufficientReadQuorum( + "bucket".to_string(), + "object".to_string(), + ))); + + let (retry_request, retry_delay, retry_error) = + retry_request_for_result(&task, &result).expect("typed read quorum should be retryable"); + + assert_eq!(retry_request.id, task.id); + assert_eq!(retry_request.retry_attempts, 1); + assert!(retry_delay > Duration::ZERO); + assert!(retry_error.contains("Storage resources are insufficient")); + } + + #[test] + fn test_retry_request_for_typed_not_found_error_is_not_retryable() { + let storage: Arc = Arc::new(MockStorage); + let task = HealTask::from_request(HealRequest::object("bucket".to_string(), "object".to_string(), None), storage); + let result = Err(Error::Storage(EcstoreError::ObjectNotFound("bucket".to_string(), "object".to_string()))); + + assert!(retry_request_for_result(&task, &result).is_none()); + } + #[test] fn test_retry_request_for_recoverable_error_stops_at_limit() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index 6e654ea9b..de8a3aa27 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -937,7 +937,7 @@ impl HealStorageAPI for ECStoreHealStorage { match self.ecstore.heal_object(bucket, object, version_id_str, opts).await { Ok((result, ecstore_error)) => { - let error = ecstore_error.map(Error::other); + let error = ecstore_error.map(Error::Storage); debug!( target: "rustfs::heal::storage", event = EVENT_HEAL_STORAGE_REPAIR_OP, @@ -968,7 +968,7 @@ impl HealStorageAPI for ECStoreHealStorage { error = %e, "Heal storage repair failed" ); - Err(Error::other(e)) + Err(Error::Storage(e)) } } } @@ -1014,7 +1014,7 @@ impl HealStorageAPI for ECStoreHealStorage { error = %e, "Heal storage repair failed" ); - Err(Error::other(e)) + Err(Error::Storage(e)) } } } @@ -1033,7 +1033,7 @@ impl HealStorageAPI for ECStoreHealStorage { match self.ecstore.heal_format(dry_run).await { Ok((result, ecstore_error)) => { - let error = ecstore_error.map(Error::other); + let error = ecstore_error.map(Error::Storage); debug!( target: "rustfs::heal::storage", event = EVENT_HEAL_STORAGE_REPAIR_OP, @@ -1058,7 +1058,7 @@ impl HealStorageAPI for ECStoreHealStorage { error = %e, "Heal storage repair failed" ); - Err(Error::other(e)) + Err(Error::Storage(e)) } } } diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 844cfd66a..919091839 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -41,7 +41,6 @@ const EVENT_HEAL_TASK_STATE: &str = "heal_task_state"; const EVENT_HEAL_OBJECT_STAGE: &str = "heal_object_stage"; const EVENT_HEAL_OBJECT_MISSING: &str = "heal_object_missing"; const EVENT_HEAL_OBJECT_RESULT: &str = "heal_object_result"; -const EVENT_HEAL_OBJECT_CLEANUP: &str = "heal_object_cleanup"; const EVENT_HEAL_BUCKET_STAGE: &str = "heal_bucket_stage"; const EVENT_HEAL_BUCKET_RESULT: &str = "heal_bucket_result"; const EVENT_HEAL_METADATA_STAGE: &str = "heal_metadata_stage"; @@ -446,6 +445,31 @@ impl HealTask { } } + fn is_object_not_found_heal_error(err: &Error) -> bool { + match err { + Error::Disk(DiskError::FileNotFound | DiskError::FileVersionNotFound) => true, + Error::Storage( + EcstoreError::FileNotFound + | EcstoreError::FileVersionNotFound + | EcstoreError::ObjectNotFound(_, _) + | EcstoreError::VersionNotFound(_, _, _), + ) => true, + Error::Other(message) | Error::IO(message) => { + message.contains("File not found") + || message.contains("file not found") + || message.contains("File version not found") + || message.contains("file version not found") + || message.contains("Object not found") + || message.contains("object not found") + } + _ => false, + } + } + + fn should_return_typed_heal_error(err: &Error) -> bool { + matches!(err, Error::Storage(_) | Error::Disk(_)) + } + async fn skip_data_usage_cache_heal_error(&self, bucket: &str, object: &str, err: &Error) -> bool { if !Self::should_skip_data_usage_cache_heal_error(bucket, object, err) { return false; @@ -748,6 +772,18 @@ impl HealTask { "Heal object recreate requested" ); return self.recreate_missing_object(bucket, object, version_id).await; + } else if self.source == HealRequestSource::Scanner { + debug!( + target: "rustfs::heal::task", + event = EVENT_HEAL_OBJECT_STAGE, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_OBJECT, + task_id = %self.id, + bucket, + object, + stage = "scanner_missing_probe", + "Heal scanner missing object will be checked by storage layer" + ); } else { return Err(Error::TaskExecutionFailed { message: format!("Object not found: {bucket}/{object}"), @@ -798,9 +834,7 @@ impl HealTask { return Ok(()); } - // Check if this is a "File not found" error during delete operations - let error_msg = format!("{e}"); - if error_msg.contains("File not found") || error_msg.contains("not found") { + if Self::is_object_not_found_heal_error(&e) { debug!( target: "rustfs::heal::task", event = EVENT_HEAL_OBJECT_RESULT, @@ -832,42 +866,15 @@ impl HealTask { "Heal object operation failed" ); - // If heal failed and remove_corrupted is enabled, delete the corrupted object - if self.options.remove_corrupted { - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_OBJECT_CLEANUP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_OBJECT, - task_id = %self.id, - bucket, - object, - action = "delete_corrupted_object", - dry_run = self.options.dry_run, - "Heal object cleanup requested" - ); - if !self.options.dry_run { - self.await_with_control(self.storage.delete_object(bucket, object)).await?; - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_OBJECT_CLEANUP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_OBJECT, - task_id = %self.id, - bucket, - object, - action = "delete_corrupted_object", - result = "deleted", - "Heal corrupted object deleted" - ); - } - } - { let mut progress = self.progress.write().await; progress.update_progress(3, 3, 0, 0); } + if Self::should_return_typed_heal_error(&e) { + return Err(e); + } + return Err(Error::TaskExecutionFailed { message: format!("Failed to heal object {bucket}/{object}: {e}"), }); @@ -914,9 +921,7 @@ impl HealTask { return Ok(()); } - // Check if this is a "File not found" error during delete operations - let error_msg = format!("{e}"); - if error_msg.contains("File not found") || error_msg.contains("not found") { + if Self::is_object_not_found_heal_error(&e) { debug!( target: "rustfs::heal::task", event = EVENT_HEAL_OBJECT_RESULT, @@ -948,45 +953,18 @@ impl HealTask { "Heal object operation failed" ); - // If heal failed and remove_corrupted is enabled, delete the corrupted object - if self.options.remove_corrupted { - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_OBJECT_CLEANUP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_OBJECT, - task_id = %self.id, - bucket, - object, - action = "delete_corrupted_object", - dry_run = self.options.dry_run, - "Heal object cleanup requested" - ); - if !self.options.dry_run { - self.await_with_control(self.storage.delete_object(bucket, object)).await?; - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_OBJECT_CLEANUP, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_OBJECT, - task_id = %self.id, - bucket, - object, - action = "delete_corrupted_object", - result = "deleted", - "Heal corrupted object deleted" - ); - } - } - { let mut progress = self.progress.write().await; progress.update_progress(3, 3, 0, 0); } - Err(Error::TaskExecutionFailed { - message: format!("Failed to heal object {bucket}/{object}: {e}"), - }) + if Self::should_return_typed_heal_error(&e) { + Err(e) + } else { + Err(Error::TaskExecutionFailed { + message: format!("Failed to heal object {bucket}/{object}: {e}"), + }) + } } } } @@ -2242,6 +2220,7 @@ mod tests { object_exists: Mutex>, object_exists_by_name: Mutex>, heal_object_outcome: Mutex>, + deleted_objects: Mutex>, format_no_heal_required: Mutex, listed_prefixes: Mutex>, truncate_without_token: Mutex, @@ -2287,7 +2266,8 @@ mod tests { Ok(()) } - async fn delete_object(&self, _bucket: &str, _object: &str) -> Result<()> { + async fn delete_object(&self, _bucket: &str, object: &str) -> Result<()> { + self.deleted_objects.lock().unwrap().push(object.to_string()); Ok(()) } @@ -2356,6 +2336,7 @@ mod tests { .lock() .unwrap() .push(version_id.map(ToString::to_string)); + self.object_heal_opts.lock().unwrap().push(*opts); if let Some(outcome) = self.heal_object_outcome.lock().unwrap().take() { return match outcome { MockHealObjectOutcome::OkWithOtherError(message) => { @@ -2376,7 +2357,6 @@ mod tests { return Ok((HealResultItem::default(), Some(Error::Disk(DiskError::FileNotFound)))); } self.healed_objects.lock().unwrap().push(object.to_string()); - self.object_heal_opts.lock().unwrap().push(*opts); Ok(( HealResultItem { object_size: 1, @@ -3010,6 +2990,68 @@ mod tests { assert!(matches!(task.get_status().await, HealTaskStatus::Failed { .. })); } + #[tokio::test] + async fn test_heal_scanner_missing_object_without_recreate_probes_storage() { + let storage = Arc::new(MockStorage { + object_exists: Mutex::new(Some(false)), + ..Default::default() + }); + let mut request = HealRequest::new( + HealType::Object { + bucket: "bucket-a".to_string(), + object: "x.rnd".to_string(), + version_id: None, + }, + HealOptions { + recreate_missing: false, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + request.source = HealRequestSource::Scanner; + let task = HealTask::from_request(request, storage.clone()); + + task.execute() + .await + .expect("scanner missing object should be checked by storage"); + + assert!(matches!(task.get_status().await, HealTaskStatus::Completed)); + assert_eq!(storage.heal_object_calls.lock().unwrap().as_slice(), ["x.rnd".to_string()]); + assert!(!storage.object_heal_opts.lock().unwrap()[0].recreate); + } + + #[tokio::test] + async fn test_heal_scanner_missing_object_without_recreate_treats_not_found_as_stale() { + let storage = Arc::new(MockStorage { + object_exists: Mutex::new(Some(false)), + heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::ErrOther("File not found"))), + ..Default::default() + }); + let mut request = HealRequest::new( + HealType::Object { + bucket: "bucket-a".to_string(), + object: "x.rnd".to_string(), + version_id: None, + }, + HealOptions { + recreate_missing: false, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + request.source = HealRequestSource::Scanner; + let task = HealTask::from_request(request, storage.clone()); + + task.execute() + .await + .expect("scanner confirmed-not-found object should be treated as stale"); + + assert!(matches!(task.get_status().await, HealTaskStatus::Completed)); + assert_eq!(storage.heal_object_calls.lock().unwrap().as_slice(), ["x.rnd".to_string()]); + } + #[tokio::test] async fn test_heal_recreate_scanner_synthetic_object_dir_disk_not_found_fails() { let storage = Arc::new(MockStorage { @@ -3106,6 +3148,37 @@ mod tests { assert_eq!(result_items[0].object_size, 1); } + #[tokio::test] + async fn test_heal_failure_with_remove_corrupted_does_not_delete_object() { + let storage = Arc::new(MockStorage { + object_exists: Mutex::new(Some(true)), + heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::OkWithOtherError( + "can not reconstruct data: not enough available shards (need 12, have 11)", + ))), + ..Default::default() + }); + let request = HealRequest::new( + HealType::Object { + bucket: "bucket-a".to_string(), + object: "x.rnd".to_string(), + version_id: None, + }, + HealOptions { + remove_corrupted: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + 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); + } + #[tokio::test] async fn test_erasure_set_heal_continues_after_format_no_heal_required() { let storage = Arc::new(MockStorage::default()); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index fd3b05038..de3d9323b 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -437,6 +437,7 @@ fn build_bucket_heal_request(bucket: String, priority: HealChannelPriority) -> H HealChannelRequest { bucket, priority, + recreate_missing: Some(false), source: HealRequestSource::Scanner, ..Default::default() } @@ -456,6 +457,7 @@ fn build_object_heal_request( priority, scan_mode: Some(scan_mode), remove_corrupted: Some(HEAL_DELETE_DANGLING), + recreate_missing: Some(false), source: HealRequestSource::Scanner, ..Default::default() } @@ -3115,6 +3117,17 @@ mod tests { assert_eq!(request.priority, HealChannelPriority::Low); assert_eq!(request.source, HealRequestSource::Scanner); assert_eq!(request.remove_corrupted, Some(HEAL_DELETE_DANGLING)); + assert_eq!(request.recreate_missing, Some(false)); + } + + #[test] + fn test_build_bucket_heal_request_disables_recreate_for_scanner() { + let request = build_bucket_heal_request("bucket".to_string(), HealChannelPriority::Low); + + assert_eq!(request.bucket, "bucket"); + assert_eq!(request.priority, HealChannelPriority::Low); + assert_eq!(request.source, HealRequestSource::Scanner); + assert_eq!(request.recreate_missing, Some(false)); } fn metadata_for_object(bucket: &str, object: &str) -> Vec {