diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index 30bd4f5c3..f0ea0530e 100644 --- a/crates/common/src/heal_channel.rs +++ b/crates/common/src/heal_channel.rs @@ -356,6 +356,8 @@ pub struct HealChannelRequest { pub recursive: Option, /// Whether to dry run pub dry_run: Option, + /// Whether to skip namespace locking + pub no_lock: Option, /// Timeout in seconds (optional) pub timeout_seconds: Option, /// Origin of the request for operational status and queue accounting @@ -560,6 +562,7 @@ pub fn create_heal_request( update_parity: None, recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, source: HealRequestSource::Internal, disk: None, @@ -718,6 +721,7 @@ pub async fn send_heal_disk(set_disk_id: String, priority: Option>> = StdMutex::new(None); +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +enum HealBucketOperation { + Make, + Delete, +} + +#[cfg(test)] +struct HealBucketOperationFailure { + bucket: String, + disk_index: usize, + operation: HealBucketOperation, +} + +#[cfg(test)] +type HealBucketOperationFailureKey = (String, usize, HealBucketOperation); + +#[cfg(test)] +fn heal_bucket_operation_failures() -> &'static StdMutex> { + static FAILURES: std::sync::OnceLock>> = std::sync::OnceLock::new(); + FAILURES.get_or_init(|| StdMutex::new(HashMap::new())) +} + +#[cfg(test)] +impl HealBucketOperationFailure { + fn install(bucket: &str, disk_index: usize, operation: HealBucketOperation, error: Error) -> Self { + let key = (bucket.to_string(), disk_index, operation); + let previous = heal_bucket_operation_failures() + .lock() + .expect("heal bucket failure registry should not poison") + .insert(key, error); + assert!(previous.is_none(), "heal bucket operation failure already installed"); + Self { + bucket: bucket.to_string(), + disk_index, + operation, + } + } +} + +#[cfg(test)] +impl Drop for HealBucketOperationFailure { + fn drop(&mut self) { + heal_bucket_operation_failures() + .lock() + .expect("heal bucket failure registry should not poison") + .remove(&(self.bucket.clone(), self.disk_index, self.operation)); + } +} + +#[cfg(test)] +fn injected_heal_bucket_operation_error(bucket: &str, disk_index: usize, operation: HealBucketOperation) -> Option { + heal_bucket_operation_failures() + .lock() + .expect("heal bucket failure registry should not poison") + .get(&(bucket.to_string(), disk_index, operation)) + .cloned() +} + +#[cfg(not(test))] +fn injected_heal_bucket_operation_error(_bucket: &str, _disk_index: usize, _operation: HealBucketOperation) -> Option { + None +} + #[cfg(test)] pub(crate) fn install_delete_bucket_empty_scan_barrier() -> Arc { let barrier = Arc::new(DeleteBucketEmptyScanBarrier::default()); @@ -1207,10 +1270,6 @@ pub(crate) async fn heal_bucket_local_on_disks( ..Default::default() }; - if opts.dry_run { - return Ok(res); - } - for (disk, state) in disks.iter().zip(before_state.read().await.iter()) { res.before.drives.push(HealDriveInfo { uuid: "".to_string(), @@ -1219,35 +1278,68 @@ pub(crate) async fn heal_bucket_local_on_disks( }); } + if opts.dry_run { + for (disk, state) in disks.iter().zip(after_state.read().await.iter()) { + res.after.drives.push(HealDriveInfo { + uuid: "".to_string(), + endpoint: disk.clone().map(|s| s.to_string()).unwrap_or_default(), + state: state.to_string(), + }); + } + return Ok(res); + } + + let mut operation_error = errs + .iter() + .filter_map(|err| match err { + Some(Error::VolumeNotFound) | None => None, + Some(err) => Some(err.clone()), + }) + .next(); + if opts.remove && !bucket.starts_with(disk::RUSTFS_META_BUCKET) && !is_all_buckets_not_found(&errs) { let mut futures = Vec::new(); - for disk in disks.iter() { - let disk = disk.clone(); + for (index, disk) in disks.iter().enumerate() { + if matches!(errs[index].as_ref(), Some(Error::DiskNotFound | Error::VolumeNotFound)) { + continue; + } + let Some(disk) = disk.clone() else { + continue; + }; let bucket = bucket.to_string(); - info!("heal_bucket_local, errs: {:?}, opts: {:?}", errs, opts); futures.push(async move { - match disk { - Some(disk) => { - // Non-force: a bucket that still holds object data refuses - // deletion (VolumeNotEmpty) instead of being recursively - // wiped, so a misclassified "dangling" bucket cannot lose - // data (backlog#799 B1). Surface that refusal instead of - // discarding it — it signals the bucket is not dangling. - match disk.delete_volume(&bucket, false).await { - Ok(()) => None, - Err(Error::VolumeNotEmpty) => { - warn!("heal declined to remove non-empty bucket {bucket} (not dangling)"); - None - } - Err(e) => Some(e), - } - } - None => Some(Error::DiskNotFound), + if let Some(err) = injected_heal_bucket_operation_error(&bucket, index, HealBucketOperation::Delete) { + return (index, Err(err)); } + (index, disk.delete_volume(&bucket, false).await) }); } - let _ = join_all(futures).await; + for (index, result) in join_all(futures).await { + match result { + Ok(()) | Err(Error::VolumeNotFound) => { + after_state.write().await[index] = DriveState::Missing.to_string(); + } + Err(Error::VolumeNotEmpty) => { + warn!( + bucket, + operation = "heal_bucket_delete_volume", + result = "preserved_non_empty_bucket", + "heal declined to remove non-empty bucket" + ); + after_state.write().await[index] = DriveState::Ok.to_string(); + } + Err(err) => { + after_state.write().await[index] = match &err { + Error::DiskNotFound => DriveState::Offline.to_string(), + _ => DriveState::Corrupt.to_string(), + }; + if operation_error.is_none() { + operation_error = Some(err); + } + } + } + } } if !opts.remove { @@ -1256,41 +1348,56 @@ pub(crate) async fn heal_bucket_local_on_disks( let disk = disk.clone(); let bucket = bucket.to_string(); let bs_clone = before_state.clone(); - let as_clone = after_state.clone(); - let errs_clone = errs.to_vec(); futures.push(async move { if bs_clone.read().await[idx] == DriveState::Missing.to_string() { let Some(disk) = disk.as_ref() else { - return Some(Error::DiskNotFound); + return (idx, Some(Error::DiskNotFound)); }; - info!("bucket not find, will recreate"); + if let Some(err) = injected_heal_bucket_operation_error(&bucket, idx, HealBucketOperation::Make) { + return (idx, Some(err)); + } match disk.make_volume(&bucket).await { - Ok(_) => { - as_clone.write().await[idx] = DriveState::Ok.to_string(); - return None; - } - Err(err) => { - return Some(err); - } + Ok(()) | Err(Error::VolumeExists) => return (idx, None), + Err(err) => return (idx, Some(err)), } } - errs_clone[idx].clone() + (idx, None) }); } - let _ = join_all(futures).await; + for (index, result) in join_all(futures).await { + match result { + None => { + if before_state.read().await[index] == DriveState::Missing.to_string() { + after_state.write().await[index] = DriveState::Ok.to_string(); + } + } + Some(err) => { + after_state.write().await[index] = match &err { + Error::DiskNotFound => DriveState::Offline.to_string(), + _ => DriveState::Corrupt.to_string(), + }; + if operation_error.is_none() { + operation_error = Some(err); + } + } + } + } } for (disk, state) in disks.iter().zip(after_state.read().await.iter()) { - res.before.drives.push(HealDriveInfo { + res.after.drives.push(HealDriveInfo { uuid: "".to_string(), endpoint: disk.clone().map(|s| s.to_string()).unwrap_or_default(), state: state.to_string(), }); } - Ok(res) + match operation_error { + Some(err) => Err(err), + None => Ok(res), + } } async fn clone_drives() -> Vec> { @@ -1756,7 +1863,7 @@ mod tests { .await .expect_err("second disk should start missing the bucket"); - heal_bucket_local( + let result = heal_bucket_local( bucket, &HealOpts { recreate: true, @@ -1766,6 +1873,25 @@ mod tests { .await .expect("bucket heal should recreate missing volumes"); + assert_eq!(result.before.drives.len(), 2); + assert_eq!(result.after.drives.len(), 2); + assert!( + result + .before + .drives + .iter() + .any(|drive| drive.state == DriveState::Missing.to_string()), + "one bucket volume must be reported missing before heal" + ); + assert!( + result + .after + .drives + .iter() + .all(|drive| drive.state == DriveState::Ok.to_string()), + "all bucket volumes must be reported healthy after heal" + ); + for disk in disks { disk.stat_volume(bucket).await.expect("bucket should exist after heal"); } @@ -1773,6 +1899,166 @@ mod tests { reset_local_disk_test_state().await; } + #[tokio::test] + #[serial] + async fn heal_bucket_local_dry_run_reports_discovered_drive_states() { + reset_local_disk_test_state().await; + + let temp_dir = TempDir::new().expect("create temp dir for bucket heal dry-run regression"); + let disks = init_test_local_disks(&temp_dir, 2, "heal-bucket-local-dry-run-reports-state").await; + let bucket = "dry-run-healed-bucket"; + disks[0] + .make_volume(bucket) + .await + .expect("bucket should exist on the first disk"); + + let result = heal_bucket_local_on_disks( + bucket, + &HealOpts { + dry_run: true, + ..Default::default() + }, + vec![Some(disks[0].clone()), Some(disks[1].clone()), None], + ) + .await + .expect("dry-run bucket heal should inspect disks"); + + assert_eq!(result.before.drives.len(), 3); + assert_eq!(result.after.drives.len(), 3); + assert_eq!(result.before.drives[0].state, DriveState::Ok.to_string()); + assert_eq!(result.before.drives[1].state, DriveState::Missing.to_string()); + assert_eq!(result.before.drives[2].state, DriveState::Offline.to_string()); + for (before, after) in result.before.drives.iter().zip(&result.after.drives) { + assert_eq!(after.endpoint, before.endpoint); + assert_eq!(after.state, before.state); + } + assert!(matches!(disks[1].stat_volume(bucket).await, Err(Error::VolumeNotFound))); + + reset_local_disk_test_state().await; + } + + #[tokio::test] + #[serial] + async fn heal_bucket_local_propagates_recreate_failure() { + reset_local_disk_test_state().await; + + let temp_dir = TempDir::new().expect("create temp dir for bucket recreate failure regression"); + let disks = init_test_local_disks(&temp_dir, 2, "heal-bucket-local-propagates-recreate-failure").await; + let bucket = "recreate-failure-bucket"; + disks[0] + .make_volume(bucket) + .await + .expect("bucket should exist on the first disk"); + let _failure = HealBucketOperationFailure::install(bucket, 1, HealBucketOperation::Make, Error::DiskAccessDenied); + + let error = heal_bucket_local_on_disks( + bucket, + &HealOpts { + recreate: true, + ..Default::default() + }, + disks.iter().cloned().map(Some).collect(), + ) + .await + .expect_err("failed volume recreation must fail bucket heal"); + + assert_eq!(error, Error::DiskAccessDenied); + assert!(matches!(disks[1].stat_volume(bucket).await, Err(Error::VolumeNotFound))); + + reset_local_disk_test_state().await; + } + + #[tokio::test] + #[serial] + async fn heal_bucket_local_propagates_delete_failure() { + reset_local_disk_test_state().await; + + let temp_dir = TempDir::new().expect("create temp dir for bucket delete failure regression"); + let disks = init_test_local_disks(&temp_dir, 2, "heal-bucket-local-propagates-delete-failure").await; + let bucket = "delete-failure-bucket"; + disks[0] + .make_volume(bucket) + .await + .expect("bucket should exist on the first disk"); + let _failure = HealBucketOperationFailure::install(bucket, 0, HealBucketOperation::Delete, Error::DiskAccessDenied); + + let error = heal_bucket_local_on_disks( + bucket, + &HealOpts { + remove: true, + ..Default::default() + }, + disks.iter().cloned().map(Some).collect(), + ) + .await + .expect_err("failed volume deletion must fail bucket heal"); + + assert_eq!(error, Error::DiskAccessDenied); + disks[0] + .stat_volume(bucket) + .await + .expect("failed deletion must leave the bucket volume present"); + + reset_local_disk_test_state().await; + } + + #[tokio::test] + #[serial] + async fn heal_bucket_local_preserves_non_empty_bucket() { + reset_local_disk_test_state().await; + + let temp_dir = TempDir::new().expect("create temp dir for non-empty bucket heal regression"); + let disks = init_test_local_disks(&temp_dir, 1, "heal-bucket-local-preserves-non-empty").await; + let bucket = "non-empty-bucket"; + disks[0] + .make_volume(bucket) + .await + .expect("bucket should exist on the first disk"); + let _failure = HealBucketOperationFailure::install(bucket, 0, HealBucketOperation::Delete, Error::VolumeNotEmpty); + + let result = heal_bucket_local_on_disks( + bucket, + &HealOpts { + remove: true, + ..Default::default() + }, + disks.iter().cloned().map(Some).collect(), + ) + .await + .expect("a non-empty bucket refusal is an expected safety result"); + + assert_eq!(result.after.drives.len(), 1); + assert_eq!(result.after.drives[0].state, DriveState::Ok.to_string()); + disks[0] + .stat_volume(bucket) + .await + .expect("the non-empty bucket must remain present"); + + reset_local_disk_test_state().await; + } + + #[tokio::test] + #[serial] + async fn heal_bucket_local_propagates_preexisting_offline_disk() { + reset_local_disk_test_state().await; + + let temp_dir = TempDir::new().expect("create temp dir for offline bucket heal regression"); + let disks = init_test_local_disks(&temp_dir, 1, "heal-bucket-local-preexisting-offline").await; + let bucket = "offline-disk-bucket"; + disks[0] + .make_volume(bucket) + .await + .expect("bucket should exist on the online disk"); + + let error = heal_bucket_local_on_disks(bucket, &HealOpts::default(), vec![Some(disks[0].clone()), None]) + .await + .expect_err("a prepass offline disk must keep the bucket heal incomplete"); + + assert_eq!(error, Error::DiskNotFound); + + reset_local_disk_test_state().await; + } + #[test] fn test_reduce_pool_write_quorum_uses_only_pool_participants() { let clients = vec![ diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 91e8570d0..ab70adc18 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -155,6 +155,60 @@ fn injected_dangling_check_parts_error(bucket: &str, object: &str, disk_index: u .cloned() } +#[cfg(test)] +struct DanglingDeleteFailure { + key: DanglingDeleteFailureKey, +} + +#[cfg(test)] +type DanglingDeleteFailureKey = (String, String, usize); + +#[cfg(test)] +type DanglingDeleteFailures = HashMap; + +#[cfg(test)] +fn dangling_delete_failures() -> &'static std::sync::Mutex { + static FAILURES: std::sync::OnceLock> = std::sync::OnceLock::new(); + FAILURES.get_or_init(|| std::sync::Mutex::new(HashMap::new())) +} + +#[cfg(test)] +impl DanglingDeleteFailure { + fn install(bucket: &str, object: &str, disk_index: usize, error: DiskError) -> Self { + let key = (bucket.to_string(), object.to_string(), disk_index); + let previous = dangling_delete_failures() + .lock() + .expect("dangling delete failure registry should not poison") + .insert(key.clone(), error); + assert!(previous.is_none(), "dangling delete failure already installed"); + Self { key } + } +} + +#[cfg(test)] +impl Drop for DanglingDeleteFailure { + fn drop(&mut self) { + dangling_delete_failures() + .lock() + .expect("dangling delete failure registry should not poison") + .remove(&self.key); + } +} + +#[cfg(test)] +fn injected_dangling_delete_error(bucket: &str, object: &str, disk_index: usize) -> Option { + dangling_delete_failures() + .lock() + .expect("dangling delete failure registry should not poison") + .get(&(bucket.to_string(), object.to_string(), disk_index)) + .cloned() +} + +#[cfg(not(test))] +fn injected_dangling_delete_error(_bucket: &str, _object: &str, _disk_index: usize) -> Option { + None +} + fn first_unhealthy_part_summary( data_errs_by_part: &HashMap>, parts: &[ObjectPartInfo], @@ -1181,30 +1235,37 @@ impl SetDisks { let errs = stat_all_dirs(&disks, bucket, object).await; let dangling_object = is_object_dir_dangling(&errs); - if dangling_object && !dry_run && remove { + let delete_errs = if dangling_object && !dry_run && remove { let mut futures = Vec::with_capacity(disks.len()); - for disk in disks.iter().flatten() { + for (disk_index, disk) in disks.iter().enumerate() { let disk = disk.clone(); - let bucket = bucket.to_string(); - let object = object.to_string(); - futures.push(tokio::spawn(async move { - let _ = disk - .delete( - &bucket, - &object, + futures.push(async move { + let Some(disk) = disk else { + return (disk_index, Some(DiskError::DiskNotFound)); + }; + if let Some(error) = injected_dangling_delete_error(bucket, object, disk_index) { + return (disk_index, Some(error)); + } + ( + disk_index, + disk.delete( + bucket, + object, DeleteOptions { recursive: false, immediate: false, ..Default::default() }, ) - .await; - })); + .await + .err(), + ) + }); } - - // ignore errors - let _ = join_all(futures).await; - } + Some(join_all(futures).await) + } else { + None + }; for (err, drive) in errs.iter().zip(self.set_endpoints.iter()) { let endpoint = drive.to_string(); @@ -1229,6 +1290,28 @@ impl SetDisks { }); } + if let Some(delete_errs) = delete_errs { + let mut delete_failure = None; + for (index, err) in delete_errs { + match err { + None | Some(DiskError::FileNotFound) => { + result.after.drives[index].state = DriveState::Missing.to_string(); + } + Some(err) => { + result.after.drives[index].state = if matches!(&err, DiskError::DiskNotFound) { + DriveState::Offline.to_string() + } else { + DriveState::Corrupt.to_string() + }; + if delete_failure.is_none() { + delete_failure = Some(err); + } + } + } + } + return Ok((result, Some(delete_failure.unwrap_or(DiskError::FileNotFound)))); + } + if dangling_object || DiskError::is_all_not_found(&errs) { return Ok((result, Some(DiskError::FileNotFound))); } @@ -1553,7 +1636,7 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks { #[cfg(test)] mod heal_result_report_tests { - use super::{DanglingCheckPartsFailure, DanglingDeleteSafety, SetDisks}; + use super::{DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, SetDisks}; use super::{HEAL_RENAME_INCOMPLETE, HealRenameFailureScope}; use crate::disk::endpoint::Endpoint; use crate::disk::error::DiskError; @@ -2020,6 +2103,67 @@ mod heal_result_report_tests { assert_eq!(result.before.drives[3].state, DriveState::Ok.to_string()); } + #[tokio::test] + async fn dangling_object_dir_delete_preserves_results_and_propagates_failure() { + let bucket = "bucket-dangling-dir-delete"; + let object = "dangling__XLDIR__"; + let mut temp_dirs = Vec::new(); + let mut endpoints = Vec::new(); + let mut disks = Vec::new(); + for _ in 0..8 { + let (temp_dir, endpoint, disk) = real_disk().await; + disk.make_volume(bucket).await.expect("test bucket should be created"); + temp_dirs.push(temp_dir); + endpoints.push(endpoint); + disks.push(Some(disk)); + } + disks[0] = None; + let set = set_disks_with(disks, endpoints, 4).await; + for disk_index in [1, 2] { + tokio::fs::create_dir_all(temp_dirs[disk_index].path().join(bucket).join(object)) + .await + .expect("dangling object directory should be created"); + } + + let _delete_failure = DanglingDeleteFailure::install(bucket, object, 2, DiskError::DiskAccessDenied); + let _file_missing = DanglingDeleteFailure::install(bucket, object, 3, DiskError::FileNotFound); + let _version_missing = DanglingDeleteFailure::install(bucket, object, 4, DiskError::FileVersionNotFound); + let _path_missing = DanglingDeleteFailure::install(bucket, object, 5, DiskError::PathNotFound); + let _volume_missing = DanglingDeleteFailure::install(bucket, object, 6, DiskError::VolumeNotFound); + let _disk_missing = DanglingDeleteFailure::install(bucket, object, 7, DiskError::DiskNotFound); + + let (result, err) = set + .heal_object_dir_locked(bucket, object, false, true) + .await + .expect("dangling directory heal should report its per-disk delete results"); + + assert_eq!(err, Some(DiskError::DiskNotFound)); + assert_eq!(result.before.drives.len(), 8); + assert_eq!(result.after.drives.len(), 8); + assert_eq!(result.before.drives[0].state, DriveState::Offline.to_string()); + assert_eq!(result.after.drives[0].state, DriveState::Offline.to_string()); + assert_eq!(result.before.drives[1].state, DriveState::Ok.to_string()); + assert_eq!(result.after.drives[1].state, DriveState::Missing.to_string()); + assert_eq!(result.before.drives[2].state, DriveState::Ok.to_string()); + assert_eq!(result.after.drives[2].state, DriveState::Corrupt.to_string()); + assert_eq!(result.before.drives[3].state, DriveState::Missing.to_string()); + assert_eq!(result.after.drives[3].state, DriveState::Missing.to_string()); + for disk_index in [4, 5, 6] { + assert_eq!(result.before.drives[disk_index].state, DriveState::Missing.to_string()); + assert_eq!(result.after.drives[disk_index].state, DriveState::Corrupt.to_string()); + } + assert_eq!(result.before.drives[7].state, DriveState::Missing.to_string()); + assert_eq!(result.after.drives[7].state, DriveState::Offline.to_string()); + assert!( + !temp_dirs[1].path().join(bucket).join(object).exists(), + "successful delete must remove the dangling directory" + ); + assert!( + temp_dirs[2].path().join(bucket).join(object).is_dir(), + "failed delete must leave the dangling directory for retry" + ); + } + #[tokio::test] async fn dangling_delete_guard_preserves_conflicting_identities_without_writing_metadata() { let bucket = "bucket-delete-guard-conflict"; diff --git a/crates/ecstore/src/set_disk/ops/heal_walk.rs b/crates/ecstore/src/set_disk/ops/heal_walk.rs index 625e01c14..9bcc0a8b1 100644 --- a/crates/ecstore/src/set_disk/ops/heal_walk.rs +++ b/crates/ecstore/src/set_disk/ops/heal_walk.rs @@ -64,6 +64,7 @@ struct HealWalkCollector { batch_objects: usize, version_budget: usize, objects: Mutex>, + decode_error: Mutex>, version_total: AtomicUsize, truncated: AtomicBool, cancel: CancellationToken, @@ -77,6 +78,22 @@ impl HealWalkCollector { }) } + fn record_decode_error(&self, error: rustfs_filemeta::Error) { + if let Ok(mut first_error) = self.decode_error.lock() + && first_error.is_none() + { + *first_error = Some(error.into()); + } + self.cancel.cancel(); + } + + fn take_decode_error(&self) -> disk::error::Result> { + self.decode_error.lock().map(|mut error| error.take()).map_err(|_| { + self.cancel.cancel(); + DiskError::FileCorrupt + }) + } + /// Expand one resolved entry into its versions and record it. Cancels the /// walk once EITHER page bound (distinct object names OR expanded versions) /// is met — always at a sorted object-key boundary so a heavily-versioned @@ -92,7 +109,7 @@ impl HealWalkCollector { let fiv = match entry.file_info_versions_with_free_versions(&self.bucket) { Ok(fiv) => fiv, Err(err) => { - debug!(entry = %entry.name, error = ?err, "heal disk-walk skipped entry with unreadable versions"); + self.record_decode_error(err); return; } }; @@ -248,6 +265,7 @@ impl SetDisks { batch_objects, version_budget: version_budget.max(1), objects: Mutex::new(Vec::new()), + decode_error: Mutex::new(None), version_total: AtomicUsize::new(0), truncated: AtomicBool::new(false), cancel: CancellationToken::new(), @@ -294,7 +312,12 @@ impl SetDisks { // Drive the walk. A tolerated missing-path / not-found is treated as an // empty page rather than an error (nothing to heal on this prefix). - match list_path_raw(collector.cancel.clone(), opts).await { + let walk_result = list_path_raw(collector.cancel.clone(), opts).await; + if let Some(err) = collector.take_decode_error()? { + return Err(err); + } + + match walk_result { Ok(()) => {} Err(DiskError::FileNotFound) | Err(DiskError::VolumeNotFound) => { debug!(bucket, prefix, "heal disk-walk treated missing path as empty page"); @@ -314,6 +337,52 @@ impl SetDisks { #[cfg(test)] mod tests { use super::*; + use crate::set_disk::ops::object::hermetic_set_disks_support::hermetic_set_disks_isolated; + use rustfs_filemeta::{ChecksumAlgo, ErasureAlgo, FileMetaVersion, MetaObject, VersionType}; + use time::OffsetDateTime; + use uuid::Uuid; + + fn test_collector() -> Arc { + Arc::new(HealWalkCollector { + bucket: "bucket".to_string(), + batch_objects: 2, + version_budget: 2, + objects: Mutex::new(Vec::new()), + decode_error: Mutex::new(None), + version_total: AtomicUsize::new(0), + truncated: AtomicBool::new(false), + cancel: CancellationToken::new(), + }) + } + + fn crc_valid_semantically_corrupt_entry(name: &str) -> MetaCacheEntry { + let mut metadata = FileMeta::new(); + metadata + .add_version_filemata(FileMetaVersion { + version_type: VersionType::Object, + object: Some(MetaObject { + version_id: Some(Uuid::new_v4()), + erasure_algorithm: ErasureAlgo::ReedSolomon, + erasure_m: 2, + erasure_n: 2, + erasure_block_size: 1 << 20, + bitrot_checksum_algo: ChecksumAlgo::HighwayHash, + part_numbers: vec![1, 2], + part_sizes: vec![10], + part_actual_sizes: vec![10, 20], + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }), + ..Default::default() + }) + .expect("corrupt test version should be accepted before semantic decoding"); + + MetaCacheEntry { + name: name.to_string(), + metadata: metadata.marshal_msg().expect("test metadata should encode with a valid CRC"), + ..Default::default() + } + } fn version(name: &str, id: &str, dm: bool) -> HealWalkVersion { HealWalkVersion { @@ -449,15 +518,7 @@ mod tests { #[test] fn poisoned_collector_state_cancels_the_walk() { - let collector = Arc::new(HealWalkCollector { - bucket: "bucket".to_string(), - batch_objects: 2, - version_budget: 2, - objects: Mutex::new(Vec::new()), - version_total: AtomicUsize::new(0), - truncated: AtomicBool::new(false), - cancel: CancellationToken::new(), - }); + let collector = test_collector(); let poison_target = Arc::clone(&collector); let _ = std::thread::spawn(move || { let _guard = poison_target.objects.lock().expect("fresh mutex should lock"); @@ -470,4 +531,45 @@ mod tests { assert!(collector.cancel.is_cancelled(), "a poisoned page collector must cancel its walk"); assert!(collector.objects.lock().is_err(), "poisoned state must remain fail-closed"); } + + #[test] + fn semantic_decode_failure_records_error_and_cancels_walk() { + let collector = test_collector(); + let entry = crc_valid_semantically_corrupt_entry("corrupt-object"); + + collector.ingest(entry); + + let error = collector + .take_decode_error() + .expect("decode error state should remain readable") + .expect("semantic metadata corruption must be recorded"); + assert_eq!(error, DiskError::FileCorrupt); + assert!(collector.cancel.is_cancelled(), "semantic metadata corruption must cancel the disk walk"); + } + + #[tokio::test] + async fn heal_walk_returns_crc_valid_semantic_decode_failure() { + let bucket = "bucket"; + let object = "corrupt-object"; + let (temp_dirs, disks, set_disks) = hermetic_set_disks_isolated(1).await; + disks[0].make_volume(bucket).await.expect("test bucket should be created"); + + let object_dir = temp_dirs[0].path().join(bucket).join(object); + tokio::fs::create_dir_all(&object_dir) + .await + .expect("test object directory should be created"); + tokio::fs::write( + object_dir.join(crate::disk::STORAGE_FORMAT_FILE), + crc_valid_semantically_corrupt_entry(object).metadata, + ) + .await + .expect("corrupt test metadata should be written"); + + let error = set_disks + .heal_walk_versions_page(bucket, "", None, 2, 2) + .await + .expect_err("semantic metadata corruption must fail the heal disk walk"); + + assert_eq!(error, DiskError::FileCorrupt); + } } diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index 8ab4ff9e5..10da452ae 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -107,13 +107,21 @@ impl Error { Error::TaskTimeout | Error::TransientSkip { .. } => true, Error::Storage(err) => { err.is_quorum_error() - || matches!(err, EcstoreError::SlowDown | EcstoreError::OperationCanceled | EcstoreError::Lock(_)) + || matches!( + err, + EcstoreError::DiskNotFound + | EcstoreError::VolumeNotFound + | EcstoreError::SlowDown + | EcstoreError::OperationCanceled + | EcstoreError::Lock(_) + ) || is_recoverable_heal_error_message(&err.to_string()) } Error::Disk(err) => { matches!( err, - DiskError::ErasureReadQuorum + DiskError::DiskNotFound + | DiskError::ErasureReadQuorum | DiskError::ErasureWriteQuorum | DiskError::Timeout | DiskError::SourceStalled @@ -159,7 +167,7 @@ impl From for std::io::Error { #[cfg(test)] mod tests { use super::Error; - use crate::heal::EcstoreError; + use crate::heal::{DiskError, EcstoreError}; #[test] fn incomplete_target_rename_is_recoverable() { @@ -173,4 +181,11 @@ mod tests { assert!(task_error.is_recoverable_heal()); assert!(storage_error.is_recoverable_heal()); } + + #[test] + fn offline_disk_errors_are_recoverable() { + assert!(Error::Disk(DiskError::DiskNotFound).is_recoverable_heal()); + assert!(Error::Storage(EcstoreError::DiskNotFound).is_recoverable_heal()); + assert!(Error::Storage(EcstoreError::VolumeNotFound).is_recoverable_heal()); + } } diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 5656db38c..a11b1b425 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -623,6 +623,7 @@ impl HealChannelProcessor { update_parity: request.update_parity.unwrap_or(true), recursive, dry_run: request.dry_run.unwrap_or(false), + no_lock: request.no_lock.unwrap_or(false), timeout: request.timeout_seconds.map(std::time::Duration::from_secs), pool_index: request.pool_index, set_index: request.set_index, @@ -852,6 +853,7 @@ mod tests { update_parity: None, recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -883,6 +885,7 @@ mod tests { update_parity: Some(true), recursive: Some(true), dry_run: Some(false), + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -914,6 +917,7 @@ mod tests { update_parity: Some(true), recursive: Some(false), dry_run: Some(false), + no_lock: Some(true), timeout_seconds: Some(300), pool_index: Some(0), set_index: Some(1), @@ -928,6 +932,7 @@ mod tests { assert_eq!(heal_request.options.scan_mode, HealScanMode::Deep); assert!(heal_request.options.remove_corrupted); assert!(heal_request.options.recreate_missing); + assert!(heal_request.options.no_lock); } #[tokio::test] @@ -948,6 +953,7 @@ mod tests { update_parity: None, recursive: Some(false), dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -984,6 +990,7 @@ mod tests { update_parity: None, recursive: Some(false), dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1022,6 +1029,7 @@ mod tests { update_parity: None, recursive: Some(false), dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1053,6 +1061,7 @@ mod tests { update_parity: Some(true), recursive: Some(true), dry_run: Some(false), + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1087,6 +1096,7 @@ mod tests { update_parity: None, recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1117,6 +1127,7 @@ mod tests { update_parity: None, recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1154,6 +1165,7 @@ mod tests { update_parity: None, recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1184,6 +1196,7 @@ mod tests { update_parity: Some(false), recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1216,6 +1229,7 @@ mod tests { update_parity: None, recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1252,6 +1266,7 @@ mod tests { update_parity: None, recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, @@ -1529,6 +1544,7 @@ mod tests { update_parity: None, recursive: None, dry_run: None, + no_lock: None, timeout_seconds: None, pool_index: None, set_index: None, diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index d08367f3f..d4b77c045 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -43,6 +43,34 @@ enum HealObjectOutcome { Failed, } +struct PageConcurrencyGuard { + in_flight: Arc, + set_label: String, +} + +impl PageConcurrencyGuard { + fn new(in_flight: Arc, set_label: String) -> Self { + let current = in_flight.fetch_add(1, Ordering::SeqCst) + 1; + gauge!( + "rustfs_heal_page_concurrency_current", + "set" => set_label.clone() + ) + .set(current as f64); + Self { in_flight, set_label } + } +} + +impl Drop for PageConcurrencyGuard { + fn drop(&mut self) { + let current = self.in_flight.fetch_sub(1, Ordering::SeqCst) - 1; + gauge!( + "rustfs_heal_page_concurrency_current", + "set" => self.set_label.clone() + ) + .set(current as f64); + } +} + const LOG_COMPONENT_HEAL: &str = "heal"; const LOG_SUBSYSTEM_ERASURE_HEALER: &str = "erasure_healer"; const EVENT_HEAL_ERASURE_RESUME_STATE: &str = "heal_erasure_resume_state"; @@ -183,35 +211,13 @@ impl ErasureSetHealer { .execute_heal_with_resume(buckets, set_disk_id, &resume_manager, &checkpoint_manager) .await; - // 4. cleanup resume state - if result.is_ok() { - if let Err(e) = resume_manager.cleanup().await { - warn!( - target: "rustfs::heal::erasure_healer", - event = EVENT_HEAL_ERASURE_RESUME_STATE, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_ERASURE_HEALER, - set_disk_id, - state = "resume_cleanup_failed", - error = %e, - "Erasure set resume cleanup failed" - ); - } - if let Err(e) = checkpoint_manager.cleanup().await { - warn!( - target: "rustfs::heal::erasure_healer", - event = EVENT_HEAL_ERASURE_RESUME_STATE, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_ERASURE_HEALER, - set_disk_id, - state = "checkpoint_cleanup_failed", - error = %e, - "Erasure set checkpoint cleanup failed" - ); - } - } + result?; - result + // The healing marker is cleared by the caller only after both cleanup + // operations succeed. Cleanup is idempotent, so a retry is safe. + checkpoint_manager.cleanup().await?; + resume_manager.cleanup().await?; + Ok(()) } /// get or create task id @@ -223,7 +229,10 @@ impl ErasureSetHealer { match ResumeManager::load_from_disk(self.disk.clone(), &task_id).await { Ok(manager) => { let state = manager.get_state().await; - if state.set_disk_id == set_disk_id && ResumeUtils::can_resume_task(&self.disk, &task_id).await { + if !state.completed + && state.set_disk_id == set_disk_id + && ResumeUtils::can_resume_task(&self.disk, &task_id).await + { debug!( target: "rustfs::heal::erasure_healer", event = EVENT_HEAL_ERASURE_RESUME_STATE, @@ -295,6 +304,21 @@ impl ErasureSetHealer { CheckpointManager::new(self.disk.clone(), task_id.to_string()).await? }; + let state = resume_manager.get_state().await; + if state.retry_count > 0 + && state.completed_buckets.is_empty() + && state.resume_cursor.is_none() + && state.processed_objects == 0 + && state.successful_objects == 0 + && state.failed_objects == 0 + && state.skipped_objects == 0 + { + // schedule_retry persists the authoritative resume reset before + // resetting the checkpoint. Reapply the checkpoint reset after + // a crash in that window so stale positions cannot skip work. + checkpoint_manager.reset_for_retry().await?; + } + Ok((resume_manager, checkpoint_manager)) } else { debug!( @@ -358,6 +382,7 @@ impl ErasureSetHealer { let mut successful_objects = state.successful_objects; let mut failed_objects = state.failed_objects; let mut skipped_objects = state.skipped_objects; + let mut failed_buckets = 0u64; // 4. process remaining buckets for (bucket_idx, bucket) in buckets.iter().enumerate().skip(current_bucket_index) { @@ -385,6 +410,10 @@ impl ErasureSetHealer { ) .await; + if matches!(bucket_result, Err(Error::TaskCancelled | Error::TaskTimeout)) { + return bucket_result; + } + // update checkpoint position checkpoint_manager.update_position(bucket_idx, current_object_index).await?; @@ -422,7 +451,9 @@ impl ErasureSetHealer { "Erasure set bucket completed" ); } + Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err), Err(e) => { + failed_buckets = failed_buckets.saturating_add(1); error!( target: "rustfs::heal::erasure_healer", event = EVENT_HEAL_ERASURE_BUCKET_STATE, @@ -453,7 +484,7 @@ impl ErasureSetHealer { // skip may be because the disk is still down, so these are deferred to a // later heal cycle via the same bounded-retry mechanism as failures — // never hot-retried in place here. - if failed_objects > 0 || skipped_objects > 0 { + if failed_objects > 0 || skipped_objects > 0 || failed_buckets > 0 { if resume_manager.schedule_retry().await? { // Both persistence layers must be reset together: schedule_retry // rewinds the resume state (cursor + counters), and the @@ -471,13 +502,14 @@ impl ErasureSetHealer { component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_ERASURE_HEALER, set_disk_id, + failed_buckets, failed_objects, skipped_objects, state = "retry_scheduled", "Erasure set heal pass finished with unhealed versions; scheduled full re-heal retry" ); return Err(Error::other(format!( - "Erasure set heal incomplete: {failed_objects} failed, {skipped_objects} skipped object(s); retry scheduled" + "Erasure set heal incomplete: {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped; retry scheduled" ))); } @@ -491,15 +523,16 @@ impl ErasureSetHealer { component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_ERASURE_HEALER, set_disk_id, + failed_buckets, failed_objects, skipped_objects, state = "failed_after_retries", "Erasure set heal exhausted retries with unrecovered versions" ); - let _ = resume_manager.cleanup().await; - let _ = checkpoint_manager.cleanup().await; + checkpoint_manager.cleanup().await?; + resume_manager.cleanup().await?; return Err(Error::other(format!( - "Erasure set heal exhausted retries with {failed_objects} failed, {skipped_objects} skipped object(s)" + "Erasure set heal exhausted retries with {failed_buckets} bucket(s) failed, {failed_objects} object(s) failed, {skipped_objects} object(s) skipped" ))); } @@ -646,12 +679,7 @@ impl ErasureSetHealer { Err(err) => return (dedup_key, object_name, version_id, Err(err)), }; - let current_in_flight = in_flight.fetch_add(1, Ordering::SeqCst) + 1; - gauge!( - "rustfs_heal_page_concurrency_current", - "set" => set_label.clone() - ) - .set(current_in_flight as f64); + let _in_flight_guard = PageConcurrencyGuard::new(in_flight, set_label); // Always go through heal_object. Genuine absence flows through // heal_object -> FileVersionNotFound/FileNotFound -> @@ -677,13 +705,6 @@ impl ErasureSetHealer { } }; - let current = in_flight.fetch_sub(1, Ordering::SeqCst) - 1; - gauge!( - "rustfs_heal_page_concurrency_current", - "set" => set_label.clone() - ) - .set(current as f64); - (dedup_key, object_name, version_id, result) }); } @@ -723,14 +744,7 @@ impl ErasureSetHealer { "Erasure set missing object treated as ok" ); } - Err(Error::TaskCancelled) => { - gauge!( - "rustfs_heal_page_concurrency_current", - "set" => set_disk_id.to_string() - ) - .set(0.0); - return Err(Error::TaskCancelled); - } + Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err), Err(Error::TransientSkip { message }) => { *skipped_objects += 1; checkpoint_manager.add_skipped_object(key).await?; @@ -783,12 +797,6 @@ impl ErasureSetHealer { let next_cursor = if is_truncated { next_token.clone() } else { None }; resume_manager.set_resume_cursor(next_cursor.clone()).await?; checkpoint_manager.complete_page(bucket_index, *current_object_index).await?; - gauge!( - "rustfs_heal_page_concurrency_current", - "set" => set_disk_id.to_string() - ) - .set(0.0); - // Check if there are more pages if !is_truncated { break; @@ -835,8 +843,30 @@ impl ErasureSetHealer { #[cfg(test)] mod tests { - use super::ErasureSetHealer; + use super::{ErasureSetHealer, PageConcurrencyGuard}; use rustfs_common::heal_channel::{HealRequestSource, HealScanMode}; + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + + #[tokio::test] + async fn dropping_pending_page_heal_releases_concurrency_slot() { + let in_flight = Arc::new(AtomicUsize::new(0)); + let mut pending_heal = Box::pin({ + let in_flight = in_flight.clone(); + async move { + let _guard = PageConcurrencyGuard::new(in_flight, "pool_0_set_0".to_string()); + std::future::pending::<()>().await; + } + }); + + assert!(futures::poll!(pending_heal.as_mut()).is_pending()); + assert_eq!(in_flight.load(Ordering::SeqCst), 1); + + drop(pending_heal); + assert_eq!(in_flight.load(Ordering::SeqCst), 0); + } #[test] fn heal_page_object_concurrency_uses_default_when_env_is_unset() { @@ -979,7 +1009,7 @@ mod resume_loop_tests { //! handling) — not merely a mock's own output. use super::ErasureSetHealer; use crate::heal::progress::HealProgress; - use crate::heal::resume::{CheckpointManager, ResumeManager, compose_key}; + use crate::heal::resume::{CheckpointManager, RESUME_CHECKPOINT_FILE, ResumeDeleteFailure, ResumeManager, compose_key}; use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI}; use crate::heal::storage_api::status::BucketInfo; use crate::heal::{ @@ -989,6 +1019,7 @@ mod resume_loop_tests { use rustfs_common::heal_channel::{HealOpts, HealRequestSource}; use rustfs_madmin::heal_commands::HealResultItem; use std::collections::HashMap; + use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use tempfile::TempDir; use tokio::sync::RwLock; @@ -1017,6 +1048,7 @@ mod resume_loop_tests { /// A transient infrastructure condition (offline disk / unmet quorum): /// the version must be recorded as skipped and retried on a later pass. Transient, + Timeout, } #[derive(Default)] @@ -1027,6 +1059,7 @@ mod resume_loop_tests { outcomes: Mutex>, /// every heal_object call recorded as (name, version_id) heal_calls: Mutex)>>, + fail_listing: AtomicBool, } impl FakeStorage { @@ -1039,6 +1072,9 @@ mod resume_loop_tests { fn calls(&self) -> Vec<(String, Option)> { self.heal_calls.lock().unwrap().clone() } + fn fail_listing(&self) { + self.fail_listing.store(true, Ordering::SeqCst); + } } #[async_trait::async_trait] @@ -1108,6 +1144,7 @@ mod resume_loop_tests { Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound)))) } HealOutcome::Transient => Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::DiskNotFound)))), + HealOutcome::Timeout => Err(Error::TaskTimeout), } } async fn heal_bucket(&self, _b: &str, _o: &HealOpts) -> Result { @@ -1125,6 +1162,9 @@ mod resume_loop_tests { _prefix: &str, continuation_token: Option<&str>, ) -> Result<(Vec, Option, bool)> { + if self.fail_listing.load(Ordering::SeqCst) { + return Err(Error::other("injected listing failure")); + } let key = continuation_token.map(str::to_string); let page = self.pages.lock().unwrap().get(&key).cloned(); match page { @@ -1233,6 +1273,126 @@ mod resume_loop_tests { assert_eq!(env.resume.resume_cursor().await, None); } + #[tokio::test] + async fn object_timeout_aborts_the_bucket_page_immediately() { + let env = make_env().await; + env.storage.set_page( + None, + Page { + items: vec![item("timed-out", None, false)], + next: None, + truncated: false, + }, + ); + env.storage.set_outcome("timed-out", None, HealOutcome::Timeout); + + let (processed, successful, failed, skipped, result) = run(&env).await; + + assert!(matches!(result, Err(Error::TaskTimeout))); + assert_eq!(processed, 0); + assert_eq!(successful, 0); + assert_eq!(failed, 0); + assert_eq!(skipped, 0); + } + + #[tokio::test] + async fn bucket_listing_failure_does_not_mark_set_completed() { + let env = make_env().await; + env.storage.fail_listing(); + + let result = env + .healer + .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint) + .await; + + assert!(result.is_err(), "a bucket listing failure must fail the set heal pass"); + let state = env.resume.get_state().await; + assert!(!state.completed, "a failed bucket must not mark the set completed"); + assert_eq!(state.retry_count, 1, "the failed bucket must schedule a bounded retry"); + assert!(state.completed_buckets.is_empty(), "the failed bucket must remain resumable"); + } + + #[tokio::test] + async fn completed_resume_state_is_not_selected_for_a_new_heal() { + let env = make_env().await; + env.resume + .mark_completed() + .await + .expect("completed resume state should persist"); + + let task_id = env + .healer + .get_or_create_task_id("pool_0_set_0") + .await + .expect("new heal should allocate a task id"); + + assert_ne!(task_id, "task", "a completed resume state must not suppress a new heal"); + } + + #[tokio::test] + async fn cleanup_failure_keeps_erasure_set_heal_incomplete() { + let env = make_env().await; + let checkpoint_path = format!("{BUCKET_META_PREFIX}/task_{RESUME_CHECKPOINT_FILE}"); + let _failure = ResumeDeleteFailure::install(checkpoint_path, crate::heal::DiskError::DiskAccessDenied); + + let error = env + .healer + .heal_erasure_set(&["b".to_string()], "pool_0_set_0") + .await + .expect_err("checkpoint cleanup failure must fail the erasure-set heal"); + + assert!(matches!(error, Error::Disk(crate::heal::DiskError::DiskAccessDenied))); + let state = ResumeManager::load_from_disk(env.healer.disk.clone(), "task") + .await + .expect("completed state must remain discoverable after cleanup failure") + .get_state() + .await; + assert!(state.completed, "successful data heal must be persisted before cleanup is attempted"); + } + + #[tokio::test] + async fn retry_resume_repairs_checkpoint_after_crash_between_resets() { + let env = make_env().await; + env.resume + .update_progress(3, 1, 1, 1) + .await + .expect("dirty resume progress should persist"); + env.resume + .complete_bucket("b") + .await + .expect("dirty completed bucket should persist"); + env.resume + .set_resume_cursor(Some("stale-cursor".to_string())) + .await + .expect("dirty resume cursor should persist"); + env.checkpoint + .add_skipped_object(compose_key("stale-object", None)) + .await + .expect("dirty checkpoint object should be recorded"); + env.checkpoint + .update_position(4, 9) + .await + .expect("dirty checkpoint position should persist"); + + assert!( + env.resume.schedule_retry().await.expect("resume retry reset should persist"), + "retry budget should remain" + ); + + let (_, checkpoint) = env + .healer + .initialize_resume_state("task", "pool_0_set_0", &["b".to_string()]) + .await + .expect("resume initialization should repair a stale checkpoint"); + let checkpoint = checkpoint.get_checkpoint().await; + + assert_eq!(checkpoint.current_bucket_index, 0); + assert_eq!(checkpoint.current_object_index, 0); + assert!(checkpoint.processed_objects.is_empty()); + assert!(checkpoint.failed_objects.is_empty()); + assert!(checkpoint.skipped_objects.is_empty()); + } + #[tokio::test] async fn test_resume_across_page_boundary_no_drop_no_double() { let env = make_env().await; diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 5f0d15c4f..a78659775 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -2495,7 +2495,8 @@ impl HealManager { queue.pop_next() }; - if let Some(request) = selected_request { + if let Some(mut request) = selected_request { + request.options.timeout.get_or_insert(config.task_timeout); let task_priority = request.priority; let task_type_label = heal_request_type_label(&request).to_string(); let task_set_label = heal_request_set_metric_label(&request); @@ -4925,6 +4926,75 @@ mod tests { assert_eq!(manager.get_queue_length().await, 0); } + #[tokio::test] + async fn configured_task_timeout_applies_only_when_request_timeout_is_absent() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new( + storage, + Some(HealConfig { + max_concurrent_heals: 1, + task_timeout: Duration::ZERO, + ..HealConfig::default() + }), + ); + + let mut defaulted = bucket_request("defaulted-timeout", HealPriority::Normal, HealRequestSource::Admin); + defaulted.options.timeout = None; + let defaulted_id = defaulted.id.clone(); + manager + .submit_heal_request(defaulted) + .await + .expect("request without timeout should be queued"); + process_manager_queue_once(&manager).await; + let defaulted_status = tokio::time::timeout(Duration::from_secs(1), async { + loop { + if let Ok(status @ HealTaskStatus::Retrying { .. }) = manager.get_task_status(&defaulted_id).await { + break status; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("configured timeout should finish the task"); + assert!(matches!(defaulted_status, HealTaskStatus::Retrying { .. })); + assert_eq!( + manager + .retrying_heals + .lock() + .await + .get(&defaulted_id) + .expect("timed out task should retain its retry request") + .request + .options + .timeout, + Some(Duration::ZERO) + ); + manager + .cancel_task(&defaulted_id) + .await + .expect("retrying timeout task should be cancelled"); + + let mut explicit = bucket_request("explicit-timeout", HealPriority::Normal, HealRequestSource::Admin); + explicit.options.timeout = Some(Duration::from_secs(60)); + let explicit_id = explicit.id.clone(); + manager + .submit_heal_request(explicit) + .await + .expect("request with explicit timeout should be queued"); + process_manager_queue_once(&manager).await; + let explicit_status = tokio::time::timeout(Duration::from_secs(1), async { + loop { + if let Ok(status @ HealTaskStatus::Failed { .. }) = manager.get_task_status(&explicit_id).await { + break status; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("explicit timeout request should finish without using the zero default"); + assert!(matches!(explicit_status, HealTaskStatus::Failed { .. })); + } + #[tokio::test] async fn test_force_start_bypasses_duplicate_and_full_admission() { let storage: Arc = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/resume.rs b/crates/heal/src/heal/resume.rs index 7c44b25f6..cd310e3c1 100644 --- a/crates/heal/src/heal/resume.rs +++ b/crates/heal/src/heal/resume.rs @@ -14,6 +14,8 @@ use crate::{Error, Result}; use serde::{Deserialize, Serialize}; +#[cfg(test)] +use std::collections::HashMap; use std::collections::HashSet; use std::path::Path; use std::sync::{Arc, Mutex}; @@ -32,7 +34,7 @@ const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state"; /// resume state file constants const RESUME_STATE_FILE: &str = "ahm_resume_state.json"; const RESUME_PROGRESS_FILE: &str = "ahm_progress.json"; -const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json"; +pub(super) const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json"; /// Current on-disk schema version for `ResumeState`. Snapshots written by an /// older schema (which tracked latest-only object names and a positional @@ -92,6 +94,64 @@ fn path_to_str(path: &Path) -> Result<&str> { .ok_or_else(|| Error::other(format!("Invalid UTF-8 path: {path:?}"))) } +#[cfg(test)] +pub(super) struct ResumeDeleteFailure { + path: String, +} + +#[cfg(test)] +fn resume_delete_failures() -> &'static Mutex> { + static FAILURES: std::sync::OnceLock>> = std::sync::OnceLock::new(); + FAILURES.get_or_init(|| Mutex::new(HashMap::new())) +} + +#[cfg(test)] +impl ResumeDeleteFailure { + pub(super) fn install(path: String, error: DiskError) -> Self { + let previous = resume_delete_failures() + .lock() + .expect("resume delete failure registry should not poison") + .insert(path.clone(), error); + assert!(previous.is_none(), "resume delete failure already installed"); + Self { path } + } +} + +#[cfg(test)] +impl Drop for ResumeDeleteFailure { + fn drop(&mut self) { + resume_delete_failures() + .lock() + .expect("resume delete failure registry should not poison") + .remove(&self.path); + } +} + +#[cfg(test)] +fn injected_resume_delete_error(path: &str) -> Option { + resume_delete_failures() + .lock() + .expect("resume delete failure registry should not poison") + .get(path) + .cloned() +} + +#[cfg(not(test))] +fn injected_resume_delete_error(_path: &str) -> Option { + None +} + +async fn delete_resume_file(disk: &DiskStore, path: &Path) -> Result<()> { + let path_str = path_to_str(path)?; + if let Some(err) = injected_resume_delete_error(path_str) { + return Err(err.into()); + } + match disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await { + Ok(()) | Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => Ok(()), + Err(err) => Err(err.into()), + } +} + /// resume state #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ResumeState { @@ -407,7 +467,7 @@ impl ResumeManager { let mut state = self.state.write().await; state.mark_completed(); drop(state); - self.save_state_throttled().await + self.save_state().await } /// set error message @@ -444,24 +504,14 @@ impl ResumeManager { /// cleanup resume state pub async fn cleanup(&self) -> Result<()> { - let state = self.state.read().await; - let task_id = &state.task_id; + let task_id = self.state.read().await.task_id.clone(); - // delete state files let state_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_STATE_FILE}")); let progress_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_PROGRESS_FILE}")); - let checkpoint_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}")); - // ignore delete errors, files may not exist - if let Ok(path_str) = path_to_str(&state_file) { - let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await; - } - if let Ok(path_str) = path_to_str(&progress_file) { - let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await; - } - if let Ok(path_str) = path_to_str(&checkpoint_file) { - let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await; - } + delete_resume_file(&self.disk, &progress_file).await?; + // Delete the state file last so a partial cleanup remains discoverable. + delete_resume_file(&self.disk, &state_file).await?; debug!( target: "rustfs::heal::resume", @@ -763,13 +813,10 @@ impl CheckpointManager { /// cleanup checkpoint pub async fn cleanup(&self) -> Result<()> { - let checkpoint = self.checkpoint.read().await; - let task_id = &checkpoint.task_id; + let task_id = self.checkpoint.read().await.task_id.clone(); let checkpoint_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}")); - if let Ok(path_str) = path_to_str(&checkpoint_file) { - let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await; - } + delete_resume_file(&self.disk, &checkpoint_file).await?; debug!( target: "rustfs::heal::resume", @@ -1211,6 +1258,83 @@ mod tests { assert!(!throttle.record(), "counter must reset after a save"); } + #[tokio::test] + async fn completion_persists_immediately_and_cleanup_propagates_delete_errors() { + use super::super::{DiskOption, Endpoint, new_disk}; + use tempfile::TempDir; + + let temp_dir = TempDir::new().expect("create resume persistence test directory"); + let endpoint = Endpoint::try_from(temp_dir.path().to_string_lossy().as_ref()).expect("create test disk endpoint"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("create resume persistence test disk"); + match disk.make_volume(RUSTFS_META_BUCKET).await { + Ok(()) | Err(DiskError::VolumeExists) => {} + Err(err) => panic!("create metadata volume for resume persistence test: {err}"), + } + + let task_id = "completion-persistence".to_string(); + let manager = ResumeManager::new( + disk.clone(), + task_id.clone(), + "erasure_set".to_string(), + "pool_0_set_0".to_string(), + vec!["bucket".to_string()], + ) + .await + .expect("create resume manager"); + manager + .update_progress(1, 1, 0, 0) + .await + .expect("buffer progress below the persistence threshold"); + manager.mark_completed().await.expect("persist completed resume state"); + + let persisted = ResumeManager::load_from_disk(disk.clone(), &task_id) + .await + .expect("reload completed resume state") + .get_state() + .await; + assert!(persisted.completed, "completion must be persisted without waiting for the throttle"); + assert_eq!(persisted.processed_objects, 1, "the completion write must include buffered progress"); + + let state_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}"); + let failure = ResumeDeleteFailure::install(state_path, DiskError::DiskAccessDenied); + let error = manager + .cleanup() + .await + .expect_err("resume cleanup must propagate a real delete failure"); + assert!(matches!(error, Error::Disk(DiskError::DiskAccessDenied))); + drop(failure); + manager.cleanup().await.expect("resume cleanup must be retryable"); + manager + .cleanup() + .await + .expect("missing resume files must be idempotent success"); + + let checkpoint = CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("create checkpoint manager"); + let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}"); + let failure = ResumeDeleteFailure::install(checkpoint_path, DiskError::DiskAccessDenied); + let error = checkpoint + .cleanup() + .await + .expect_err("checkpoint cleanup must propagate a real delete failure"); + assert!(matches!(error, Error::Disk(DiskError::DiskAccessDenied))); + drop(failure); + checkpoint.cleanup().await.expect("checkpoint cleanup must be retryable"); + checkpoint + .cleanup() + .await + .expect("missing checkpoint must be idempotent success"); + } + #[tokio::test] async fn test_resume_utils() { let task_id1 = ResumeUtils::generate_task_id(); diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 8057f6967..8ac104215 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -144,6 +144,9 @@ pub struct HealOptions { pub recursive: bool, /// Whether to dry run pub dry_run: bool, + /// Whether to skip namespace locking + #[serde(default)] + pub no_lock: bool, /// Timeout pub timeout: Option, /// pool index @@ -161,7 +164,8 @@ impl Default for HealOptions { update_parity: true, recursive: false, dry_run: false, - timeout: Some(Duration::from_secs(300)), // 5 minutes default timeout + no_lock: false, + timeout: None, pool_index: None, set_index: None, } @@ -896,7 +900,7 @@ impl HealTask { recreate: self.options.recreate_missing, scan_mode: self.options.scan_mode, update_parity: self.options.update_parity, - no_lock: false, + no_lock: self.options.no_lock, pool: self.options.pool_index, set: self.options.set_index, }; @@ -1104,7 +1108,7 @@ impl HealTask { recreate: true, scan_mode: HealScanMode::Deep, update_parity: true, - no_lock: false, + no_lock: self.options.no_lock, pool: None, set: None, }; @@ -1260,7 +1264,7 @@ impl HealTask { recreate: self.options.recreate_missing, scan_mode: self.options.scan_mode, update_parity: self.options.update_parity, - no_lock: false, + no_lock: self.options.no_lock, pool: self.options.pool_index, set: self.options.set_index, }; @@ -1426,7 +1430,7 @@ impl HealTask { recreate: self.options.recreate_missing, scan_mode: self.options.scan_mode, update_parity: self.options.update_parity, - no_lock: false, + no_lock: self.options.no_lock, pool: self.options.pool_index, set: self.options.set_index, }; @@ -1680,7 +1684,7 @@ impl HealTask { recreate: false, scan_mode: HealScanMode::Deep, update_parity: false, - no_lock: false, + no_lock: self.options.no_lock, pool: self.options.pool_index, set: self.options.set_index, }; @@ -1809,7 +1813,7 @@ impl HealTask { recreate: self.options.recreate_missing, scan_mode: HealScanMode::Deep, update_parity: true, - no_lock: false, + no_lock: self.options.no_lock, pool: None, set: None, }; @@ -1973,7 +1977,7 @@ impl HealTask { recreate: true, scan_mode: HealScanMode::Deep, update_parity: true, - no_lock: false, + no_lock: self.options.no_lock, pool: None, set: None, }; @@ -2214,7 +2218,7 @@ impl HealTask { recreate: self.options.recreate_missing, scan_mode: self.options.scan_mode, update_parity: self.options.update_parity, - no_lock: false, + no_lock: self.options.no_lock, pool: self.options.pool_index, set: self.options.set_index, }; @@ -2230,10 +2234,6 @@ impl HealTask { self.record_result_item(result).await; } Err(err) => { - // Check if error is due to cancellation or timeout - if matches!(err, Error::TaskCancelled | Error::TaskTimeout) { - return Err(err); - } warn!( target: "rustfs::heal::task", event = EVENT_HEAL_ERASURE_SET_RESULT, @@ -2246,6 +2246,7 @@ impl HealTask { error = %err, "Heal erasure set bucket prepass failed" ); + return Err(err); } } } @@ -2268,7 +2269,7 @@ impl HealTask { recreate: self.options.recreate_missing, scan_mode: self.options.scan_mode, update_parity: self.options.update_parity, - no_lock: false, + no_lock: self.options.no_lock, pool: self.options.pool_index, set: self.options.set_index, }; @@ -2297,7 +2298,9 @@ impl HealTask { stage = "execute_resumable_heal", "Heal erasure set stage entered" ); - let result = erasure_healer.heal_erasure_set(&buckets, &set_disk_id).await; + let result = self + .await_with_control(erasure_healer.heal_erasure_set(&buckets, &set_disk_id)) + .await; // Keep the markers on failure: the resume state also persists, and the // next run of this set heal re-marks and eventually clears them. @@ -2360,12 +2363,13 @@ impl std::fmt::Debug for HealTask { #[cfg(test)] mod tests { - use super::super::{DiskStore, Endpoint}; + use super::super::{DiskOption, DiskStore, Endpoint, HealDiskExt as _, new_disk}; use super::*; use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo}; use rustfs_madmin::heal_commands::HealResultItem; use std::collections::{HashMap, VecDeque}; use std::sync::Mutex; + use tempfile::TempDir; use super::super::storage_api::status::BucketInfo; #[derive(Default)] @@ -2388,6 +2392,8 @@ mod tests { listed_buckets: Mutex>>, bucket_heal_errors: Mutex>>, bucket_heal_calls: Mutex>, + block_heal_object: Mutex, + resume_disk: Mutex>, } /// Build a latest, non-delete-marker heal list item with no version id. @@ -2520,6 +2526,10 @@ mod tests { .unwrap() .push(version_id.map(ToString::to_string)); self.object_heal_opts.lock().unwrap().push(*opts); + let block_heal_object = *self.block_heal_object.lock().unwrap(); + if block_heal_object { + std::future::pending::<()>().await; + } if let Some(outcome) = self .heal_object_outcomes .lock() @@ -2635,10 +2645,35 @@ mod tests { } async fn get_disk_for_resume(&self, _set_disk_id: &str) -> Result { - Err(Error::other("not implemented in tests")) + self.resume_disk + .lock() + .unwrap() + .clone() + .ok_or_else(|| Error::other("not implemented in tests")) } } + async fn make_resume_disk(temp: &TempDir) -> DiskStore { + let disk_path = temp.path().join("test_disk"); + std::fs::create_dir_all(&disk_path).expect("test disk directory should be created"); + let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).expect("test disk endpoint should be valid"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("test disk should initialize"); + let metadata_volume = disk.make_volume(RUSTFS_META_BUCKET).await; + assert!( + matches!(metadata_volume, Ok(()) | Err(DiskError::VolumeExists)), + "metadata volume should exist: {metadata_volume:?}" + ); + disk + } + #[tokio::test] async fn test_recursive_bucket_heal_visits_objects() { let storage = Arc::new(MockStorage::default()); @@ -2932,6 +2967,7 @@ mod tests { remove_corrupted: true, recreate_missing: true, scan_mode: HealScanMode::Deep, + no_lock: true, timeout: None, ..Default::default() }, @@ -2948,12 +2984,14 @@ mod tests { assert!(!bucket_opts[0].remove); assert!(bucket_opts[0].recreate); assert_eq!(bucket_opts[0].scan_mode, HealScanMode::Deep); + assert!(bucket_opts[0].no_lock); let object_opts = storage.object_heal_opts.lock().unwrap(); assert_eq!(object_opts.len(), 2); assert!(object_opts.iter().all(|opts| opts.remove)); assert!(object_opts.iter().all(|opts| opts.recreate)); assert!(object_opts.iter().all(|opts| opts.scan_mode == HealScanMode::Deep)); + assert!(object_opts.iter().all(|opts| opts.no_lock)); } #[tokio::test] @@ -3584,4 +3622,87 @@ mod tests { "erasure-set heal should continue past NoHealRequired format result, got: {err}" ); } + + #[tokio::test] + async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() { + let temp = TempDir::new().expect("temporary directory should be created"); + let disk = make_resume_disk(&temp).await; + let storage = Arc::new(MockStorage { + bucket_heal_errors: Mutex::new(HashMap::from([( + "bucket-a".to_string(), + VecDeque::from(["injected bucket prepass failure"]), + )])), + resume_disk: Mutex::new(Some(disk)), + ..Default::default() + }); + let request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket-a".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + let error = task + .heal_erasure_set(vec!["bucket-a".to_string()], "pool_0_set_0".to_string()) + .await + .expect_err("bucket prepass failure must stop the erasure-set heal"); + + assert!(error.to_string().contains("injected bucket prepass failure")); + assert_eq!(storage.bucket_heal_calls.lock().unwrap().as_slice(), ["bucket-a".to_string()]); + assert!(storage.object_heal_opts.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn resumable_erasure_set_execution_is_cancelled_while_object_heal_is_pending() { + let temp = TempDir::new().expect("temporary directory should be created"); + let disk = make_resume_disk(&temp).await; + let storage = Arc::new(MockStorage { + block_heal_object: Mutex::new(true), + resume_disk: Mutex::new(Some(disk)), + ..Default::default() + }); + let request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket-a".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + no_lock: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = Arc::new(HealTask::from_request(request, storage.clone())); + let execution = tokio::spawn({ + let task = task.clone(); + async move { task.execute().await } + }); + + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if !storage.object_heal_opts.lock().unwrap().is_empty() { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("resumable object heal should start"); + task.cancel().await.expect("task cancellation should succeed"); + + let result = tokio::time::timeout(Duration::from_secs(1), execution) + .await + .expect("cancellation should interrupt the pending resumable heal") + .expect("task execution should join"); + assert!(matches!(result, Err(Error::TaskCancelled))); + assert!(storage.bucket_heal_opts.lock().unwrap().iter().all(|opts| opts.no_lock)); + assert!(storage.object_heal_opts.lock().unwrap().iter().all(|opts| opts.no_lock)); + } } diff --git a/crates/heal/tests/heal_integration_test.rs b/crates/heal/tests/heal_integration_test.rs index 35bf55fc5..7094614db 100644 --- a/crates/heal/tests/heal_integration_test.rs +++ b/crates/heal/tests/heal_integration_test.rs @@ -306,6 +306,7 @@ mod serial_tests { recreate_missing: false, scan_mode: HealScanMode::Normal, update_parity: false, + no_lock: false, timeout: Some(Duration::from_secs(300)), pool_index: None, set_index: None, diff --git a/crates/protos/src/heal_control.rs b/crates/protos/src/heal_control.rs index 7847f3bc0..df89126d2 100644 --- a/crates/protos/src/heal_control.rs +++ b/crates/protos/src/heal_control.rs @@ -27,7 +27,7 @@ use rustfs_common::heal_channel::{ use serde::{Deserialize, Serialize, de::SeqAccess, de::Visitor}; use std::{fmt, io::Cursor, io::Write}; -const ENVELOPE_VERSION: u8 = 1; +const ENVELOPE_VERSION: u8 = 2; pub const ENVELOPE_MAX_SIZE: usize = 64 * 1024; pub const RESULT_MAX_SIZE: usize = 16 * 1024 * 1024; pub const NONCE_SIZE: usize = 16; @@ -101,6 +101,8 @@ pub struct StartCommand { update_parity: Option, recursive: Option, dry_run: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + no_lock: Option, timeout_seconds: Option, source: HealRequestSource, } @@ -132,6 +134,7 @@ impl TryFrom for StartCommand { update_parity: request.update_parity, recursive: request.recursive, dry_run: request.dry_run, + no_lock: request.no_lock, timeout_seconds: request.timeout_seconds, source: request.source, }) @@ -164,6 +167,7 @@ impl StartCommand { update_parity: self.update_parity, recursive: self.recursive, dry_run: self.dry_run, + no_lock: self.no_lock, timeout_seconds: self.timeout_seconds, source: self.source, }) @@ -557,8 +561,8 @@ fn encode_bounded(value: &impl Serialize, value_name: &str, max_size: usize) -> #[cfg(test)] mod tests { use super::{ - Admission, ENVELOPE_MAX_SIZE, Envelope, Outcome, RESULT_MAX_SIZE, RequestMetadata, ResultEnvelope, decode_envelope, - decode_result, encode_result, + Admission, ENVELOPE_MAX_SIZE, Envelope, ExecutableCommand, Outcome, RESULT_MAX_SIZE, RequestMetadata, ResultEnvelope, + decode_envelope, decode_result, encode_result, }; use rustfs_common::heal_channel::{HealChannelRequest, HealChannelResponse, HealRequestSource}; use serde::de::{DeserializeSeed, SeqAccess, Visitor, value::Error as ValueError}; @@ -632,7 +636,26 @@ mod tests { } #[test] - fn v1_query_and_result_wire_fixtures_are_stable() { + fn v2_start_round_trips_no_lock() { + for no_lock in [None, Some(false), Some(true)] { + let request_id = uuid::Uuid::new_v4().to_string(); + let mut request = test_request(request_id); + request.no_lock = no_lock; + + let envelope = Envelope::start(request, metadata(1, 7)).expect("v2 start should encode nolock"); + let encoded = super::encode_envelope(&envelope).expect("v2 start should serialize"); + let decoded = super::decode_envelope(&encoded).expect("v2 start should deserialize"); + let (_, _, ExecutableCommand::Start { request }) = decoded.into_execution().expect("v2 start command should decode") + else { + panic!("expected start command"); + }; + + assert_eq!(request.no_lock, no_lock); + } + } + + #[test] + fn v2_command_and_result_wire_fixtures_are_stable() { let request_id = "00112233-4455-6677-8899-aabbccddeeff".to_string(); let start = Envelope::start( test_request(request_id.clone()), @@ -700,23 +723,23 @@ mod tests { .collect::(); assert_eq!( start_hex, - "87a776657273696f6e01a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6482a6616374696f6ea57374617274a772657175657374de0010a46469736bc0a66275636b6574a66275636b6574ac6f626a656374507265666978a6707265666978af6f626a65637456657273696f6e4964c0aa666f7263655374617274c2a87072696f72697479a66e6f726d616ca9706f6f6c496e64657801a8736574496e64657802a87363616e4d6f6465c0af72656d6f7665436f72727570746564c0af72656372656174654d697373696e67c0ac757064617465506172697479c0a9726563757273697665c0a664727952756ec0ae74696d656f75745365636f6e6473c0a6736f75726365a561646d696e" + "87a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6482a6616374696f6ea57374617274a772657175657374de0010a46469736bc0a66275636b6574a66275636b6574ac6f626a656374507265666978a6707265666978af6f626a65637456657273696f6e4964c0aa666f7263655374617274c2a87072696f72697479a66e6f726d616ca9706f6f6c496e64657801a8736574496e64657802a87363616e4d6f6465c0af72656d6f7665436f72727570746564c0af72656372656174654d697373696e67c0ac757064617465506172697479c0a9726563757273697665c0a664727952756ec0ae74696d656f75745365636f6e6473c0a6736f75726365a561646d696e" ); assert_eq!( cancel_hex, - "87a776657273696f6e01a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6483a6616374696f6ea663616e63656ca96865616c5f70617468ad6275636b65742f707265666978ac636c69656e745f746f6b656eac636c69656e742d746f6b656e" + "87a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6483a6616374696f6ea663616e63656ca96865616c5f70617468ad6275636b65742f707265666978ac636c69656e745f746f6b656eac636c69656e742d746f6b656e" ); assert_eq!( start_result_hex, - "84a776657273696f6e01a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666b0636f6f7264696e61746f7245706f636809a76f7574636f6d6583a6726573756c74a57374617274a77461736b5f6964d92466666565646463632d626261612d393938382d373736362d353534343333323231313030a961646d697373696f6eae64726f707065645f706f6c696379" + "84a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666b0636f6f7264696e61746f7245706f636809a76f7574636f6d6583a6726573756c74a57374617274a77461736b5f6964d92466666565646463632d626261612d393938382d373736362d353534343333323231313030a961646d697373696f6eae64726f707065645f706f6c696379" ); assert_eq!( envelope_hex, - "87a776657273696f6e01a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6483a6616374696f6ea57175657279a96865616c5f70617468ad6275636b65742f707265666978ac636c69656e745f746f6b656eac636c69656e742d746f6b656e" + "87a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666a56e6f6e6365dc001011111111111111111111111111111111ae6973737565644174556e69784d73cf0000018bcfe56800af657870697265734174556e69784d73cf0000018bcfe5dd30b0636f6f7264696e61746f7245706f636809a7636f6d6d616e6483a6616374696f6ea57175657279a96865616c5f70617468ad6275636b65742f707265666978ac636c69656e745f746f6b656eac636c69656e742d746f6b656e" ); assert_eq!( result_hex, - "84a776657273696f6e01a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666b0636f6f7264696e61746f7245706f636809a76f7574636f6d6584a6726573756c74a76368616e6e656ca773756363657373c3a46461746193010203a56572726f72c0" + "84a776657273696f6e02a9726571756573744964d92430303131323233332d343435352d363637372d383839392d616162626363646465656666b0636f6f7264696e61746f7245706f636809a76f7574636f6d6584a6726573756c74a76368616e6e656ca773756363657373c3a46461746193010203a56572726f72c0" ); } diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 5e26be200..8a9a39adc 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -168,9 +168,9 @@ pub fn internode_rpc_max_message_size() -> usize { } pub const HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE: usize = heal_control::RESULT_MAX_SIZE + 1024; -pub const HEAL_CONTROL_PROTOCOL_VERSION: u32 = 2; +pub const HEAL_CONTROL_PROTOCOL_VERSION: u32 = 3; pub const DYNAMIC_CONFIG_PROTOCOL_VERSION: u32 = 1; -pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v2\0"; +pub const HEAL_CONTROL_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-heal-control-capability-v3\0"; pub const REMOTE_VERSION_STATE_CAPABILITY_PROBE_PREFIX: &[u8] = b"rustfs-tier-remote-version-state-capability-v1\0"; pub const TIER_MUTATION_RPC_MAX_PREPARE_PAYLOAD_SIZE: usize = 64 * 1024; pub const TIER_MUTATION_RPC_MAX_COMMIT_PAYLOAD_SIZE: usize = 1024; @@ -250,7 +250,7 @@ pub fn canonical_heal_control_request_body( topology_fingerprint: &str, command: &[u8], ) -> Result, std::num::TryFromIntError> { - const DOMAIN: &[u8] = b"rustfs-heal-control-v2\0"; + const DOMAIN: &[u8] = b"rustfs-heal-control-v3\0"; let fingerprint = topology_fingerprint.as_bytes(); let mut body = Vec::with_capacity(DOMAIN.len() + 4 + 8 + fingerprint.len() + 8 + command.len()); @@ -270,7 +270,7 @@ pub fn canonical_heal_control_capability_ack( topology_fingerprint: &str, probe: &[u8], ) -> Result, std::num::TryFromIntError> { - const DOMAIN: &[u8] = b"rustfs-heal-control-capability-ack-v2\0"; + const DOMAIN: &[u8] = b"rustfs-heal-control-capability-ack-v3\0"; let fingerprint = topology_fingerprint.as_bytes(); let mut body = Vec::with_capacity(DOMAIN.len() + 4 + 8 + fingerprint.len() + 8 + probe.len()); @@ -289,7 +289,7 @@ pub fn canonical_heal_control_response_body( command: &[u8], result: &[u8], ) -> Result, std::num::TryFromIntError> { - const DOMAIN: &[u8] = b"rustfs-heal-control-response-v2\0"; + const DOMAIN: &[u8] = b"rustfs-heal-control-response-v3\0"; let fingerprint = topology_fingerprint.as_bytes(); let mut body = Vec::with_capacity(DOMAIN.len() + 4 + 8 + fingerprint.len() + 8 + command.len() + 8 + result.len()); @@ -1713,7 +1713,7 @@ mod heal_control_tests { #[test] fn canonical_heal_control_body_binds_every_field_and_boundary() { let baseline = canonical_heal_control_request_body(1, "ab", b"c").expect("small request should encode"); - let mut golden = b"rustfs-heal-control-v2\0".to_vec(); + let mut golden = b"rustfs-heal-control-v3\0".to_vec(); golden.extend_from_slice(&1_u32.to_be_bytes()); golden.extend_from_slice(&2_u64.to_be_bytes()); golden.extend_from_slice(b"ab"); @@ -1741,11 +1741,11 @@ mod heal_control_tests { #[test] fn canonical_capability_ack_binds_version_and_topology() { - assert_eq!(HEAL_CONTROL_PROTOCOL_VERSION, 2); - assert!(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX.starts_with(b"rustfs-heal-control-capability-v2")); + assert_eq!(HEAL_CONTROL_PROTOCOL_VERSION, 3); + assert!(HEAL_CONTROL_CAPABILITY_PROBE_PREFIX.starts_with(b"rustfs-heal-control-capability-v3")); let probe = heal_control_capability_probe(&[7; 16]); let ack = canonical_heal_control_capability_ack(1, "ab", &probe).expect("small acknowledgement should encode"); - let mut golden = b"rustfs-heal-control-capability-ack-v2\0".to_vec(); + let mut golden = b"rustfs-heal-control-capability-ack-v3\0".to_vec(); golden.extend_from_slice(&1_u32.to_be_bytes()); golden.extend_from_slice(&2_u64.to_be_bytes()); golden.extend_from_slice(b"ab"); diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 70e45fdd4..c8a87483d 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -845,6 +845,7 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { heal_request.update_parity = Some(hip.hs.update_parity); heal_request.recursive = Some(recursive || root_cluster_target); heal_request.dry_run = Some(hip.hs.dry_run); + heal_request.no_lock = Some(hip.hs.no_lock); heal_request.source = HealRequestSource::Admin; heal_request } @@ -2123,6 +2124,23 @@ mod tests { assert_eq!(request.update_parity, Some(false)); assert_eq!(request.recursive, Some(true)); assert_eq!(request.dry_run, Some(true)); + assert_eq!(request.no_lock, Some(true)); + + let envelope = rustfs_protos::heal_control::Envelope::start( + request, + rustfs_protos::heal_control::RequestMetadata::new([1; 16], 1_000, 2_000, 1), + ) + .expect("admin heal request should encode through heal-control v2"); + let encoded = + rustfs_protos::heal_control::encode_envelope(&envelope).expect("admin heal-control envelope should serialize"); + let decoded = + rustfs_protos::heal_control::decode_envelope(&encoded).expect("admin heal-control envelope should deserialize"); + let (_, _, rustfs_protos::heal_control::ExecutableCommand::Start { request }) = + decoded.into_execution().expect("admin heal start should decode") + else { + panic!("expected admin heal start command"); + }; + assert_eq!(request.no_lock, Some(true)); } #[test] diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index e1f00af66..5aec4de94 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -20,6 +20,7 @@ checked_files=( "rustfs/src/admin/handlers/table_catalog/refs.rs" "rustfs/src/admin/handlers/table_catalog/routes.rs" "rustfs/src/admin/handlers/table_catalog/table.rs" + "rustfs/src/admin/handlers/table_catalog/tests.rs" "rustfs/src/admin/handlers/table_catalog/view.rs" "rustfs/src/admin/handlers/service_account.rs" "rustfs/src/admin/handlers/kms_audit.rs"