From caa7eeac0b9574888935ba70dfb5e854771addb9 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 01:34:10 +0800 Subject: [PATCH] fix(ecstore): retain source-set lock during cleanup --- crates/ecstore/src/store/init.rs | 104 +++++++++++++++++++++++++++++ crates/ecstore/src/store/object.rs | 5 +- 2 files changed, 107 insertions(+), 2 deletions(-) diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 11877ee25..6207bc72b 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -2930,6 +2930,110 @@ mod tests { shutdown.cancel(); } + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn decommission_source_cleanup_holds_hashed_set_lock_across_preflight() { + let temp_dir = tempfile::tempdir().expect("create multi-set decommission cleanup store dir"); + let (_ctx, store, shutdown) = without_storage_class_env(build_isolated_test_store_with_layout( + temp_dir.path(), + "multi-set-decommission-source-cleanup", + &[(2, 4)], + CancellationToken::new(), + )) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let bucket = format!("multi-set-decommission-source-cleanup-{}", uuid::Uuid::new_v4()); + let object = (0..128) + .map(|index| format!("object-{index}.bin")) + .find(|candidate| store.pools[0].get_disks_by_key(candidate).set_index == 1) + .expect("the deterministic object search should select source set 1"); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create multi-set decommission cleanup bucket"); + let mut source = PutObjReader::from_vec(b"source generation".to_vec()); + store.pools[0] + .put_object(&bucket, &object, &mut source, &ObjectOptions::default()) + .await + .expect("write the source generation to set 1"); + let source_set = store.pools[0].get_disks_by_key(&object); + assert_eq!(source_set.set_index, 1, "the source must not share the fixed set-0 namespace"); + let expected_source_versions = source_set + .load_file_info_versions_exact(&bucket, &object) + .await + .expect("source versions should be readable") + .expect("the source generation should exist"); + + let cleanup_barrier = crate::data_movement::SourceCleanupDeleteBarrier::install(&bucket, &object); + let cleanup_store = Arc::clone(&store); + let cleanup_bucket = bucket.clone(); + let cleanup_object = object.clone(); + let cleanup = tokio::spawn(async move { + let mutation_fence = cleanup_store + .acquire_decommission_source_cleanup_fence(&cleanup_bucket, &cleanup_object, source_set.as_ref()) + .await?; + crate::data_movement::cleanup_source_entry_if_unchanged( + source_set, + &cleanup_bucket, + &cleanup_object, + &expected_source_versions, + &[], + crate::data_movement::SourceCleanupBucketFence { + object_mutation_fence: Some(&mutation_fence), + ..Default::default() + }, + "test_multi_set_decommission_source_cleanup", + ) + .await + }); + cleanup_barrier.wait_until_paused().await; + + let put_barrier = crate::set_disk::PutObjectCommitBarrier::install( + &bucket, + &object, + crate::set_disk::PutObjectCommitPause::BeforeNamespace, + ); + let mutation_pool = Arc::clone(&store.pools[0]); + let mutation_bucket = bucket.clone(); + let mutation_object = object.clone(); + let replacement = b"replacement generation".to_vec(); + let expected_replacement = replacement.clone(); + let mutation = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(replacement); + mutation_pool + .put_object(&mutation_bucket, &mutation_object, &mut reader, &ObjectOptions::default()) + .await + }); + put_barrier.wait_until_paused().await; + put_barrier.release_and_wait_until_namespace_pending().await; + assert!(!mutation.is_finished(), "a source mutation must wait behind cleanup's set-1 write lock"); + + cleanup_barrier.release(); + cleanup + .await + .expect("source cleanup task should join") + .expect("source cleanup should remove only the preflight generation"); + mutation + .await + .expect("source mutation task should join") + .expect("source mutation should commit after cleanup releases the set lock"); + + let mut reader = store.pools[0] + .get_object_reader(&bucket, &object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("the replacement generation must remain readable"); + let mut actual = Vec::new(); + reader + .stream + .read_to_end(&mut actual) + .await + .expect("read the replacement generation"); + assert_eq!(actual, expected_replacement, "cleanup must not delete the replacement generation"); + + shutdown.cancel(); + } + #[tokio::test] #[serial_test::serial(storage_class_env)] async fn versioned_delete_marker_survives_decommission_source_cleanup() { diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 16cc40935..ac644695f 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -1745,9 +1745,10 @@ impl ECStore { #[cfg(test)] crate::data_movement::notify_source_cleanup_mutation_fence_pending(bucket, object); let object = encode_dir_object(object); - let distributed = self.ctx.is_dist_erasure().await; let fixed_set = Arc::clone(&self.pools[0].disk_set[0]); - let source_lock_covered = !distributed || same_distributed_lock_domain(&fixed_set.lockers, &source_set.lockers); + // SetDisks namespaces include pool/set identity, so only the canonical + // fixed set is covered by the fixed mutation guard. + let source_lock_covered = std::ptr::eq(fixed_set.as_ref(), source_set); // Lock order: fixed store mutation domain first; source cleanup takes its // hashed source-domain lock second only when this guard does not cover it. let guard = self