From aeae6f9980d4ebee140a3adcef6fa61c7dacd4f1 Mon Sep 17 00:00:00 2001 From: Hauser Date: Sun, 18 Oct 2026 00:03:00 +0800 Subject: [PATCH] fix(ecstore): reclaim rollback backups after object heal (#8641) * 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 Co-Authored-By: zhi22915 * 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 Co-Authored-By: zhi22915 --------- Co-authored-by: heihutu Co-authored-by: zhi22915 --- .../src/set_disk/ops/delete_recovery.rs | 2 +- crates/ecstore/src/set_disk/ops/heal.rs | 452 +++++++++++++++--- 2 files changed, 394 insertions(+), 60 deletions(-) diff --git a/crates/ecstore/src/set_disk/ops/delete_recovery.rs b/crates/ecstore/src/set_disk/ops/delete_recovery.rs index c4ab3bdc5..25ef35049 100644 --- a/crates/ecstore/src/set_disk/ops/delete_recovery.rs +++ b/crates/ecstore/src/set_disk/ops/delete_recovery.rs @@ -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() diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index e07d01ca5..f617efed6 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -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>, 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> = 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>, ) -> 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); 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 [