fix(pool): guard decommission lifecycle recovery

src/reconcile/pool_lifecycle.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 use std::time::Duration;
 
 use k8s_openapi::api::apps::v1::StatefulSet;
@@ -59,6 +59,12 @@ struct MatchedRustfsPool {
     expected_endpoint_set_hash: String,
 }
 
+struct PoolLifecycleSnapshot {
+    state: Option<PoolLifecycleState>,
+    decommission: Option<PoolDecommissionStatus>,
+    block_start_decommission: bool,
+}
+
 enum PoolMappingError {
     ListFailed(String),
     NotFound(String),
@@ -125,6 +131,7 @@ pub(super) async fn reconcile_pool_lifecycle(
     namespace: &str,
 ) -> Result<PoolLifecycleDecisions, Error> {
     let mut decisions = PoolLifecycleDecisions::default();
+    let blocked_start_pools = start_decommission_requests_that_would_remove_all_live_pools(tenant);
 
     for pool in &tenant.spec.pools {
         let request = tenant
@@ -134,6 +141,11 @@ pub(super) async fn reconcile_pool_lifecycle(
             .and_then(|lifecycle| lifecycle.request_for_pool(&pool.name));
         let existing = existing_decommission_status(tenant, &pool.name);
         let existing_state = existing_lifecycle_state(tenant, &pool.name);
+        let snapshot = PoolLifecycleSnapshot {
+            state: existing_state.clone(),
+            decommission: existing,
+            block_start_decommission: blocked_start_pools.contains(&pool.name),
+        };
 
         let should_continue = matches!(
             existing_state,
@@ -149,16 +161,8 @@ pub(super) async fn reconcile_pool_lifecycle(
             continue;
         }
 
-        let decision = reconcile_single_pool_lifecycle(
-            ctx,
-            tenant,
-            namespace,
-            pool,
-            request,
-            existing_state,
-            existing,
-        )
-        .await;
+        let decision =
+            reconcile_single_pool_lifecycle(ctx, tenant, namespace, pool, request, snapshot).await;
 
         decisions.insert(pool.name.clone(), decision);
     }
@@ -172,9 +176,14 @@ async fn reconcile_single_pool_lifecycle(
     namespace: &str,
     pool: &Pool,
     request: Option<&DecommissionRequest>,
-    existing_state: Option<PoolLifecycleState>,
-    existing: Option<PoolDecommissionStatus>,
+    existing: PoolLifecycleSnapshot,
 ) -> PoolLifecycleDecision {
+    let PoolLifecycleSnapshot {
+        state: existing_state,
+        decommission: existing,
+        block_start_decommission,
+    } = existing;
+
     if matches!(existing_state, Some(PoolLifecycleState::Decommissioned)) {
         let status = existing.unwrap_or_else(|| PoolDecommissionStatus {
             phase: Some(PoolDecommissionPhase::Complete),
@@ -227,11 +236,11 @@ async fn reconcile_single_pool_lifecycle(
         );
     }
 
-    if request.action == DecommissionAction::Start && tenant.spec.pools.len() <= 1 {
+    if should_block_last_live_pool_decommission(request, block_start_decommission) {
         return failed_decision(
             Some(request.request_id.clone()),
             "LastPoolBlocked",
-            "cannot decommission the last pool in a tenant",
+            "cannot decommission the last live pool in a tenant",
         );
     }
 
@@ -893,13 +902,78 @@ fn terminal_decision_from_existing(
     existing: Option<PoolDecommissionStatus>,
 ) -> PoolLifecycleDecision {
     let state = existing_state.unwrap_or(PoolLifecycleState::Active);
+    if matches!(
+        state,
+        PoolLifecycleState::DecommissionCanceled | PoolLifecycleState::DecommissionFailed
+    ) {
+        return active_lifecycle_decision();
+    }
+
     PoolLifecycleDecision {
         skip_workload_reconcile: !matches!(state, PoolLifecycleState::Active),
         state,
         decommission: existing,
     }
 }
 
+fn start_decommission_requests_that_would_remove_all_live_pools(
+    tenant: &Tenant,
+) -> BTreeSet<String> {
+    let live_pool_names = live_pool_names_for_decommission_guard(tenant);
+    if live_pool_names.is_empty() {
+        return BTreeSet::new();
+    }
+
+    let requested_live_starts = tenant
+        .spec
+        .pool_lifecycle
+        .as_ref()
+        .map(|lifecycle| {
+            lifecycle
+                .decommission_requests
+                .iter()
+                .filter(|request| {
+                    request.action == DecommissionAction::Start
+                        && live_pool_names.contains(&request.pool_name)
+                })
+                .map(|request| request.pool_name.clone())
+                .collect::<BTreeSet<_>>()
+        })
+        .unwrap_or_default();
+
+    if requested_live_starts.len() >= live_pool_names.len() {
+        requested_live_starts
+    } else {
+        BTreeSet::new()
+    }
+}
+
+fn should_block_last_live_pool_decommission(
+    request: &DecommissionRequest,
+    block_start_decommission: bool,
+) -> bool {
+    request.action == DecommissionAction::Start && block_start_decommission
+}
+
+fn live_pool_names_for_decommission_guard(tenant: &Tenant) -> BTreeSet<String> {
+    tenant
+        .spec
+        .pools
+        .iter()
+        .filter(|pool| {
+            pool_counts_as_live_for_decommission_guard(existing_lifecycle_state(tenant, &pool.name))
+        })
+        .map(|pool| pool.name.clone())
+        .collect()
+}
+
+fn pool_counts_as_live_for_decommission_guard(state: Option<PoolLifecycleState>) -> bool {
+    !matches!(
+        state,
+        Some(PoolLifecycleState::Decommissioning | PoolLifecycleState::Decommissioned)
+    )
+}
+
 fn active_lifecycle_decision() -> PoolLifecycleDecision {
     PoolLifecycleDecision {
         state: PoolLifecycleState::Active,
@@ -1101,20 +1175,69 @@ mod tests {
     }
 
     fn test_tenant(pool: Pool) -> Tenant {
+        test_tenant_with_pools(vec![pool])
+    }
+
+    fn test_tenant_with_pools(pools: Vec<Pool>) -> Tenant {
         Tenant {
             metadata: ObjectMeta {
                 name: Some("logs".to_string()),
                 namespace: Some("rustfs-system".to_string()),
                 ..Default::default()
             },
             spec: TenantSpec {
-                pools: vec![pool],
+                pools,
                 ..Default::default()
             },
             status: None,
         }
     }
 
+    fn test_pool_status(
+        pool_name: &str,
+        lifecycle_state: PoolLifecycleState,
+    ) -> crate::types::v1alpha1::status::pool::Pool {
+        crate::types::v1alpha1::status::pool::Pool {
+            name: Some(pool_name.to_string()),
+            ss_name: format!("logs-{pool_name}"),
+            state: crate::types::v1alpha1::status::pool::PoolState::RolloutComplete,
+            lifecycle_state: Some(lifecycle_state),
+            workload_state: Some(crate::types::v1alpha1::status::pool::PoolState::RolloutComplete),
+            decommission: None,
+            replicas: Some(4),
+            ready_replicas: Some(4),
+            current_replicas: Some(4),
+            updated_replicas: Some(4),
+            current_revision: Some("rev-a".to_string()),
+            update_revision: Some("rev-a".to_string()),
+            last_update_time: None,
+        }
+    }
+
+    fn start_request(pool_name: &str) -> DecommissionRequest {
+        start_request_with_id(pool_name, "request-1")
+    }
+
+    fn start_request_with_id(pool_name: &str, request_id: &str) -> DecommissionRequest {
+        DecommissionRequest {
+            pool_name: pool_name.to_string(),
+            request_id: request_id.to_string(),
+            action: DecommissionAction::Start,
+            requested_at: Some("2026-05-20T00:00:05Z".to_string()),
+            cancel_requested_at: None,
+            reason: None,
+        }
+    }
+
+    fn set_decommission_requests(tenant: &mut Tenant, requests: Vec<DecommissionRequest>) {
+        tenant.spec.pool_lifecycle =
+            Some(crate::types::v1alpha1::pool_lifecycle::PoolLifecycleSpec {
+                pvc_retention_policy:
+                    crate::types::v1alpha1::pool_lifecycle::PvcRetentionPolicy::Retain,
+                decommission_requests: requests,
+            });
+    }
+
     #[test]
     fn lifecycle_phase_maps_current_rustfs_terminal_flags() {
         let complete = RustfsPoolDecommissionInfo {
@@ -1176,6 +1299,84 @@ mod tests {
         assert_eq!(decision.state, PoolLifecycleState::Decommissioned);
     }
 
+    #[test]
+    fn canceled_or_failed_terminal_state_without_request_restores_active_reconcile() {
+        for state in [
+            PoolLifecycleState::DecommissionCanceled,
+            PoolLifecycleState::DecommissionFailed,
+        ] {
+            let decision =
+                terminal_decision_from_existing(Some(state), Some(empty_decommission_status()));
+
+            assert_eq!(decision.state, PoolLifecycleState::Active);
+            assert!(!decision.skip_workload_reconcile);
+            assert!(decision.decommission.is_none());
+        }
+    }
+
+    #[test]
+    fn start_decommission_blocks_last_live_pool() {
+        let old_pool = test_pool("old-pool");
+        let active_pool = test_pool("active-pool");
+        let mut tenant = test_tenant_with_pools(vec![old_pool, active_pool.clone()]);
+        let request = start_request("active-pool");
+        set_decommission_requests(&mut tenant, vec![request.clone()]);
+        tenant.status = Some(crate::types::v1alpha1::status::Status {
+            pools: vec![
+                test_pool_status("old-pool", PoolLifecycleState::Decommissioned),
+                test_pool_status("active-pool", PoolLifecycleState::Active),
+            ],
+            ..Default::default()
+        });
+
+        assert_eq!(live_pool_names_for_decommission_guard(&tenant).len(), 1);
+        let blocked_pools = start_decommission_requests_that_would_remove_all_live_pools(&tenant);
+        assert!(blocked_pools.contains("active-pool"));
+        assert!(should_block_last_live_pool_decommission(
+            &request,
+            blocked_pools.contains(&active_pool.name)
+        ));
+    }
+
+    #[test]
+    fn start_decommission_allows_one_of_multiple_live_pools() {
+        let pool_a = test_pool("pool-a");
+        let pool_b = test_pool("pool-b");
+        let mut tenant = test_tenant_with_pools(vec![pool_a.clone(), pool_b]);
+        let request = start_request("pool-a");
+        set_decommission_requests(&mut tenant, vec![request.clone()]);
+
+        assert_eq!(live_pool_names_for_decommission_guard(&tenant).len(), 2);
+        let blocked_pools = start_decommission_requests_that_would_remove_all_live_pools(&tenant);
+        assert!(blocked_pools.is_empty());
+        assert!(!should_block_last_live_pool_decommission(
+            &request,
+            blocked_pools.contains(&pool_a.name)
+        ));
+    }
+
+    #[test]
+    fn start_decommission_blocks_batch_that_would_remove_all_live_pools() {
+        let pool_a = test_pool("pool-a");
+        let pool_b = test_pool("pool-b");
+        let mut tenant = test_tenant_with_pools(vec![pool_a.clone(), pool_b.clone()]);
+        let request_a = start_request_with_id("pool-a", "request-a");
+        let request_b = start_request_with_id("pool-b", "request-b");
+        set_decommission_requests(&mut tenant, vec![request_a.clone(), request_b.clone()]);
+
+        assert_eq!(live_pool_names_for_decommission_guard(&tenant).len(), 2);
+        let blocked_pools = start_decommission_requests_that_would_remove_all_live_pools(&tenant);
+        assert_eq!(blocked_pools.len(), 2);
+        assert!(should_block_last_live_pool_decommission(
+            &request_a,
+            blocked_pools.contains(&pool_a.name)
+        ));
+        assert!(should_block_last_live_pool_decommission(
+            &request_b,
+            blocked_pools.contains(&pool_b.name)
+        ));
+    }
+
     #[test]
     fn decommissioned_cleanup_pending_keeps_tenant_reconciling() {
         let mut status = empty_decommission_status();