From e14ea59dbbbc172b7a16c2965c02980eb3917177 Mon Sep 17 00:00:00 2001 From: GatewayJ <18332154+GatewayJ@users.noreply.github.com> Date: Sat, 11 Jul 2026 11:59:07 +0800 Subject: [PATCH 1/2] fix(pool): guard decommission lifecycle recovery --- src/reconcile/pool_lifecycle.rs | 127 +++++++++++++++++++++++++++++++- 1 file changed, 124 insertions(+), 3 deletions(-) diff --git a/src/reconcile/pool_lifecycle.rs b/src/reconcile/pool_lifecycle.rs index b97ec63..ac4a659 100644 --- a/src/reconcile/pool_lifecycle.rs +++ b/src/reconcile/pool_lifecycle.rs @@ -227,11 +227,11 @@ async fn reconcile_single_pool_lifecycle( ); } - if request.action == DecommissionAction::Start && tenant.spec.pools.len() <= 1 { + if should_block_last_live_pool_decommission(tenant, pool, request) { 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,6 +893,13 @@ fn terminal_decision_from_existing( existing: Option, ) -> 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, @@ -900,6 +907,34 @@ fn terminal_decision_from_existing( } } +fn should_block_last_live_pool_decommission( + tenant: &Tenant, + pool: &Pool, + request: &DecommissionRequest, +) -> bool { + request.action == DecommissionAction::Start + && pool_counts_as_live_for_decommission_guard(existing_lifecycle_state(tenant, &pool.name)) + && live_pool_count_for_decommission_guard(tenant) <= 1 +} + +fn live_pool_count_for_decommission_guard(tenant: &Tenant) -> usize { + tenant + .spec + .pools + .iter() + .filter(|pool| { + pool_counts_as_live_for_decommission_guard(existing_lifecycle_state(tenant, &pool.name)) + }) + .count() +} + +fn pool_counts_as_live_for_decommission_guard(state: Option) -> bool { + !matches!( + state, + Some(PoolLifecycleState::Decommissioning | PoolLifecycleState::Decommissioned) + ) +} + fn active_lifecycle_decision() -> PoolLifecycleDecision { PoolLifecycleDecision { state: PoolLifecycleState::Active, @@ -1101,6 +1136,10 @@ mod tests { } fn test_tenant(pool: Pool) -> Tenant { + test_tenant_with_pools(vec![pool]) + } + + fn test_tenant_with_pools(pools: Vec) -> Tenant { Tenant { metadata: ObjectMeta { name: Some("logs".to_string()), @@ -1108,13 +1147,45 @@ mod tests { ..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 { + DecommissionRequest { + pool_name: pool_name.to_string(), + request_id: "request-1".to_string(), + action: DecommissionAction::Start, + requested_at: Some("2026-05-20T00:00:05Z".to_string()), + cancel_requested_at: None, + reason: None, + } + } + #[test] fn lifecycle_phase_maps_current_rustfs_terminal_flags() { let complete = RustfsPoolDecommissionInfo { @@ -1176,6 +1247,56 @@ 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()]); + 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_count_for_decommission_guard(&tenant), 1); + assert!(should_block_last_live_pool_decommission( + &tenant, + &active_pool, + &start_request("active-pool") + )); + } + + #[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 tenant = test_tenant_with_pools(vec![pool_a.clone(), pool_b]); + + assert_eq!(live_pool_count_for_decommission_guard(&tenant), 2); + assert!(!should_block_last_live_pool_decommission( + &tenant, + &pool_a, + &start_request("pool-a") + )); + } + #[test] fn decommissioned_cleanup_pending_keeps_tenant_reconciling() { let mut status = empty_decommission_status(); From fba6d598f7a25451ef3ee9b5d56cd2506fa58279 Mon Sep 17 00:00:00 2001 From: GatewayJ <18332154+GatewayJ@users.noreply.github.com> Date: Sat, 11 Jul 2026 12:13:45 +0800 Subject: [PATCH 2/2] fix(pool): block batch live pool decommission --- src/reconcile/pool_lifecycle.rs | 142 +++++++++++++++++++++++++------- 1 file changed, 111 insertions(+), 31 deletions(-) diff --git a/src/reconcile/pool_lifecycle.rs b/src/reconcile/pool_lifecycle.rs index ac4a659..bdce995 100644 --- a/src/reconcile/pool_lifecycle.rs +++ b/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, + decommission: Option, + block_start_decommission: bool, +} + enum PoolMappingError { ListFailed(String), NotFound(String), @@ -125,6 +131,7 @@ pub(super) async fn reconcile_pool_lifecycle( namespace: &str, ) -> Result { 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, - existing: Option, + 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,7 +236,7 @@ async fn reconcile_single_pool_lifecycle( ); } - if should_block_last_live_pool_decommission(tenant, pool, request) { + if should_block_last_live_pool_decommission(request, block_start_decommission) { return failed_decision( Some(request.request_id.clone()), "LastPoolBlocked", @@ -907,17 +916,46 @@ fn terminal_decision_from_existing( } } -fn should_block_last_live_pool_decommission( +fn start_decommission_requests_that_would_remove_all_live_pools( tenant: &Tenant, - pool: &Pool, +) -> BTreeSet { + 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::>() + }) + .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 - && pool_counts_as_live_for_decommission_guard(existing_lifecycle_state(tenant, &pool.name)) - && live_pool_count_for_decommission_guard(tenant) <= 1 + request.action == DecommissionAction::Start && block_start_decommission } -fn live_pool_count_for_decommission_guard(tenant: &Tenant) -> usize { +fn live_pool_names_for_decommission_guard(tenant: &Tenant) -> BTreeSet { tenant .spec .pools @@ -925,7 +963,8 @@ fn live_pool_count_for_decommission_guard(tenant: &Tenant) -> usize { .filter(|pool| { pool_counts_as_live_for_decommission_guard(existing_lifecycle_state(tenant, &pool.name)) }) - .count() + .map(|pool| pool.name.clone()) + .collect() } fn pool_counts_as_live_for_decommission_guard(state: Option) -> bool { @@ -1176,9 +1215,13 @@ mod tests { } 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-1".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, @@ -1186,6 +1229,15 @@ mod tests { } } + fn set_decommission_requests(tenant: &mut Tenant, requests: Vec) { + 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 { @@ -1267,6 +1319,8 @@ mod tests { 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), @@ -1275,11 +1329,12 @@ mod tests { ..Default::default() }); - assert_eq!(live_pool_count_for_decommission_guard(&tenant), 1); + 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( - &tenant, - &active_pool, - &start_request("active-pool") + &request, + blocked_pools.contains(&active_pool.name) )); } @@ -1287,13 +1342,38 @@ mod tests { fn start_decommission_allows_one_of_multiple_live_pools() { let pool_a = test_pool("pool-a"); let pool_b = test_pool("pool-b"); - let tenant = test_tenant_with_pools(vec![pool_a.clone(), 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_count_for_decommission_guard(&tenant), 2); + 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( - &tenant, - &pool_a, - &start_request("pool-a") + &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) )); }