diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 8db359d3f..ce9f95d46 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -303,7 +303,7 @@ use crate::object_api::{ }; use crate::services::notification_sys::RemoteVersionStateFleetProofToken; use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata}; -use crate::set_disk::core::io_primitives::{RenameTailCleanup, finish_rename_tail_heal}; +use crate::set_disk::core::io_primitives::{RenameRollbackReceipt, RenameTailCleanup, finish_rename_tail_heal}; #[cfg(test)] use crate::storage_api_contracts::namespace::NamespaceLocking; #[cfg(test)] @@ -3548,6 +3548,7 @@ impl SetDisks { (None, None, None) }; let mut tmp_cleanup_owned = false; + let rollback_receipt = RenameRollbackReceipt::default(); let operation = async { let erasure = Arc::new(erasure_from_file_info(&fi, false)?); @@ -4256,6 +4257,7 @@ impl SetDisks { let commit_bucket = bucket.to_owned(); let commit_object = object.to_owned(); let commit_tmp_dir = tmp_dir.clone(); + let commit_rollback_receipt = rollback_receipt.clone(); let commit_object_lock_guard = object_lock_guard.take(); let commit_decommission_object_lock_guard = decommission_object_lock_guard.take(); let commit_publication_guard = publication_commit_guard.take(); @@ -4456,7 +4458,8 @@ impl SetDisks { write_quorum, commit_scanner_publication_lease_tokens.as_ref(), ) - .with_publication_scope(commit_scanner_publication_scope.clone()), + .with_publication_scope(commit_scanner_publication_scope.clone()) + .with_rollback_receipt(commit_rollback_receipt.clone()), ) .await; if let Some(scope) = commit_scanner_publication_scope.as_ref() { @@ -4589,6 +4592,11 @@ impl SetDisks { let rename_commit = match rename_result { Ok(commit) => commit, Err(err) => { + if commit_rollback_receipt.is_incomplete() { + // Incomplete undo retains the staging source and + // rollback backup for recovery; cleanup is unsafe. + return Err(err.into()); + } if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await { warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data"); } else if issue3031_diag_enabled() { @@ -4888,7 +4896,7 @@ impl SetDisks { ); } }); - } else { + } else if !rollback_receipt.is_incomplete() { // Failure path (quorum loss / rollback): keep the cleanup inline so // a failed PUT never returns while its tmp shards are still on disk // (state-residue hardening tracked by backlog#864 / backlog#898). @@ -18358,6 +18366,99 @@ mod put_object_tmp_cleanup_tests { ); } + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn put_incomplete_rollback_preserves_staging_and_old_version_backup() { + use crate::set_disk::core::io_primitives::rollback_fault_injection; + + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] { + let (dirs, disks, set) = hermetic_set_disks(4).await; + let bucket = "put-incomplete-undo"; + let object = "incomplete-undo-object"; + make_completion_test_bucket(&disks, bucket).await; + let mut old_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]); + set.put_object( + bucket, + object, + &mut old_reader, + &ObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("old generation should be completely committed"); + wait_for_tmp_workspace_to_drain(&dirs, "old PUT must leave no unrelated staging").await; + let old = disks[0] + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("old metadata must be readable"); + let old_data_dir = old.data_dir.expect("non-inline old version needs a data directory"); + let tasks = rename_fanout_barrier::observe_tasks(object); + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]); + let _undo_fault = rollback_fault_injection::arm(object, 0, rollback_fault_injection::Fault::Io); + let writer = Arc::clone(&set); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + writer + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + write_completion, + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("overwrite must enter the actual rename fan-out before failure injection"); + barrier.release(); + let err = tokio::time::timeout(Duration::from_secs(30), put) + .await + .expect("incomplete undo must return without hanging") + .expect("PUT task should join") + .expect_err("two renamed disks cannot satisfy write quorum three"); + assert!( + matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)), + "original quorum error expected: {err}" + ); + assert_eq!(tasks.running(), 0, "every rename and undo task must be reaped before return"); + let leftovers = non_trash_tmp_entries(&dirs).await; + assert!(!leftovers.is_empty(), "incomplete undo must retain the new staging source for recovery"); + let backups = dirs + .iter() + .filter(|dir| { + dir.path() + .join(bucket) + .join(object) + .join(old_data_dir.to_string()) + .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP) + .exists() + }) + .count(); + assert_eq!(backups, 1, "exactly the failed undo disk must retain its old-version backup"); + // The remaining three disks still serve the old generation; + // the failed minority must never become an acknowledged write. + let mut read = set + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("old generation must remain readable after incomplete rollback"); + let mut body = Vec::new(); + read.stream + .read_to_end(&mut body) + .await + .expect("old generation should stream"); + assert_eq!(body, vec![b'0'; TEST_OBJECT_SIZE]); + } + }) + .await; + } + #[tokio::test] #[serial_test::serial(capacity_dirty_scope)] async fn tail_drained_put_owned_commit_survives_waiter_cancellation() {