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

fix(ecstore): reclaim rollback backups after object heal (#8641)

Browse Source 
* fix(ecstore): reclaim rollback backups after object heal

Retire only the backup returned by a successful, fenced heal publication.
Keep child keys, live data directories, and delete recovery evidence safe.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(ecstore): retain heal cleanup ownership through cancellation

Forward publication and repair key owners to the existing owned delete
executor so local unlink cannot outlive its exact-key serialization.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-10-18 00:03:00 +08:00
committed by GitHub
parent f64cb816db
commit aeae6f9980
2 changed files with 394 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/ops/delete_recovery.rs
+1 -1
View File
@@ -528,7 +528,7 @@ impl SetDisks {
None
};
if let Some((transaction, source)) = prerequisite {
Box::pin(self.heal_delete_recovery_source(bucket, object, &source, can_commit)).await?;
Box::pin(self.heal_delete_recovery_source(bucket, object, &source, can_commit, namespace_owner.as_ref())).await?;
let refreshed = inspect(&disks, bucket, object, transaction)
.await
.into_iter()
crates/ecstore/src/set_disk/ops/heal.rs
+393 -59
View File
@@ -33,6 +33,7 @@ struct ExplicitVersionHeal<'a> {
allow_regeneration: bool,
preserve_unrepairable: bool,
commit_fence: Option<&'a (dyn Fn() -> bool + Sync)>,
namespace_owner: Option<&'a Arc<dyn Send + Sync>>,
retirement: Option<&'a MarkerRetirementContext<'a>>,
}
@@ -775,6 +776,7 @@ impl SetDisks {
allow_regeneration: true,
preserve_unrepairable: false,
commit_fence: None,
namespace_owner: None,
retirement: None,
},
&mut None,
@@ -921,6 +923,7 @@ impl SetDisks {
allow_regeneration: allow_explicit_version_regen,
preserve_unrepairable,
commit_fence,
namespace_owner,
retirement,
} = run;
trace!(
@@ -983,10 +986,10 @@ impl SetDisks {
// Bound, not `_`: this guard must live to the end of the scope. A bare
// `_` would drop it here and release the namespace write lock.
let _write_lock_guard = if !opts.no_lock && !read_repair_uses_shared_lock {
Some(
Some(Arc::new(
self.acquire_heal_object_lock(bucket, object, version_id, HealObjectLockKind::Write, "heal_object")
.await?,
)
))
} else {
None
};
@@ -1870,7 +1873,7 @@ impl SetDisks {
)
.await
{
Ok(Some(guard)) => Some(guard),
Ok(Some(guard)) => Some(Arc::new(guard)),
Ok(None) => {
let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
return Ok((result, Some(DiskError::Timeout)));
@@ -1891,6 +1894,7 @@ impl SetDisks {
let mut rename_attempts = 0usize;
let mut rename_successes = 0usize;
let mut healed_disks = vec![None; out_dated_disks.len()];
let mut backup_cleanup_owner: Option<Arc<dyn Send + Sync>> = None;
for (index, outdated_disk) in out_dated_disks.iter().enumerate() {
if let Some(disk) = outdated_disk {
if let Some(scope) = &bucket_heal_scope {
@@ -1919,59 +1923,114 @@ impl SetDisks {
.await
};
if let Err(err) = &rename_result {
warn!(
event = EVENT_HEAL_OBJECT_RENAME,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_HEAL,
bucket,
object,
version_id,
disk_index = index,
endpoint = %disk.endpoint(),
tmp_id,
result = "failed",
error = %err,
"Heal object rename failed"
);
} else {
rename_successes += 1;
healed_disks[index] = Some(disk.clone());
if parts_metadata[index].is_remote() {
let rm_data_dir =
parts_metadata[index].data_dir.expect("operation should succeed").to_string();
let d_path = Path::new(&encode_dir_object(object)).join(rm_data_dir);
if let Err(e) = disk
.delete(
bucket,
d_path.to_str().expect("operation should succeed"),
DeleteOptions {
immediate: true,
recursive: true,
..Default::default()
},
)
.await
{
// The healed shard has already been renamed into place; a
// failure cleaning up the old remote data dir must not abort
// the heal and leak the tmp shards (backlog#799 B20).
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_HEAL,
bucket,
object,
error = %e,
"Heal remote data-dir cleanup failed"
);
}
match rename_result {
Err(err) => {
warn!(
event = EVENT_HEAL_OBJECT_RENAME,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_HEAL,
result = "failed",
bucket,
object,
version_id,
disk_index = index,
endpoint = %disk.endpoint(),
tmp_id,
error = %err,
"Heal object rename failed"
);
}
Ok(rename_response) => {
rename_successes += 1;
healed_disks[index] = Some(disk.clone());
// Heal can reuse a live data directory, and a synthetic
// rollback directory can also contain a child key.
// Retire only this publication's backup while fenced.
if let Some(rollback_dir) =
rename_response.rollback_data_dir.or(rename_response.old_data_dir)
&& bucket_heal_scope.as_ref().is_none_or(|scope| scope.check().is_ok())
&& commit_fence.is_none_or(|check| check())
&& !_write_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
&& !_read_repair_write_lock_guard
.as_ref()
.is_some_and(|guard| guard.is_lock_lost())
{
let backup_path = format!(
"{}/{rollback_dir}/{}",
rustfs_utils::path::encode_dir_object_ref(object),
disk::STORAGE_FORMAT_FILE_BACKUP,
);
// The syscall can outlive this future; retain the exact-key
// locks until physical cleanup has finished.
let cleanup_owner = backup_cleanup_owner.get_or_insert_with(|| {
Arc::new((
namespace_owner.cloned(),
_write_lock_guard.clone(),
_read_repair_write_lock_guard.clone(),
bucket_heal_scope.clone(),
self.ctx.begin_namespace_commit(),
))
});
match disk
.delete_with_namespace_owner(
bucket,
&backup_path,
DeleteOptions::default(),
Some(cleanup_owner.clone()),
)
.await
{
Ok(()) | Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
Err(err) => warn!(
event = EVENT_HEAL_OBJECT_RENAME,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_HEAL,
result = "cleanup_failed",
bucket,
object,
version_id,
disk_index = index,
error = %err,
"Heal rollback backup cleanup failed"
),
}
}
if parts_metadata[index].is_remote() {
let rm_data_dir =
parts_metadata[index].data_dir.expect("operation should succeed").to_string();
for (i, v) in result.before.drives.iter().enumerate() {
if v.endpoint == disk.endpoint().to_string() {
result.after.drives[i].state = DriveState::Ok.to_string();
let d_path = Path::new(&encode_dir_object(object)).join(rm_data_dir);
if let Err(e) = disk
.delete(
bucket,
d_path.to_str().expect("operation should succeed"),
DeleteOptions {
immediate: true,
recursive: true,
..Default::default()
},
)
.await
{
// The healed shard has already been renamed into place; a
// failure cleaning up the old remote data dir must not abort
// the heal and leak the tmp shards (backlog#799 B20).
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_HEAL,
bucket,
object,
error = %e,
"Heal remote data-dir cleanup failed"
);
}
}
for (i, v) in result.before.drives.iter().enumerate() {
if v.endpoint == disk.endpoint().to_string() {
result.after.drives[i].state = DriveState::Ok.to_string();
}
}
}
}
@@ -3385,6 +3444,7 @@ impl SetDisks {
object: &str,
source: &crate::disk::DeleteRecoveryVersionIdentity,
can_commit: &(dyn Fn() -> bool + Sync),
namespace_owner: Option<&Arc<dyn Send + Sync>>,
) -> disk::error::Result<()> {
if !can_commit() {
return Err(DiskError::DiskOngoingReq);
@@ -3430,6 +3490,7 @@ impl SetDisks {
allow_regeneration: false,
preserve_unrepairable: true,
commit_fence: Some(can_commit),
namespace_owner,
retirement: None,
},
&mut None,
@@ -3619,6 +3680,8 @@ impl SetDisks {
// Pass no_lock=true since we already obtained write lock (or are already called with no_lock=true)
let mut inner_opts = *opts;
inner_opts.no_lock = true;
let commit_fence = || !_write_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost());
let namespace_owner = _write_lock_guard.as_ref().map(|guard| guard.clone() as Arc<dyn Send + Sync>);
let (result, err) = self
.heal_object_with_explicit_version_regen(
bucket,
@@ -3628,7 +3691,8 @@ impl SetDisks {
opts: &inner_opts,
allow_regeneration: true,
preserve_unrepairable: false,
commit_fence: None,
commit_fence: Some(&commit_fence),
namespace_owner: namespace_owner.as_ref(),
retirement: proof.retirement,
},
absence,
@@ -3650,7 +3714,8 @@ impl SetDisks {
opts: &inner_opts,
allow_regeneration: true,
preserve_unrepairable: false,
commit_fence: None,
commit_fence: Some(&commit_fence),
namespace_owner: namespace_owner.as_ref(),
retirement: proof.retirement,
},
absence,
@@ -3703,8 +3768,9 @@ impl SetDisks {
#[cfg(test)]
mod heal_result_report_tests {
use super::{
AbsenceProofRequest, DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, HealWriterFailures,
ReadRepairCommitFingerprint, ReadRepairPauseScope, SetDisks, heal_writer_error_summary, is_retryable_heal_writer_error,
AbsenceProofRequest, DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, ExplicitVersionHeal,
HealWriterFailures, ReadRepairCommitFingerprint, ReadRepairPauseScope, SetDisks, heal_writer_error_summary,
is_retryable_heal_writer_error,
};
use super::{HEAL_RENAME_INCOMPLETE, HealRenameFailureScope, HealWriterFailureScope};
use crate::disk::endpoint::Endpoint;
@@ -3719,6 +3785,7 @@ mod heal_result_report_tests {
};
use crate::storage_api_contracts::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
use crate::storage_api_contracts::heal::HealOperations as _;
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
use crate::{
config::storageclass,
@@ -4517,6 +4584,11 @@ mod heal_result_report_tests {
expected.to_string(),
"{case}: after.drives must reflect the actual rename outcome at index {index}"
);
let data_path = temp_dirs[index].path().join(&bucket).join(object).join(data_dir.to_string());
assert!(
!data_path.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP).exists(),
"{case}: heal must retire the backup even when it shares the live data directory"
);
assert_eq!(
temp_dirs[index]
.path()
@@ -5420,6 +5492,254 @@ mod heal_result_report_tests {
}
}
#[cfg(not(windows))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cancelled_heal_backup_cleanup_retains_key_owner_until_unlink_finishes() {
use crate::disk::os::prepared_publication_test_hooks::{Stage, install_at};
use std::time::Duration;
for mode in ["outer", "inner", "read-repair"] {
let (dirs, disks, set) = hermetic_set_disks_isolated(4).await;
let bucket = format!("heal-cleanup-cancel-{mode}");
let object = "object";
for disk in &disks {
disk.make_volume(&bucket).await.expect("create bucket volume");
}
let options = ObjectOptions {
no_lock: true,
..Default::default()
};
set.put_object(&bucket, object, &mut PutObjReader::from_vec(vec![0x11; 592]), &options)
.await
.expect("write old generation on every disk");
let metadata_path = dirs[0].path().join(&bucket).join(object).join(STORAGE_FORMAT_FILE);
let old_metadata = tokio::fs::read(&metadata_path).await.expect("save stale replica");
let payload = vec![0x22; 592];
set.put_object(&bucket, object, &mut PutObjReader::from_vec(payload.clone()), &options)
.await
.expect("write current generation on every disk");
tokio::fs::write(&metadata_path, old_metadata)
.await
.expect("restore stale target");
let rollback_dir = crate::disk::local::inline_metadata_rollback_dir(Uuid::nil(), &rustfs_filemeta::FileMeta::new());
let backup_path = disks[0]
.get_object_path_for_io_if_local(
&bucket,
&format!("{object}/{rollback_dir}/{}", crate::disk::STORAGE_FORMAT_FILE_BACKUP),
)
.expect("fixture disk is local")
.expect("physical backup path");
let (arrived_tx, arrived_rx) = tokio::sync::oneshot::channel();
let (release_tx, release_rx) = std::sync::mpsc::channel();
let _hook = install_at(Stage::Remove, &backup_path, move || {
let _ = arrived_tx.send(());
let _ = release_rx.recv_timeout(Duration::from_secs(30));
});
let task_set = set.clone();
let task_bucket = bucket.clone();
let heal = tokio::spawn(async move {
let opts = HealOpts {
read_repair: mode == "read-repair",
scan_mode: HealScanMode::Deep,
..Default::default()
};
if mode == "outer" {
let (_, error) = task_set
.heal_object_with_retirement(
&task_bucket,
object,
"",
&opts,
&mut None,
AbsenceProofRequest {
retirement: None,
allow_unversioned: false,
},
)
.await
.expect("production wrapper heal should reach cleanup");
assert!(error.is_none(), "production wrapper heal error: {error:?}");
} else {
let (_, error) = task_set
.heal_object(&task_bucket, object, "", &opts)
.await
.expect("inner/read-repair heal should reach cleanup");
assert!(error.is_none(), "inner/read-repair heal error: {error:?}");
}
});
let arrived = tokio::time::timeout(Duration::from_secs(10), arrived_rx).await;
if !matches!(arrived, Ok(Ok(()))) {
heal.abort();
let outcome = heal.await;
let _ = release_tx.send(());
panic!("{mode}: cleanup must reach the owned physical unlink hook; task: {outcome:?}");
}
heal.abort();
assert!(heal.await.expect_err("heal must be cancelled at the unlink").is_cancelled());
assert!(
set.ctx.namespace_commits_pending(),
"{mode}: physical cleanup must retain namespace ownership"
);
let namespace = set.new_ns_lock(&bucket, object).await.expect("successor namespace");
let mut successor = Box::pin(namespace.get_write_lock(Duration::from_secs(30)));
assert!(
tokio::time::timeout(Duration::from_secs(1), &mut successor).await.is_err(),
"{mode}: a successor must not enter while the old unlink is pending"
);
release_tx.send(()).expect("release physical cleanup");
let successor = tokio::time::timeout(Duration::from_secs(10), successor)
.await
.expect("physical completion should release the key lock")
.expect("successor should acquire the released lock");
assert!(!backup_path.exists(), "{mode}: owned unlink must finish before releasing the key");
drop(successor);
let mut reader = set
.get_object_reader(&bucket, object, None, Default::default(), &options)
.await
.expect("committed object remains readable after cancellation");
let mut body = Vec::new();
tokio::io::copy(&mut reader, &mut body)
.await
.expect("read full committed body");
assert_eq!(body, payload, "{mode}: cancellation must preserve committed data");
}
}
#[tokio::test]
async fn heal_reclaims_published_inline_backup_and_preserves_child_key() {
let (dirs, disks, set) = hermetic_set_disks_isolated(4).await;
let bucket = "heal-inline-rollback";
let object = "nested/object";
for disk in &disks {
disk.make_volume(bucket).await.expect("create bucket volume");
}
let options = ObjectOptions {
no_lock: true,
..Default::default()
};
let rollback_dir = crate::disk::local::inline_metadata_rollback_dir(Uuid::nil(), &rustfs_filemeta::FileMeta::new());
let child = format!("{object}/{rollback_dir}");
let child_payload = b"a child key must survive parent repair".to_vec();
set.put_object(bucket, &child, &mut PutObjReader::from_vec(child_payload.clone()), &options)
.await
.expect("write child key in the synthetic rollback directory");
set.put_object(bucket, object, &mut PutObjReader::from_vec(vec![0x11; 592]), &options)
.await
.expect("write old inline generation on every disk");
let metadata_path = dirs[0].path().join(bucket).join(object).join(STORAGE_FORMAT_FILE);
let old_metadata = tokio::fs::read(&metadata_path).await.expect("save stale target metadata");
let payload = vec![0x22; 592];
set.put_object(bucket, object, &mut PutObjReader::from_vec(payload.clone()), &options)
.await
.expect("write current inline generation on every disk");
let read_options = ReadOptions {
read_data: true,
..Default::default()
};
let expected = disks[0]
.read_version("", bucket, object, "", &read_options)
.await
.expect("read current target generation");
assert!(expected.data.is_some(), "fixture must exercise inline publication");
tokio::fs::write(&metadata_path, &old_metadata)
.await
.expect("restore one stale replica without a pre-existing delete transaction");
let backup_path = metadata_path
.parent()
.expect("object directory")
.join(rollback_dir.to_string())
.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP);
let opts = HealOpts {
no_lock: true,
scan_mode: HealScanMode::Deep,
..Default::default()
};
let (_, error) = set
.heal_object(bucket, object, "", &HealOpts { dry_run: true, ..opts })
.await
.expect("dry-run stale inline repair");
assert!(error.is_none(), "dry-run error: {error:?}");
assert_eq!(tokio::fs::read(&metadata_path).await.expect("dry-run target"), old_metadata);
assert!(!backup_path.exists(), "dry-run must not publish a backup");
let failure_scope = HealRenameFailureScope::install(bucket, object, &[expected.erasure.distribution[0] - 1]);
let (_, error) = set
.heal_object(bucket, object, "", &opts)
.await
.expect("report failed publication");
drop(failure_scope);
assert!(error.is_some(), "failed rename must remain retryable");
assert_eq!(tokio::fs::read(&metadata_path).await.expect("failed target"), old_metadata);
assert!(!backup_path.exists(), "failed publication must not reclaim or manufacture a backup");
let (result, error) = set
.heal_object(bucket, object, "", &opts)
.await
.expect("repair stale inline replica");
assert!(error.is_none(), "repair error: {error:?}");
assert_eq!(result.drives_healed(), Some(1));
let repaired = disks[0]
.read_version("", bucket, object, "", &read_options)
.await
.expect("read repaired physical target");
assert_eq!(repaired.mod_time, expected.mod_time);
assert_eq!(repaired.data_dir, expected.data_dir);
assert_eq!(repaired.data, expected.data);
assert!(!backup_path.exists(), "successful heal must retire its old-generation backup");
let evidence = disks[0]
.discover_delete_transactions(bucket, Some(object), None, 128)
.await
.expect("discover delete evidence without treating a child key as a transaction");
assert!(evidence.transactions.is_empty(), "successful repair must leave no delete blocker");
for (key, expected_body) in [(object, payload), (child.as_str(), child_payload)] {
let mut reader = set
.get_object_reader(bucket, key, None, Default::default(), &options)
.await
.expect("read repaired parent and preserved child");
let mut body = Vec::new();
tokio::io::copy(&mut reader, &mut body)
.await
.expect("read complete object body");
assert_eq!(body, expected_body, "object {key}");
}
let (healthy, error) = set.heal_object(bucket, object, "", &opts).await.expect("repeat repair");
assert!(error.is_none(), "repeat repair error: {error:?}");
assert_eq!(healthy.drives_healed(), Some(0));
tokio::fs::write(&metadata_path, &old_metadata)
.await
.expect("restore stale replica for fence probe");
// The fence is valid up to publication, then fails before backup removal.
let can_commit = || !backup_path.exists();
let (_, error) = set
.heal_object_with_explicit_version_regen(
bucket,
object,
"",
ExplicitVersionHeal {
opts: &opts,
allow_regeneration: true,
preserve_unrepairable: false,
commit_fence: Some(&can_commit),
namespace_owner: None,
retirement: None,
},
&mut None,
)
.await
.expect("report committed repair after fence loss");
assert!(
error.is_none(),
"post-publication fence loss must not undo the committed target: {error:?}"
);
assert!(backup_path.exists(), "a lost fence must preserve the rollback evidence");
let repaired = disks[0]
.read_version("", bucket, object, "", &read_options)
.await
.expect("committed target survives skipped cleanup");
assert_eq!(repaired.data, expected.data);
}
#[tokio::test]
async fn deep_heal_inline_bitrot_rebuilds_physical_data_and_parity_shards() {
use crate::storage_api_contracts::range::HTTPRangeSpec;
@@ -5437,12 +5757,13 @@ mod heal_result_report_tests {
read_data: true,
..Default::default()
};
let object_options = ObjectOptions {
let mut object_options = ObjectOptions {
no_lock: true,
..Default::default()
};
for parity in [false, true] {
object_options.versioned = parity;
for corrupt_prefix in [false, true] {
let object = format!("parity-{parity}-prefix-{corrupt_prefix}.bin");
set.put_object(bucket, &object, &mut PutObjReader::from_vec(payload.clone()), &object_options)
@@ -5516,6 +5837,19 @@ mod heal_result_report_tests {
assert_eq!(repaired.data, original_meta.data, "heal must restore the exact encoded shard");
assert_eq!(repaired.version_id, original_meta.version_id);
assert_eq!(repaired.data_dir, original_meta.data_dir);
let rollback_dir = crate::disk::local::inline_metadata_rollback_dir(
repaired.version_id.unwrap_or_default(),
&rustfs_filemeta::FileMeta::new(),
);
assert!(
!metadata_path
.parent()
.expect("object directory")
.join(rollback_dir.to_string())
.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP)
.exists(),
"inline bitrot repair must not retain a rollback backup"
);
}
for (range, expected) in [