diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index da7786df0..2a9edecc3 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -1415,7 +1415,7 @@ impl SetDisks { ); } - if read_repair_uses_shared_lock || preserve_unrepairable { + if read_repair_uses_shared_lock || preserve_unrepairable || !opts.remove { return Ok((result, Some(cannot_heal_err))); } @@ -2166,6 +2166,13 @@ impl SetDisks { } } Err(err) => { + // All replicas explicitly missing was handled before the + // quorum check. A partial read cannot prove object absence. + let err = if matches!(err, DiskError::FileNotFound | DiskError::FileVersionNotFound) { + DiskError::ErasureReadQuorum + } else { + err + }; if let Some(retirement) = retirement && parts_metadata .iter() @@ -2205,7 +2212,8 @@ impl SetDisks { } return Ok((item, cleanup.err())); } - if allow_explicit_version_regen + if !opts.dry_run + && allow_explicit_version_regen && !version_id.is_empty() && self .try_regenerate_explicit_version_meta(bucket, object, version_id, &parts_metadata, &errs, &disks) @@ -2224,7 +2232,7 @@ impl SetDisks { .await; } - if opts.dry_run || preserve_unrepairable { + if opts.dry_run || preserve_unrepairable || !opts.remove { return Ok(( self.default_heal_result(FileInfo::default(), &errs, bucket, object, version_id) .await, @@ -7205,12 +7213,13 @@ mod heal_result_report_tests { "", &HealOpts { no_lock: true, + remove: true, ..Default::default() }, ) .await .expect("unsafe dangling state should be reported without deletion"); - assert_eq!(err, Some(DiskError::FileNotFound)); + assert_eq!(err, Some(DiskError::ErasureReadQuorum)); let surviving = disks[0] .as_ref() .expect("first test disk should be online") @@ -7251,6 +7260,7 @@ mod heal_result_report_tests { .expect("seed exact stale historical version"); let opts = HealOpts { no_lock: true, + remove: true, scan_mode: HealScanMode::Deep, ..Default::default() }; @@ -7326,6 +7336,443 @@ mod heal_result_report_tests { .await; } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn heal_without_remove_preserves_unrecoverable_unversioned_shards() { + use crate::disk::{CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS}; + use crate::object_api::WriteCompletion; + use futures::FutureExt as _; + use std::panic::AssertUnwindSafe; + use std::time::Duration; + + // Exercise the same on-disk layout with explicit removal first, so + // an unrelated early failure cannot masquerade as preservation. + for remove in [true, false] { + let (dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let outcome = AssertUnwindSafe(tokio::time::timeout(Duration::from_secs(45), async { + let bucket = if remove { + "heal-remove-unversioned" + } else { + "heal-preserve-unversioned" + }; + let object = "object.bin"; + let old_time = OffsetDateTime::from_unix_timestamp(1).expect("fixture modification time"); + let elapsed = (OffsetDateTime::now_utc() - old_time).whole_seconds(); + let grace = rustfs_utils::get_env_u64("RUSTFS_HEAL_DANGLING_DELETE_GRACE_SECS", 3600); + assert!( + elapsed > i64::try_from(grace).unwrap_or(i64::MAX), + "fixture must be outside the configured dangling-delete grace period" + ); + set.make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("create fixture bucket"); + let options = ObjectOptions { + mod_time: Some(old_time), + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }; + set.put_object(bucket, object, &mut PutObjReader::from_vec(vec![0x5a; 1024 * 1024]), &options) + .await + .expect("write all four fixture shards"); + + let mut metadata = Vec::new(); + let mut infos = Vec::new(); + for (dir, disk) in dirs.iter().zip(&disks) { + metadata.push( + tokio::fs::read(dir.path().join(bucket).join(object).join(STORAGE_FORMAT_FILE)) + .await + .expect("read each committed metadata file"), + ); + infos.push( + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("read each committed unversioned metadata replica"), + ); + } + let selected = &infos[0]; + let data_dir = selected.data_dir.expect("fixture must be non-inline"); + assert!(selected.version_id.is_none_or(|id| id.is_nil())); + assert_eq!(selected.mod_time, Some(old_time)); + let missing_count = selected.erasure.parity_blocks + 1; + assert!(missing_count < disks.len(), "at least one physical shard must survive"); + assert!(infos.iter().all(|info| { + info.data_dir == Some(data_dir) + && info.version_id.is_none_or(|id| id.is_nil()) + && info.erasure.parity_blocks == selected.erasure.parity_blocks + })); + let part_path = |dir: &TempDir| dir.path().join(bucket).join(object).join(data_dir.to_string()).join("part.1"); + let surviving_shard = tokio::fs::read(part_path(&dirs[missing_count])) + .await + .expect("read the shard that must survive"); + for dir in dirs.iter().take(missing_count) { + tokio::fs::remove_file(part_path(dir)).await.expect("remove one exact shard"); + } + for (index, (disk, info)) in disks.iter().zip(&infos).enumerate() { + let checked = disk.check_parts(bucket, object, info).await.expect("check fixture shards"); + let expected = if index < missing_count { + CHECK_PART_FILE_NOT_FOUND + } else { + CHECK_PART_SUCCESS + }; + assert_eq!(checked.results, vec![expected], "disk {index} has the intended shard state"); + } + let visible = set + .get_object_info( + bucket, + object, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("metadata quorum remains visible before heal"); + assert_eq!(visible.data_dir, Some(data_dir)); + + let (result, error) = ::heal_object( + set.as_ref(), + bucket, + object, + "", + &HealOpts { + remove, + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + ) + .await + .expect("heal returns an object result"); + if remove { + assert!(error.is_none(), "explicit dangling cleanup must succeed: {error:?}"); + for dir in &dirs { + assert!( + tokio::fs::metadata(dir.path().join(bucket)) + .await + .expect("bucket remains readable") + .is_dir() + ); + assert!( + matches!( + tokio::fs::symlink_metadata(dir.path().join(bucket).join(object).join(STORAGE_FORMAT_FILE)).await, + Err(error) if error.kind() == std::io::ErrorKind::NotFound + ), + "explicit removal must remove the metadata replica" + ); + } + assert!( + matches!(tokio::fs::symlink_metadata(part_path(&dirs[missing_count])).await, + Err(error) if error.kind() == std::io::ErrorKind::NotFound), + "explicit removal must remove the surviving shard" + ); + assert_eq!(result.after.drives.len(), disks.len()); + assert!( + result + .after + .drives + .iter() + .all(|drive| drive.state == DriveState::Missing.to_string()) + ); + } else { + for (index, (dir, before)) in dirs.iter().zip(&metadata).enumerate() { + let after = tokio::fs::read(dir.path().join(bucket).join(object).join(STORAGE_FORMAT_FILE)) + .await + .expect("remove=false must keep each metadata replica"); + assert_eq!(&after, before, "remove=false changed metadata on disk {index}"); + } + assert_eq!( + tokio::fs::read(part_path(&dirs[missing_count])) + .await + .expect("remove=false must keep the surviving shard"), + surviving_shard + ); + assert!(matches!(&error, Some(Error::ErasureReadQuorum)), "unexpected heal result: {error:?}"); + assert_eq!(result.before.drives.len(), disks.len()); + assert_eq!( + result + .before + .drives + .iter() + .filter(|drive| drive.state == DriveState::Missing.to_string()) + .count(), + missing_count + ); + let mut absence = None; + let (_, proof_error) = set + .heal_object_with_absence( + bucket, + object, + "", + &HealOpts { + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + &mut absence, + AbsenceProofRequest { + retirement: None, + allow_unversioned: true, + }, + ) + .await + .expect("repeat through the absence-proof path"); + assert!(matches!(proof_error, Some(Error::ErasureReadQuorum))); + assert!(absence.is_none(), "surviving shards cannot produce an absence proof"); + } + })) + .catch_unwind() + .await; + drop(set); + drop(disks); + if !matches!(&outcome, Ok(Ok(()))) { + let retained: Vec<_> = dirs.into_iter().map(TempDir::keep).collect(); + eprintln!("unrecoverable-shard fixture retained: {retained:?}"); + match outcome { + Ok(Err(_)) => panic!("unrecoverable-shard fixture timed out"), + Err(panic) => std::panic::resume_unwind(panic), + Ok(Ok(())) => unreachable!(), + } + } + } + } + + #[tokio::test] + #[serial_test::serial] + async fn heal_without_remove_preserves_explicit_version_metadata_during_dry_run() { + use crate::object_api::{ShardIntegrityWriteMode, WriteCompletion}; + use futures::FutureExt as _; + use std::panic::AssertUnwindSafe; + use std::time::Duration; + + let (dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let outcome = AssertUnwindSafe(tokio::time::timeout(Duration::from_secs(45), async { + let bucket = "heal-dry-run-versioned"; + let object = "object.bin"; + set.make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("create fixture bucket"); + let original = set + .put_object( + bucket, + object, + &mut PutObjReader::from_vec(vec![0x67; 1024 * 1024]), + &ObjectOptions { + versioned: true, + shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Legacy), + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("write a real version on all four disks"); + let version = original.version_id.expect("explicit version"); + assert!(!version.is_nil()); + let generation = original.data_dir.expect("external data generation"); + let version_string = version.to_string(); + let metadata_path = |dir: &TempDir| dir.path().join(bucket).join(object).join(STORAGE_FORMAT_FILE); + let shard_path = |dir: &TempDir| { + dir.path() + .join(bucket) + .join(object) + .join(generation.to_string()) + .join("part.1") + }; + let mut shards = Vec::new(); + for (dir, disk) in dirs.iter().zip(&disks) { + let info = disk + .read_version("", bucket, object, &version_string, &ReadOptions::default()) + .await + .expect("initial metadata on each disk"); + assert_eq!(info.version_id, Some(version)); + assert_eq!(info.data_dir, Some(generation)); + assert!(info.parts.iter().all(|part| part.integrity.is_none())); + assert!( + rustfs_filemeta::shard_integrity::descriptor_from_metadata(&info.metadata) + .expect("metadata protection descriptor") + .is_none() + ); + shards.push(tokio::fs::read(shard_path(dir)).await.expect("save original shard")); + } + let surviving_metadata = tokio::fs::read(metadata_path(&dirs[0])) + .await + .expect("save surviving metadata"); + for dir in dirs.iter().skip(1) { + tokio::fs::remove_file(metadata_path(dir)) + .await + .expect("remove one metadata replica"); + } + for disk in disks.iter().skip(1) { + assert!(matches!( + disk.read_version("", bucket, object, &version_string, &ReadOptions::default()) + .await, + Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) + )); + } + + let mut absence = None; + let (_, error) = set + .heal_object_with_absence( + bucket, + object, + &version_string, + &HealOpts { + dry_run: true, + ..Default::default() + }, + &mut absence, + AbsenceProofRequest { + retirement: None, + allow_unversioned: false, + }, + ) + .await + .expect("dry-run reports the metadata quorum failure"); + assert!(matches!(error, Some(Error::ErasureReadQuorum)), "partial metadata must not look absent"); + assert!(absence.is_none()); + assert_eq!( + tokio::fs::read(metadata_path(&dirs[0])).await.expect("surviving metadata"), + surviving_metadata + ); + for (index, dir) in dirs.iter().enumerate() { + if index != 0 { + assert!(matches!( + tokio::fs::symlink_metadata(metadata_path(dir)).await, + Err(error) if error.kind() == std::io::ErrorKind::NotFound + )); + } + assert_eq!(tokio::fs::read(shard_path(dir)).await.expect("original shard"), shards[index]); + } + + let (_, error) = set + .heal_object_with_absence( + bucket, + object, + &version_string, + &HealOpts::default(), + &mut absence, + AbsenceProofRequest { + retirement: None, + allow_unversioned: false, + }, + ) + .await + .expect("normal heal attempts metadata regeneration"); + assert!(error.is_none(), "normal heal must repair the same layout: {error:?}"); + assert!(absence.is_none()); + for (index, (dir, disk)) in dirs.iter().zip(&disks).enumerate() { + let info = disk + .read_version("", bucket, object, &version_string, &ReadOptions::default()) + .await + .expect("normal heal restored metadata"); + assert_eq!(info.version_id, Some(version)); + assert_eq!(info.data_dir, Some(generation)); + assert_eq!(tokio::fs::read(shard_path(dir)).await.expect("original shard"), shards[index]); + } + })) + .catch_unwind() + .await; + drop(set); + drop(disks); + if !matches!(&outcome, Ok(Ok(()))) { + let retained: Vec<_> = dirs.into_iter().map(TempDir::keep).collect(); + eprintln!("dry-run versioned fixture retained: {retained:?}"); + match outcome { + Ok(Err(_)) => panic!("dry-run versioned fixture timed out"), + Err(panic) => std::panic::resume_unwind(panic), + Ok(Ok(())) => unreachable!(), + } + } + } + + #[tokio::test] + #[serial_test::serial] + async fn heal_without_remove_preserves_dangling_inline_metadata_and_emits_no_absence_proof() { + use futures::FutureExt as _; + use std::panic::AssertUnwindSafe; + use std::time::Duration; + + temp_env::async_with_vars([("RUSTFS_HEAL_DANGLING_DELETE_GRACE_SECS", Some("0"))], async { + for remove in [true, false] { + let bucket = if remove { + "inline-dangling-remove" + } else { + "inline-dangling-preserve" + }; + let object = "stale.txt"; + let (dirs, set, disks) = + dangling_inline_test_fixture(bucket, object, OffsetDateTime::now_utc() - time::Duration::hours(2)).await; + let outcome = AssertUnwindSafe(tokio::time::timeout(Duration::from_secs(45), async { + let metadata = dirs[0].path().join(bucket).join(object).join(STORAGE_FORMAT_FILE); + let before = tokio::fs::read(&metadata).await.expect("read surviving inline metadata"); + let mut absence = None; + let (_, error) = set + .heal_object_with_absence( + bucket, + object, + "", + &HealOpts { + no_lock: true, + remove, + ..Default::default() + }, + &mut absence, + AbsenceProofRequest { + retirement: None, + allow_unversioned: true, + }, + ) + .await + .expect("dangling inline heal returns an object result"); + if remove { + assert!(error.is_none(), "explicit cleanup must succeed: {error:?}"); + assert!( + absence.as_ref().is_some_and(|proof| proof.removed), + "successful cleanup must prove removal" + ); + assert!(matches!( + tokio::fs::symlink_metadata(metadata).await, + Err(error) if error.kind() == std::io::ErrorKind::NotFound + )); + let (_, missing_error) = set + .heal_object_with_absence( + bucket, + object, + "", + &HealOpts { + no_lock: true, + ..Default::default() + }, + &mut absence, + AbsenceProofRequest { + retirement: None, + allow_unversioned: true, + }, + ) + .await + .expect("fully absent object returns a result"); + assert!(matches!(missing_error, Some(Error::FileNotFound))); + assert!(absence.is_some_and(|proof| !proof.removed)); + } else { + assert!(matches!(error, Some(Error::ErasureReadQuorum)), "partial metadata must not look absent"); + assert!(absence.is_none(), "a surviving candidate cannot prove authoritative absence"); + assert_eq!(tokio::fs::read(metadata).await.expect("metadata must remain"), before); + } + })) + .catch_unwind() + .await; + drop(set); + drop(disks); + if !matches!(&outcome, Ok(Ok(()))) { + let retained: Vec<_> = dirs.into_iter().map(TempDir::keep).collect(); + eprintln!("dangling-inline fixture retained: {retained:?}"); + match outcome { + Ok(Err(_)) => panic!("dangling-inline fixture timed out"), + Err(panic) => std::panic::resume_unwind(panic), + Ok(Ok(())) => unreachable!(), + } + } + } + }) + .await; + } + #[tokio::test] #[serial_test::serial] async fn heal_reports_success_after_dangling_inline_cleanup() { @@ -7342,6 +7789,7 @@ mod heal_result_report_tests { "", &HealOpts { no_lock: true, + remove: true, ..Default::default() }, ) @@ -7372,6 +7820,7 @@ mod heal_result_report_tests { "", &HealOpts { no_lock: true, + remove: true, ..Default::default() }, ) @@ -7422,13 +7871,14 @@ mod heal_result_report_tests { &HealOpts { dry_run: true, no_lock: true, + remove: true, ..Default::default() }, ) .await .expect("dry-run should report the dangling candidate without mutation"); - assert_eq!(error, Some(DiskError::FileNotFound)); + assert_eq!(error, Some(DiskError::ErasureReadQuorum)); assert!( temp_dirs[0] .path() @@ -7452,8 +7902,9 @@ mod heal_result_report_tests { seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata).await; set.disks.write().await[1] = None; - let (_, err) = set - .heal_object( + let mut absence = None; + let (_, error) = set + .heal_object_with_absence( bucket, object, "", @@ -7461,10 +7912,31 @@ mod heal_result_report_tests { no_lock: true, ..Default::default() }, + &mut absence, + AbsenceProofRequest { + retirement: None, + allow_unversioned: true, + }, + ) + .await + .expect("offline shard state should be reported without removal"); + assert!(matches!(error, Some(Error::ErasureReadQuorum))); + assert!(absence.is_none(), "offline members cannot prove absence"); + + let (_, err) = set + .heal_object( + bucket, + object, + "", + &HealOpts { + no_lock: true, + remove: true, + ..Default::default() + }, ) .await .expect("offline shard state should be reported without deletion"); - assert_eq!(err, Some(DiskError::FileNotFound)); + assert_eq!(err, Some(DiskError::ErasureReadQuorum)); let surviving = disks[0] .as_ref() .expect("first test disk should be online") @@ -7501,12 +7973,13 @@ mod heal_result_report_tests { "", &HealOpts { no_lock: true, + remove: true, ..Default::default() }, ) .await .expect("part probe timeout should be reported without deletion"); - assert_eq!(err, Some(DiskError::FileNotFound)); + assert_eq!(err, Some(DiskError::ErasureReadQuorum)); let surviving = disks[0] .as_ref() .expect("first test disk should be online")