fix(controller): re-apply worker resources when tracked object is missing

internal/controller/execplan.go

@@ -23,6 +23,7 @@
 	appsv1 "k8s.io/api/apps/v1"
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -323,7 +324,7 @@
 
 //nolint:revive // cyclomatic complexity acceptable given breadth of plan execution
 func (r *WorkerDeploymentReconciler) executePlan(ctx context.Context, l logr.Logger, workerDeploy *temporaliov1alpha1.WorkerDeployment, temporalClient sdkclient.Client, p *plan) error {
 	deletedWorkerResources, err := r.executeK8sOperations(ctx, l, workerDeploy, p)
 	if err != nil {
 		return err
 	}
@@ -351,9 +352,14 @@
 		}
 	}
 
-	// Apply worker resource templates via Server-Side Apply.
-	// Partial failure isolation: all resources are attempted even if some fail;
-	// errors are collected and returned together.
+	return r.applyWorkerResourceTemplates(ctx, l, p)
+}
+
+// applyWorkerResourceTemplates applies rendered worker resource templates via
+// Server-Side Apply and records per-Build-ID results in each WRT's status.
+// Partial failure isolation: all resources are attempted even if some fail;
+// errors are collected and returned together.
+func (r *WorkerDeploymentReconciler) applyWorkerResourceTemplates(ctx context.Context, l logr.Logger, p *plan) error {
 	type wrtKey struct{ namespace, name string }
 	type applyResult struct {
 		buildID      string
@@ -384,14 +390,47 @@
 		// successfully applied. This avoids unnecessary API server load at scale
 		// (hundreds of TWDs × hundreds of versions × multiple WRTs).
 		// An empty RenderedHash means hashing failed; always apply in that case.
+		//
+		// The hash match alone is not sufficient: the tracked resource may have been
+		// deleted since the hash was recorded. In particular, when a version is
+		// sunset its rendered resources are deleted (DeleteWorkerResources), but the
+		// WRT status entry for that build can survive — e.g. the status update that
+		// would drop it hits a conflict, or the version is re-registered before the
+		// next successful status write. If the same build ID then comes back, the
+		// stale LastAppliedHash matches the unchanged render and the resource is
+		// never re-created, leaving the returning version without its scaler
+		// (observed as a current version pinned at 1 replica with no ScaledObject).
+		// Guard the skip with an existence check on the rendered object.
 		if apply.RenderedHash != "" && apply.RenderedHash == apply.LastAppliedHash {
-			wrtResults[key] = append(wrtResults[key], applyResult{
-				buildID:      apply.BuildID,
-				resourceName: apply.Resource.GetName(),
-				hash:         apply.RenderedHash,
-				skipped:      true,
-			})
-			continue
+			existing := &unstructured.Unstructured{}
+			existing.SetGroupVersionKind(apply.Resource.GroupVersionKind())
+			getErr := r.Get(ctx, types.NamespacedName{
+				Namespace: apply.Resource.GetNamespace(),
+				Name:      apply.Resource.GetName(),
+			}, existing)
+			if getErr == nil {
+				wrtResults[key] = append(wrtResults[key], applyResult{
+					buildID:      apply.BuildID,
+					resourceName: apply.Resource.GetName(),
+					hash:         apply.RenderedHash,
+					skipped:      true,
+				})
+				continue
+			}
+			if !apierrors.IsNotFound(getErr) {
+				l.Error(getErr, "unable to confirm worker resource exists; re-applying",
+					"name", apply.Resource.GetName(),
+					"kind", apply.Resource.GetKind(),
+				)
+			} else {
+				l.Info("worker resource missing despite unchanged hash; re-applying",
+					"name", apply.Resource.GetName(),
+					"kind", apply.Resource.GetKind(),
+					"buildID", apply.BuildID,
+				)
+			}
+			// Fall through to the SSA apply: it is create-or-update, so re-applying
+			// is safe in both the NotFound and the indeterminate-error case.
 		}
 
 		l.Info("applying rendered worker resource template",
@@ -442,7 +481,7 @@
 	// pruning it on success is what allows a later redeploy of the same build ID to be
 	// re-applied instead of being skipped against a stale hash.
 	deletedBuildIDs := make(map[wrtKey]map[string]struct{})
 	for _, ref := range deletedWorkerResources {
 		if ref.WRTName == "" || ref.BuildID == "" {
 			continue
 		}

internal/controller/execplan_wrt_test.go

@@ -0,0 +1,121 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the MIT License.
+//
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2024 Datadog, Inc.
+
+package controller
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1"
+	"github.com/temporalio/temporal-worker-controller/internal/planner"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
+)
+
+// newRenderedConfigMap returns an unstructured ConfigMap standing in for a
+// rendered WRT resource (the controller treats rendered objects generically).
+func newRenderedConfigMap(namespace, name string) *unstructured.Unstructured {
+	u := &unstructured.Unstructured{}
+	u.SetAPIVersion("v1")
+	u.SetKind("ConfigMap")
+	u.SetNamespace(namespace)
+	u.SetName(name)
+	return u
+}
+
+func newTestWRT(namespace, name string) *temporaliov1alpha1.WorkerResourceTemplate {
+	return &temporaliov1alpha1.WorkerResourceTemplate{
+		ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name},
+	}
+}
+
+func newTestTWD(namespace, name string) *temporaliov1alpha1.WorkerDeployment {
+	return &temporaliov1alpha1.WorkerDeployment{
+		ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name},
+	}
+}
+
+// A version whose rendered hash matches the recorded LastAppliedHash but whose
+// rendered resource is absent from the cluster (version sunset deletes rendered
+// resources, and the build ID can be re-registered while its stale status entry
+// survives) must be re-applied, not skipped — otherwise the returning build ID
+// runs without its rendered resources.
+func TestExecutePlan_WRTApply_ReappliesWhenResourceMissing(t *testing.T) {
+	const ns = "default"
+	wrt := newTestWRT(ns, "test-wrt")
+	twd := newTestTWD(ns, "test-twd")
+
+	r, _ := newTestReconciler([]client.Object{wrt, twd})
+
+	rendered := newRenderedConfigMap(ns, "test-wrt-rendered")
+	p := &plan{
+		ApplyWorkerResources: []planner.WorkerResourceApply{{
+			Resource:        rendered,
+			WRTName:         wrt.Name,
+			WRTNamespace:    ns,
+			BuildID:         "build-1",
+			RenderedHash:    "hash-1",
+			LastAppliedHash: "hash-1", // matches, but the resource is gone
+		}},
+	}
+
+	err := r.applyWorkerResourceTemplates(context.Background(), ctrl.Log, p)
+	require.NoError(t, err)
+
+	// The rendered resource must have been re-created despite the hash match.
+	got := &corev1.ConfigMap{}
+	require.NoError(t, r.Get(context.Background(),
+		types.NamespacedName{Namespace: ns, Name: "test-wrt-rendered"}, got))
+
+	// The WRT status must record the build as applied (not skipped-with-stale-state).
+	gotWRT := &temporaliov1alpha1.WorkerResourceTemplate{}
+	require.NoError(t, r.Get(context.Background(),
+		types.NamespacedName{Namespace: ns, Name: wrt.Name}, gotWRT))
+	require.Len(t, gotWRT.Status.Versions, 1)
+	assert.Equal(t, "build-1", gotWRT.Status.Versions[0].BuildID)
+	assert.Equal(t, "hash-1", gotWRT.Status.Versions[0].LastAppliedHash)
+}
+
+// The skip fast-path must still hold when the resource exists and the hash is
+// unchanged: no SSA apply call is made.
+func TestExecutePlan_WRTApply_SkipsWhenResourceExistsAndHashUnchanged(t *testing.T) {
+	const ns = "default"
+	wrt := newTestWRT(ns, "test-wrt")
+	twd := newTestTWD(ns, "test-twd")
+	existing := newRenderedConfigMap(ns, "test-wrt-rendered")
+
+	patchCalls := 0
+	r, _ := newTestReconcilerWithInterceptors(
+		[]client.Object{wrt, twd, existing},
+		interceptor.Funcs{
+			Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
+				patchCalls++
+				return c.Patch(ctx, obj, patch, opts...)
+			},
+		},
+	)
+
+	p := &plan{
+		ApplyWorkerResources: []planner.WorkerResourceApply{{
+			Resource:        newRenderedConfigMap(ns, "test-wrt-rendered"),
+			WRTName:         wrt.Name,
+			WRTNamespace:    ns,
+			BuildID:         "build-1",
+			RenderedHash:    "hash-1",
+			LastAppliedHash: "hash-1",
+		}},
+	}
+
+	err := r.applyWorkerResourceTemplates(context.Background(), ctrl.Log, p)
+	require.NoError(t, err)
+	assert.Zero(t, patchCalls, "hash-unchanged apply with existing resource must be skipped")
+}