From f842f0dc37052427ea2ba7f3bf92893a3bf81b40 Mon Sep 17 00:00:00 2001 From: Mitch Ross Date: Wed, 9 Sep 2026 09:05:29 -0400 Subject: [PATCH 1/5] fix(controller): propose pruning superseded inactive versions --- internal/controller/execplan.go | 29 +++++++++++--- internal/controller/execplan_test.go | 58 ++++++++++++++++++++++++++++ internal/planner/planner.go | 12 +++++- internal/planner/planner_test.go | 45 ++++++++++++++++++++- 4 files changed, 137 insertions(+), 7 deletions(-) diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go index 06b6d255..54b5a53a 100644 --- a/internal/controller/execplan.go +++ b/internal/controller/execplan.go @@ -20,6 +20,7 @@ import ( commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" sdkclient "go.temporal.io/sdk/client" "go.temporal.io/sdk/converter" "go.temporal.io/sdk/worker" @@ -662,13 +663,16 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( return errors.Join(append(applyErrs, statusErrs...)...) } -// deleteDrainedVersions prunes the Temporal server-side Worker Deployment Version +// deleteDeprecatedVersions prunes the Temporal server-side Worker Deployment Version // record for each k8s Deployment in DeleteDeployments, before executeK8sOperations // deletes them. It is mutated to remove the k8s Deployments that should not be // deleted because their Temporal server-side WDV record could not be removed or // because the k8s Deployment has no build ID label. The removed k8s deployments // stay in the cluster so that a later reconcile retries the pruning. // +// DeleteVersion does not check pinned execution visibility for Inactive versions, +// which can receive workflows through VersioningOverride. Check visibility first. +// // The planner only adds a drained version to DeleteDeployments once it is // EligibleForDeletion (see planner.getDeleteDeployments): drained past the sunset // delays with no active worker pods. Deleting the Kubernetes Deployment alone would @@ -687,11 +691,12 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( // phase without reaching back into k8sState. NotRegistered Deployments are also carried // in DeleteDeployments; they have no server-side version, so they skip DeleteVersion and // are retained for deletion. -func (r *WorkerDeploymentReconciler) deleteDrainedVersions( +func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( ctx context.Context, l logr.Logger, workerDeploy *temporaliov1alpha1.WorkerDeployment, depHandle sdkclient.WorkerDeploymentHandle, + temporalClient sdkclient.Client, p *plan, ) { identity := getControllerIdentity() @@ -710,6 +715,20 @@ func (r *WorkerDeploymentReconciler) deleteDrainedVersions( markedForDeletion = append(markedForDeletion, d) continue } + if slices.ContainsFunc(workerDeploy.Status.DeprecatedVersions, func(v *temporaliov1alpha1.DeprecatedWorkerDeploymentVersion) bool { + return v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive + }) { + // Visibility can contain either the legacy dot separator or the newer colon form. + legacyVersion := strings.ReplaceAll(p.WorkerDeploymentName+"."+buildID, "'", "''") + version := strings.ReplaceAll(p.WorkerDeploymentName+":"+buildID, "'", "''") + count, err := temporalClient.CountWorkflow(ctx, &workflowservice.CountWorkflowExecutionsRequest{ + Query: fmt.Sprintf("TemporalWorkerDeploymentVersion IN ('%s', '%s') AND TemporalWorkflowVersioningBehavior = 'Pinned' AND ExecutionStatus = 'Running'", legacyVersion, version), + }) + if err != nil || count == nil || count.Count != 0 { + l.Info("could not confirm inactive version has no running pinned workflows, keeping its Deployment", "buildID", buildID, "error", err) + continue + } + } _, err := depHandle.DeleteVersion( ctx, sdkclient.WorkerDeploymentDeleteVersionOptions{ @@ -726,7 +745,7 @@ func (r *WorkerDeploymentReconciler) deleteDrainedVersions( } l.Info("worker deployment version already deleted", "buildID", buildID) } else { - l.Info("deleted drained worker deployment version", "buildID", buildID) + l.Info("deleted deprecated worker deployment version", "buildID", buildID) } markedForDeletion = append(markedForDeletion, d) } @@ -757,8 +776,8 @@ func (r *WorkerDeploymentReconciler) executePlan( // Prune the Temporal server-side version records before their k8s Deployments are // deleted, and narrow the plan to the versions the server confirmed gone. A Deployment // held back here keeps its version nominated for deletion, so a failed deletion is retried - // on the next reconcile instead of orphaning the record; see deleteDrainedVersions. - r.deleteDrainedVersions(ctx, l, workerDeploy, deploymentHandler, p) + // on the next reconcile instead of orphaning the record; see deleteDeprecatedVersions. + r.deleteDeprecatedVersions(ctx, l, workerDeploy, deploymentHandler, temporalClient, p) deletedWorkerResources, err := r.executeK8sOperations(ctx, l, workerDeploy, p) if err != nil { return err diff --git a/internal/controller/execplan_test.go b/internal/controller/execplan_test.go index 1cc7db27..6fe710d6 100644 --- a/internal/controller/execplan_test.go +++ b/internal/controller/execplan_test.go @@ -17,6 +17,7 @@ import ( "github.com/temporalio/temporal-worker-controller/internal/k8s" "github.com/temporalio/temporal-worker-controller/internal/temporal" "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" sdkclient "go.temporal.io/sdk/client" "go.temporal.io/sdk/converter" appsv1 "k8s.io/api/apps/v1" @@ -608,3 +609,60 @@ func TestGeneratePlan_CarriesEncodingAndMessageType(t *testing.T) { require.Equal(t, "my.package.DeployRequest", wf.messageType) require.Equal(t, []byte(`{"service":"checkout"}`), wf.input) } + +// A visibility failure must retain the Deployment so the next reconciliation can retry. +type inactivePruneClient struct { + *stubTemporalClient + response *workflowservice.CountWorkflowExecutionsResponse + err error + query string +} + +func (c *inactivePruneClient) CountWorkflow(_ context.Context, request *workflowservice.CountWorkflowExecutionsRequest) (*workflowservice.CountWorkflowExecutionsResponse, error) { + c.query = request.Query + return c.response, c.err +} + +func TestExecutePlan_InactiveVersionDeletion(t *testing.T) { + for _, tc := range []struct { + name string + response *workflowservice.CountWorkflowExecutionsResponse + countErr error + deleteErr error + wantAttempt bool + wantDelete bool + }{ + {name: "unused version", response: &workflowservice.CountWorkflowExecutionsResponse{}, wantAttempt: true, wantDelete: true}, + {name: "running pinned workflow", response: &workflowservice.CountWorkflowExecutionsResponse{Count: 1}}, + {name: "visibility failure", countErr: errors.New("visibility unavailable")}, + {name: "missing response"}, + {name: "server rejects deletion", response: &workflowservice.CountWorkflowExecutionsResponse{}, deleteErr: serviceerror.NewFailedPrecondition("active pollers"), wantAttempt: true}, + } { + t.Run(tc.name, func(t *testing.T) { + connection := temporaliov1alpha1.ConnectionSpec{HostPort: "test:7233"} + twd := makeExecplanTWD("my-worker", "default") + old := makeVersionedDeployment(twd, "old", 0, connection) + current := makeVersionedDeployment(twd, "current", 1, connection) + r, _ := newTestReconciler([]client.Object{twd, old, current}) + handle := newPruneStubHandle(tc.deleteErr) + c := &inactivePruneClient{stubTemporalClient: newStubTemporalClientWithHandle(handle), response: tc.response, err: tc.countErr} + status := statusWithDeprecated("current", current, &temporaliov1alpha1.DeprecatedWorkerDeploymentVersion{ + BaseWorkerDeploymentVersion: baseVersion("old", old, temporaliov1alpha1.VersionStatusInactive), + }) + twd.Status = status + state := &temporal.TemporalWorkerState{Versions: map[string]*temporal.VersionInfo{ + "old": {Status: temporaliov1alpha1.VersionStatusInactive}, + "current": {Status: temporaliov1alpha1.VersionStatusCurrent}, + }} + p, err := r.generatePlan(context.Background(), logr.Discard(), twd, connection, state) + require.NoError(t, err) + require.NoError(t, r.executePlan(context.Background(), logr.Discard(), twd, c, p)) + require.Equal(t, tc.wantAttempt, len(handle.deletedVersions) == 1) + require.Equal(t, tc.wantDelete, len(p.DeleteDeployments) == 1) + require.Equal(t, !tc.wantDelete, deploymentExists(t, r, "default", old.Name)) + require.Contains(t, c.query, "TemporalWorkerDeploymentVersion IN ('default/my-worker.old', 'default/my-worker:old')") + require.Contains(t, c.query, "TemporalWorkflowVersioningBehavior = 'Pinned'") + require.Contains(t, c.query, "ExecutionStatus = 'Running'") + }) + } +} diff --git a/internal/planner/planner.go b/internal/planner/planner.go index d74289ee..d78bb542 100644 --- a/internal/planner/planner.go +++ b/internal/planner/planner.go @@ -744,6 +744,16 @@ func getDeleteDeployments( } switch version.Status { + case temporaliov1alpha1.VersionStatusInactive: + // Superseded versions that never received routed traffic never become Drained. + // Wait for scale-down to finish; execution checks pinned workflows before pruning. + if foundDeploymentInTemporal && status.TargetVersion.BuildID != version.BuildID && + (status.CurrentVersion == nil || status.CurrentVersion.BuildID != version.BuildID) && + d.Spec.Replicas != nil && *d.Spec.Replicas == 0 && + d.Status.ObservedGeneration >= d.Generation && d.Status.Replicas == 0 && + (d.Status.TerminatingReplicas == nil || *d.Status.TerminatingReplicas == 0) { + deleteDeployments = append(deleteDeployments, d) + } case temporaliov1alpha1.VersionStatusDrained: // Deleting a deployment is only possible when: // 1. The deployment has been drained for deleteDelay + scaledownDelay. @@ -754,7 +764,7 @@ func getDeleteDeployments( // reconcile as the Deployment delete: EligibleForDeletion is only // computable while the Deployment (and thus this DeprecatedVersions // entry) still exists, so this is the only point that can reliably - // prune it. See execplan.deleteDrainedVersions. + // prune it. See execplan.deleteDeprecatedVersions. if version.DrainedSince != nil && (time.Since(version.DrainedSince.Time) > spec.SunsetStrategy.DeleteDelay.Duration+spec.SunsetStrategy.ScaledownDelay.Duration) && d.Spec.Replicas != nil && *d.Spec.Replicas == 0 && diff --git a/internal/planner/planner_test.go b/internal/planner/planner_test.go index 32d71edc..173226c3 100644 --- a/internal/planner/planner_test.go +++ b/internal/planner/planner_test.go @@ -734,7 +734,7 @@ func TestGetDeleteDeployments(t *testing.T) { // EligibleForDeletion: worker pods have not fully terminated (Status.Replicas > 0), // so versioned pollers may still be registered and the Temporal-side DeleteVersion // would fail. Deleting the Deployment now would strand that server-side version - // record with no way to retry (see execplan.deleteDrainedVersions), so hold off. + // record with no way to retry (see execplan.deleteDeprecatedVersions), so hold off. name: "drained long enough and scaled to zero in spec, but not eligible for deletion - not deleted", k8sState: &k8s.DeploymentState{ Deployments: map[string]*appsv1.Deployment{ @@ -4673,3 +4673,46 @@ func TestGetSunsetScaleDownBuildIDs(t *testing.T) { }) } } + +func TestGetDeleteDeployments_Inactive(t *testing.T) { + for _, tc := range []struct { + name string + mutate func(*appsv1.Deployment, *temporaliov1alpha1.WorkerDeploymentStatus) + wantDelete bool + }{ + {name: "superseded and fully scaled down", wantDelete: true}, + {name: "scale down requested but pods remain", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { d.Status.Replicas = 1 }}, + {name: "terminating pods remain", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { + n := int32(1) + d.Status.TerminatingReplicas = &n + }}, + {name: "scale down not observed", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { d.Generation = 1 }}, + {name: "replicas unspecified", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { d.Spec.Replicas = nil }}, + {name: "replicas positive", mutate: func(d *appsv1.Deployment, _ *temporaliov1alpha1.WorkerDeploymentStatus) { *d.Spec.Replicas = 1 }}, + {name: "target retained", mutate: func(_ *appsv1.Deployment, s *temporaliov1alpha1.WorkerDeploymentStatus) { + s.TargetVersion.BuildID = "old" + }}, + {name: "current retained", mutate: func(_ *appsv1.Deployment, s *temporaliov1alpha1.WorkerDeploymentStatus) { + s.CurrentVersion = &temporaliov1alpha1.CurrentWorkerDeploymentVersion{BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{BuildID: "old"}} + }}, + } { + t.Run(tc.name, func(t *testing.T) { + d := createDeploymentWithDefaultConnectionSpecHash(0) + s := &temporaliov1alpha1.WorkerDeploymentStatus{ + DeprecatedVersions: []*temporaliov1alpha1.DeprecatedWorkerDeploymentVersion{{ + BaseWorkerDeploymentVersion: temporaliov1alpha1.BaseWorkerDeploymentVersion{ + BuildID: "old", Status: temporaliov1alpha1.VersionStatusInactive, + Deployment: &corev1.ObjectReference{Name: "old"}, + }, + }}, + } + if tc.mutate != nil { + tc.mutate(d, s) + } + state := &k8s.DeploymentState{Deployments: map[string]*appsv1.Deployment{"old": d}} + deleted := getDeleteDeployments(state, s, &temporaliov1alpha1.WorkerDeploymentSpec{}, true) + assert.Equal(t, tc.wantDelete, len(deleted) == 1) + assert.Empty(t, getDeleteDeployments(state, s, &temporaliov1alpha1.WorkerDeploymentSpec{}, false)) + }) + } +} From 0d1aee7ffe23cee1b1ac6242cc0e4b1ac4dc8b52 Mon Sep 17 00:00:00 2001 From: Mitch Ross Date: Thu, 10 Sep 2026 17:59:58 -0400 Subject: [PATCH 2/5] test: verify inactive retirement against Temporal and preserve retained resources --- docs/concepts.md | 11 ++ internal/controller/execplan.go | 12 ++ internal/controller/execplan_test.go | 19 ++- .../internal/deletion_integration_test.go | 6 + .../tests/internal/deployment_controller.go | 1 + .../inactive_retirement_integration_test.go | 158 ++++++++++++++++++ 6 files changed, 206 insertions(+), 1 deletion(-) create mode 100644 internal/tests/internal/inactive_retirement_integration_test.go diff --git a/docs/concepts.md b/docs/concepts.md index 05e9e197..96d44961 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -125,6 +125,17 @@ Defines how Drained versions are cleaned up: - **scaledownDelay**: How long to wait after a version has been Drained before scaling pods to zero - **deleteDelay**: How long to wait after a version has been Drained before deleting the Kubernetes `Deployment` +Versions superseded before ever becoming Current or Ramping remain Inactive in Temporal; +they never acquire a drainage timestamp. The controller retires these versions after +their Deployment has fully scaled to zero, visibility reports no running pinned workflows, +and Temporal accepts the normal version deletion request. The drainage-based sunset delays +do not apply to these unused versions. API failures or active pollers defer deletion and are +retried on subsequent reconciliations. + +Stop sending pinned version overrides to a version being retired. Visibility is eventually +consistent, so this check cannot exclude concurrent workflow starts or override changes. +Temporal's drained status has the [same limitation for newly pinned overrides](https://typescript.temporal.io/api/interfaces/proto.temporal.api.deployment.v1.IWorkerDeploymentVersionInfo#drainageinfo). + ### Template The pod template used for the target version of this worker deployment. Similar to the pod template used in a standar Kubernetes `Deployment`, but managed by the controller. diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go index 54b5a53a..d6b54ce7 100644 --- a/internal/controller/execplan.go +++ b/internal/controller/execplan.go @@ -672,6 +672,9 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( // // DeleteVersion does not check pinned execution visibility for Inactive versions, // which can receive workflows through VersioningOverride. Check visibility first. +// Like Temporal drainage, visibility is eventually consistent: callers must stop +// sending new pinned overrides to a version being retired. This is not an atomic +// exclusion against concurrent workflow starts or override updates. // // The planner only adds a drained version to DeleteDeployments once it is // EligibleForDeletion (see planner.getDeleteDeployments): drained past the sunset @@ -701,6 +704,7 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( ) { identity := getControllerIdentity() markedForDeletion := make([]*appsv1.Deployment, 0, len(p.DeleteDeployments)) + retainedInactiveBuilds := make(map[string]bool) for _, d := range p.DeleteDeployments { buildID, ok := d.GetLabels()[k8s.BuildIDLabel] if !ok { @@ -718,6 +722,7 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( if slices.ContainsFunc(workerDeploy.Status.DeprecatedVersions, func(v *temporaliov1alpha1.DeprecatedWorkerDeploymentVersion) bool { return v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive }) { + retainedInactiveBuilds[buildID] = true // Visibility can contain either the legacy dot separator or the newer colon form. legacyVersion := strings.ReplaceAll(p.WorkerDeploymentName+"."+buildID, "'", "''") version := strings.ReplaceAll(p.WorkerDeploymentName+":"+buildID, "'", "''") @@ -748,8 +753,15 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( l.Info("deleted deprecated worker deployment version", "buildID", buildID) } markedForDeletion = append(markedForDeletion, d) + delete(retainedInactiveBuilds, buildID) } p.DeleteDeployments = markedForDeletion + // Resources nominated with an inactive Deployment must survive if its deletion + // was refused. Otherwise, for example, its ConfigMaps could disappear underneath + // a version retained for pinned workflows. + p.DeleteWorkerResources = slices.DeleteFunc(p.DeleteWorkerResources, func(ref planner.WorkerResourceRef) bool { + return retainedInactiveBuilds[ref.BuildID] + }) } // isVersionNotRegistered checks whether the Temporal server had no record of buildID when diff --git a/internal/controller/execplan_test.go b/internal/controller/execplan_test.go index 6fe710d6..f70b404a 100644 --- a/internal/controller/execplan_test.go +++ b/internal/controller/execplan_test.go @@ -637,13 +637,22 @@ func TestExecutePlan_InactiveVersionDeletion(t *testing.T) { {name: "visibility failure", countErr: errors.New("visibility unavailable")}, {name: "missing response"}, {name: "server rejects deletion", response: &workflowservice.CountWorkflowExecutionsResponse{}, deleteErr: serviceerror.NewFailedPrecondition("active pollers"), wantAttempt: true}, + {name: "Temporal version already deleted", response: &workflowservice.CountWorkflowExecutionsResponse{}, deleteErr: serviceerror.NewNotFound("version"), wantAttempt: true, wantDelete: true}, } { t.Run(tc.name, func(t *testing.T) { connection := temporaliov1alpha1.ConnectionSpec{HostPort: "test:7233"} twd := makeExecplanTWD("my-worker", "default") old := makeVersionedDeployment(twd, "old", 0, connection) current := makeVersionedDeployment(twd, "current", 1, connection) - r, _ := newTestReconciler([]client.Object{twd, old, current}) + wrt := makeExecplanWRT("config", twd) + wrt.Spec.Template.Raw = []byte(`{"apiVersion":"v1","kind":"ConfigMap","data":{"test":"present"}}`) + oldConfig, oldHash := renderWRT(t, wrt, old, "old", testTemporalNamespace) + currentConfig, currentHash := renderWRT(t, wrt, current, "current", testTemporalNamespace) + wrt.Status.Versions = []temporaliov1alpha1.WorkerResourceTemplateVersionStatus{ + k8s.WorkerResourceTemplateVersionStatusForBuildID("old", oldConfig.GetName(), 1, oldHash, ""), + k8s.WorkerResourceTemplateVersionStatusForBuildID("current", currentConfig.GetName(), 1, currentHash, ""), + } + r, _ := newTestReconciler([]client.Object{twd, old, current, wrt, oldConfig, currentConfig}) handle := newPruneStubHandle(tc.deleteErr) c := &inactivePruneClient{stubTemporalClient: newStubTemporalClientWithHandle(handle), response: tc.response, err: tc.countErr} status := statusWithDeprecated("current", current, &temporaliov1alpha1.DeprecatedWorkerDeploymentVersion{ @@ -660,6 +669,14 @@ func TestExecutePlan_InactiveVersionDeletion(t *testing.T) { require.Equal(t, tc.wantAttempt, len(handle.deletedVersions) == 1) require.Equal(t, tc.wantDelete, len(p.DeleteDeployments) == 1) require.Equal(t, !tc.wantDelete, deploymentExists(t, r, "default", old.Name)) + require.True(t, deploymentExists(t, r, "default", current.Name)) + configErr := r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: oldConfig.GetName()}, &corev1.ConfigMap{}) + if tc.wantDelete { + require.True(t, apierrors.IsNotFound(configErr)) + } else { + require.NoError(t, configErr, "retained version must keep its rendered resources") + } + require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: currentConfig.GetName()}, &corev1.ConfigMap{})) require.Contains(t, c.query, "TemporalWorkerDeploymentVersion IN ('default/my-worker.old', 'default/my-worker:old')") require.Contains(t, c.query, "TemporalWorkflowVersioningBehavior = 'Pinned'") require.Contains(t, c.query, "ExecutionStatus = 'Running'") diff --git a/internal/tests/internal/deletion_integration_test.go b/internal/tests/internal/deletion_integration_test.go index 31eb0d5a..3491ea5d 100644 --- a/internal/tests/internal/deletion_integration_test.go +++ b/internal/tests/internal/deletion_integration_test.go @@ -49,6 +49,12 @@ func runDeletionTests( t.Run("drained-version-pruned-from-temporal-on-sunset", func(t *testing.T) { testDrainedVersionPrunedOnSunset(t, k8sClient, ts, testNamespace) }) + + for _, pinned := range []bool{false, true} { + t.Run(fmt.Sprintf("inactive-version-retirement/pinned-%t", pinned), func(t *testing.T) { + testInactiveVersionRetirement(t, k8sClient, ts, testNamespace, pinned) + }) + } } // testDeletionSetsCurrentToUnversioned verifies the core fix: when a WD is deleted, diff --git a/internal/tests/internal/deployment_controller.go b/internal/tests/internal/deployment_controller.go index 160c9fa2..7cc4cb27 100644 --- a/internal/tests/internal/deployment_controller.go +++ b/internal/tests/internal/deployment_controller.go @@ -177,6 +177,7 @@ func scaleDeploymentToZero(t *testing.T, ctx context.Context, k8sClient client.C dep.Status.ReadyReplicas = 0 dep.Status.AvailableReplicas = 0 dep.Status.UpdatedReplicas = 0 + dep.Status.ObservedGeneration = dep.Generation return k8sClient.Status().Update(ctx, &dep) }); err != nil { t.Fatalf("failed to zero status of deployment %s: %v", name, err) diff --git a/internal/tests/internal/inactive_retirement_integration_test.go b/internal/tests/internal/inactive_retirement_integration_test.go new file mode 100644 index 00000000..b1050c3c --- /dev/null +++ b/internal/tests/internal/inactive_retirement_integration_test.go @@ -0,0 +1,158 @@ +package internal + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1" + "github.com/temporalio/temporal-worker-controller/internal/k8s" + "github.com/temporalio/temporal-worker-controller/internal/testhelpers" + "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" + sdkclient "go.temporal.io/sdk/client" + sdkworker "go.temporal.io/sdk/worker" + "go.temporal.io/sdk/workflow" + "go.temporal.io/server/temporaltest" + appsv1 "k8s.io/api/apps/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// A superseded rollout that was never Current or Ramping stays Inactive forever. +// Exercise real version registration, visibility, poller expiry and DeleteVersion, +// including a pinned override on a version which has never received routed traffic. +func testInactiveVersionRetirement(t *testing.T, k8sClient client.Client, ts *temporaltest.TestServer, namespace string, pinned bool) { + ctx := context.Background() + name := fmt.Sprintf("inactive-retire-%t", pinned) + tc := testhelpers.NewTestCase().WithInput(testhelpers.NewWorkerDeploymentBuilder(). + WithManualStrategy().WithTargetTemplate("v1.0")). + BuildWithValues(name, namespace, ts.GetDefaultNamespace()) + twd := tc.GetTWD() + connection := &temporaliov1alpha1.Connection{ + ObjectMeta: metav1.ObjectMeta{Name: twd.Spec.WorkerOptions.ConnectionRef.Name, Namespace: namespace}, + Spec: temporaliov1alpha1.ConnectionSpec{HostPort: ts.GetFrontendHostPort()}, + } + if err := k8sClient.Create(ctx, connection); err != nil { + t.Fatal(err) + } + if err := k8sClient.Create(ctx, twd); err != nil { + t.Fatal(err) + } + deploymentName := k8s.ComputeWorkerDeploymentName(twd) + buildID := k8s.ComputeBuildID(twd) + oldKey := types.NamespacedName{Namespace: namespace, Name: k8s.ComputeVersionedDeploymentName(twd.Name, buildID)} + var old appsv1.Deployment + eventually(t, 30*time.Second, time.Second, func() error { return k8sClient.Get(ctx, oldKey, &old) }) + w, stop, err := testhelpers.NewWorker(ctx, deploymentName, buildID, name, ts.GetFrontendHostPort(), ts.GetDefaultNamespace(), true) + if err != nil { + t.Fatal(err) + } + stopOnce := sync.OnceFunc(stop) + defer stopOnce() + w.RegisterWorkflowWithOptions(func(ctx workflow.Context) error { + return workflow.Await(ctx, func() bool { return false }) + }, workflow.RegisterOptions{Name: "inactivePinnedWorkflow"}) + if err := w.Start(); err != nil { + t.Fatal(err) + } + setHealthyDeploymentStatus(t, ctx, k8sClient, old) + version := sdkworker.WorkerDeploymentVersion{DeploymentName: deploymentName, BuildID: buildID} + waitForVersionRegistrationInDeployment(t, ctx, ts, &version) + handle := ts.GetDefaultClient().WorkerDeploymentClient().GetHandle(deploymentName) + + var run sdkclient.WorkflowRun + if pinned { + run, err = ts.GetDefaultClient().ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{ + ID: name, TaskQueue: name, VersioningOverride: &sdkclient.PinnedVersioningOverride{Version: version}, + }, "inactivePinnedWorkflow") + if err != nil { + t.Fatal(err) + } + defer func() { _ = ts.GetDefaultClient().TerminateWorkflow(ctx, run.GetID(), run.GetRunID(), "test cleanup") }() + // Wait for the actual visibility index, not just the start RPC, before retiring. + eventually(t, 30*time.Second, time.Second, func() error { + count, err := ts.GetDefaultClient().CountWorkflow(ctx, &workflowservice.CountWorkflowExecutionsRequest{ + Query: fmt.Sprintf("WorkflowId = '%s' AND TemporalWorkflowVersioningBehavior = 'Pinned' AND ExecutionStatus = 'Running'", name), + }) + if err != nil { + return err + } + if count.Count != 1 { + return fmt.Errorf("pinned workflow not yet visible") + } + return nil + }) + } + + // Replace the target before ever making v1 current or ramping. + var next temporaliov1alpha1.WorkerDeployment + key := types.NamespacedName{Name: twd.Name, Namespace: namespace} + if err := k8sClient.Get(ctx, key, &next); err != nil { + t.Fatal(err) + } + next.Spec.Template.Spec.Containers[0].Image = "v2.0" + newBuildID := k8s.ComputeBuildID(&next) + if err := k8sClient.Update(ctx, &next); err != nil { + t.Fatal(err) + } + newKey := types.NamespacedName{Name: k8s.ComputeVersionedDeploymentName(twd.Name, newBuildID), Namespace: namespace} + eventually(t, 30*time.Second, time.Second, func() error { + var dep appsv1.Deployment + return k8sClient.Get(ctx, newKey, &dep) + }) + stops := applyDeployment(t, ctx, k8sClient, newKey.Name, namespace) + defer handleStopFuncs(stops) + setCurrentVersion(t, ctx, ts, deploymentName, newBuildID) + eventually(t, 30*time.Second, time.Second, func() error { + if err := k8sClient.Get(ctx, key, &next); err != nil { + return err + } + for _, v := range next.Status.DeprecatedVersions { + if v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive && v.DrainedSince == nil { + return nil + } + } + return fmt.Errorf("superseded version is not yet Inactive") + }) + stopOnce() + scaleDeploymentToZero(t, ctx, k8sClient, oldKey.Name, namespace) + if pinned { + // Longer than both the poller TTL and several controller reconciliations. + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Second) { + if err := k8sClient.Get(ctx, oldKey, &old); err != nil { + t.Fatalf("deleted a version with a pinned workflow: %v", err) + } + if _, err := handle.DescribeVersion(ctx, sdkclient.WorkerDeploymentDescribeVersionOptions{BuildID: buildID}); err != nil { + t.Fatal(err) + } + } + if err := ts.GetDefaultClient().TerminateWorkflow(ctx, run.GetID(), run.GetRunID(), "release pinned version"); err != nil { + t.Fatal(err) + } + } + eventually(t, 90*time.Second, time.Second, func() error { + if err := k8sClient.Get(ctx, oldKey, &old); err == nil { + return fmt.Errorf("superseded Inactive Deployment still exists") + } else if client.IgnoreNotFound(err) != nil { + return err + } + _, err := handle.DescribeVersion(ctx, sdkclient.WorkerDeploymentDescribeVersionOptions{BuildID: buildID}) + var notFound *serviceerror.NotFound + if !errors.As(err, ¬Found) { + return fmt.Errorf("expected retired Temporal version, got %v", err) + } + return nil + }) + var current appsv1.Deployment + if err := k8sClient.Get(ctx, newKey, ¤t); err != nil { + t.Fatal(err) + } + if _, err := handle.DescribeVersion(ctx, sdkclient.WorkerDeploymentDescribeVersionOptions{BuildID: newBuildID}); err != nil { + t.Fatal(err) + } +} From 456a3bae422959b1a4a73f7b7757cf25ac032293 Mon Sep 17 00:00:00 2001 From: Mitch Ross Date: Wed, 16 Sep 2026 09:06:12 -0400 Subject: [PATCH 3/5] refactor(controller): address inactive retirement review Extract the pinned-workflow visibility query, clarify its formatting and safety comment, and distinguish confirmed running workflows from failed visibility checks in logs. Shorten the concepts documentation into a note. Rename retainedInactiveBuilds to retainedResourceBuilds and explain its purpose: skipping a Deployment does not filter the separate worker-resource deletion list. Preserve the existing protection for retained versions. Validation: go test ./... and go vet ./... passed. Server-backed integration tests passed for normal drained retirement and inactive retirement with and without a pinned workflow. Existing unit cases verify rendered ConfigMaps survive visibility failures, pinned workflows, and rejected deletion. --- docs/concepts.md | 15 ++++------- internal/controller/execplan.go | 47 ++++++++++++++++++++++++--------- 2 files changed, 39 insertions(+), 23 deletions(-) diff --git a/docs/concepts.md b/docs/concepts.md index 96d44961..673d21b5 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -125,16 +125,11 @@ Defines how Drained versions are cleaned up: - **scaledownDelay**: How long to wait after a version has been Drained before scaling pods to zero - **deleteDelay**: How long to wait after a version has been Drained before deleting the Kubernetes `Deployment` -Versions superseded before ever becoming Current or Ramping remain Inactive in Temporal; -they never acquire a drainage timestamp. The controller retires these versions after -their Deployment has fully scaled to zero, visibility reports no running pinned workflows, -and Temporal accepts the normal version deletion request. The drainage-based sunset delays -do not apply to these unused versions. API failures or active pollers defer deletion and are -retried on subsequent reconciliations. - -Stop sending pinned version overrides to a version being retired. Visibility is eventually -consistent, so this check cannot exclude concurrent workflow starts or override changes. -Temporal's drained status has the [same limitation for newly pinned overrides](https://typescript.temporal.io/api/interfaces/proto.temporal.api.deployment.v1.IWorkerDeploymentVersionInfo#drainageinfo). +> **NOTE**: Versions superseded before ever becoming Current or Ramping remain Inactive in Temporal; +> they never acquire a drainage timestamp. The controller retires these versions after +> their Deployment has fully scaled to zero, visibility reports no running pinned workflows, +> and Temporal accepts the normal version deletion request. The drainage-based sunset delays +> do not apply to these unused versions. ### Template The pod template used for the target version of this worker deployment. Similar to the pod template used in a standar Kubernetes `Deployment`, but managed by the controller. diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go index d6b54ce7..d5d3d574 100644 --- a/internal/controller/execplan.go +++ b/internal/controller/execplan.go @@ -670,8 +670,9 @@ func (r *WorkerDeploymentReconciler) executeWRTOperations( // because the k8s Deployment has no build ID label. The removed k8s deployments // stay in the cluster so that a later reconcile retries the pruning. // -// DeleteVersion does not check pinned execution visibility for Inactive versions, -// which can receive workflows through VersioningOverride. Check visibility first. +// Because DeleteVersion does not check to see if there are open workflows using +// pinned execution for Inactive versions, we query the Temporal visibility service +// for open pinned workflows before deleting Inactive versions. // Like Temporal drainage, visibility is eventually consistent: callers must stop // sending new pinned overrides to a version being retired. This is not an atomic // exclusion against concurrent workflow starts or override updates. @@ -704,7 +705,9 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( ) { identity := getControllerIdentity() markedForDeletion := make([]*appsv1.Deployment, 0, len(p.DeleteDeployments)) - retainedInactiveBuilds := make(map[string]bool) + // Worker resources have a separate deletion list. Track retained inactive + // versions so skipping their Deployments also preserves their resources. + retainedResourceBuilds := make(map[string]bool) for _, d := range p.DeleteDeployments { buildID, ok := d.GetLabels()[k8s.BuildIDLabel] if !ok { @@ -722,17 +725,16 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( if slices.ContainsFunc(workerDeploy.Status.DeprecatedVersions, func(v *temporaliov1alpha1.DeprecatedWorkerDeploymentVersion) bool { return v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive }) { - retainedInactiveBuilds[buildID] = true - // Visibility can contain either the legacy dot separator or the newer colon form. - legacyVersion := strings.ReplaceAll(p.WorkerDeploymentName+"."+buildID, "'", "''") - version := strings.ReplaceAll(p.WorkerDeploymentName+":"+buildID, "'", "''") - count, err := temporalClient.CountWorkflow(ctx, &workflowservice.CountWorkflowExecutionsRequest{ - Query: fmt.Sprintf("TemporalWorkerDeploymentVersion IN ('%s', '%s') AND TemporalWorkflowVersioningBehavior = 'Pinned' AND ExecutionStatus = 'Running'", legacyVersion, version), - }) - if err != nil || count == nil || count.Count != 0 { + retainedResourceBuilds[buildID] = true + count, err := getOpenPinnedWorkflowExecutions(ctx, temporalClient, p.WorkerDeploymentName, buildID) + if err != nil || count == nil { l.Info("could not confirm inactive version has no running pinned workflows, keeping its Deployment", "buildID", buildID, "error", err) continue } + if count.Count != 0 { + l.Info("inactive version has running pinned workflows, keeping its Deployment", "buildID", buildID, "count", count.Count) + continue + } } _, err := depHandle.DeleteVersion( ctx, @@ -753,17 +755,36 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( l.Info("deleted deprecated worker deployment version", "buildID", buildID) } markedForDeletion = append(markedForDeletion, d) - delete(retainedInactiveBuilds, buildID) + delete(retainedResourceBuilds, buildID) } p.DeleteDeployments = markedForDeletion // Resources nominated with an inactive Deployment must survive if its deletion // was refused. Otherwise, for example, its ConfigMaps could disappear underneath // a version retained for pinned workflows. p.DeleteWorkerResources = slices.DeleteFunc(p.DeleteWorkerResources, func(ref planner.WorkerResourceRef) bool { - return retainedInactiveBuilds[ref.BuildID] + return retainedResourceBuilds[ref.BuildID] }) } +func getOpenPinnedWorkflowExecutions( + ctx context.Context, + temporalClient sdkclient.Client, + deploymentName string, + buildID string, +) (*workflowservice.CountWorkflowExecutionsResponse, error) { + // Visibility can contain either the legacy dot separator or the newer colon form. + legacyVersion := strings.ReplaceAll(deploymentName+"."+buildID, "'", "''") + version := strings.ReplaceAll(deploymentName+":"+buildID, "'", "''") + qs := fmt.Sprintf( + "TemporalWorkerDeploymentVersion IN ('%s', '%s') "+ + "AND TemporalWorkflowVersioningBehavior = 'Pinned' "+ + "AND ExecutionStatus = 'Running'", + legacyVersion, version, + ) + req := &workflowservice.CountWorkflowExecutionsRequest{Query: qs} + return temporalClient.CountWorkflow(ctx, req) +} + // isVersionNotRegistered checks whether the Temporal server had no record of buildID when // the status was generated. func isVersionNotRegistered(workerDeploy *temporaliov1alpha1.WorkerDeployment, buildID string) bool { From b45efe50c79e561e5e24858e04082143aab180a5 Mon Sep 17 00:00:00 2001 From: Mitch Ross Date: Thu, 17 Sep 2026 11:30:36 -0400 Subject: [PATCH 4/5] refactor(controller): reuse deletion candidates for resource retention Remove retainedResourceBuilds as requested in review. Use markedForDeletion to identify rejected Inactive Deployment deletions and preserve their worker resources without maintaining a separate map. Keep independent resource deletions, including orphan cleanup and drained autoscaler sunset, unchanged. Extend the existing regression cases to check orphan cleanup alongside retained Inactive versions. Report the actual observed state when the inactive-retirement integration test times out. Validation: unit tests, go vet, import formatting, and the three targeted server-backed retirement integration scenarios passed locally. --- internal/controller/execplan.go | 30 ++++++++++++------- internal/controller/execplan_test.go | 9 +++++- .../inactive_retirement_integration_test.go | 9 ++++-- 3 files changed, 33 insertions(+), 15 deletions(-) diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go index d5d3d574..6f9ca954 100644 --- a/internal/controller/execplan.go +++ b/internal/controller/execplan.go @@ -705,9 +705,6 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( ) { identity := getControllerIdentity() markedForDeletion := make([]*appsv1.Deployment, 0, len(p.DeleteDeployments)) - // Worker resources have a separate deletion list. Track retained inactive - // versions so skipping their Deployments also preserves their resources. - retainedResourceBuilds := make(map[string]bool) for _, d := range p.DeleteDeployments { buildID, ok := d.GetLabels()[k8s.BuildIDLabel] if !ok { @@ -725,7 +722,6 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( if slices.ContainsFunc(workerDeploy.Status.DeprecatedVersions, func(v *temporaliov1alpha1.DeprecatedWorkerDeploymentVersion) bool { return v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive }) { - retainedResourceBuilds[buildID] = true count, err := getOpenPinnedWorkflowExecutions(ctx, temporalClient, p.WorkerDeploymentName, buildID) if err != nil || count == nil { l.Info("could not confirm inactive version has no running pinned workflows, keeping its Deployment", "buildID", buildID, "error", err) @@ -755,15 +751,27 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( l.Info("deleted deprecated worker deployment version", "buildID", buildID) } markedForDeletion = append(markedForDeletion, d) - delete(retainedResourceBuilds, buildID) + } + // Keep resources belonging to inactive Deployments whose deletion was refused. + // Other resource deletions, such as orphan cleanup, remain in the plan. + for _, d := range p.DeleteDeployments { + if slices.Contains(markedForDeletion, d) { + continue + } + buildID, ok := d.Labels[k8s.BuildIDLabel] + if !ok { + continue + } + if !slices.ContainsFunc(workerDeploy.Status.DeprecatedVersions, func(v *temporaliov1alpha1.DeprecatedWorkerDeploymentVersion) bool { + return v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive + }) { + continue + } + p.DeleteWorkerResources = slices.DeleteFunc(p.DeleteWorkerResources, func(ref planner.WorkerResourceRef) bool { + return ref.BuildID == buildID + }) } p.DeleteDeployments = markedForDeletion - // Resources nominated with an inactive Deployment must survive if its deletion - // was refused. Otherwise, for example, its ConfigMaps could disappear underneath - // a version retained for pinned workflows. - p.DeleteWorkerResources = slices.DeleteFunc(p.DeleteWorkerResources, func(ref planner.WorkerResourceRef) bool { - return retainedResourceBuilds[ref.BuildID] - }) } func getOpenPinnedWorkflowExecutions( diff --git a/internal/controller/execplan_test.go b/internal/controller/execplan_test.go index f70b404a..a1fe75d2 100644 --- a/internal/controller/execplan_test.go +++ b/internal/controller/execplan_test.go @@ -648,11 +648,16 @@ func TestExecutePlan_InactiveVersionDeletion(t *testing.T) { wrt.Spec.Template.Raw = []byte(`{"apiVersion":"v1","kind":"ConfigMap","data":{"test":"present"}}`) oldConfig, oldHash := renderWRT(t, wrt, old, "old", testTemporalNamespace) currentConfig, currentHash := renderWRT(t, wrt, current, "current", testTemporalNamespace) + // This resource outlived its Deployment and must still be cleaned up, + // even when the old version's Deployment and resources are retained. + orphan := makeVersionedDeployment(twd, "orphan", 0, connection) + orphanConfig, orphanHash := renderWRT(t, wrt, orphan, "orphan", testTemporalNamespace) wrt.Status.Versions = []temporaliov1alpha1.WorkerResourceTemplateVersionStatus{ k8s.WorkerResourceTemplateVersionStatusForBuildID("old", oldConfig.GetName(), 1, oldHash, ""), k8s.WorkerResourceTemplateVersionStatusForBuildID("current", currentConfig.GetName(), 1, currentHash, ""), + k8s.WorkerResourceTemplateVersionStatusForBuildID("orphan", orphanConfig.GetName(), 1, orphanHash, ""), } - r, _ := newTestReconciler([]client.Object{twd, old, current, wrt, oldConfig, currentConfig}) + r, _ := newTestReconciler([]client.Object{twd, old, current, wrt, oldConfig, currentConfig, orphanConfig}) handle := newPruneStubHandle(tc.deleteErr) c := &inactivePruneClient{stubTemporalClient: newStubTemporalClientWithHandle(handle), response: tc.response, err: tc.countErr} status := statusWithDeprecated("current", current, &temporaliov1alpha1.DeprecatedWorkerDeploymentVersion{ @@ -677,6 +682,8 @@ func TestExecutePlan_InactiveVersionDeletion(t *testing.T) { require.NoError(t, configErr, "retained version must keep its rendered resources") } require.NoError(t, r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: currentConfig.GetName()}, &corev1.ConfigMap{})) + orphanErr := r.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: orphanConfig.GetName()}, &corev1.ConfigMap{}) + require.True(t, apierrors.IsNotFound(orphanErr), "orphan cleanup must continue when another version is retained") require.Contains(t, c.query, "TemporalWorkerDeploymentVersion IN ('default/my-worker.old', 'default/my-worker:old')") require.Contains(t, c.query, "TemporalWorkflowVersioningBehavior = 'Pinned'") require.Contains(t, c.query, "ExecutionStatus = 'Running'") diff --git a/internal/tests/internal/inactive_retirement_integration_test.go b/internal/tests/internal/inactive_retirement_integration_test.go index b1050c3c..d0b6c380 100644 --- a/internal/tests/internal/inactive_retirement_integration_test.go +++ b/internal/tests/internal/inactive_retirement_integration_test.go @@ -113,11 +113,14 @@ func testInactiveVersionRetirement(t *testing.T, k8sClient client.Client, ts *te return err } for _, v := range next.Status.DeprecatedVersions { - if v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive && v.DrainedSince == nil { - return nil + if v.BuildID == buildID { + if v.Status == temporaliov1alpha1.VersionStatusInactive && v.DrainedSince == nil { + return nil + } + return fmt.Errorf("superseded version has status %s and drainedSince %v", v.Status, v.DrainedSince) } } - return fmt.Errorf("superseded version is not yet Inactive") + return fmt.Errorf("superseded version %s is not yet in deprecated versions", buildID) }) stopOnce() scaleDeploymentToZero(t, ctx, k8sClient, oldKey.Name, namespace) From 6e7f182743938dbfb119b3925f9c8e86abb8dd5d Mon Sep 17 00:00:00 2001 From: Mitch Ross Date: Fri, 18 Sep 2026 17:18:37 -0400 Subject: [PATCH 5/5] refactor(controller): remove reviewed resource retention block Remove the additional worker-resource filtering block as requested in review. Keep the existing markedForDeletion handling for Deployments. Validation: go vet ./... and git diff --check passed. go test ./... fails four existing TestExecutePlan_InactiveVersionDeletion cases with a missing ConfigMap: running pinned workflow, visibility failure, missing response, and rejected server deletion. Test expectations are unchanged. --- internal/controller/execplan.go | 19 ------------------- 1 file changed, 19 deletions(-) diff --git a/internal/controller/execplan.go b/internal/controller/execplan.go index 6f9ca954..badeb9e2 100644 --- a/internal/controller/execplan.go +++ b/internal/controller/execplan.go @@ -752,25 +752,6 @@ func (r *WorkerDeploymentReconciler) deleteDeprecatedVersions( } markedForDeletion = append(markedForDeletion, d) } - // Keep resources belonging to inactive Deployments whose deletion was refused. - // Other resource deletions, such as orphan cleanup, remain in the plan. - for _, d := range p.DeleteDeployments { - if slices.Contains(markedForDeletion, d) { - continue - } - buildID, ok := d.Labels[k8s.BuildIDLabel] - if !ok { - continue - } - if !slices.ContainsFunc(workerDeploy.Status.DeprecatedVersions, func(v *temporaliov1alpha1.DeprecatedWorkerDeploymentVersion) bool { - return v.BuildID == buildID && v.Status == temporaliov1alpha1.VersionStatusInactive - }) { - continue - } - p.DeleteWorkerResources = slices.DeleteFunc(p.DeleteWorkerResources, func(ref planner.WorkerResourceRef) bool { - return ref.BuildID == buildID - }) - } p.DeleteDeployments = markedForDeletion }