gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): align scanner repair safety with quorum errors (#3887)

Browse Source 
* fix(heal): avoid task-level delete on repair failure

* fix(scanner): avoid recreate for scanner heal checks

* fix(heal): preserve typed quorum failures

* test(heal): satisfy clippy bool assertion
This commit is contained in:
cxymds
2026-06-26 08:58:27 +08:00
committed by GitHub
parent 97dce107f0
commit 0b9c8a7731
8 changed files with 392 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/erasure_coding/heal.rs
+41 -5
View File
@@ -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<R>(
@@ -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::<Vec<_>>();
let mut writers = (0..erasure.total_shard_count()).map(|_| None).collect::<Vec<_>>();
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));
}
}
crates/ecstore/src/set_disk.rs
+11 -3
View File
@@ -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())));
}
_ => {}
crates/ecstore/src/set_disk/heal.rs
+16 -8
View File
@@ -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),
))
}
};
crates/heal/src/heal/channel.rs
+110 -2
View File
@@ -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();
crates/heal/src/heal/manager.rs
+49 -8
View File
@@ -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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = Arc::new(MockStorage);
crates/heal/src/heal/storage.rs
+5 -5
View File
@@ -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))
}
}
}
crates/heal/src/heal/task.rs
+147 -74
View File
@@ -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<Option<bool>>,
object_exists_by_name: Mutex<HashMap<String, MockObjectExists>>,
heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
deleted_objects: Mutex<Vec<String>>,
format_no_heal_required: Mutex<bool>,
listed_prefixes: Mutex<Vec<String>>,
truncate_without_token: Mutex<bool>,
@@ -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());
crates/scanner/src/scanner_folder.rs
+13
View File
@@ -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<u8> {