diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 43147c1f2..301c67fe4 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -3491,12 +3491,26 @@ impl RenameConvergence { pub(in crate::set_disk) struct RenameDataCommit { pub(in crate::set_disk) online_disks: Vec>, + /// Disks whose capacity must be marked for this commit. Under early ACK, + /// this includes the unresolved tail beyond the quorum snapshot. + pub(in crate::set_disk) capacity_disks: Vec>, pub(in crate::set_disk) convergence: RenameConvergence, pub(in crate::set_disk) data_dir: Option, pub(in crate::set_disk) cleanup_disks: Vec>, pub(in crate::set_disk) old_current_size: Option, pub(in crate::set_disk) committed_file_info: FileInfo, - pub(in crate::set_disk) tail_drain: Option>, + pub(in crate::set_disk) tail_drain: Option>>, +} + +pub(in crate::set_disk) struct RenameTailCleanup { + pub(in crate::set_disk) disk_index: usize, + pub(in crate::set_disk) disk: DiskStore, + pub(in crate::set_disk) old_data_dir: Uuid, +} + +pub(in crate::set_disk) struct RenameTailOutcome { + pub(in crate::set_disk) convergence: RenameConvergence, + pub(in crate::set_disk) cleanup: Vec, } /// Options shared by the normal and early-ack rename fanouts. Keeping the @@ -3533,6 +3547,13 @@ fn put_rename_early_ack_enabled() -> bool { } impl RenameDataCommit { + /// Whether the quorum snapshot is final enough to admit heal immediately. + /// An early-ACK snapshot deliberately leaves the unresolved tail as + /// `DiskNotFound`; only the tail's final observations may decide heal. + pub(in crate::set_disk) fn needs_immediate_heal(&self) -> bool { + self.tail_drain.is_none() && self.convergence.needs_heal() + } + #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")] fn into_legacy_tuple(self) -> RenameDataLegacyTuple { ( @@ -3545,6 +3566,80 @@ impl RenameDataCommit { } } +pub(in crate::set_disk) async fn finish_rename_tail_heal< + Guards, + Finalize, + FinalizeFuture, + Cleanup, + CleanupFuture, + Submit, + SubmitFuture, +>( + tail_drain: tokio::task::JoinHandle>, + guard_release: tokio::sync::oneshot::Receiver, + guards: Guards, + request: rustfs_common::heal_channel::HealChannelRequest, + finalize: Finalize, + cleanup: Cleanup, + submit: Submit, +) where + Guards: Send, + Finalize: FnOnce() -> FinalizeFuture + Send, + FinalizeFuture: Future + Send, + Cleanup: FnOnce(Guards, Vec) -> CleanupFuture + Send, + CleanupFuture: Future + Send, + Submit: FnOnce(rustfs_common::heal_channel::HealChannelRequest) -> SubmitFuture + Send, + SubmitFuture: Future + Send, +{ + let (needs_heal, tail_cleanup, tail_complete) = match tail_drain.await { + Ok(Some(outcome)) => (outcome.convergence.needs_heal(), outcome.cleanup, true), + Ok(None) => { + warn!( + event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + state = "missing_convergence", + bucket = %request.bucket, + object = request.object_prefix.as_deref().unwrap_or_default(), + "committed rename tail drain completed without convergence" + ); + (true, Vec::new(), false) + } + Err(err) => { + warn!( + event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + state = "failed", + bucket = %request.bucket, + object = request.object_prefix.as_deref().unwrap_or_default(), + error = %err, + "rename tail drain failed" + ); + (true, Vec::new(), false) + } + }; + // A dropped sender means the request continuation was cancelled. Cleanup + // must still run; only the explicit `false` used by the hard-crash harness + // suppresses all in-process continuation work. + if matches!(guard_release.await, Ok(false)) { + drop(guards); + return; + } + finalize().await; + if tail_complete { + cleanup(guards, tail_cleanup).await; + } else { + // The outer coordinator may fail while a disk rename still owns the + // staging source. Preserve it for later healing/GC instead of racing + // that in-flight task with eager cleanup. + drop(guards); + } + if needs_heal { + submit(request).await; + } +} + impl SetDisks { pub(in crate::set_disk) fn default_read_quorum(&self) -> usize { self.set_drive_count - self.default_parity_count @@ -3663,6 +3758,7 @@ impl SetDisks { }; Ok(RenameDataCommit { + capacity_disks: disks.to_vec(), online_disks, convergence, data_dir, @@ -3705,9 +3801,16 @@ impl SetDisks { dst_object: &str, write_quorum: usize, ) -> disk::error::Result { - Self::rename_data_owned(disks, src_bucket, src_object, file_infos.to_vec(), dst_bucket, dst_object, write_quorum) - .await - .map(RenameDataCommit::into_legacy_tuple) + Self::rename_data_owned( + disks, + (src_bucket, src_object), + file_infos.to_vec(), + (dst_bucket, dst_object), + write_quorum, + false, + ) + .await + .map(RenameDataCommit::into_legacy_tuple) } pub(in crate::set_disk) fn scanner_publication_lease_token_for_disk( @@ -3876,6 +3979,8 @@ impl SetDisks { let mut data_dirs = vec![None; disk_count]; let mut cleanup_data_dirs = vec![None; disk_count]; let mut old_current_sizes = vec![None; disk_count]; + let mut capacity_scope_generation = None; + let mut cleanup_at_snapshot = vec![false; disk_count]; while let Some(joined) = tasks.join_next().await { results_seen += 1; @@ -3910,6 +4015,10 @@ impl SetDisks { &old_current_sizes, write_quorum, ); + if let Ok(commit) = snapshot_commit.as_ref() { + capacity_scope_generation = Some(current_dirty_generation()); + cleanup_at_snapshot = commit.cleanup_disks.iter().map(Option::is_some).collect(); + } if let Some(commit_tx) = commit_tx.take() { let _ = commit_tx.send(snapshot_commit); } @@ -3965,9 +4074,29 @@ impl SetDisks { if let Some(commit_tx) = commit_tx.take() { let _ = commit_tx.send(Err(ret_err)); } - return; + return None; } + let final_convergence = Self::classify_rename_convergence(&disk_versions, &errs); + let final_data_dir = Self::reduce_common_data_dir(&cleanup_data_dirs, write_quorum); + let tail_cleanup = coordinator_disks + .iter() + .zip(errs.iter()) + .zip(cleanup_data_dirs.iter()) + .zip(cleanup_at_snapshot.iter()) + .enumerate() + .filter_map(|(disk_index, (((disk, err), old_data_dir), cleanup_at_snapshot))| { + if *cleanup_at_snapshot || err.is_some() || *old_data_dir != final_data_dir { + return None; + } + Some(RenameTailCleanup { + disk_index, + disk: disk.clone()?, + old_data_dir: (*old_data_dir)?, + }) + }) + .collect(); + let mut backup_reclaims = Vec::new(); for (idx, disk) in coordinator_disks.iter().enumerate() { if errs[idx].is_some() { @@ -4011,6 +4140,22 @@ impl SetDisks { } } } + + // The caller marks every candidate disk before returning the + // quorum ACK. If a refresh drains that mark while the tail is + // still mutating disks, re-mark after the tail is complete so + // the next refresh cannot miss those capacity changes. + if capacity_scope_generation.is_some_and(|generation| current_dirty_generation() != generation) { + let scope = capacity_scope_from_disks(&coordinator_disks); + if !scope.disks.is_empty() { + let _ = record_global_dirty_scope(scope); + } + } + + Some(RenameTailOutcome { + convergence: final_convergence, + cleanup: tail_cleanup, + }) } }); @@ -4029,20 +4174,18 @@ impl SetDisks { #[tracing::instrument(level = "debug", skip(disks, file_infos))] pub(in crate::set_disk) async fn rename_data_owned( disks: &[Option], - src_bucket: &str, - src_object: &str, + src: (&str, &str), file_infos: Vec, - dst_bucket: &str, - dst_object: &str, + dst: (&str, &str), write_quorum: usize, + allow_early_ack: bool, ) -> disk::error::Result { Self::rename_data_owned_with_fence( disks, - src_bucket, - src_object, + src, file_infos, - dst_bucket, - dst_object, + dst, + allow_early_ack, RenameDataFenceOptions::new(write_quorum, None), ) .await @@ -4051,14 +4194,15 @@ impl SetDisks { #[tracing::instrument(level = "debug", skip(disks, file_infos, fence_options))] pub(in crate::set_disk) async fn rename_data_owned_with_fence( disks: &[Option], - src_bucket: &str, - src_object: &str, + src: (&str, &str), file_infos: Vec, - dst_bucket: &str, - dst_object: &str, + dst: (&str, &str), + allow_early_ack: bool, fence_options: RenameDataFenceOptions<'_>, ) -> disk::error::Result { - if put_rename_early_ack_enabled() { + let (src_bucket, src_object) = src; + let (dst_bucket, dst_object) = dst; + if allow_early_ack && put_rename_early_ack_enabled() { return Self::rename_data_owned_early_ack_with_fence( disks, src_bucket, @@ -4423,6 +4567,7 @@ impl SetDisks { }; Ok(RenameDataCommit { + capacity_disks: online_disks.clone(), online_disks, convergence, data_dir, @@ -4576,6 +4721,11 @@ impl SetDisks { let disk = disk.clone(); let object_for_fault = object_for_fault.clone(); tokio::spawn(async move { + let Some(disk) = disk else { + // `None` slot: ignored placeholder. It is not `attempted`, so + // classification excludes it from residue regardless. + return (false, Some(DiskError::DiskNotFound)); + }; // Test-only introspection guard + awaitable pause point for the // old-data-dir cleanup fan-out. Both compile away in production. #[allow(clippy::let_unit_value)] @@ -4585,26 +4735,20 @@ impl SetDisks { if let Some(err) = Self::cleanup_injected_error(&object_for_fault, idx) { return (false, Some(err)); } - if let Some(disk) = disk { - match disk - .delete_data_dir( - &bucket, - &file_path, - DeleteOptions { - recursive: true, - ..Default::default() - }, - ) - .await - { - Ok(DataDirDeleteStatus::Deleted) => (false, None), - Ok(DataDirDeleteStatus::Deferred) => (true, None), - Err(err) => (false, Some(err)), - } - } else { - // `None` slot: ignored placeholder. It is not `attempted`, so - // classification excludes it from residue regardless. - (false, Some(DiskError::DiskNotFound)) + match disk + .delete_data_dir( + &bucket, + &file_path, + DeleteOptions { + recursive: true, + ..Default::default() + }, + ) + .await + { + Ok(DataDirDeleteStatus::Deleted) => (false, None), + Ok(DataDirDeleteStatus::Deferred) => (true, None), + Err(err) => (false, Some(err)), } }) }); @@ -4627,6 +4771,22 @@ impl SetDisks { cleanup } + pub(in crate::set_disk) async fn commit_rename_data_dir_and_mark_capacity( + &self, + disks: &[Option], + bucket: &str, + object: &str, + old_data_dir: &str, + committed_data_dir: &str, + write_quorum: usize, + ) -> OldDataDirCleanup { + let cleanup = self + .commit_rename_data_dir(disks, bucket, object, old_data_dir, committed_data_dir, write_quorum) + .await; + self.record_capacity_scope_if_needed(None, disks); + cleanup + } + /// Test-only fault-injection seam for the old-data-dir cleanup path /// (backlog#898 §5). In production this is inlined to `None` and adds no /// behavior; only the `#[cfg(test)]` variant consults the fault registry. @@ -4823,6 +4983,8 @@ impl SetDisks { } } + self.record_capacity_scope_if_needed(None, &disks); + if errs.iter().any(|e| e.is_some()) { warn!("cleanup_multipart_path errs {:?}", &errs); } @@ -8651,6 +8813,54 @@ mod tests { /// hang into a deterministic failure instead of an infinite wait. const BARRIER_PAUSE_GUARD: std::time::Duration = std::time::Duration::from_secs(10); + #[tokio::test] + async fn rename_tail_join_error_fails_safe_to_heal() { + let submissions = Arc::new(std::sync::Mutex::new(Vec::new())); + let captured = Arc::clone(&submissions); + let cleanup_called = Arc::new(std::sync::Mutex::new(false)); + let cleanup_captured = Arc::clone(&cleanup_called); + let finalize_called = Arc::new(std::sync::Mutex::new(false)); + let finalize_captured = Arc::clone(&finalize_called); + let tail_drain = tokio::spawn(async { + panic!("injected rename tail join failure"); + #[allow(unreachable_code)] + None + }); + let (release, released) = tokio::sync::oneshot::channel::(); + drop(release); + + finish_rename_tail_heal( + tail_drain, + released, + (), + rustfs_common::heal_channel::HealChannelRequest::default(), + move || async move { + *finalize_captured.lock().expect("finalize recorder should not poison") = true; + }, + move |(), _| async move { + *cleanup_captured.lock().expect("cleanup recorder should not poison") = true; + }, + |request| async move { + captured.lock().expect("submission recorder should not poison").push(request); + }, + ) + .await; + + assert_eq!( + submissions.lock().expect("submission recorder should not poison").len(), + 1, + "a tail JoinError must fail safe to heal" + ); + assert!( + !*cleanup_called.lock().expect("cleanup recorder should not poison"), + "a tail JoinError must preserve staging that an in-flight disk task may still need" + ); + assert!( + *finalize_called.lock().expect("finalize recorder should not poison"), + "a completed tail JoinError must release resources needed only by disk rename tasks" + ); + } + fn rename_barrier_fileinfos(object: &str, count: usize) -> Vec { (0..count).map(|_| metadata_test_fileinfo(object)).collect() } @@ -8774,6 +8984,7 @@ mod tests { /// reclaimed — leftover residue keeps DeleteBucket failing with /// BucketNotEmpty long after the object itself is deleted. #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_reclaims_synthetic_inline_rollback_dir_after_commit() { let bucket = "rename-inline-rollback-bucket"; let object = "object"; @@ -8796,17 +9007,18 @@ mod tests { // Use rename_data_owned so we can await the tail_drain for cleanup. let commit = SetDisks::rename_data_owned( std::slice::from_ref(&online_disk), - RUSTFS_META_TMP_BUCKET, - "tmp-inline", + (RUSTFS_META_TMP_BUCKET, "tmp-inline"), vec![inline_fi], - bucket, - object, + (bucket, object), 1, + true, ) .await .expect("inline version should commit"); if let Some(td) = commit.tail_drain { - td.await.expect("inline commit tail drain must succeed"); + td.await + .expect("inline commit tail drain must succeed") + .expect("committed inline tail must report convergence"); } // Overwrite the same (nil) version with a non-inline one. @@ -8822,17 +9034,18 @@ mod tests { std::fs::write(staged_data_dir.join("part.1"), b"streamed-body").expect("staged part should be written"); let commit = SetDisks::rename_data_owned( std::slice::from_ref(&online_disk), - RUSTFS_META_TMP_BUCKET, - "tmp-streaming", + (RUSTFS_META_TMP_BUCKET, "tmp-streaming"), vec![streaming_fi], - bucket, - object, + (bucket, object), 1, + true, ) .await .expect("non-inline overwrite should commit"); if let Some(td) = commit.tail_drain { - td.await.expect("non-inline overwrite tail drain must succeed"); + td.await + .expect("non-inline overwrite tail drain must succeed") + .expect("committed non-inline tail must report convergence"); } let mut leftovers: Vec = std::fs::read_dir(disk_root.join(bucket).join(object)) @@ -8946,7 +9159,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_fanout_drains_after_caller_cancellation() { const DISKS: usize = 4; let bucket = "rename-cancel-bucket"; @@ -8992,11 +9205,11 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] - async fn rename_data_waits_for_tail_disk_after_write_quorum() { - // Explicitly test the serial (join_all) path: early ack is now the - // default, so disable it to verify the legacy behaviour still works. - temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("false"))], async { + #[serial_test::serial(capacity_dirty_scope)] + async fn rename_data_without_tail_handoff_waits_after_write_quorum() { + // The environment enables early ACK, but this legacy/no-owner entry + // point cannot transfer a namespace guard and must stay serial. + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { const DISKS: usize = 4; let bucket = "rename-tail-success-bucket"; let object = "rename-tail-success-object"; @@ -9045,7 +9258,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_early_ack_returns_after_write_quorum_and_drains_tail_success() { temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { const DISKS: usize = 4; @@ -9057,40 +9270,54 @@ mod tests { let tracker = rename_fanout_barrier::observe_tasks(object); let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); - let mut rename = Box::pin(SetDisks::rename_data( + let mut rename = Box::pin(SetDisks::rename_data_owned( &disks, - RUSTFS_META_TMP_BUCKET, - "source", - &file_infos, - bucket, - object, + (RUSTFS_META_TMP_BUCKET, "source"), + file_infos, + (bucket, object), 3, + true, )); tokio::time::timeout(BARRIER_PAUSE_GUARD, async { tokio::select! { () = barrier.wait_until_paused() => {} - result = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier: {result:?}"), + _ = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier"), } }) .await .expect("paused disk must reach the armed rename barrier"); - rename + let commit = rename .await .expect("early ACK must return once the other three disks satisfy write quorum"); assert!( tracker.running() >= 1, "the paused tail disk must continue in the background after early ACK" ); + assert_eq!( + commit.convergence, + RenameConvergence::PartialCommit, + "the quorum snapshot sees the unresolved tail as missing" + ); + assert!( + !commit.needs_immediate_heal(), + "a pending tail must defer heal admission until its final observations" + ); + let tail_drain = commit.tail_drain.expect("early ACK must return a tail drain"); barrier.release(); - tokio::time::timeout(BARRIER_PAUSE_GUARD, async { - while tracker.running() != 0 { - tokio::task::yield_now().await; - } - }) - .await - .expect("background tail disk must drain after release"); + let tail_outcome = tokio::time::timeout(BARRIER_PAUSE_GUARD, tail_drain) + .await + .expect("background tail disk must drain after release") + .expect("background tail task must join") + .expect("a committed early-ACK tail must return final convergence"); + let final_convergence = tail_outcome.convergence; + assert_eq!( + final_convergence, + RenameConvergence::AllSuccessIdentical, + "a successful tail must clear the quorum snapshot's false partial commit" + ); + assert!(!final_convergence.needs_heal(), "a successful tail must not enqueue heal after release"); for (idx, dir) in dirs.iter().enumerate() { let reopened = reopen_local_disk(dir).await; @@ -9109,7 +9336,97 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] + async fn rename_data_early_ack_tail_recovers_final_old_data_dir_quorum() { + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + const DISKS: usize = 4; + let bucket = "rename-early-ack-old-dir-quorum-bucket"; + let object = "rename-early-ack-old-dir-quorum-object"; + let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; + prepare_rename_source_dirs(&dirs, &disks, "source").await; + let minority_old_dir = Uuid::new_v4(); + let majority_old_dir = Uuid::new_v4(); + let old_dirs = [minority_old_dir, majority_old_dir, majority_old_dir, majority_old_dir]; + let old_mod_time = OffsetDateTime::now_utc(); + for (idx, disk) in disks.iter().flatten().enumerate() { + let mut old = metadata_test_fileinfo(object); + old.data_dir = Some(old_dirs[idx]); + old.mod_time = Some(old_mod_time); + disk.write_all(bucket, &format!("{object}/{}/part.1", old_dirs[idx]), Bytes::from_static(b"old-body")) + .await + .expect("old data shard should be written"); + disk.write_metadata(bucket, bucket, object, old) + .await + .expect("old metadata should be written"); + } + + let file_infos = rename_commit_fileinfos(object, DISKS, "early-old-dir-quorum-etag"); + let barrier = rename_fanout_barrier::arm(object, 3, rename_fanout_barrier::PHASE_RENAME); + let mut rename = Box::pin(SetDisks::rename_data_owned( + &disks, + (RUSTFS_META_TMP_BUCKET, "source"), + file_infos, + (bucket, object), + 3, + true, + )); + tokio::time::timeout(BARRIER_PAUSE_GUARD, async { + tokio::select! { + () = barrier.wait_until_paused() => {} + _ = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier"), + } + }) + .await + .expect("the final majority-vote disk must pause before rename"); + + let commit = rename.await.expect("the first three disks must satisfy write quorum"); + assert_eq!(commit.data_dir, None, "two matching old dirs are not a write quorum"); + assert!( + commit.cleanup_disks.iter().all(Option::is_none), + "the quorum snapshot must not schedule cleanup without an old-dir quorum" + ); + let tail_drain = commit.tail_drain.expect("early ACK must return a tail drain"); + + barrier.release(); + let outcome = tokio::time::timeout(BARRIER_PAUSE_GUARD, tail_drain) + .await + .expect("the paused majority-vote disk must drain after release") + .expect("background tail task must join") + .expect("a committed early-ACK tail must return final observations"); + let cleanup_indexes: Vec<_> = outcome.cleanup.iter().map(|target| target.disk_index).collect(); + assert_eq!( + cleanup_indexes, + vec![1, 2, 3], + "the final old-dir quorum must recover both snapshot and tail cleanup targets" + ); + + let set = io_primitives_test_set(disks, 2).await; + set.cleanup_rename_tail(outcome.cleanup, bucket, object, None, None).await; + assert!( + dirs[0] + .path() + .join(bucket) + .join(object) + .join(minority_old_dir.to_string()) + .exists(), + "the minority old data dir must not be reclaimed" + ); + for dir in &dirs[1..] { + assert!( + !dir.path() + .join(bucket) + .join(object) + .join(majority_old_dir.to_string()) + .exists(), + "every disk in the final old-dir quorum must be reclaimed" + ); + } + }) + .await; + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_early_ack_tail_failure_does_not_expose_partial_fresh_after_reopen() { temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { const DISKS: usize = 4; @@ -9118,39 +9435,47 @@ mod tests { let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await; prepare_rename_source_dirs(&dirs, &disks, "source").await; let file_infos = rename_commit_fileinfos(object, DISKS, "early-tail-failure-etag"); - let tracker = rename_fanout_barrier::observe_tasks(object); let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); let _fault = rename_fault_injection::fail_rename_on(object, &[0]); - let mut rename = Box::pin(SetDisks::rename_data( + let mut rename = Box::pin(SetDisks::rename_data_owned( &disks, - RUSTFS_META_TMP_BUCKET, - "source", - &file_infos, - bucket, - object, + (RUSTFS_META_TMP_BUCKET, "source"), + file_infos, + (bucket, object), 3, + true, )); tokio::time::timeout(BARRIER_PAUSE_GUARD, async { tokio::select! { () = barrier.wait_until_paused() => {} - result = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier: {result:?}"), + _ = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier"), } }) .await .expect("paused disk must reach the armed rename barrier"); - rename + let commit = rename .await .expect("early ACK must return after write quorum before the tail failure"); + assert!( + !commit.needs_immediate_heal(), + "a pending failed tail must still defer heal until the failure is observed" + ); + let tail_drain = commit.tail_drain.expect("early ACK must return a tail drain"); barrier.release(); - tokio::time::timeout(BARRIER_PAUSE_GUARD, async { - while tracker.running() != 0 { - tokio::task::yield_now().await; - } - }) - .await - .expect("background tail failure must drain after release"); + let tail_outcome = tokio::time::timeout(BARRIER_PAUSE_GUARD, tail_drain) + .await + .expect("background tail failure must drain after release") + .expect("background tail task must join") + .expect("a committed early-ACK tail must return final convergence"); + let final_convergence = tail_outcome.convergence; + assert_eq!( + final_convergence, + RenameConvergence::PartialCommit, + "the observed tail failure must require heal" + ); + assert!(final_convergence.needs_heal(), "a real tail failure must enqueue heal after drain"); for (idx, dir) in dirs.iter().enumerate() { let reopened = reopen_local_disk(dir).await; @@ -9176,7 +9501,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_early_ack_tail_failure_preserves_overwrite_after_reopen() { temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { const DISKS: usize = 4; @@ -9199,19 +9524,18 @@ mod tests { let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); let _fault = rename_fault_injection::fail_rename_on(object, &[0]); - let mut rename = Box::pin(SetDisks::rename_data( + let mut rename = Box::pin(SetDisks::rename_data_owned( &disks, - RUSTFS_META_TMP_BUCKET, - "source", - &file_infos, - bucket, - object, + (RUSTFS_META_TMP_BUCKET, "source"), + file_infos, + (bucket, object), 3, + true, )); tokio::time::timeout(BARRIER_PAUSE_GUARD, async { tokio::select! { () = barrier.wait_until_paused() => {} - result = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier: {result:?}"), + _ = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier"), } }) .await @@ -9247,7 +9571,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_early_ack_background_drains_after_caller_cancellation() { temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { const DISKS: usize = 4; @@ -9259,19 +9583,18 @@ mod tests { let tracker = rename_fanout_barrier::observe_tasks(object); let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); - let mut rename = Box::pin(SetDisks::rename_data( + let mut rename = Box::pin(SetDisks::rename_data_owned( &disks, - RUSTFS_META_TMP_BUCKET, - "source", - &file_infos, - bucket, - object, + (RUSTFS_META_TMP_BUCKET, "source"), + file_infos, + (bucket, object), 3, + true, )); tokio::time::timeout(BARRIER_PAUSE_GUARD, async { tokio::select! { () = barrier.wait_until_paused() => {} - result = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier: {result:?}"), + _ = rename.as_mut() => panic!("rename_data returned before the armed fan-out barrier"), } }) .await @@ -9304,7 +9627,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_early_ack_strict_quorum_failure_rolls_back_fresh_after_reopen() { temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { const DISKS: usize = 4; @@ -9315,9 +9638,12 @@ mod tests { let file_infos = rename_commit_fileinfos(object, DISKS, "early-strict-rollback-etag"); let _fault = rename_fault_injection::fail_rename_on(object, &[0]); - SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 4) - .await - .expect_err("three successful disks must fail an early-ACK strict write quorum of four"); + assert!( + SetDisks::rename_data_owned(&disks, (RUSTFS_META_TMP_BUCKET, "source"), file_infos, (bucket, object), 4, true,) + .await + .is_err(), + "three successful disks must fail an early-ACK strict write quorum of four" + ); for (idx, dir) in dirs.iter().enumerate() { let reopened = reopen_local_disk(dir).await; @@ -9332,7 +9658,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_commits_fresh_object_when_tail_disk_fails_after_write_quorum() { const DISKS: usize = 4; let bucket = "rename-tail-failure-fresh-bucket"; @@ -9361,7 +9687,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_overwrite_tail_failure_preserves_old_tail_version_after_reopen() { const DISKS: usize = 4; let bucket = "rename-tail-failure-overwrite-bucket"; @@ -9401,7 +9727,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_strict_quorum_failure_rolls_back_fresh_object_after_reopen() { let _mode = durability_mode_override::set(DurabilityMode::Strict); const DISKS: usize = 4; @@ -9427,7 +9753,7 @@ mod tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn rename_data_strict_quorum_failure_restores_overwrite_after_reopen() { let _mode = durability_mode_override::set(DurabilityMode::Strict); const DISKS: usize = 4; @@ -10067,7 +10393,10 @@ mod tests { } #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] async fn commit_rename_data_dir_reclaims_old_data_dir_and_reports_receipt() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + let bucket = "commit-rename-bucket"; let object = "object"; let old_data_dir = "11111111-1111-1111-1111-111111111111"; @@ -10076,16 +10405,12 @@ mod tests { let (_dir1, disk1) = read_multiple_test_disk(bucket, &[(&path, b"one".as_slice())]).await; let (_dir2, disk2) = read_multiple_test_disk(bucket, &[(&path, b"two".as_slice())]).await; let set = io_primitives_test_set(vec![Some(disk1.clone()), Some(disk2.clone())], 1).await; + let disks = [Some(disk1.clone()), Some(disk2.clone())]; + let expected = capacity_scope_from_disks(&disks).disks.into_iter().collect::>(); + let _ = drain_global_dirty_scopes(); let cleanup = set - .commit_rename_data_dir( - &[Some(disk1.clone()), Some(disk2.clone())], - bucket, - object, - old_data_dir, - committed_data_dir, - 2, - ) + .commit_rename_data_dir_and_mark_capacity(&disks, bucket, object, old_data_dir, committed_data_dir, 2) .await; assert_eq!(cleanup.attempted, 2); assert_eq!(cleanup.reclaimed, 2); @@ -10093,6 +10418,11 @@ mod tests { assert!(matches!(disk1.read_all(bucket, &path).await, Err(DiskError::FileNotFound))); assert!(matches!(disk2.read_all(bucket, &path).await, Err(DiskError::FileNotFound))); + let marked = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&marked), + "old-data cleanup must mark capacity before callers can await residue reporting" + ); let missing = set .commit_rename_data_dir(&[None, None], bucket, object, old_data_dir, committed_data_dir, 1) diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index a24e18bb0..165f8e099 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -620,8 +620,9 @@ impl SetDisks { } } -/// Build an ad-hoc, deduplicated dirty scope from `disks`. Used by the heal -/// path where disks are not in physical-slot order (backlog#1315). +/// Build an ad-hoc, deduplicated dirty scope from `disks`. Used where the +/// per-set generation fast path is intentionally bypassed: heal rewrites and +/// an early-ACK tail whose first mark was drained before it completed. fn capacity_scope_from_disks(disks: &[Option]) -> CapacityScope { let mut unique = HashSet::with_capacity(disks.len()); let mut scoped_disks = Vec::with_capacity(disks.len()); @@ -3109,6 +3110,9 @@ pub struct SetDisks { capacity_dirty_generation: Arc, #[cfg(test)] storage_class_config_override: Arc>>>, + #[cfg(test)] + rename_tail_heal_capture: + Arc>>>, } // DistributedLock sends the raw ObjectKey to its clients; LockRegistry clones @@ -3384,6 +3388,37 @@ impl DiskHealthEntry { } impl SetDisks { + pub(in crate::set_disk) async fn submit_rename_tail_heal(&self, request: rustfs_common::heal_channel::HealChannelRequest) { + #[cfg(test)] + { + let capture = self + .rename_tail_heal_capture + .lock() + .expect("rename tail heal capture mutex should not poison") + .clone(); + if let Some(capture) = capture { + let _ = capture.send(request); + return; + } + } + + let _ = rustfs_common::heal_channel::send_heal_request(request).await; + } + + #[cfg(test)] + pub(in crate::set_disk) fn capture_test_rename_tail_heals( + &self, + ) -> tokio::sync::mpsc::UnboundedReceiver { + let (capture, requests) = tokio::sync::mpsc::unbounded_channel(); + let mut slot = self + .rename_tail_heal_capture + .lock() + .expect("rename tail heal capture mutex should not poison"); + assert!(slot.is_none(), "only one rename tail heal capture may be installed per set"); + *slot = Some(capture); + requests + } + fn storage_class_config_snapshot(&self) -> Arc { #[cfg(test)] if let Some(config) = self @@ -3687,6 +3722,8 @@ impl SetDisks { capacity_dirty_generation: Arc::new(AtomicU64::new(u64::MAX)), #[cfg(test)] storage_class_config_override: Arc::new(std::sync::RwLock::new(None)), + #[cfg(test)] + rename_tail_heal_capture: Arc::new(std::sync::Mutex::new(None)), }) } @@ -7594,7 +7631,7 @@ mod tests { // the ad-hoc scope the previous per-write construction produced, otherwise // dirty-disk keys diverge from the disk-cache keys and capacity counts drift. #[tokio::test] - #[serial] + #[serial(capacity_dirty_scope)] async fn capacity_scope_memo_matches_adhoc_and_is_reused() { use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; @@ -7620,7 +7657,7 @@ mod tests { // write of each generation; steady-state writes skip it. Reverting the // generation skip makes the upgrade count grow per write and fails this test. #[tokio::test] - #[serial] + #[serial(capacity_dirty_scope)] async fn record_capacity_scope_upgrades_registry_once_per_generation() { use rustfs_object_capacity::capacity_scope::{drain_global_dirty_scopes, global_dirty_upgrade_count}; @@ -7669,7 +7706,7 @@ mod tests { // backlog#1315: an offline slot must not force the per-write slow path, and // the resolved scope must still cover every online disk. #[tokio::test] - #[serial] + #[serial(capacity_dirty_scope)] async fn capacity_scope_tolerates_offline_slot_without_reallocating() { use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; diff --git a/crates/ecstore/src/set_disk/ops/list.rs b/crates/ecstore/src/set_disk/ops/list.rs index d6786ddd5..59dc1401a 100644 --- a/crates/ecstore/src/set_disk/ops/list.rs +++ b/crates/ecstore/src/set_disk/ops/list.rs @@ -27,13 +27,19 @@ use super::super::*; impl SetDisks { #[tracing::instrument(skip(self))] pub async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> { - ListOperations::new(self.ctx()).delete_all(bucket, prefix).await + let (result, disks) = ListOperations::new(self.ctx()) + .delete_all_observed(bucket, prefix, None) + .await; + self.record_capacity_scope_if_needed(None, &disks); + result } pub(crate) async fn delete_all_with_quorum(&self, bucket: &str, prefix: &str, write_quorum: usize) -> Result<()> { - ListOperations::new(self.ctx()) - .delete_all_with_quorum(bucket, prefix, write_quorum) - .await + let (result, disks) = ListOperations::new(self.ctx()) + .delete_all_observed(bucket, prefix, Some(write_quorum)) + .await; + self.record_capacity_scope_if_needed(None, &disks); + result } } @@ -53,19 +59,24 @@ impl<'a> ListOperations<'a> { Self { ctx } } - pub(crate) async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> { - self.delete_all_inner(bucket, prefix, None).await + pub(crate) async fn delete_all_observed( + &self, + bucket: &str, + prefix: &str, + write_quorum: Option, + ) -> (Result<()>, Vec>) { + let disks = self.ctx.disks().read().await.clone(); + let result = self.delete_all_inner(bucket, prefix, write_quorum, disks.clone()).await; + (result, disks) } - async fn delete_all_with_quorum(&self, bucket: &str, prefix: &str, write_quorum: usize) -> Result<()> { - self.delete_all_inner(bucket, prefix, Some(write_quorum)).await - } - - async fn delete_all_inner(&self, bucket: &str, prefix: &str, write_quorum: Option) -> Result<()> { - let disks = self.ctx.disks().read().await; - - let disks = disks.clone(); - + async fn delete_all_inner( + &self, + bucket: &str, + prefix: &str, + write_quorum: Option, + disks: Vec>, + ) -> Result<()> { let mut futures = Vec::with_capacity(disks.len()); let mut errors = Vec::with_capacity(disks.len()); diff --git a/crates/ecstore/src/set_disk/ops/locking.rs b/crates/ecstore/src/set_disk/ops/locking.rs index 2c82c9150..be33c0c43 100644 --- a/crates/ecstore/src/set_disk/ops/locking.rs +++ b/crates/ecstore/src/set_disk/ops/locking.rs @@ -922,10 +922,11 @@ mod tests { // The List family runs through the borrow handle with unchanged // behavior: delete_all reports success even when the prefix is absent. - ListOperations::new(set_disks.ctx()) - .delete_all("nonexistent-bucket", "nonexistent-prefix") - .await - .expect("delete_all via borrow handle should succeed"); + let (result, observed_disks) = ListOperations::new(set_disks.ctx()) + .delete_all_observed("nonexistent-bucket", "nonexistent-prefix", None) + .await; + result.expect("delete_all via borrow handle should succeed"); + assert_eq!(observed_disks.len(), disk_count); set_disks .delete_all("nonexistent-bucket", "nonexistent-prefix") .await diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 1aaa00340..871c7afd4 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -30,6 +30,7 @@ use super::object::{ use crate::bucket::quota::reservation; use crate::crash_inject::{self, CrashPoint}; use crate::multipart_listing::paginate_multipart_listing; +use crate::set_disk::core::io_primitives::finish_rename_tail_heal; use futures::{StreamExt, stream}; use std::future::Future; #[cfg(test)] @@ -2621,7 +2622,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let commit_is_versioned = opts.versioned || opts.version_suspended; let commit_capacity_scope_token = opts.capacity_scope_token; let commit_object_lock_guard = object_lock_guard.take(); - let detach_commit_owner = commit_object_lock_guard.is_some() || upload_guard.is_some() || quota_mutation_fence; + let commit_allows_early_ack = commit_object_lock_guard.is_some(); + let detach_commit_owner = commit_allows_early_ack || upload_guard.is_some() || quota_mutation_fence; let commit = async move { let mut _object_lock_guard = commit_object_lock_guard; let mut _upload_guard = upload_guard; @@ -2728,17 +2730,101 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit // `get_object_info` lookup, so the backfill has no consumer here yet. Self::assign_rename_data_indexes(&mut parts_metadatas); - let rename_result = SetDisks::rename_data_owned( + let mut rename_result = SetDisks::rename_data_owned( &commit_disks, - RUSTFS_META_MULTIPART_BUCKET, - &commit_upload_id_path, + (RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path), parts_metadatas, - &commit_bucket, - &commit_object, + (&commit_bucket, &commit_object), write_quorum, + commit_allows_early_ack, ) .await; - if quota_mutation_fence { + let mut rename_guard_release = None; + let mut needs_immediate_heal = false; + let mut tail_owns_staging_cleanup = false; + if let Ok(rename_commit) = rename_result.as_mut() { + commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &rename_commit.capacity_disks); + // Install the tail watcher before any post-commit await. The + // latch keeps namespace guards through their prior handoff point. + needs_immediate_heal = rename_commit.needs_immediate_heal(); + if let Some(rename_tail_drain) = rename_commit.tail_drain.take() { + tail_owns_staging_cleanup = true; + let mut request = rustfs_common::heal_channel::create_heal_request_with_options( + commit_bucket.clone(), + Some(commit_object.clone()), + false, + Some(HealChannelPriority::Normal), + Some(commit_set.pool_index), + Some(commit_set.set_index), + ); + request.object_version_id = fi + .version_id + .or_else(|| commit_version_suspended.then(Uuid::nil)) + .map(|version_id| version_id.to_string()); + let object_lock_guard = _object_lock_guard.take(); + let upload_guard = _upload_guard.take(); + let cleanup_bucket = commit_bucket.clone(); + let cleanup_object = commit_object.clone(); + let heal_set = commit_set.clone(); + let cleanup_set = commit_set.clone(); + let committed_data_dir = fi.data_dir; + let cleanup_parts = parts.clone(); + let cleanup_upload_path = commit_upload_id_path.clone(); + let cleanup_upload_id = commit_upload_id.clone(); + let fence_disks = commit_disks.clone(); + let fence_tokens = quota_fence_tokens.clone(); + let fence_bucket = commit_bucket.clone(); + let fence_object = commit_object.clone(); + let (guard_release_tx, guard_release_rx) = tokio::sync::oneshot::channel(); + rename_guard_release = Some(guard_release_tx); + tokio::spawn(finish_rename_tail_heal( + rename_tail_drain, + guard_release_rx, + (object_lock_guard, upload_guard), + request, + move || async move { + if quota_mutation_fence { + let _ = SetDisks::release_quota_mutation_fences( + &fence_disks, + &fence_tokens, + &fence_bucket, + &fence_object, + write_quorum, + ) + .await; + } + }, + move |(object_lock_guard, upload_guard), targets| async move { + drop(object_lock_guard); + cleanup_set.cleanup_multipart_path(&cleanup_parts).await; + cleanup_set + .cleanup_rename_tail( + targets, + &cleanup_bucket, + &cleanup_object, + committed_data_dir, + transaction_epoch, + ) + .await; + if let Err(err) = cleanup_set + .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_path, write_quorum) + .await + { + warn!( + bucket = %cleanup_bucket, + object = %cleanup_object, + upload_id = %cleanup_upload_id, + error = ?err, + "completed multipart upload staging cleanup did not reach write quorum" + ); + } + drop(upload_guard); + }, + |request| async move { heal_set.submit_rename_tail_heal(request).await }, + )); + } + } + if quota_mutation_fence && !tail_owns_staging_cleanup { let _ = SetDisks::release_quota_mutation_fences( &commit_disks, "a_fence_tokens, @@ -2755,16 +2841,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { Ok(result) => result, Err(err) => return Err(err.into()), }; - let online_disks = rename_commit.online_disks; - let convergence = rename_commit.convergence; let op_old_dir = rename_commit.data_dir; let cleanup_disks = rename_commit.cleanup_disks; let committed_file_info = rename_commit.committed_file_info; - let rename_tail_drain = rename_commit.tail_drain; - // Detach admission before any post-commit await: client cancellation - // must not couple durable convergence repair to cleanup work. - if convergence.needs_heal() { + if needs_immediate_heal { let mut request = rustfs_common::heal_channel::create_heal_request_with_options( commit_bucket.clone(), Some(commit_object.clone()), @@ -2802,14 +2883,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // parts are swept by a retried completion or upload GC (rustfs/backlog#946). // Compiles to a no-op outside `#[cfg(test)]`. if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, &commit_object) { + if let Some(release) = rename_guard_release.take() { + let _ = release.send(false); + } return Err(StorageError::Unexpected); } fi = committed_file_info; let committed_dir = fi.data_dir.unwrap_or_default().to_string(); - commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks); - fi.is_latest = true; #[cfg(test)] @@ -2819,30 +2901,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object) .await; - if let Some(rename_tail_drain) = rename_tail_drain { - let object_lock_guard = _object_lock_guard.take(); - let upload_guard = _upload_guard.take(); - let tail_bucket = commit_bucket.clone(); - let tail_object = commit_object.clone(); - tokio::spawn(async move { - let _object_lock_guard = object_lock_guard; - let _upload_guard = upload_guard; - if let Err(err) = rename_tail_drain.await { - warn!( - event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_SET_DISK, - state = "failed", - bucket = %tail_bucket, - object = %tail_object, - error = %err, - "rename tail drain failed" - ); - } - }); - } else { - drop(_object_lock_guard.take()); // release the object lock before multipart cleanup tail IO. + if let Some(release) = rename_guard_release.take() { + let _ = release.send(true); } + drop(_object_lock_guard.take()); // release the object lock before multipart cleanup tail IO. #[cfg(test)] pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterObjectPublication).await; @@ -2855,14 +2917,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // parts; deleting them before the commit would strand the upload // permanently. This mirrors the "clean up only after commit" pattern // already used for the old data-dir GC and the upload-dir delete_all below. - commit_set.cleanup_multipart_path(&parts).await; + if !tail_owns_staging_cleanup { + commit_set.cleanup_multipart_path(&parts).await; + } if let Some(old_dir) = op_old_dir { // backlog#898: best-effort reclaim of the dereferenced old data dir. // Returns a receipt (never `Err`); a failed GC must not turn an // already-committed multipart completion into a 503. let cleanup = commit_set - .commit_rename_data_dir( + .commit_rename_data_dir_and_mark_capacity( &cleanup_disks, &commit_bucket, &commit_object, @@ -2886,9 +2950,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { #[cfg(test)] pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterRename).await; - if let Err(err) = commit_set - .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum) - .await + if !tail_owns_staging_cleanup + && let Err(err) = commit_set + .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum) + .await { warn!( bucket = %commit_bucket, @@ -2937,6 +3002,7 @@ mod tests { use crate::disk::{endpoint::Endpoint, format::FormatV3}; use crate::layout::endpoints::SetupType; use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test; + use crate::set_disk::core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier}; // No-locker helpers resolve to the isolated-context variants (see // `hermetic_set_disks_isolated`); the guard-based tests build through // `hermetic_set_disks_with_lockers`, which stays on the bootstrap context @@ -2944,6 +3010,7 @@ mod tests { use crate::set_disk::ops::object::hermetic_set_disks_support::{ hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity, hermetic_set_disks_isolated as hermetic_set_disks, hermetic_set_disks_with_lockers, + hermetic_set_disks_with_lockers_and_ctx, }; use crate::set_disk::ops::object::{PutObjectCommitBarrier, PutObjectCommitPause}; use crate::storage_api_contracts::namespace::NamespaceLocking as _; @@ -2951,7 +3018,11 @@ mod tests { use rustfs_config::server_config::KVS; use rustfs_lock::{LockClient, client::local::LocalClient}; use serial_test::serial; - use std::sync::atomic::{AtomicUsize, Ordering}; + use std::{ + collections::HashSet, + sync::atomic::{AtomicUsize, Ordering}, + time::Duration, + }; use tempfile::TempDir; use tokio::sync::{Notify, RwLock}; @@ -3324,6 +3395,203 @@ mod tests { ) } + #[tokio::test] + #[serial(capacity_dirty_scope)] + async fn recovered_disk_multipart_cleanup_marks_each_snapshot_it_mutates() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let recovered = disk_stores[0].clone(); + let part_path = "recovered-multipart/part.1.meta"; + recovered + .write_all(RUSTFS_META_MULTIPART_BUCKET, part_path, Bytes::from_static(b"stale part metadata")) + .await + .expect("recovered disk should contain staged part metadata"); + + set_disks.disks.write().await[0] = None; + let stale_snapshot = set_disks.get_disks_internal().await; + assert!(stale_snapshot[0].is_none(), "the pre-cleanup snapshot must exclude the disk"); + set_disks.disks.write().await[0] = Some(recovered.clone()); + let expected = capacity_scope_from_disks(&[Some(recovered.clone())]) + .disks + .into_iter() + .collect::>(); + let _ = drain_global_dirty_scopes(); + + set_disks.cleanup_multipart_path(&[part_path.to_string()]).await; + assert!(matches!( + recovered.read_all(RUSTFS_META_MULTIPART_BUCKET, part_path).await, + Err(DiskError::FileNotFound) + )); + let part_marked = drain_global_dirty_scopes().into_iter().collect::>(); + assert!(expected.is_subset(&part_marked), "part cleanup must mark the recovered disk it mutated"); + + let upload_path = "recovered-multipart/upload/part.1"; + recovered + .write_all(RUSTFS_META_MULTIPART_BUCKET, upload_path, Bytes::from_static(b"stale upload shard")) + .await + .expect("recovered disk should contain staged upload data"); + set_disks + .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, "recovered-multipart/upload", 3) + .await + .expect("upload cleanup should use the recovered disk"); + assert!(matches!( + recovered.read_all(RUSTFS_META_MULTIPART_BUCKET, upload_path).await, + Err(DiskError::FileNotFound) + )); + let upload_marked = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&upload_marked), + "upload cleanup must mark the recovered disk it actually mutated" + ); + } + + #[tokio::test] + #[serial(capacity_dirty_scope)] + async fn early_ack_multipart_holds_quota_fences_and_re_marks_capacity_after_tail_drain() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager)))); + let lockers: Vec> = vec![signaling.clone()]; + let instance_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + instance_ctx.update_erasure_type(SetupType::DistErasure).await; + let (_temp_dirs, disk_stores, set_disks) = + hermetic_set_disks_with_lockers_and_ctx(4, 0, 2, lockers, instance_ctx).await; + let bucket = "multipart-early-ack-capacity-scope"; + let object = "multipart-early-ack-capacity-scope-object"; + make_bucket_on_all(&disk_stores, bucket).await; + let mut initial = PutObjReader::from_vec(vec![b'0'; 1 << 20]); + set_disks + .put_object(bucket, object, &mut initial, &ObjectOptions::default()) + .await + .expect("initial object should commit before the multipart overwrite"); + drop( + set_disks + .acquire_write_lock_diag("multipart_capacity_fixture_ready", bucket, object) + .await + .expect("initial object tail should drain before the multipart overwrite"), + ); + let multipart_body = vec![b'1'; 1 << 20]; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, &multipart_body, &ObjectOptions::default()).await; + let (staged, _) = set_disks + .check_upload_id_exists(bucket, object, &upload_id, true) + .await + .expect("staged upload metadata should be readable"); + let competing_upload_id = upload_id.clone(); + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + let staged_part = format!( + "{upload_id_path}/{}/part.1", + staged.data_dir.expect("staged multipart data dir should exist") + ); + signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path.clone())); + let candidate_disks = disk_stores.iter().cloned().map(Some).collect::>(); + let expected = capacity_scope_from_disks(&candidate_disks) + .disks + .into_iter() + .collect::>(); + let _ = drain_global_dirty_scopes(); + + let rename_tasks = rename_fanout_barrier::observe_tasks(object); + let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let complete_store = Arc::clone(&set_disks); + let complete = tokio::spawn(async move { + let mut opts = ObjectOptions::default(); + assert!(opts.set_quota_admission(0, u64::MAX)); + complete_store + .complete_multipart_upload(bucket, object, &upload_id, parts, &opts) + .await + }); + tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused()) + .await + .expect("multipart completion should pause one tail disk during rename"); + let cleanup_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_CLEANUP); + complete + .await + .expect("early-ACK multipart task should join before tail release") + .expect("multipart completion should return after write quorum"); + assert!( + rename_tasks.running() >= 1, + "the paused multipart tail disk must remain in flight after quorum ACK" + ); + + let initial = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&initial), + "the multipart quorum ACK must mark every candidate disk dirty" + ); + + let abort_store = Arc::clone(&set_disks); + let abort = tokio::spawn(async move { + abort_store + .abort_multipart_upload(bucket, object, &competing_upload_id, &ObjectOptions::default()) + .await + }); + signaling.wait_for_attempts(2).await; + assert!(!abort.is_finished(), "the detached tail owner must retain the multipart upload guard"); + + let retained_staging = futures::future::join_all( + disk_stores + .iter() + .map(|disk| disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &staged_part)), + ) + .await + .into_iter() + .filter(|result| result.is_ok()) + .count(); + assert_eq!( + retained_staging, 1, + "only the paused tail disk should still retain the multipart rename source" + ); + + signaling.set_target(rustfs_lock::ObjectKey::new(bucket, object)); + let object_attempt = signaling.attempts.load(Ordering::Acquire) + 1; + let probe_store = Arc::clone(&set_disks); + let object_probe = tokio::spawn(async move { + probe_store + .acquire_write_lock_diag("multipart_tail_object_guard_probe", bucket, object) + .await + }); + signaling.wait_for_attempts(object_attempt).await; + assert!(!object_probe.is_finished(), "the detached tail owner must retain the object guard"); + + rename_barrier.release(); + object_probe + .await + .expect("object guard probe should join after the tail releases") + .expect("object guard probe should acquire after the tail releases"); + tokio::time::timeout(Duration::from_secs(30), cleanup_barrier.wait_until_paused()) + .await + .expect("the multipart tail should pause before reclaiming its old body"); + let after_tail = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&after_tail), + "the multipart rename tail must re-mark capacity after the first scope was drained" + ); + cleanup_barrier.release(); + let abort_err = abort + .await + .expect("abort task should join after the tail releases") + .expect_err("the committed upload should no longer exist"); + assert!(matches!(abort_err, StorageError::InvalidUploadID(..))); + + for (disk_index, disk) in disk_stores.iter().enumerate() { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} must claim its quota fence and finish the rename: {err}")); + } + + let after_cleanup = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&after_cleanup), + "the multipart tail cleanup must re-mark capacity after its preceding scope was drained" + ); + }) + .await; + } + async fn put_test_part( set_disks: &Arc, bucket: &str, @@ -7510,13 +7778,30 @@ mod tests { let new = payload(0xC3); let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await; let parts_retry = parts_new.clone(); + let rename_tasks = rename_fanout_barrier::observe_tasks(object); + let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); crash_inject::arm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); let crashed = complete(&set_disks, bucket, object, &u_new, parts_new).await; assert!( matches!(crashed, Err(StorageError::Unexpected)), "the armed post-commit crash point must be the failure that surfaced, got {crashed:?}" ); + assert!(rename_tasks.running() >= 1, "the crash must interrupt an actual early-ACK tail handoff"); crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); + rename_barrier.release(); + tokio::time::timeout(Duration::from_secs(30), async { + while rename_tasks.running() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the crash-interrupted rename tail should drain after release"); + drop( + set_disks + .acquire_write_lock_diag("post_commit_crash_tail_probe", bucket, object) + .await + .expect("the crash-interrupted tail should release its object guard"), + ); // The commit landed: the new version reads back whole and correct. let (body, _etag) = read_object(&set_disks, bucket, object).await; @@ -7560,7 +7845,7 @@ mod tests { #[tokio::test] #[serial(storage_class_env)] - async fn post_commit_crash_receipt_reclaims_old_data_after_restart() { + async fn post_commit_crash_reclaims_old_data_after_restart() { let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "multipart-crash-old-data-receipt"; @@ -7578,30 +7863,55 @@ mod tests { complete(&set_disks, bucket, object, &u_old, parts_old) .await .expect("the old version should commit"); + drop( + set_disks + .acquire_write_lock_diag("post_commit_receipt_fixture_ready", bucket, object) + .await + .expect("the initial multipart tail should drain before inspecting one disk"), + ); let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; let new = payload(0x52); let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await; + let rename_tasks = rename_fanout_barrier::observe_tasks(object); + let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); crash_inject::arm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); let crashed = complete(&set_disks, bucket, object, &u_new, parts_new).await; assert!( matches!(crashed, Err(StorageError::Unexpected)), "the post-commit crash point must surface as unexpected, got {crashed:?}" ); + assert!(rename_tasks.running() >= 1, "the crash must interrupt an actual early-ACK tail handoff"); crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); + rename_barrier.release(); + tokio::time::timeout(Duration::from_secs(30), async { + while rename_tasks.running() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the crash-interrupted rename tail should drain after release"); + drop( + set_disks + .acquire_write_lock_diag("post_commit_receipt_tail_probe", bucket, object) + .await + .expect("the crash-interrupted tail should release its object guard"), + ); let (body, _) = read_object(&set_disks, bucket, object).await; assert_eq!(body, new, "the committed replacement must remain readable after the crash"); + let mut receipts = 0; for disk in &disk_stores { - assert!( - cleanup_receipt_exists(disk, bucket, object, old_dir).await, - "post-commit crash must leave a durable old-data cleanup receipt" - ); + receipts += usize::from(cleanup_receipt_exists(disk, bucket, object, old_dir).await); assert!( data_dir_exists(disk, bucket, object, old_dir).await, "post-commit crash must leave old data for restart reconciliation" ); } + assert_eq!( + receipts, 3, + "the committed quorum must persist receipts while the crash-interrupted tail preserves staging" + ); let restarted_endpoints = temp_dirs .iter() @@ -7646,7 +7956,12 @@ mod tests { .reconcile_old_data_cleanup_receipts(bucket, object) .await .expect("restart receipt reconciliation should succeed"); - assert_eq!(removed, 4, "restart reconciler should delete all receipt targets"); + assert_eq!(removed, 3, "restart receipt reconciliation should delete the committed quorum's targets"); + let reclaimed = restarted_set + .reclaim_orphan_data_dirs(bucket, object) + .await + .expect("restart orphan reconciliation should succeed"); + assert_eq!(reclaimed, 1, "the late commit without a receipt must remain reclaimable as an orphan"); for disk in &reloaded { assert!( !data_dir_exists(disk, bucket, object, old_dir).await, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 8633c9590..6b06f7c12 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -53,6 +53,7 @@ use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppress use crate::object_api::{NamespaceLockFence, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY}; use crate::services::notification_sys::RemoteVersionStateFleetProofToken; use crate::services::tier::tier::{TierConfigMgr, TierOperationLease}; +use crate::set_disk::core::io_primitives::{RenameTailCleanup, finish_rename_tail_heal}; use crate::store::ECStore; use crate::store::utils::clean_metadata; use futures::FutureExt as _; @@ -2040,6 +2041,34 @@ pub(in crate::set_disk) fn merge_replication_metadata_lww( } impl SetDisks { + pub(in crate::set_disk) async fn cleanup_rename_tail( + &self, + targets: Vec, + bucket: &str, + object: &str, + committed_data_dir: Option, + epoch: Option, + ) { + for target in targets { + let mut disks = vec![None; self.set_drive_count]; + disks[target.disk_index] = Some(target.disk); + self.persist_old_data_cleanup_receipts(&disks, bucket, object, target.old_data_dir, committed_data_dir, epoch) + .await; + let cleanup = self + .commit_rename_data_dir_and_mark_capacity( + &disks, + bucket, + object, + &target.old_data_dir.to_string(), + &committed_data_dir.unwrap_or_default().to_string(), + 1, + ) + .await; + self.report_old_data_dir_cleanup(bucket, object, &target.old_data_dir.to_string(), &cleanup) + .await; + } + } + pub(in crate::set_disk) async fn persist_old_data_cleanup_receipts( &self, disks: &[Option], @@ -2940,8 +2969,8 @@ impl SetDisks { let commit_tmp_dir = tmp_dir.clone(); let commit_object_lock_guard = object_lock_guard.take(); let commit_bucket_lifecycle_guard = bucket_lifecycle_guard.take(); - let detach_commit_owner = - commit_object_lock_guard.is_some() || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence; + let commit_allows_early_ack = commit_object_lock_guard.is_some(); + let detach_commit_owner = commit_allows_early_ack || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence; let commit_write_path_label = write_path.metric_label(); let commit_is_versioned = opts.versioned || opts.version_suspended; let commit_versioned = opts.versioned; @@ -2955,8 +2984,8 @@ impl SetDisks { tmp_cleanup_owned = true; let commit = move |cancellation: Option| async move { - let _object_lock_guard = commit_object_lock_guard; - let _bucket_lifecycle_guard = commit_bucket_lifecycle_guard; + let mut _object_lock_guard = commit_object_lock_guard; + let mut _bucket_lifecycle_guard = commit_bucket_lifecycle_guard; let mut quota_reservation = quota_reservation; let rename_stage_start = Instant::now(); let pre_rename = async { @@ -3071,20 +3100,104 @@ impl SetDisks { } Self::assign_rename_data_indexes(&mut parts_metadatas); - let rename_result = SetDisks::rename_data_owned_with_fence( + let mut rename_result = SetDisks::rename_data_owned_with_fence( &commit_disks, - RUSTFS_META_TMP_BUCKET, - commit_tmp_dir.as_str(), + (RUSTFS_META_TMP_BUCKET, commit_tmp_dir.as_str()), parts_metadatas, - &commit_bucket, - &commit_object, + (&commit_bucket, &commit_object), + commit_allows_early_ack, crate::set_disk::core::io_primitives::RenameDataFenceOptions::new( write_quorum, commit_scanner_publication_lease_tokens.as_ref(), ), ) .await; - if quota_mutation_fence { + #[cfg(any(test, feature = "test-util"))] + if rename_result.is_ok() { + pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::AfterRenameQuorum).await; + } + let mut rename_guard_release = None; + let mut needs_immediate_heal = false; + let mut tail_owns_tmp_cleanup = false; + if let Ok(rename_commit) = rename_result.as_mut() { + commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &rename_commit.capacity_disks); + // Install the tail watcher before any post-commit await. The + // latch keeps namespace guards through their prior handoff point. + needs_immediate_heal = rename_commit.needs_immediate_heal(); + if let Some(rename_tail_drain) = rename_commit.tail_drain.take() { + tail_owns_tmp_cleanup = true; + let mut request = rustfs_common::heal_channel::create_heal_request_with_options( + commit_bucket.clone(), + Some(commit_object.clone()), + false, + Some(HealChannelPriority::Normal), + Some(commit_set.pool_index), + Some(commit_set.set_index), + ); + request.object_version_id = committed_version_id + .or_else(|| commit_version_suspended.then(Uuid::nil)) + .map(|version_id| version_id.to_string()); + let object_lock_guard = _object_lock_guard.take(); + let bucket_lifecycle_guard = _bucket_lifecycle_guard.take(); + let cleanup_bucket = commit_bucket.clone(); + let cleanup_object = commit_object.clone(); + let heal_set = commit_set.clone(); + let cleanup_set = commit_set.clone(); + let cleanup_tmp_dir = commit_tmp_dir.clone(); + let fence_disks = commit_disks.clone(); + let fence_tokens = quota_fence_tokens.clone(); + let fence_bucket = commit_bucket.clone(); + let fence_object = commit_object.clone(); + let (guard_release_tx, guard_release_rx) = tokio::sync::oneshot::channel(); + rename_guard_release = Some(guard_release_tx); + tokio::spawn(finish_rename_tail_heal( + rename_tail_drain, + guard_release_rx, + (object_lock_guard, bucket_lifecycle_guard), + request, + move || async move { + if quota_mutation_fence { + let _ = SetDisks::release_quota_mutation_fences( + &fence_disks, + &fence_tokens, + &fence_bucket, + &fence_object, + write_quorum, + ) + .await; + } + }, + move |(object_lock_guard, bucket_lifecycle_guard), targets| async move { + drop(object_lock_guard); + drop(bucket_lifecycle_guard); + cleanup_set + .cleanup_rename_tail( + targets, + &cleanup_bucket, + &cleanup_object, + committed_data_dir, + transaction_epoch, + ) + .await; + if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await { + warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data"); + } else if issue3031_diag_enabled() { + warn!( + target: "rustfs_ecstore::set_disk", + tmp_dir = %cleanup_tmp_dir, + "issue3031_put_object_tmp_cleanup_done" + ); + } + }, + |request| async move { heal_set.submit_rename_tail_heal(request).await }, + )); + } + } + #[cfg(any(test, feature = "test-util"))] + if rename_result.is_ok() { + pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::AfterRenameHandoff).await; + } + if quota_mutation_fence && !tail_owns_tmp_cleanup { let _ = SetDisks::release_quota_mutation_fences( &commit_disks, "a_fence_tokens, @@ -3115,16 +3228,12 @@ impl SetDisks { } }; let online_disks = rename_commit.online_disks; - let convergence = rename_commit.convergence; let op_old_dir = rename_commit.data_dir; let cleanup_disks = rename_commit.cleanup_disks; let old_current_size = rename_commit.old_current_size; let mut fi = rename_commit.committed_file_info; - let rename_tail_drain = rename_commit.tail_drain; - // Do this before any post-commit await so request cancellation cannot - // bypass best-effort admission. A process crash before admission - // remains subject to the existing scanner reconciliation path. - if convergence.needs_heal() { + + if needs_immediate_heal { let mut request = rustfs_common::heal_channel::create_heal_request_with_options( commit_bucket.clone(), Some(commit_object.clone()), @@ -3133,7 +3242,9 @@ impl SetDisks { Some(commit_set.pool_index), Some(commit_set.set_index), ); - request.object_version_id = committed_version_id.map(|version_id| version_id.to_string()); + request.object_version_id = committed_version_id + .or_else(|| commit_version_suspended.then(Uuid::nil)) + .map(|version_id| version_id.to_string()); tokio::spawn(async move { let _ = rustfs_common::heal_channel::send_heal_request(request).await; }); @@ -3159,38 +3270,13 @@ impl SetDisks { .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object) .await; - // `rename_data` has completed the authoritative quorum commit. With - // the default-off early-ACK experiment, tail disk rename tasks may - // still be draining after quorum. Keep the namespace guards alive - // until that drain completes so the next same-object mutation cannot - // race a background tail rename. - if let Some(rename_tail_drain) = rename_tail_drain { - let object_lock_guard = _object_lock_guard; - let bucket_lifecycle_guard = _bucket_lifecycle_guard; - let tail_bucket = commit_bucket.clone(); - let tail_object = commit_object.clone(); - tokio::spawn(async move { - let _object_lock_guard = object_lock_guard; - let _bucket_lifecycle_guard = bucket_lifecycle_guard; - if let Err(err) = rename_tail_drain.await { - warn!( - event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_SET_DISK, - state = "failed", - bucket = %tail_bucket, - object = %tail_object, - error = %err, - "rename tail drain failed" - ); - } - }); - } else { - // The exact old-data-dir reclamation below is best-effort space - // cleanup; it must not serialize the next operation on this object. - drop(_object_lock_guard); - drop(_bucket_lifecycle_guard); + if let Some(release) = rename_guard_release.take() { + let _ = release.send(true); } + // The exact old-data-dir reclamation below is best-effort space + // cleanup; it must not serialize the next operation on this object. + drop(_object_lock_guard.take()); + drop(_bucket_lifecycle_guard.take()); rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", duration_millis_f64(rename_stage_elapsed)); if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { @@ -3219,7 +3305,7 @@ impl SetDisks { // deliberately do NOT `?`-propagate it into a 503. On residue the // report path emits the leak metric and enqueues a heal. let cleanup = commit_set - .commit_rename_data_dir( + .commit_rename_data_dir_and_mark_capacity( &cleanup_disks, &commit_bucket, &commit_object, @@ -3255,12 +3341,9 @@ impl SetDisks { ); } } - if is_compressed { record_compression_total_memory(actual_size as u64, w_size as u64).await; } - commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks); - fi.replication_state_internal = Some(commit_replication_state); fi.is_latest = true; @@ -3317,19 +3400,21 @@ impl SetDisks { ); } - let cleanup_set = commit_set.clone(); - let cleanup_tmp_dir = commit_tmp_dir.clone(); - tokio::spawn(async move { - if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await { - warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data"); - } else if issue3031_diag_enabled() { - warn!( - target: "rustfs_ecstore::set_disk", - tmp_dir = %cleanup_tmp_dir, - "issue3031_put_object_tmp_cleanup_done" - ); - } - }); + if !tail_owns_tmp_cleanup { + let cleanup_set = commit_set.clone(); + let cleanup_tmp_dir = commit_tmp_dir.clone(); + tokio::spawn(async move { + if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await { + warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data"); + } else if issue3031_diag_enabled() { + warn!( + target: "rustfs_ecstore::set_disk", + tmp_dir = %cleanup_tmp_dir, + "issue3031_put_object_tmp_cleanup_done" + ); + } + }); + } Ok(( ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned), @@ -4824,6 +4909,8 @@ pub enum PutObjectCommitPause { BeforeQuotaRename, BeforeMetadata, BeforeTransactionEpochVerify, + AfterRenameQuorum, + AfterRenameHandoff, } #[cfg(any(test, feature = "test-util"))] @@ -13986,8 +14073,10 @@ mod put_object_tmp_cleanup_tests { use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks; use super::*; use crate::disk::DiskAPI as _; - use crate::set_disk::core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier}; - use std::time::Duration; + use crate::set_disk::core::io_primitives::{ + ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier, rename_fault_injection, + }; + use std::{collections::HashSet, time::Duration}; use tempfile::TempDir; use tokio::io::AsyncReadExt; @@ -14028,6 +14117,45 @@ mod put_object_tmp_cleanup_tests { } } + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn recovered_disk_tmp_cleanup_marks_the_snapshot_it_mutates() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let recovered = disk_stores[0].clone(); + let tmp_path = "recovered-disk-cleanup/part.1"; + recovered + .write_all(RUSTFS_META_TMP_BUCKET, tmp_path, Bytes::from_static(b"stale tmp shard")) + .await + .expect("recovered disk should contain staged tmp data"); + + set_disks.disks.write().await[0] = None; + let stale_snapshot = set_disks.get_disks_internal().await; + assert!(stale_snapshot[0].is_none(), "the pre-cleanup snapshot must exclude the disk"); + set_disks.disks.write().await[0] = Some(recovered.clone()); + let _ = drain_global_dirty_scopes(); + + set_disks + .delete_all(RUSTFS_META_TMP_BUCKET, "recovered-disk-cleanup") + .await + .expect("tmp cleanup should use the recovered disk"); + + assert!(matches!( + recovered.read_all(RUSTFS_META_TMP_BUCKET, tmp_path).await, + Err(DiskError::FileNotFound) + )); + let expected = capacity_scope_from_disks(&[Some(recovered)]) + .disks + .into_iter() + .collect::>(); + let marked = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&marked), + "tmp cleanup must mark the recovered disk it actually mutated" + ); + } + #[tokio::test] async fn put_object_success_eventually_cleans_tmp_workspace() { let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; @@ -14190,19 +14318,35 @@ mod put_object_tmp_cleanup_tests { } #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] async fn cancelled_post_commit_cleanup_does_not_retain_namespace_lock() { + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("false"))], async { + assert_cancelled_post_commit_cleanup_does_not_retain_namespace_lock().await; + }) + .await; + } + + async fn assert_cancelled_post_commit_cleanup_does_not_retain_namespace_lock() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "put-commit-lock-cancelled-cleanup"; let object = "commit-lock-cancelled-cleanup-object"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } + let candidate_disks = disk_stores.iter().cloned().map(Some).collect::>(); + let expected_scope = capacity_scope_from_disks(&candidate_disks) + .disks + .into_iter() + .collect::>(); let mut initial_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]); set_disks .put_object(bucket, object, &mut initial_reader, &ObjectOptions::default()) .await .expect("initial object should be committed"); + let _ = drain_global_dirty_scopes(); let cleanup_tasks = rename_fanout_barrier::observe_tasks(object); let cleanup_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_CLEANUP); @@ -14236,6 +14380,11 @@ mod put_object_tmp_cleanup_tests { .expect_err("the first request should be cancelled during cleanup") .is_cancelled() ); + let committed_scope = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected_scope.is_subset(&committed_scope), + "a committed overwrite must mark every candidate disk before caller cancellation can interrupt post-commit awaits" + ); assert!( cleanup_tasks.running() >= 1, "cancelled cleanup must remain observable until its disk task drains" @@ -14249,6 +14398,17 @@ mod put_object_tmp_cleanup_tests { .await .expect("cancelled cleanup disk tasks should drain"); drop(cleanup_barrier); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let after_cleanup = drain_global_dirty_scopes().into_iter().collect::>(); + if expected_scope.is_subset(&after_cleanup) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("the completed old-data cleanup must re-mark every candidate disk after a refresh drain"); second_commit_barrier.release(); second @@ -14263,6 +14423,7 @@ mod put_object_tmp_cleanup_tests { let mut body = Vec::new(); reader.stream.read_to_end(&mut body).await.expect("latest body should drain"); assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]); + let _ = drain_global_dirty_scopes(); } #[tokio::test] @@ -14334,7 +14495,7 @@ mod put_object_tmp_cleanup_tests { } #[tokio::test] - #[serial_test::serial(rename_quorum_ack)] + #[serial_test::serial(capacity_dirty_scope)] async fn early_ack_tail_drain_retains_namespace_lock_until_background_rename_finishes() { temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; @@ -14405,6 +14566,368 @@ mod put_object_tmp_cleanup_tests { .await; } + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn early_ack_successful_tail_reclaims_its_old_non_inline_data_dir() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-early-ack-tail-cleanup"; + let object = "early-ack-tail-cleanup-object"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + let expected_scope = capacity_scope_from_disks(&disk_stores.iter().cloned().map(Some).collect::>()) + .disks + .into_iter() + .collect::>(); + + let mut initial = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]); + set_disks + .put_object(bucket, object, &mut initial, &ObjectOptions::default()) + .await + .expect("initial non-inline object should commit"); + drop( + set_disks + .acquire_write_lock_diag("tail_cleanup_fixture_ready", bucket, object) + .await + .expect("initial PUT tail should drain before inspecting one disk"), + ); + let old_data_dir = disk_stores[0] + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("initial object metadata should be readable") + .data_dir + .expect("one-megabyte test object should have a data directory"); + let old_part = format!("{object}/{old_data_dir}/part.1"); + + let mut heal_requests = set_disks.capture_test_rename_tail_heals(); + let tail_tasks = rename_fanout_barrier::observe_tasks(object); + let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let overwrite_set = Arc::clone(&set_disks); + let overwrite = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + overwrite_set + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + }); + + tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused()) + .await + .expect("one overwrite tail should pause during rename"); + let cleanup_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_CLEANUP); + overwrite + .await + .expect("early-ACK overwrite task should join") + .expect("overwrite should return after write quorum"); + let retained_before_tail = futures::future::join_all(disk_stores.iter().map(|disk| disk.read_all(bucket, &old_part))) + .await + .into_iter() + .filter(|result| result.is_ok()) + .count(); + assert_eq!( + retained_before_tail, 1, + "only the paused tail disk should still retain the referenced old body" + ); + let _ = drain_global_dirty_scopes(); + + rename_barrier.release(); + tokio::time::timeout(Duration::from_secs(30), cleanup_barrier.wait_until_paused()) + .await + .expect("the successful tail should pause before reclaiming its old body"); + let after_tail = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected_scope.is_subset(&after_tail), + "the completed rename tail must re-mark capacity after the first scope was drained" + ); + cleanup_barrier.release(); + let cleanup_wait = tokio::time::timeout(Duration::from_secs(30), async { + loop { + let retained = futures::future::join_all(disk_stores.iter().map(|disk| disk.read_all(bucket, &old_part))) + .await + .into_iter() + .filter(|result| result.is_ok()) + .count(); + if retained == 0 && tail_tasks.running() == 0 { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + if cleanup_wait.is_err() { + let retained = futures::future::join_all(disk_stores.iter().map(|disk| disk.read_all(bucket, &old_part))) + .await + .into_iter() + .filter(|result| result.is_ok()) + .count(); + panic!( + "the successful tail must reclaim its old body and drain: retained={retained}, running={}", + tail_tasks.running() + ); + } + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let after_cleanup = drain_global_dirty_scopes().into_iter().collect::>(); + if expected_scope.is_subset(&after_cleanup) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("the completed tail cleanup must re-mark capacity after its preceding scope was drained"); + + for (idx, disk) in disk_stores.iter().enumerate() { + assert!( + matches!(disk.read_all(bucket, &old_part).await, Err(DiskError::FileNotFound)), + "disk {idx} must reclaim the dereferenced old body" + ); + } + let mut reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("the overwrite should remain readable after tail cleanup"); + let mut body = Vec::new(); + reader + .stream + .read_to_end(&mut body) + .await + .expect("overwritten body should drain"); + assert_eq!(body, vec![b'1'; TEST_OBJECT_SIZE]); + assert!( + matches!(heal_requests.try_recv(), Err(tokio::sync::mpsc::error::TryRecvError::Empty)), + "successful tail cleanup must not submit heal" + ); + }) + .await; + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn early_ack_put_holds_quota_fences_and_re_marks_capacity_after_tail_drain() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-early-ack-capacity-scope"; + let object = "early-ack-capacity-scope-object"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + let candidate_disks = disk_stores.iter().cloned().map(Some).collect::>(); + let expected = capacity_scope_from_disks(&candidate_disks) + .disks + .into_iter() + .collect::>(); + let _ = drain_global_dirty_scopes(); + let mut heal_requests = set_disks.capture_test_rename_tail_heals(); + + let rename_tasks = rename_fanout_barrier::observe_tasks(object); + let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let put_store = Arc::clone(&set_disks); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + let mut opts = ObjectOptions::default(); + assert!(opts.set_quota_admission(0, u64::MAX)); + put_store.put_object(bucket, object, &mut reader, &opts).await + }); + tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused()) + .await + .expect("PUT should pause one tail disk during rename"); + put.await + .expect("early-ACK PUT task should join before tail release") + .expect("early-ACK PUT should return after write quorum"); + assert!(rename_tasks.running() >= 1, "the paused tail disk must remain in flight after quorum ACK"); + + let initial = drain_global_dirty_scopes().into_iter().collect::>(); + assert!(expected.is_subset(&initial), "the quorum ACK must mark every candidate disk dirty"); + + rename_barrier.release(); + let tail_done = tokio::time::timeout( + Duration::from_secs(30), + set_disks.acquire_write_lock_diag("capacity_scope_tail_probe", bucket, object), + ) + .await + .expect("the object lock should become available after the tail drain") + .expect("the tail completion probe should acquire the object lock"); + drop(tail_done); + + let after_tail = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&after_tail), + "a tail completing after refresh drained the first mark must re-mark every candidate disk" + ); + for (disk_index, disk) in disk_stores.iter().enumerate() { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} must claim its quota fence and finish the rename: {err}")); + } + assert!( + matches!(heal_requests.try_recv(), Err(tokio::sync::mpsc::error::TryRecvError::Empty)), + "a successful rename tail must not submit heal" + ); + }) + .await; + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn detached_early_ack_handoff_survives_caller_cancellation() { + use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes; + + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-early-ack-cancelled-handoff"; + let object = "early-ack-cancelled-handoff-object"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + let candidate_disks = disk_stores.iter().cloned().map(Some).collect::>(); + let expected = capacity_scope_from_disks(&candidate_disks) + .disks + .into_iter() + .collect::>(); + let _ = drain_global_dirty_scopes(); + let mut heal_requests = set_disks.capture_test_rename_tail_heals(); + let expected_version_id = Uuid::nil().to_string(); + + let tail_tasks = rename_fanout_barrier::observe_tasks(object); + let tail_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let _fault = rename_fault_injection::fail_rename_on(object, &[0]); + let quorum_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterRenameQuorum); + let handoff_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterRenameHandoff); + let put_store = Arc::clone(&set_disks); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + put_store + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + version_suspended: true, + ..Default::default() + }, + ) + .await + }); + + tokio::time::timeout(Duration::from_secs(30), tail_barrier.wait_until_paused()) + .await + .expect("one rename tail should pause while the other disks publish quorum"); + quorum_barrier.wait_until_paused().await; + put.abort(); + assert!( + put.await.expect_err("the request task should be cancelled").is_cancelled(), + "the caller should be cancelled after quorum publication" + ); + + quorum_barrier.release(); + handoff_barrier.wait_until_paused().await; + assert!(tail_tasks.running() >= 1, "the detached owner must retain the paused tail"); + let initial = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&initial), + "the detached continuation must install the full capacity scope" + ); + assert!( + tokio::time::timeout( + Duration::from_millis(50), + set_disks.acquire_write_lock_diag("cancelled_handoff_probe", bucket, object), + ) + .await + .is_err(), + "the watcher must own the namespace guard before the detached continuation awaits" + ); + + handoff_barrier.release(); + tail_barrier.release(); + let tail_done = tokio::time::timeout( + Duration::from_secs(30), + set_disks.acquire_write_lock_diag("cancelled_handoff_tail_probe", bucket, object), + ) + .await + .expect("the namespace guard should release after the detached tail drains") + .expect("the post-tail probe should acquire the namespace guard"); + drop(tail_done); + + let after_tail = drain_global_dirty_scopes().into_iter().collect::>(); + assert!( + expected.is_subset(&after_tail), + "the detached tail must re-mark capacity after the first scope was drained" + ); + let request = tokio::time::timeout(Duration::from_secs(30), heal_requests.recv()) + .await + .expect("a failed tail should submit heal") + .expect("the per-set heal capture should stay connected"); + assert_eq!(request.bucket, bucket); + assert_eq!(request.object_prefix.as_deref(), Some(object)); + assert_eq!(request.object_version_id.as_deref(), Some(expected_version_id.as_str())); + assert_eq!(request.pool_index, Some(set_disks.pool_index)); + assert_eq!(request.set_index, Some(set_disks.set_index)); + }) + .await; + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn no_lock_put_waits_for_rename_tail_under_outer_guard() { + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-no-lock-serial-rename"; + let object = "put-no-lock-serial-rename-object"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let outer_guard = set_disks + .acquire_write_lock_diag("outer_no_lock_put", bucket, object) + .await + .expect("the outer caller should hold the namespace guard"); + let rename_tasks = rename_fanout_barrier::observe_tasks(object); + let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let put_store = Arc::clone(&set_disks); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + put_store + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + }); + + tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused()) + .await + .expect("no-lock PUT should reach the paused rename disk"); + tokio::time::timeout(Duration::from_secs(30), async { + while rename_tasks.running() != 1 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the other rename disks should finish while one disk remains paused"); + assert!( + !put.is_finished(), + "a no-lock caller that cannot hand off its outer guard must await the full rename fanout" + ); + + rename_barrier.release(); + put.await + .expect("no-lock PUT task should join") + .expect("no-lock PUT should commit after the rename tail releases"); + drop(outer_guard); + }) + .await; + } + #[tokio::test] async fn put_object_no_lock_aborts_after_outer_namespace_lock_loss() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 6ad051c0a..1631ba0c7 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -842,6 +842,7 @@ mod tests { use crate::runtime::instance::InstanceContext; use crate::storage_api_contracts::{ bucket::{BucketOperations as _, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp}, + list::ListOperations as _, object::{ObjectIO as _, ObjectOperations as _}, }; use crate::store::{ECStore, init_local_disks_with_instance_ctx}; @@ -1664,6 +1665,70 @@ mod tests { .expect("DeleteBucket must succeed once the client has drained the bucket"); } + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn bucket_delete_succeeds_after_listing_and_deleting_an_unversioned_overwrite() { + let (disk_paths, ecstore) = setup_bucket_delete_test_env().await; + let bucket = format!("bucket-delete-after-overwrite-{}", Uuid::new_v4().simple()); + let object = "object.txt"; + + ecstore + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("unversioned bucket should be created"); + + let mut first_reader = PutObjReader::from_vec(b"version A".to_vec()); + let first = ecstore + .put_object(&bucket, object, &mut first_reader, &ObjectOptions::default()) + .await + .expect("version A should be written"); + let mut second_reader = PutObjReader::from_vec(b"version B".to_vec()); + let second = ecstore + .put_object(&bucket, object, &mut second_reader, &ObjectOptions::default()) + .await + .expect("version B should overwrite version A"); + assert_ne!(first.data_dir, second.data_dir, "the overwrite must publish a new body generation"); + + let listing = ecstore + .clone() + .list_object_versions(&bucket, "", None, None, None, 1000) + .await + .expect("the overwritten object should remain listable for teardown"); + assert_eq!(listing.objects.len(), 1, "an unversioned overwrite should expose one current version"); + let current = &listing.objects[0]; + assert_eq!(current.name, object); + assert!(current.is_latest, "the listed null version must be current"); + assert_eq!(current.version_id, None, "an unversioned object must be exposed as the null version"); + assert_eq!( + current.data_dir, second.data_dir, + "listing must expose version B, not the overwritten body" + ); + + for version in listing.objects { + let version_id = version.version_id.map(|version_id| version_id.to_string()); + ecstore + .delete_object( + &bucket, + &version.name, + ObjectOptions { + version_id, + ..Default::default() + }, + ) + .await + .expect("each version returned by teardown listing should be deletable"); + } + + assert!( + !any_disk_has_object_metadata(&disk_paths, &bucket).await, + "deleting the listed null version must remove every xl.meta" + ); + ecstore + .delete_bucket(&bucket, &DeleteBucketOptions::default()) + .await + .expect("DeleteBucket should succeed after the listed overwrite is deleted"); + } + #[tokio::test] #[serial] async fn bucket_delete_default_s3_delete_still_rejects_non_empty_bucket() {