fix(ecstore): pause retirement when capacity growth no longer fits

crates/ecstore/src/core/pools.rs

@@ -2208,9 +2208,14 @@ fn ensure_decommission_capacity_reservations_available(
             continue;
         }
         metrics::counter!(METRIC_DECOMMISSION_CAPACITY_CONFLICTS_TOTAL, "phase" => phase).increment(1);
-        return Err(decommission_capacity_blocked_error(format!(
+        let detail = format!(
             "target pool {target_pool_index} reservation requires {required} physical bytes, but only {available} bytes remain during {phase}"
-        )));
+        );
+        return Err(if growth_pool.is_some() {
+            decommission_target_no_longer_fits_error(target_pool_index, detail)
+        } else {
+            decommission_capacity_blocked_error(detail)
+        });
     }
     Ok(())
 }
@@ -2341,6 +2346,13 @@ fn grow_decommission_target_reservation(
         let existing_remaining = existing_target
             .map(|target| target.remaining_reserved_physical_bytes(reservation.temporary_copies))
             .unwrap_or_default();
+        let existing_reserved = existing_target
+            .map(|target| target.reserved_physical_bytes)
+            .unwrap_or_default();
+        let allocation_limit = existing_target
+            .map(|target| target.physical_free_at_reservation)
+            .unwrap_or_default()
+            .max(target_capacity.physical_free);
         if required_peak <= existing_remaining {
             return Ok(false);
         }
@@ -2376,6 +2388,15 @@ fn grow_decommission_target_reservation(
             .predicted_physical_bytes
             .saturating_add(reservation.temporary_physical_bytes);
         let allocation_growth = reservation.remaining_peak_physical_bytes().saturating_sub(current_peak);
+        let required_allocation = existing_reserved.saturating_add(allocation_growth);
+        if required_allocation > allocation_limit {
+            return Err(decommission_target_no_longer_fits_error(
+                target_pool_index,
+                format!(
+                    "target pool {target_pool_index} reservation requires {required_allocation} physical bytes, but its observed allocation limit is {allocation_limit} bytes"
+                ),
+            ));
+        }
         if let Some(target) = reservation
             .targets
             .iter_mut()
@@ -37890,10 +37911,9 @@ mod pools_tests {
         );
         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"
-        );
+        let error = grow_decommission_target_reservation(&mut meta, &shortage, 0, 1, 32, uuid::Uuid::new_v4(), now)
+            .expect_err("real free-space shortage still rejects growth");
+        assert_eq!(super::decommission_capacity_stale_target_index(&error), Some(1));
         assert_eq!(serde_json::to_value(&meta).expect("snapshot after rejected growth"), before);
     }
 

crates/ecstore/src/core/pools_test.rs

@@ -1616,7 +1616,7 @@ mod decommission_lock_order_tests {
             run_routine_to_a_deferral_pause(
                 &store,
                 1,
-                "target pool 1 no longer fits the selected mutation: source pool 0 reservation estimate is exhausted",
+                "target pool 1 no longer fits the selected mutation: target pool 1 reservation requires",
             )
             .await;
         });
@@ -1647,7 +1647,7 @@ mod decommission_lock_order_tests {
             run_routine_to_a_deferral_pause(
                 &store,
                 1,
-                "target pool 1 no longer fits the selected mutation: source pool 0 reservation estimate is exhausted",
+                "target pool 1 no longer fits the selected mutation: target pool 1 reservation requires",
             )
             .await;
         });
@@ -6062,6 +6062,12 @@ mod decommission_lock_order_tests {
             )
             .await
             .expect("read the next source object for the multipart capacity oracle");
+        // Reconciled bytes remain charged. With no physical room for a new
+        // reservation, a distinct mutation must fail instead of reusing them.
+        let mut exhausted_capacity = capacity_snapshot();
+        exhausted_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, 0, target_total, target_total);
+        set_decommission_capacity_info_overrides_for_test(other_store.id, vec![exhausted_capacity]);
+
         let next_result = tokio::time::timeout(
             Duration::from_secs(30),
             data_movement::migrate_decommission_object(
@@ -7827,6 +7833,12 @@ mod decommission_lock_order_tests {
             .expect("same-mutation retry should publish a readable target");
         assert_eq!(retry_info.size, i64::try_from(body.len()).expect("test body size fits i64"));
 
+        // Reconciled bytes remain charged. With no physical room for a new
+        // reservation, a distinct mutation must fail instead of reusing them.
+        let mut exhausted_capacity = capacity_snapshot();
+        exhausted_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, 0, target_total, target_total);
+        set_decommission_capacity_info_overrides_for_test(other_store.id, vec![exhausted_capacity]);
+
         let next_object = "data-movement-put-after-retry.bin";
         let next_version_id = uuid::Uuid::new_v4().to_string();
         let mut next_opts = ObjectOptions {
@@ -7841,7 +7853,11 @@ mod decommission_lock_order_tests {
         let next_result = other_store
             .put_object_for_data_movement(RUSTFS_META_BUCKET, next_object, &mut next_data, &next_opts, None)
             .await;
-        assert!(next_result.is_err(), "a different mutation must not reuse A's recovered capacity");
+        let next_error = match next_result {
+            Err(error) | Ok((_, Err(error))) => error,
+            Ok((_, Ok(_))) => panic!("a different mutation must not reuse A's recovered capacity"),
+        };
+        assert!(next_error.to_string().contains("decommission_capacity_blocked"), "{next_error}");
         let next_target_err = other_store.pools[2]
             .get_object_info(
                 RUSTFS_META_BUCKET,
@@ -9929,6 +9945,12 @@ mod decommission_lock_order_tests {
             )
             .await
             .expect("read the next source object for the capacity oracle");
+        // Reconciled bytes remain charged. With no physical room for a new
+        // reservation, a distinct mutation must fail instead of reusing them.
+        let mut exhausted_capacity = capacity_snapshot();
+        exhausted_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, 0, target_total, target_total);
+        set_decommission_capacity_info_overrides_for_test(other_store.id, vec![exhausted_capacity]);
+
         let next_result = tokio::time::timeout(
             Duration::from_secs(30),
             data_movement::migrate_decommission_object(
@@ -10190,11 +10212,20 @@ mod decommission_lock_order_tests {
         assert_eq!(reconciled_reservation.pending_target_physical_bytes, 0);
         assert_eq!(reconciled_reservation.consumed_target_physical_bytes, 1);
 
+        // Reconciled bytes remain charged. With no physical room for a new
+        // reservation, a distinct mutation must fail instead of reusing them.
+        let mut exhausted_capacity = capacity_snapshot();
+        exhausted_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, 1, 2, 1);
+        set_decommission_capacity_info_overrides_for_test(other_store.id, vec![exhausted_capacity]);
+
         let next_after_reconcile = other_store
             .delete_object(&bucket, next_object, next_opts)
             .await
             .expect_err("the consumed exact-fit capacity must reject the next marker");
-        assert!(next_after_reconcile.to_string().contains("decommission_capacity_blocked"));
+        assert!(
+            next_after_reconcile.to_string().contains("decommission_capacity_blocked"),
+            "{next_after_reconcile}"
+        );
         let next_target_err = other_store.pools[2]
             .get_object_info(
                 &bucket,