From 57b95b8ca3031aa3e1d699a1c24e4c03ab89b679 Mon Sep 17 00:00:00 2001 From: Yevgen Polyak Date: Fri, 31 Jul 2026 10:43:51 +1200 Subject: [PATCH 1/2] fix stuck finalizer on delete --- .../172-fix-stuck-finalizer-on-delete.md | 36 +++++++++ .../reconcilers/replicaset.rs | 73 +++++++++++++++++-- 2 files changed, 102 insertions(+), 7 deletions(-) create mode 100644 release-notes/unreleased/172-fix-stuck-finalizer-on-delete.md diff --git a/release-notes/unreleased/172-fix-stuck-finalizer-on-delete.md b/release-notes/unreleased/172-fix-stuck-finalizer-on-delete.md new file mode 100644 index 00000000..723d0971 --- /dev/null +++ b/release-notes/unreleased/172-fix-stuck-finalizer-on-delete.md @@ -0,0 +1,36 @@ +# Release Notes for Issue #172: Fix RestateDeployment finalizer stuck in CleanupFailed loop on delete + +## Bug Fix + +### What Changed + +When deleting a `RestateDeployment`, the finalizer no longer gets permanently +stuck with `CleanupFailed(DeploymentInUse)`. Previously, `cleanup_old_replicasets` +treated the latest ReplicaSet's Restate deployment as an unconditional blocker +because it is always `active = true` in `sys_service`. Since no new version +ever registers during deletion, the `active_count > 0` check caused an infinite +retry loop with no way out. + +When `rsd.deletion_timestamp.is_some()`, active deployments are now scheduled +for drain (respecting `spec.restate.drainDelaySeconds`) rather than treated as +a permanent blocker. After the drain period they proceed through the existing +force-delete path. + +### Why This Matters + +This is the common case in **ephemeral PR/preview environments**: a service is +deployed, registered with Restate, but no workflows are ever invoked before the +environment is torn down. Without this fix, deleting the `RestateDeployment` +would stall forever and require manual intervention to remove the finalizer. + +### Impact on Users + +- **Existing deployments being deleted:** Stuck `RestateDeployment` objects will + make progress on the next reconcile after upgrading. +- **New deletions:** Behave as expected — drain delay is respected, then the + Restate deployment is force-deleted and the finalizer released. +- **No migration required.** + +### Related Issues + +- Issue #172: RestateDeployment finalizer stuck in CleanupFailed(DeploymentInUse) loop when no invocations ran diff --git a/src/controllers/restatedeployment/reconcilers/replicaset.rs b/src/controllers/restatedeployment/reconcilers/replicaset.rs index 23266d40..dde4797b 100644 --- a/src/controllers/restatedeployment/reconcilers/replicaset.rs +++ b/src/controllers/restatedeployment/reconcilers/replicaset.rs @@ -341,7 +341,70 @@ pub async fn cleanup_old_replicasets( let deployment_exists = deployment.is_some(); let deployment_active = deployment.unwrap_or(false); - if deployment_active { + // During deletion, once the drain period has passed for an active version, + // fall through to the force-delete path below (same as inactive versions). + let is_deleting = rsd.metadata.deletion_timestamp.is_some(); + let drain_annotation = rs + .annotations() + .get(RESTATE_REMOVE_VERSION_AT_ANNOTATION) + .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok().map(|t| t.to_utc())); + let drain_due = drain_annotation.is_some_and(|t| t < now); + + if deployment_active && !(is_deleting && drain_due) { + if is_deleting { + // During deletion: schedule drain for active versions (latest-in-sys_service + // or still-pinned) so the finalizer can eventually complete. Don't reset any + // existing timer — unlike the rollback case, we're committed to removal. + match drain_annotation { + None => { + let drain_delay_seconds = rsd.spec.restate.drain_delay_seconds(); + info!( + replicaset = %rs_name, + namespace = %namespace, + drain_delay_seconds, + "RestateDeployment being deleted; scheduling active version for drain" + ); + let remove_at = chrono::Utc::now() + .checked_add_signed(chrono::TimeDelta::seconds(drain_delay_seconds)) + .expect("remove_version_at in bounds"); + + rs_api + .patch_metadata( + &rs_name, + &PatchParams::apply("restate-operator/remove-version-at").force(), + &Patch::Apply( + ObjectMeta { + annotations: Some( + [( + RESTATE_REMOVE_VERSION_AT_ANNOTATION.to_string(), + remove_at.to_rfc3339(), + )] + .into(), + ), + ..Default::default() + } + .into_request_partial::(), + ), + ) + .await?; + + next_removal = match next_removal { + None => Some(remove_at), + Some(nr) if nr > remove_at => Some(remove_at), + els => els, + }; + } + Some(remove_at) => { + next_removal = match next_removal { + None => Some(remove_at), + Some(nr) if nr > remove_at => Some(remove_at), + els => els, + }; + } + } + continue; + } + active_count += 1; // Per-version autoscaling: a non-latest version has an operator HPA @@ -350,7 +413,7 @@ pub async fn cleanup_old_replicasets( // handled unconditionally below, before scale-down.) match super::autoscaling::plan_active_version_hpa( rsd.spec.autoscaling.is_some(), - rsd.metadata.deletion_timestamp.is_some(), + false, // not deleting (is_deleting is false here) ) { HpaPlan::Ensure => { if let Some(template) = rsd.spec.autoscaling.as_ref() @@ -420,11 +483,7 @@ pub async fn cleanup_old_replicasets( HpaPlan::Skip => {} } - if rs - .annotations() - .get(RESTATE_REMOVE_VERSION_AT_ANNOTATION) - .is_none() - { + if drain_annotation.is_none() { // not scheduled for removal; all good. continue; } From 4ae83c3b0cedc15ba05ab121d585be138bd134d7 Mon Sep 17 00:00:00 2001 From: Yevgen Polyak Date: Tue, 11 Aug 2026 08:26:51 +1200 Subject: [PATCH 2/2] RestateDeployment: only wait for pinned invocations when deleting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit list_deployments returned a single flag that conflated two signals, so the latest version — always the latest entry in sys_service — blocked its own finalizer forever. Nothing can supersede it once the object is being deleted, so the retry loop had no way out. Split it into and . Outside deletion either signal keeps a version alive, unchanged. While deleting, only live pinned invocations block: they are held for drainDelaySeconds and then force-deregistered, while everything else is torn down immediately. Also bypass revisionHistoryLimit during deletion, which was skipping the admin-API deregistration and orphaning registrations. Mirrored in the Knative Configuration cleanup path. --- CLAUDE.md | 26 +-- .../172-fix-stuck-finalizer-on-delete.md | 35 +++- .../restatedeployment/controller.rs | 111 +++++++++--- .../restatedeployment/reconcilers/knative.rs | 152 ++++++++++------ .../reconcilers/replicaset.rs | 164 +++++++++--------- 5 files changed, 303 insertions(+), 185 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index d913fdad..70c156cd 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -206,24 +206,24 @@ Both RestateCluster and RestateDeployment use finalizers for cleanup: ### Restate Invocation Lifecycle and Deployment Status -**Important**: The operator aligns with Restate's invocation retention model when determining deployment activity: +**Important**: `list_deployments` (`src/controllers/restatedeployment/controller.rs`) queries the admin API for two independent signals per registered deployment, returning them as a `DeploymentState`: -- **Invocation retention**: Completed invocations remain in `sys_invocation_status` for 24 hours (default) before automatic purging -- **Deployment status**: A deployment is considered "active" if it has ANY invocation in `sys_invocation_status`, including completed ones -- **Cleanup timing**: Configurations tied to "active" deployments are retained until Restate purges the invocations -- **Manual override**: Use `restate invocations purge ` to immediately purge completed invocations for testing +- **`latest_endpoint`**: the deployment serves the latest revision of at least one service (`sys_service`) +- **`has_pinned_invocations`**: at least one *non-completed* invocation is pinned to it (`sys_invocation_status WHERE status != 'completed'`) -**Deployment states** (`restate deployments list`): -- `Active`: Has latest service revision -- `Draining`: Superseded but has pinned invocations (including completed) -- `Drained`: Superseded and all invocations purged (active_inv == 0) +Only one of them is a reason to wait, and which one depends on whether the RestateDeployment is being deleted: + +- **Normal reconcile**: either signal makes a version "active" — it is kept, and any pending removal timer is reset. +- **Being deleted** (`deletion_timestamp` set): only `has_pinned_invocations` blocks. Nothing can supersede the latest endpoint once the object is going away, so waiting on it would wedge the finalizer forever (issue #172). A version with live pinned invocations is held for `spec.restate.drainDelaySeconds` and then force-deregistered; `spec.revisionHistoryLimit` is bypassed so every version is actually deregistered before the finalizer is released. + +Completed invocations do **not** keep a deployment alive — the operator does not wait for Restate's 24h retention purge (changed in #71). Note this differs from what `restate deployments list` reports: it shows a superseded deployment as `Draining` while completed invocations are still retained, and `Drained` only once they are purged. **SQL tables**: -- `sys_invocation_status`: ALL invocations including completed (used by operator's `list_deployments` query) -- `sys_invocation`: Same content as sys_invocation_status -- Both tables include a `status` column to filter by invocation state +- `sys_service`: current service revisions, one row per service +- `sys_invocation_status`: ALL invocations including completed; the `status` column is what the operator filters on +- `sys_invocation`: same content as `sys_invocation_status` -The operator's cleanup logic intentionally waits for Restate's invocation purge before considering a deployment truly inactive, ensuring Configuration lifecycle aligns with Restate's internal state management. +For testing, `restate invocations purge ` immediately purges a completed invocation. ## Helm Chart diff --git a/release-notes/unreleased/172-fix-stuck-finalizer-on-delete.md b/release-notes/unreleased/172-fix-stuck-finalizer-on-delete.md index 723d0971..1d9f96fe 100644 --- a/release-notes/unreleased/172-fix-stuck-finalizer-on-delete.md +++ b/release-notes/unreleased/172-fix-stuck-finalizer-on-delete.md @@ -5,16 +5,31 @@ ### What Changed When deleting a `RestateDeployment`, the finalizer no longer gets permanently -stuck with `CleanupFailed(DeploymentInUse)`. Previously, `cleanup_old_replicasets` -treated the latest ReplicaSet's Restate deployment as an unconditional blocker -because it is always `active = true` in `sys_service`. Since no new version +stuck with `CleanupFailed(DeploymentInUse)`. Previously, cleanup treated the +latest version's Restate deployment as an unconditional blocker because it is +always "active" — it is the latest entry in `sys_service`. Since no new version ever registers during deletion, the `active_count > 0` check caused an infinite retry loop with no way out. -When `rsd.deletion_timestamp.is_some()`, active deployments are now scheduled -for drain (respecting `spec.restate.drainDelaySeconds`) rather than treated as -a permanent blocker. After the drain period they proceed through the existing -force-delete path. +The operator now tracks the two liveness signals separately instead of one +`active` flag: + +- `latest_endpoint` — serves the latest revision of a service (`sys_service`) +- `has_pinned_invocations` — has a non-completed invocation pinned to it + (`sys_invocation_status`) + +Outside deletion nothing changes: either signal keeps a version alive. During +deletion only pinned invocations are worth waiting for — a version that is +merely the latest endpoint has nothing to drain, so it is deregistered and +removed immediately. A version with live pinned invocations is held for +`spec.restate.drainDelaySeconds` first, then force-deregistered as before. + +Cleanup during deletion also bypasses `spec.revisionHistoryLimit`. Retaining +versions for rollback made no sense once the object is going away, and it left +the Restate deployments registered forever with nothing to deregister them +after the finalizer was released. + +Both deployment modes are covered — ReplicaSet and Knative (Configurations). ### Why This Matters @@ -27,8 +42,10 @@ would stall forever and require manual intervention to remove the finalizer. - **Existing deployments being deleted:** Stuck `RestateDeployment` objects will make progress on the next reconcile after upgrading. -- **New deletions:** Behave as expected — drain delay is respected, then the - Restate deployment is force-deleted and the finalizer released. +- **New deletions with no live invocations:** Complete immediately, without + waiting out the drain delay. +- **New deletions with live pinned invocations:** Drain delay is respected, then + the Restate deployment is force-deleted and the finalizer released. - **No migration required.** ### Related Issues diff --git a/src/controllers/restatedeployment/controller.rs b/src/controllers/restatedeployment/controller.rs index ee13d561..5532abad 100644 --- a/src/controllers/restatedeployment/controller.rs +++ b/src/controllers/restatedeployment/controller.rs @@ -211,6 +211,36 @@ fn error_policy(_rs: Arc, _: &Error, _ctx: C) -> Action { Action::requeue(Duration::from_secs(30)) } +/// Why Restate still considers a registered deployment live. The two signals are +/// kept apart because only one of them is a reason to wait: a version can be the +/// latest endpoint of a service and have never served an invocation. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct DeploymentState { + /// Serves the latest revision of at least one service (`sys_service`). + pub latest_endpoint: bool, + /// At least one non-completed invocation is pinned to it (`sys_invocation_status`). + pub has_pinned_invocations: bool, +} + +impl DeploymentState { + pub fn active(&self) -> bool { + self.latest_endpoint || self.has_pinned_invocations + } + + /// Whether this version must be left alone this reconcile. + /// + /// While the RestateDeployment is being deleted only live pinned invocations + /// count: nothing will ever supersede the latest endpoint now, so waiting on + /// it wedges the finalizer forever. Otherwise either signal keeps it. + pub fn blocks_removal(&self, is_deleting: bool) -> bool { + if is_deleting { + self.has_pinned_invocations + } else { + self.active() + } + } +} + impl RestateDeployment { /// Resolve the RestateCloudEnvironment values a `tunnelMode: in-process` /// deployment derives its identity from (None for every other mode). They feed @@ -442,8 +472,7 @@ impl RestateDeployment { if existing_deployment_id.is_none_or(|existing_deployment_id| { !deployments .get(existing_deployment_id) - .cloned() - .unwrap_or_default() + .is_some_and(DeploymentState::active) }) { let valid = async { if let Some(cluster_name) = &self.spec.restate.register.cluster { @@ -494,9 +523,15 @@ impl RestateDeployment { self.spec.restate.use_http11.as_ref().cloned(), ) .await?; - // if registration succeeded, treat this as an active endpoint + // if registration succeeded, treat this as the latest endpoint // if we fail after this point we will re-register and should get the same deployment id - deployments.insert(deployment_id.clone(), true); + deployments.insert( + deployment_id.clone(), + DeploymentState { + latest_endpoint: true, + has_pinned_invocations: false, + }, + ); debug!( "Updating deployment-id annotation of ReplicaSet/Service {versioned_name} in namespace {namespace}" @@ -940,22 +975,29 @@ impl RestateDeployment { Ok(resp.id) } - pub(super) async fn list_deployments(&self, ctx: &Context) -> Result> { - // This query finds deployments, noting those that are the latest for a particular service, or have an active invocation + pub(super) async fn list_deployments( + &self, + ctx: &Context, + ) -> Result> { + // This query finds deployments, noting separately those that are the latest for a + // particular service, and those that still have a live invocation pinned to them let sql_query = r#" - WITH active_deployments AS ( + WITH latest_deployments AS ( SELECT DISTINCT deployment_id as id FROM sys_service WHERE deployment_id IS NOT NULL - UNION + ), + pinned_deployments AS ( SELECT DISTINCT pinned_deployment_id as id FROM sys_invocation_status WHERE pinned_deployment_id IS NOT NULL AND status != 'completed' ) SELECT d.id as deployment_id, - a.id IS NOT NULL as active + l.id IS NOT NULL as latest_endpoint, + p.id IS NOT NULL as has_pinned_invocations FROM sys_deployment d - LEFT JOIN active_deployments a ON d.id = a.id; + LEFT JOIN latest_deployments l ON d.id = l.id + LEFT JOIN pinned_deployments p ON d.id = p.id; "#; #[derive(Deserialize)] @@ -966,7 +1008,8 @@ impl RestateDeployment { #[derive(Deserialize)] struct DeploymentQueryResultRow { deployment_id: String, - active: bool, + latest_endpoint: bool, + has_pinned_invocations: bool, } let resp = ctx @@ -984,21 +1027,15 @@ impl RestateDeployment { .await .map_err(Error::AdminCallFailed)?; - let mut endpoints = HashMap::with_capacity(response.rows.len()); + let mut endpoints: HashMap = + HashMap::with_capacity(response.rows.len()); for row in response.rows { - match endpoints.entry(row.deployment_id) { - std::collections::hash_map::Entry::Occupied(mut entry) => { - // two rows with same deployment id shouldnt happen... - // we treat the deployment as active if any row is active - if !entry.get() { - entry.insert(row.active); - } - } - std::collections::hash_map::Entry::Vacant(entry) => { - entry.insert(row.active); - } - } + // two rows with same deployment id shouldnt happen, but if they do, + // any row asserting a signal wins + let entry = endpoints.entry(row.deployment_id).or_default(); + entry.latest_endpoint |= row.latest_endpoint; + entry.has_pinned_invocations |= row.has_pinned_invocations; } Ok(endpoints) @@ -1557,4 +1594,30 @@ mod tests { let s1_again = latest_version_label_selector(&v1, None).expect("v1 selector again"); assert_eq!(s1, s1_again, "selector should be deterministic"); } + + #[test] + fn deletion_only_waits_for_pinned_invocations() { + let latest_only = DeploymentState { + latest_endpoint: true, + has_pinned_invocations: false, + }; + // outside deletion, the latest endpoint is the live version and is kept + assert!(latest_only.active()); + assert!(latest_only.blocks_removal(false)); + // ...but nothing can supersede it once the RestateDeployment is going away, + // and with no invocations there is nothing to drain (issue #172) + assert!(!latest_only.blocks_removal(true)); + + let pinned = DeploymentState { + latest_endpoint: false, + has_pinned_invocations: true, + }; + assert!(pinned.blocks_removal(false)); + assert!(pinned.blocks_removal(true)); + + let drained = DeploymentState::default(); + assert!(!drained.active()); + assert!(!drained.blocks_removal(false)); + assert!(!drained.blocks_removal(true)); + } } diff --git a/src/controllers/restatedeployment/reconcilers/knative.rs b/src/controllers/restatedeployment/reconcilers/knative.rs index 4ea59935..8820f2e9 100644 --- a/src/controllers/restatedeployment/reconcilers/knative.rs +++ b/src/controllers/restatedeployment/reconcilers/knative.rs @@ -9,7 +9,7 @@ use tracing::*; use url::Url; use crate::controllers::restatedeployment::controller::{ - Context, RESTATE_DEPLOYMENT_ID_ANNOTATION, + Context, DeploymentState, RESTATE_DEPLOYMENT_ID_ANNOTATION, }; use crate::controllers::restatedeployment::reconcilers::replicaset::generate_pod_template_hash; use crate::resources::knative::{ @@ -743,7 +743,7 @@ pub async fn cleanup_old_configurations( ctx: &Context, rsd_uid: &str, rsd: &RestateDeployment, - deployments: &std::collections::HashMap, + deployments: &std::collections::HashMap, active_tag: Option<&str>, ) -> Result<(i32, Option>)> { // Use reflector cache instead of API list() call @@ -807,6 +807,7 @@ pub async fn cleanup_old_configurations( let mut next_removal = None; let now = chrono::Utc::now(); + let is_deleting = rsd.metadata.deletion_timestamp.is_some(); for config in configurations { let config_name = config.name_any(); @@ -817,22 +818,29 @@ pub async fn cleanup_old_configurations( .as_ref() .and_then(|a| a.get(RESTATE_DEPLOYMENT_ID_ANNOTATION)); - // Skip active deployments let deployment = config_deployment_id - .and_then(|config_deployment_id| deployments.get(config_deployment_id).cloned()); + .and_then(|config_deployment_id| deployments.get(config_deployment_id).copied()); let deployment_exists = deployment.is_some(); - let deployment_active = deployment.unwrap_or(false); + let deployment = deployment.unwrap_or_default(); - if deployment_active { + let current_remove_at = config + .metadata + .annotations + .as_ref() + .and_then(|a| a.get(RESTATE_REMOVE_VERSION_AT_ANNOTATION)) + .and_then(|remove_at| { + chrono::DateTime::parse_from_rfc3339(remove_at) + .map(|t| t.to_utc()) + .ok() + }); + let current_remove_at_in_past = current_remove_at.is_some_and(|c| c < now); + let blocks_removal = deployment.blocks_removal(is_deleting); + + // Skip active deployments + if blocks_removal && !is_deleting { active_count += 1; - if config - .metadata - .annotations - .as_ref() - .and_then(|a| a.get(RESTATE_REMOVE_VERSION_AT_ANNOTATION)) - .is_none() - { + if current_remove_at.is_none() { // not scheduled for removal; all good. trace!( "Keeping active Configuration {} in namespace {namespace}", @@ -867,27 +875,51 @@ pub async fn cleanup_old_configurations( continue; } - let current_remove_at = config - .metadata - .annotations - .as_ref() - .and_then(|a| a.get(RESTATE_REMOVE_VERSION_AT_ANNOTATION)) - .and_then(|remove_at| { - chrono::DateTime::parse_from_rfc3339(remove_at) - .map(|t| t.to_utc()) - .ok() - }); + // Deleting, and the only thing left worth waiting for: a version with live + // pinned invocations, which the force-deregistration below would kill. Hold + // it for the drain delay, then tear it down like any other version. Unlike + // the rollback case above we never reset an existing timer — we're committed. + if is_deleting && blocks_removal && !current_remove_at_in_past { + let remove_at = match current_remove_at { + Some(remove_at) => remove_at, + None => { + let drain_delay_seconds = rsd.spec.restate.drain_delay_seconds(); + info!( + configuration = %config_name, + namespace = %namespace, + drain_delay_seconds, + "RestateDeployment being deleted; draining version with pinned invocations" + ); + let remove_at = now + .checked_add_signed(chrono::TimeDelta::seconds(drain_delay_seconds)) + .expect("remove_version_at in bounds"); - let current_remove_at_in_past = current_remove_at.is_some_and(|c| c < now); + schedule_configuration_removal(ctx, namespace, &config_name, remove_at).await?; + + remove_at + } + }; + + next_removal = match next_removal { + None => Some(remove_at), + Some(next_removal) if next_removal > remove_at => Some(remove_at), + els => els, + }; + + continue; + } + + // A deleting RestateDeployment never waits here: it either had nothing + // pinned, or its drain already elapsed above. + let teardown_due = current_remove_at_in_past || is_deleting; - match ( - current_remove_at, - current_remove_at_in_past, - deployment_exists, - ) { + match (current_remove_at, teardown_due, deployment_exists) { (_, true, _) | (_, _, false) => { - // we are past the remove-at time, or the endpoint was removed by other means; can now delete it (subject to the history limit) - if historic_count < rsd.spec.revision_history_limit { + // we are past the remove-at time, or the endpoint was removed by other means; can + // now delete it (subject to the history limit — except while deleting, where + // keeping it for rollback would just leave the Restate deployment registered with + // nothing left to deregister it once the finalizer is released) + if !is_deleting && historic_count < rsd.spec.revision_history_limit { historic_count += 1; trace!( "Keeping old Configuration {} in namespace {namespace} (within revision history limit: {}/{})", @@ -946,32 +978,11 @@ pub async fn cleanup_old_configurations( "Scheduling removal of old Configuration (after drain delay)" ); - let remove_at = chrono::Utc::now() + let remove_at = now .checked_add_signed(chrono::TimeDelta::seconds(drain_delay_seconds)) .expect("remove_version_at in bounds"); - let config_api: Api = Api::namespaced(ctx.client.clone(), namespace); - let params = PatchParams::apply("restate-operator/remove-version-at").force(); - - config_api - .patch_metadata( - &config_name, - ¶ms, - &Patch::Apply( - ObjectMeta { - annotations: Some( - [( - RESTATE_REMOVE_VERSION_AT_ANNOTATION.to_string(), - remove_at.to_rfc3339(), - )] - .into(), - ), - ..Default::default() - } - .into_request_partial::(), - ), - ) - .await?; + schedule_configuration_removal(ctx, namespace, &config_name, remove_at).await?; // ensure we keep track of the soonest remove_at next_removal = match next_removal { @@ -999,6 +1010,39 @@ pub async fn cleanup_old_configurations( Ok((active_count, next_removal)) } +/// Stamp the drain deadline after which a Configuration may be removed +async fn schedule_configuration_removal( + ctx: &Context, + namespace: &str, + config_name: &str, + remove_at: chrono::DateTime, +) -> Result<()> { + let config_api: Api = Api::namespaced(ctx.client.clone(), namespace); + let params = PatchParams::apply("restate-operator/remove-version-at").force(); + + config_api + .patch_metadata( + config_name, + ¶ms, + &Patch::Apply( + ObjectMeta { + annotations: Some( + [( + RESTATE_REMOVE_VERSION_AT_ANNOTATION.to_string(), + remove_at.to_rfc3339(), + )] + .into(), + ), + ..Default::default() + } + .into_request_partial::(), + ), + ) + .await?; + + Ok(()) +} + /// Get tag from Configuration annotation fn get_configuration_tag(config: &Configuration) -> Option { config diff --git a/src/controllers/restatedeployment/reconcilers/replicaset.rs b/src/controllers/restatedeployment/reconcilers/replicaset.rs index dde4797b..c7a45ce4 100644 --- a/src/controllers/restatedeployment/reconcilers/replicaset.rs +++ b/src/controllers/restatedeployment/reconcilers/replicaset.rs @@ -16,7 +16,8 @@ use serde_json::json; use tracing::*; use crate::controllers::restatedeployment::controller::{ - APP_MANAGED_BY_LABEL, Context, OWNED_BY_LABEL, RESTATE_DEPLOYMENT_ID_ANNOTATION, + APP_MANAGED_BY_LABEL, Context, DeploymentState, OWNED_BY_LABEL, + RESTATE_DEPLOYMENT_ID_ANNOTATION, }; use crate::resources::restatecloudenvironments::InProcessTunnelParams; use crate::resources::restatedeployments::RestateDeployment; @@ -274,7 +275,7 @@ pub async fn cleanup_old_replicasets( rs_api: &Api, rsd_uid: &str, rsd: &RestateDeployment, - deployments: &HashMap, + deployments: &HashMap, except_rs: Option<&str>, ) -> Result<(i32, Option>)> { let replicasets_cell = std::cell::Cell::new(Vec::new()); @@ -329,82 +330,31 @@ pub async fn cleanup_old_replicasets( let mut next_removal = None; let now = chrono::Utc::now(); + let is_deleting = rsd.metadata.deletion_timestamp.is_some(); for rs in replicasets { let rs_name = rs.name_any(); let rs_deployment_id = rs.annotations().get(RESTATE_DEPLOYMENT_ID_ANNOTATION); - // Skip active deployments let deployment = rs_deployment_id - .and_then(|rs_deployment_id| deployments.get(rs_deployment_id).cloned()); + .and_then(|rs_deployment_id| deployments.get(rs_deployment_id).copied()); let deployment_exists = deployment.is_some(); - let deployment_active = deployment.unwrap_or(false); + let deployment = deployment.unwrap_or_default(); - // During deletion, once the drain period has passed for an active version, - // fall through to the force-delete path below (same as inactive versions). - let is_deleting = rsd.metadata.deletion_timestamp.is_some(); - let drain_annotation = rs + let current_remove_at = rs .annotations() .get(RESTATE_REMOVE_VERSION_AT_ANNOTATION) - .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok().map(|t| t.to_utc())); - let drain_due = drain_annotation.is_some_and(|t| t < now); - - if deployment_active && !(is_deleting && drain_due) { - if is_deleting { - // During deletion: schedule drain for active versions (latest-in-sys_service - // or still-pinned) so the finalizer can eventually complete. Don't reset any - // existing timer — unlike the rollback case, we're committed to removal. - match drain_annotation { - None => { - let drain_delay_seconds = rsd.spec.restate.drain_delay_seconds(); - info!( - replicaset = %rs_name, - namespace = %namespace, - drain_delay_seconds, - "RestateDeployment being deleted; scheduling active version for drain" - ); - let remove_at = chrono::Utc::now() - .checked_add_signed(chrono::TimeDelta::seconds(drain_delay_seconds)) - .expect("remove_version_at in bounds"); - - rs_api - .patch_metadata( - &rs_name, - &PatchParams::apply("restate-operator/remove-version-at").force(), - &Patch::Apply( - ObjectMeta { - annotations: Some( - [( - RESTATE_REMOVE_VERSION_AT_ANNOTATION.to_string(), - remove_at.to_rfc3339(), - )] - .into(), - ), - ..Default::default() - } - .into_request_partial::(), - ), - ) - .await?; - - next_removal = match next_removal { - None => Some(remove_at), - Some(nr) if nr > remove_at => Some(remove_at), - els => els, - }; - } - Some(remove_at) => { - next_removal = match next_removal { - None => Some(remove_at), - Some(nr) if nr > remove_at => Some(remove_at), - els => els, - }; - } - } - continue; - } + .and_then(|remove_at| { + chrono::DateTime::parse_from_rfc3339(remove_at) + .map(|t| t.to_utc()) + .ok() + }); + let current_remove_at_in_past = current_remove_at.is_some_and(|c| c < now); + let blocks_removal = deployment.blocks_removal(is_deleting); + // Skip active deployments + if blocks_removal && !is_deleting { active_count += 1; // Per-version autoscaling: a non-latest version has an operator HPA @@ -413,7 +363,7 @@ pub async fn cleanup_old_replicasets( // handled unconditionally below, before scale-down.) match super::autoscaling::plan_active_version_hpa( rsd.spec.autoscaling.is_some(), - false, // not deleting (is_deleting is false here) + is_deleting, ) { HpaPlan::Ensure => { if let Some(template) = rsd.spec.autoscaling.as_ref() @@ -483,7 +433,7 @@ pub async fn cleanup_old_replicasets( HpaPlan::Skip => {} } - if drain_annotation.is_none() { + if current_remove_at.is_none() { // not scheduled for removal; all good. continue; } @@ -515,6 +465,58 @@ pub async fn cleanup_old_replicasets( continue; } + // Deleting, and the only thing left worth waiting for: a version with live + // pinned invocations, which the force-deregistration below would kill. Hold + // it for the drain delay, then tear it down like any other version. Unlike + // the rollback case above we never reset an existing timer — we're committed. + if is_deleting && blocks_removal && !current_remove_at_in_past { + let remove_at = match current_remove_at { + Some(remove_at) => remove_at, + None => { + let drain_delay_seconds = rsd.spec.restate.drain_delay_seconds(); + info!( + replicaset = %rs_name, + namespace = %namespace, + drain_delay_seconds, + "RestateDeployment being deleted; draining version with pinned invocations" + ); + let remove_at = now + .checked_add_signed(chrono::TimeDelta::seconds(drain_delay_seconds)) + .expect("remove_version_at in bounds"); + + rs_api + .patch_metadata( + &rs_name, + &PatchParams::apply("restate-operator/remove-version-at").force(), + &Patch::Apply( + ObjectMeta { + annotations: Some( + [( + RESTATE_REMOVE_VERSION_AT_ANNOTATION.to_string(), + remove_at.to_rfc3339(), + )] + .into(), + ), + ..Default::default() + } + .into_request_partial::(), + ), + ) + .await?; + + remove_at + } + }; + + next_removal = match next_removal { + None => Some(remove_at), + Some(next_removal) if next_removal > remove_at => Some(remove_at), + els => els, + }; + + continue; + } + // Non-active version: remove any operator HPA before scaling it down. Its // minReplicas floor (>= 1) would otherwise fight the operator scaling the // ReplicaSet to zero, and would hold the version at the floor through the @@ -528,22 +530,11 @@ pub async fn cleanup_old_replicasets( super::autoscaling::delete_version_hpa(&ctx.client, namespace, &rs_name).await?; } - let current_remove_at = rs - .annotations() - .get(RESTATE_REMOVE_VERSION_AT_ANNOTATION) - .and_then(|remove_at| { - chrono::DateTime::parse_from_rfc3339(remove_at) - .map(|t| t.to_utc()) - .ok() - }); - - let current_remove_at_in_past = current_remove_at.is_some_and(|c| c < now); + // A deleting RestateDeployment never waits here: it either had nothing + // pinned, or its drain already elapsed above. + let teardown_due = current_remove_at_in_past || is_deleting; - match ( - current_remove_at, - current_remove_at_in_past, - deployment_exists, - ) { + match (current_remove_at, teardown_due, deployment_exists) { (_, true, _) | (_, _, false) => { // we are past the remove at time, or the endpoint was removed by other means; can now scale it down @@ -574,8 +565,11 @@ pub async fn cleanup_old_replicasets( .await?; } - // If we are here, there is a 0 sized replicaset which should be subject to the history limit - if historic_count < rsd.spec.revision_history_limit { + // If we are here, there is a 0 sized replicaset which should be subject to the + // history limit — except while deleting, where keeping it for rollback would + // just leave the Restate deployment registered with nothing left to + // deregister it once the finalizer is released. + if !is_deleting && historic_count < rsd.spec.revision_history_limit { historic_count += 1; // we haven't hit that limit yet, so we don't need to delete this rs continue;