From e3ed2cf2f41e6ec7e44656d555c653f705ca499a Mon Sep 17 00:00:00 2001 From: Chris An Date: Sat, 7 Nov 2026 01:49:42 -0500 Subject: [PATCH] fix(decommission): reserve owned source usage during activation (#9220) --- crates/ecstore/src/core/pools.rs | 383 ++++++++++++++++++++++++++++--- 1 file changed, 345 insertions(+), 38 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index f85691532..9a32330ed 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -2344,33 +2344,31 @@ fn grow_decommission_target_reservation( if required_peak <= existing_remaining { return Ok(false); } - if reservation.source_data_equivalent_bytes > 0 { - return Err(decommission_target_no_longer_fits_error( - target_pool_index, - format!("source pool {source_pool_index} reservation estimate is exhausted"), - )); - } - if let Some(target) = existing_target - && target_capacity.physical_free <= target.physical_free_at_reservation - { - return Err(decommission_capacity_blocked_error(format!( - "source pool {source_pool_index} target pool {target_pool_index} reservation is exhausted" - ))); - } + // Scanner estimates can predate source writes. Grow against current + // target capacity, then validate the aggregate reservation below. let current_peak = reservation.remaining_peak_physical_bytes(); let additional_peak = required_peak.saturating_sub(existing_remaining); let additional_predicted = additional_peak.div_ceil(1usize.saturating_add(reservation.temporary_copies)); let additional_source_data = capacity_source_data_equivalent(additional_predicted, target_layout)?; - reservation.source_data_equivalent_bytes = reservation - .source_data_equivalent_bytes - .saturating_add(additional_source_data); reservation.source_physical_bytes = reservation .source_physical_bytes .saturating_add(capacity_target_physical_bytes(additional_source_data, reservation.source_layout)?); + // Recompute from the physical total so EC rounding matches the + // persisted reader after every growth, including repeated growth. + reservation.source_data_equivalent_bytes = + capacity_source_data_equivalent(reservation.source_physical_bytes, reservation.source_layout)?; reservation.predicted_physical_bytes = reservation - .predicted_physical_bytes - .saturating_add(additional_predicted) - .max(capacity_target_physical_bytes(reservation.source_data_equivalent_bytes, target_layout)?); + .targets + .iter() + .map(|target| target.layout) + .chain(std::iter::once(target_layout)) + .try_fold( + reservation.predicted_physical_bytes.saturating_add(additional_predicted), + |predicted, layout| { + capacity_target_physical_bytes(reservation.source_data_equivalent_bytes, layout) + .map(|required| predicted.max(required)) + }, + )?; reservation.temporary_physical_bytes = reservation .predicted_physical_bytes .saturating_mul(reservation.temporary_copies); @@ -2402,6 +2400,7 @@ fn grow_decommission_target_reservation( }); } renew_decommission_capacity_reservation(reservation, now, true); + validate_decommission_capacity_reservation(Some(reservation))?; pool.last_update = pool.last_update.max(now); Ok(true) })(); @@ -12145,8 +12144,13 @@ async fn pause_decommission_start_for_test(store_id: uuid::Uuid, point: Decommis } #[cfg(any(test, feature = "test-util"))] -type DecommissionCapacityInfoOverrides = - std::sync::Mutex>>>; +struct DecommissionCapacityInfoOverride { + snapshots: std::collections::VecDeque>, + owned_snapshots: Option>>, +} + +#[cfg(any(test, feature = "test-util"))] +type DecommissionCapacityInfoOverrides = std::sync::Mutex>; #[cfg(any(test, feature = "test-util"))] static DECOMMISSION_CAPACITY_INFO_OVERRIDES: std::sync::OnceLock = std::sync::OnceLock::new(); @@ -12163,16 +12167,30 @@ pub(crate) fn set_decommission_capacity_info_overrides_for_test( .get_or_init(|| std::sync::Mutex::new(HashMap::new())) .lock() .expect("decommission capacity info override should not be poisoned") - .insert(store_id, snapshots.into()); + .insert( + store_id, + DecommissionCapacityInfoOverride { + snapshots: snapshots.into(), + owned_snapshots: None, + }, + ); } #[cfg(any(test, feature = "test-util"))] -fn take_decommission_capacity_info_override_for_test(store_id: uuid::Uuid) -> Option> { +fn take_decommission_capacity_info_override_for_test( + store_id: uuid::Uuid, + use_owned_usage: bool, +) -> Option> { let mut overrides = DECOMMISSION_CAPACITY_INFO_OVERRIDES .get_or_init(|| std::sync::Mutex::new(HashMap::new())) .lock() .expect("decommission capacity info override should not be poisoned"); - let queue = overrides.get_mut(&store_id)?; + let entry = overrides.get_mut(&store_id)?; + let queue = if use_owned_usage { + entry.owned_snapshots.as_mut().unwrap_or(&mut entry.snapshots) + } else { + &mut entry.snapshots + }; if queue.len() > 1 { queue.pop_front() } else { @@ -15242,7 +15260,7 @@ impl ECStore { ensure_decommission_start_pool_states(&latest_pool_meta, indices)?; ensure_decommission_ledger_persistence_supported(&latest_pool_meta)?; ensure_decommission_capacity_writer_supported(&latest_pool_meta)?; - let capacity_infos = self.get_decommission_all_pool_capacity_infos().await?; + let capacity_infos = self.get_decommission_all_pool_capacity_infos_with_owned_usage().await?; activation_fence.ensure_held()?; let target_fence_proof = crate::services::notification_sys::acquire_decommission_target_fence_fleet_proof(); @@ -15604,7 +15622,7 @@ impl ECStore { async fn get_decommission_prewrite_free_capacity(&self, target_pool_index: usize) -> Result { #[cfg(any(test, feature = "test-util"))] - if let Some(capacity_infos) = take_decommission_capacity_info_override_for_test(self.id) { + if let Some(capacity_infos) = take_decommission_capacity_info_override_for_test(self.id, false) { return capacity_infos .iter() .find(|capacity| capacity.pool_index == target_pool_index) @@ -15627,7 +15645,7 @@ impl ECStore { async fn get_decommission_pool_capacity_infos(&self, use_owned_usage: bool) -> Result> { #[cfg(any(test, feature = "test-util"))] - if let Some(capacity_infos) = take_decommission_capacity_info_override_for_test(self.id) { + if let Some(capacity_infos) = take_decommission_capacity_info_override_for_test(self.id, use_owned_usage) { return Ok(capacity_infos); } @@ -15845,14 +15863,8 @@ impl ECStore { log_decommission_adopted_target_choice(reservation, owner, pool_index, &over_budget); return Ok(pool_index); } - if reservation.source_data_equivalent_bytes > 0 { - let mut skipped = over_budget.iter().copied().collect::>(); - skipped.sort_unstable(); - return Err(decommission_capacity_blocked_error(format!( - "source pool {} has no target allocation for {expected_data_bytes} data bytes; target pools over their temporary-copy budget: {skipped:?}", - owner.source_pool_index - ))); - } + // A stale nonzero estimate can exhaust its allocation too. Selection + // is provisional until fenced growth persists the new reservation. let capacities = self.get_decommission_all_pool_capacity_infos().await?; let active_sources = active_decommission_source_indices(&pool_meta); capacities @@ -25784,6 +25796,55 @@ mod tests { ); } + #[tokio::test] + #[serial_test::serial] + async fn decommission_activation_reserves_owned_usage_on_a_shared_filesystem() { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let target = DecommissionPoolCapacityInfo::for_test(1, layout, 60, 60, 0); + let filesystem_source = DecommissionPoolCapacityInfo::for_test(0, layout, 0, 400, 400); + let owned_source = DecommissionPoolCapacityInfo::for_test(0, layout, 0, 400, 30); + set_decommission_capacity_info_overrides_for_test(store.id, vec![vec![filesystem_source, target]]); + { + let mut overrides = super::DECOMMISSION_CAPACITY_INFO_OVERRIDES + .get() + .expect("capacity overrides initialized") + .lock() + .expect("capacity overrides lock"); + overrides + .get_mut(&store.id) + .expect("this store has an override") + .owned_snapshots = Some(vec![vec![owned_source, target]].into()); + } + + store + .save_current_pool_meta_for_decommission_start(&[0], Vec::new()) + .await + .expect("activation should reserve the owned bytes rather than unrelated mount usage"); + + let local = store.pool_meta.read().await; + let reservation = local.pools[0] + .decommission + .as_ref() + .and_then(|state| state.capacity_reservation.as_ref()) + .expect("activation must retain its capacity reservation"); + assert_eq!(reservation.source_physical_bytes, 30); + assert_eq!(reservation.peak_physical_bytes, 60); + drop(local); + let mut persisted = PoolMeta::default(); + persisted + .load_no_lock_from_replicas(store.pools.clone()) + .await + .expect("activation should be durable"); + let reservation = persisted.pools[0] + .decommission + .as_ref() + .and_then(|state| state.capacity_reservation.as_ref()) + .expect("persisted activation must retain its capacity reservation"); + assert_eq!(reservation.source_physical_bytes, 30); + assert_eq!(reservation.peak_physical_bytes, 60); + } + #[tokio::test] #[serial_test::serial] async fn decommission_activation_lock_recheck_rejects_sudden_space_drop_without_persisting() { @@ -25818,6 +25879,173 @@ mod tests { assert!(persisted.pools[0].decommission.is_none()); } + #[tokio::test] + #[serial_test::serial] + async fn decommission_copies_native_object_after_nonzero_usage_estimate_is_exhausted() { + use tokio::io::AsyncReadExt as _; + + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + let bucket = format!("stale-owned-{}", uuid::Uuid::new_v4()); + let object = "new-after-scan.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create source bucket"); + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let capacities = vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 1 << 30, 1), + DecommissionPoolCapacityInfo::for_test(1, layout, 1 << 30, 1 << 30, 0), + ]; + set_decommission_capacity_info_overrides_for_test(store.id, vec![capacities]); + let expected = vec![0x57; 16 * 1024]; + let mut body = crate::object_api::PutObjReader::from_vec(expected.clone()); + store.pools[0] + .put_object(&bucket, object, &mut body, &ObjectOptions::default()) + .await + .expect("write an object larger than the earlier scanner estimate"); + store + .save_current_pool_meta_for_decommission_start( + &[0], + vec![DecomBucketInfo { + name: bucket.clone(), + prefix: String::new(), + }], + ) + .await + .expect("activate the stale nonzero reservation"); + let canceler = DecommissionCanceler::new(CancellationToken::new()); + store.decommission_cancelers.write().await[0] = Some(canceler.clone()); + store + .do_decommission_in_routine(canceler, 0, Arc::new(Semaphore::new(1))) + .await + .expect("run the real native-disk mover"); + + let mut reader = store.pools[1] + .get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("the newer object must reach the target despite the stale estimate"); + let mut actual = Vec::new(); + reader.stream.read_to_end(&mut actual).await.expect("read migrated payload"); + assert_eq!(actual, expected); + assert!( + store.pools[0] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .is_err(), + "source cleanup follows the verified target copy" + ); + let meta = store.pool_meta.read().await; + let state = meta.pools[0].decommission.as_ref().expect("retirement state is retained"); + assert!(state.complete && !state.failed); + assert_eq!(state.items_decommission_failed, 0); + } + + #[tokio::test] + #[cfg(feature = "test-util")] + #[serial_test::serial] + async fn decommission_native_ec_growth_keeps_all_metadata_replicas_readable() { + use rustfs_data_usage::{DataUsageCache, DataUsageEntry}; + use tokio::io::AsyncReadExt as _; + + let (_temp_dirs, store, _other_store) = + crate::services::rebalance::test_three_pool_stores_with_four_disk_sets_with_isolated_node_contexts(None).await; + let bucket = format!("stale-ec-{}", uuid::Uuid::new_v4()); + let object = "new-after-scan.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create bucket"); + for set in &store.pools[0].disk_set { + let encoded = rmp_serde::to_vec_named(&serde_json::json!({ + "info": { + "name": DATA_USAGE_ROOT, + "next_cycle": 1, + "last_update": std::time::SystemTime::now(), + "skip_healing": false, + "snapshot_complete": true + }, + "cache": { + DATA_USAGE_ROOT: DataUsageEntry { size: 3, objects: 1, ..Default::default() } + } + })) + .expect("encode completed scanner fixture"); + let cache = DataUsageCache::unmarshal(&encoded).expect("read scanner fixture"); + assert!(cache.dui(DATA_USAGE_ROOT, &[]).is_complete_bucket_usage_snapshot()); + set.put_object( + RUSTFS_META_BUCKET, + &format!("{}/{}", crate::disk::BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME), + &mut crate::object_api::PutObjReader::from_vec(encoded), + &ObjectOptions::default(), + ) + .await + .expect("persist the earlier complete cache"); + } + let expected = vec![0x37; 16 * 1024]; + store.pools[0] + .put_object( + &bucket, + object, + &mut crate::object_api::PutObjReader::from_vec(expected.clone()), + &ObjectOptions::default(), + ) + .await + .expect("write after the scan"); + store + .save_current_pool_meta_for_decommission_start( + &[0], + vec![DecomBucketInfo { + name: bucket.clone(), + prefix: String::new(), + }], + ) + .await + .expect("activate from the earlier owned estimate"); + { + let meta = store.pool_meta.read().await; + let reservation = meta.pools[0] + .decommission + .as_ref() + .unwrap() + .capacity_reservation + .as_ref() + .unwrap(); + assert_eq!(reservation.source_layout, DecommissionErasureLayout { data: 3, parity: 1 }); + assert_eq!(reservation.source_physical_bytes, 8); + } + let canceler = DecommissionCanceler::new(CancellationToken::new()); + store.decommission_cancelers.write().await[0] = Some(canceler.clone()); + store + .do_decommission_in_routine(canceler, 0, Arc::new(Semaphore::new(1))) + .await + .expect("run native EC migration"); + for pool in &store.pools { + let mut reloaded = PoolMeta::default(); + reloaded + .load_no_lock_from_replicas(vec![pool.clone()]) + .await + .expect("each saved metadata replica must reload"); + let state = reloaded.pools[0].decommission.as_ref().expect("saved retirement state"); + assert!(state.complete && !state.failed); + } + let mut reader = store + .get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("target object must remain readable"); + let mut actual = Vec::new(); + reader + .stream + .read_to_end(&mut actual) + .await + .expect("read migrated EC payload"); + assert_eq!(actual, expected); + assert!( + store.pools[0] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .is_err() + ); + } + #[tokio::test] #[serial_test::serial] async fn decommission_worker_pauses_on_runtime_capacity_shortage_without_source_side_effect() { @@ -37625,10 +37853,89 @@ mod pools_tests { .as_ref() .and_then(|info| info.capacity_reservation.as_ref()) .unwrap(); - assert_eq!(reservation.source_data_equivalent_bytes, 2); - assert_eq!(reservation.predicted_physical_bytes, 3); + assert_eq!(reservation.source_data_equivalent_bytes, 3); + assert_eq!(reservation.predicted_physical_bytes, 4); assert_eq!(reservation.targets.len(), 1); - assert_eq!(reservation.targets[0].reserved_physical_bytes, 6); + assert_eq!(reservation.targets[0].reserved_physical_bytes, 8); + validate_decommission_capacity_reservation(Some(reservation)).expect("empty-snapshot growth must remain readable"); + } + + #[test] + fn decommission_capacity_grows_a_stale_nonzero_estimate_and_rolls_back_if_space_is_short() { + let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(1); + let layout = DecommissionErasureLayout { data: 3, parity: 1 }; + let initial = vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 64, 2), + DecommissionPoolCapacityInfo::for_test(1, layout, 64, 64, 0), + ]; + let mut meta = PoolMeta { + version: POOL_META_VERSION, + pools: vec![decommission_test_pool_status(0, None), decommission_test_pool_status(1, None)], + ..Default::default() + }; + meta.decommission(0, initial[0].space).expect("activate source"); + reserve_decommission_start_target_capacity( + &mut meta, + &[0], + &initial, + uuid::Uuid::new_v4(), + 1, + now, + DECOMMISSION_CAPACITY_MODEL_VERSION, + ) + .expect("reserve the stale estimate"); + assert!( + grow_decommission_target_reservation(&mut meta, &initial, 0, 1, 16, uuid::Uuid::new_v4(), now) + .expect("a nonzero stale estimate may grow against real free space") + ); + let before = serde_json::to_value(&meta).expect("snapshot reservation before rejected growth"); + let shortage = vec![initial[0], DecommissionPoolCapacityInfo::for_test(1, layout, 43, 64, 21)]; + assert!( + grow_decommission_target_reservation(&mut meta, &shortage, 0, 1, 32, uuid::Uuid::new_v4(), now).is_err(), + "real free-space shortage still rejects growth" + ); + assert_eq!(serde_json::to_value(&meta).expect("snapshot after rejected growth"), before); + } + + #[test] + fn decommission_capacity_growth_keeps_rounded_and_mixed_layout_estimates_loadable() { + let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(1); + for source_layout in [ + DecommissionErasureLayout { data: 3, parity: 1 }, + DecommissionErasureLayout { data: 2, parity: 2 }, + ] { + let capacities = vec![ + DecommissionPoolCapacityInfo::for_test(0, source_layout, 0, 8, 8), + DecommissionPoolCapacityInfo::for_test(1, DecommissionErasureLayout { data: 2, parity: 2 }, 16, 16, 0), + DecommissionPoolCapacityInfo::for_test(2, DecommissionErasureLayout { data: 3, parity: 1 }, 1024, 1024, 0), + ]; + let mut meta = PoolMeta { + version: POOL_META_VERSION, + pools: (0..3).map(|index| decommission_test_pool_status(index, None)).collect(), + ..Default::default() + }; + meta.decommission(0, capacities[0].space).unwrap(); + reserve_decommission_start_target_capacity( + &mut meta, + &[0], + &capacities, + uuid::Uuid::new_v4(), + 1, + now, + DECOMMISSION_CAPACITY_MODEL_VERSION, + ) + .expect("initial reservation"); + grow_decommission_target_reservation(&mut meta, &capacities, 0, 2, 16, uuid::Uuid::new_v4(), now) + .expect("grow against the target's free space"); + let reservation = meta.pools[0] + .decommission + .as_ref() + .unwrap() + .capacity_reservation + .as_ref() + .unwrap(); + validate_decommission_capacity_reservation(Some(reservation)).expect("growth must satisfy the persisted reader"); + } } #[test]