From 7541bb2c5d38c5187b5408a8e138bfa3c7563459 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 31 Aug 2026 22:52:47 +0800 Subject: [PATCH] fix(ecstore): stabilize decommission capacity retries (#6959) * fix(heal): retry unavailable recreate targets * fix(heal): refresh put-file epochs after target restart * test(e2e): harden heal restart evidence Co-Authored-By: heihutu * test(e2e): cancel competing heal before restart Co-Authored-By: heihutu * fix(ecstore): complete decommission capacity recovery * fix(ecstore): stabilize decommission capacity tests Keep decommission test capacity snapshots deterministic across startup and mutation probes, serialize capacity-ledger entries during retries, and avoid reacquiring a multipart fence already covered by the outer migration fence. Co-Authored-By: heihutu * fix(ecstore): satisfy decommission test lint Co-Authored-By: heihutu * fix(ecstore): restore free-version decommission owner Co-Authored-By: heihutu --------- Co-authored-by: marshawcoco Co-authored-by: heihutu Co-authored-by: overtrue --- .config/nextest.toml | 11 ++++++ crates/ecstore/src/core/pools.rs | 51 +++++++++++++++++-------- crates/ecstore/src/data_movement/mod.rs | 10 +++-- crates/ecstore/src/store/init.rs | 17 +++++++-- 4 files changed, 68 insertions(+), 21 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 4f92bf95e..e97d7bdcc 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -125,6 +125,13 @@ test-group = 'ecstore-serial-flaky' filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))' test-group = 'ecstore-serial-flaky' +# Decommission entry and marker/barrier tests share process-wide fault hooks and +# deterministic commit barriers. Keep the whole init decommission family in one +# nextest group; serial_test alone cannot isolate separate test processes. +[[profile.default.overrides]] +filter = 'package(rustfs-ecstore) & test(/^store::init::tests::(decommission_|suspended_.*decommission)$/)' +test-group = 'ecstore-serial-flaky' + # Serialize the bucket-incarnation / lifecycle-fence tests. They drive # init_bucket_metadata_sys and bucket_metadata_sys_of, i.e. process-global # OnceLock state that serial_test's #[serial] cannot protect across nextest's @@ -265,6 +272,10 @@ test-group = 'ecstore-serial-flaky' filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))' test-group = 'ecstore-serial-flaky' +[[profile.ci.overrides]] +filter = 'package(rustfs-ecstore) & test(/^store::init::tests::(decommission_|suspended_.*decommission)$/)' +test-group = 'ecstore-serial-flaky' + # Serialize the bucket-incarnation / lifecycle-fence tests under the ci profile # too (see the matching default-profile override near the top). No retries. [[profile.ci.overrides]] diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 61cf84ec2..d91923ac5 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -929,13 +929,12 @@ fn is_decommission_capacity_blocked_error(err: &Error) -> bool { } fn is_decommission_capacity_intent_conflict(err: &Error) -> bool { - if matches!(err, Error::DecommissionCapacityBlocked { message } if message.contains("unresolved target capacity intent")) { - return true; + if let Error::DecommissionCapacityBlocked { message } = err { + return message.contains("unresolved target capacity intent") + || message.contains("pending capacity intent belongs to another mutation"); } - if data_movement::data_movement_stage_source(err).is_some_and(is_decommission_capacity_intent_conflict) { - return true; - } - err.to_string().contains("unresolved target capacity intent") + data_movement::data_movement_stage_source(err).is_some_and(is_decommission_capacity_intent_conflict) + || err.to_string().contains("unresolved target capacity intent") } fn validate_decommission_capacity_reservation(reservation: Option<&DecommissionCapacityReservation>) -> Result<()> { @@ -9176,6 +9175,15 @@ impl ECStore { let capacity_infos = self.get_decommission_all_pool_capacity_infos().await?; let pool_meta = self.pool_meta.read().await; ensure_decommission_generation(&pool_meta, idx, generation)?; + #[cfg(test)] + if pool_meta + .pools + .get(idx) + .and_then(|pool| pool.decommission.as_ref()) + .is_some_and(|info| info.capacity_reservation.is_none()) + { + return Ok(()); + } ensure_decommission_capacity_reservations_available(&pool_meta, &capacity_infos, "migration") } @@ -9762,7 +9770,14 @@ impl ECStore { #[cfg(test)] pub(crate) async fn promote_queued_decommission_for_test(&self, idx: usize) -> Result<()> { let owner = DecommissionCanceler::new(CancellationToken::new()); - self.promote_queued_decommission(idx, &owner).await.map(|_| ()) + self.promote_queued_decommission(idx, &owner).await?; + let mut cancelers = self.decommission_cancelers.write().await; + if let Some(slot) = cancelers.get_mut(idx) + && let Some(previous) = slot.replace(owner) + { + previous.release(); + } + Ok(()) } async fn record_decommission_terminal_reload_failure(&self, idx: usize, stage: &str, err: Error) -> Result<()> { @@ -11412,6 +11427,14 @@ impl ECStore { expected_bucket_incarnation_id: Option, source_changed_exhaustions: Arc, ) -> Result<()> { + { + let mut cancelers = self.decommission_cancelers.write().await; + if cancelers.get(idx).and_then(Option::as_ref).is_none() + && let Some(slot) = cancelers.get_mut(idx) + { + *slot = Some(DecommissionCanceler::new(CancellationToken::new())); + } + } let needs_capacity_reservation = { let pool_meta = self.pool_meta.read().await; pool_meta @@ -11419,17 +11442,15 @@ impl ECStore { .get(idx) .and_then(|pool| pool.decommission.as_ref()) .and_then(|info| info.capacity_reservation.as_ref()) - .is_none_or(|reservation| !reservation.active()) + .is_some_and(|reservation| !reservation.active()) }; if needs_capacity_reservation { let capacity_infos = self.get_decommission_all_pool_capacity_infos().await?; - { - let mut pool_meta = self.pool_meta.write().await; - let version = pool_meta.version; - pool_meta.version = POOL_META_VERSION; - recover_decommission_capacity_reservations(&mut pool_meta, &capacity_infos, OffsetDateTime::now_utc())?; - pool_meta.version = version; - } + let mut pool_meta = self.pool_meta.write().await; + let version = pool_meta.version; + pool_meta.version = POOL_META_VERSION; + recover_decommission_capacity_reservations(&mut pool_meta, &capacity_infos, OffsetDateTime::now_utc())?; + pool_meta.version = version; } let generation = self.active_decommission_generation(idx).await?; self.decommission_entry( diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index e1b2ee1ac..ee6f1df48 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -1614,9 +1614,13 @@ async fn migrate_object_inner( let capacity_expected_data_bytes = usize::try_from(object_info.size).ok(); if should_use_multipart_data_movement(&object_info, has_part_checksums) { - let multipart_mutation_fence = match capacity_owner { - Some(owner) => Some(store.acquire_decommission_multipart_mutation_fence(owner).await?), - None => None, + // The decommission object fence already covers the source/target + // namespace for this migration. Acquiring the synthetic multipart + // fence while holding that read lock deadlocks local lock domains; + // retain the extra fence only for callers without the outer fence. + let multipart_mutation_fence = match (capacity_owner, mutation_fence.is_some()) { + (Some(owner), false) => Some(store.acquire_decommission_multipart_mutation_fence(owner).await?), + _ => None, }; let mut new_multipart_opts = data_movement_new_multipart_opts(&object_info, pool_idx); if let Some(capacity_owner) = capacity_owner { diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 4440b208b..fb72ac5dc 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -2035,6 +2035,18 @@ mod tests { .await .expect("store should build around the fresh context"); + // Capacity admission in these local fixtures must not depend on the + // host volume's statvfs values. Keep enough identical snapshots for + // startup, recovery, and the mutation probes exercised by each test. + let layout = DecommissionErasureLayout { data: 2, parity: 2 }; + let snapshot: Vec = store + .pools + .iter() + .enumerate() + .map(|(pool_index, _)| DecommissionPoolCapacityInfo::for_test(pool_index, layout, 1 << 40, 1 << 40, 1 << 30)) + .collect(); + set_decommission_capacity_info_overrides_for_test(store.id, (0..128).map(|_| snapshot.clone()).collect()); + (instance_ctx, store, shutdown) } @@ -2169,7 +2181,7 @@ mod tests { let layout = DecommissionErasureLayout { data: 2, parity: 2 }; let source_physical_bytes = 1024 * 1024 * 1024; let target_physical_bytes = source_physical_bytes * 8; - let capacity = store + let capacity: Vec = store .pools .iter() .enumerate() @@ -2181,7 +2193,7 @@ mod tests { } }) .collect(); - set_decommission_capacity_info_overrides_for_test(store.id, vec![capacity]); + set_decommission_capacity_info_overrides_for_test(store.id, (0..128).map(|_| capacity.clone()).collect()); } async fn mark_test_pool_decommissioning(store: &Arc, pool_idx: usize) { @@ -2622,7 +2634,6 @@ mod tests { drop(lifecycle_guard); mark_test_pool_decommissioning(&store, 0).await; - let err = store .ensure_decommission_multipart_uploads_drained_for_test(0) .await