fix(heal): preserve data when removal is disabled

This commit is contained in:
overtrue
2026-10-28 22:56:32 +08:00
parent c232f1552d
commit a4222509a1
+482 -9
View File
@@ -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) = <SetDisks as crate::storage_api_contracts::heal::HealOperations>::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")