feat(upgradepipeline): avoid stale object updates in status sync

Author: ndk

controllers/kafkaschema_controller.go

@@ -14,7 +14,6 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/util/retry"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -255,11 +254,6 @@ func (r *KafkaSchemaController) applySchema(ctx context.Context, schema *v1alpha
 	// ID is used by Observe to look up the version, which may take some time to appear.
 	schema.Status.ID = schemaID
 
-	// TODO: workaround for a stale-cache race in the managed. Remove once the reconciler fix lands.
-	if err := r.persistStatusID(ctx, schema); err != nil {
-		return fmt.Errorf("persisting Status.ID: %w", err)
-	}
-
 	if schema.Spec.CompatibilityLevel != "" {
 		if _, err := r.avnGen.ServiceSchemaRegistrySubjectConfigPut(
 			ctx,
@@ -339,26 +333,6 @@ func referencesEqual(desired []kafkaschemaregistry.ReferenceIn, got []kafkaschem
 	return true
 }
 
-// persistStatusID writes schema.Status.ID to the API server in its own status
-// subresource update, retrying on optimistic-concurrency conflicts.
-//
-// This is a workaround for the reconciler race issue.
-// It should be removed when updateStatus in the managed reconciler no longer clobbers
-// concurrently-written status fields.
-func (r *KafkaSchemaController) persistStatusID(ctx context.Context, schema *v1alpha1.KafkaSchema) error {
-	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		latest := &v1alpha1.KafkaSchema{}
-		if err := r.Get(ctx, client.ObjectKeyFromObject(schema), latest); err != nil {
-			return err
-		}
-		latest.Status.ID = schema.Status.ID
-		if latest.Status.Conditions == nil {
-			latest.Status.Conditions = []metav1.Condition{}
-		}
-		return r.Status().Update(ctx, latest)
-	})
-}
-
 func (r *KafkaSchemaController) Delete(ctx context.Context, schema *v1alpha1.KafkaSchema) error {
 	// Block delete if any KafkaSchema in this namespace still imports us via kafkaSchemaRef.
 	// Only catches kafkaSchemaRef dependents in the same namespace.

controllers/reconciler.go

@@ -2,6 +2,7 @@ package controllers
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"maps"
@@ -17,7 +18,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/util/retry"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -232,43 +232,51 @@ func (r *Reconciler[T]) resolveK8sRefs(ctx context.Context, obj T) (requeue bool
 	return false, nil
 }
 
-// updateStatus persists spec/metadata and status of obj.
-//
-// KNOWN ISSUE — stale-status race:
-// The Status().Update below sends obj.Status verbatim. If obj was built from
-// a stale informer-cache snapshot, this can clobber a status field that a
-// concurrent reconcile pass just persisted.
-//
-// For controllers built: do not use .status as an input to control flow. Treat .status as user-observable output only.
+// updateStatus persists status and controller-owned readiness annotations.
 func (r *Reconciler[T]) updateStatus(ctx context.Context, orig v1alpha1.AivenManagedObject, obj v1alpha1.AivenManagedObject) error {
 	if equality.Semantic.DeepEqual(orig, obj) {
 		return nil
 	}
 
-	// Order matters.
-	// First need to update the object, and then update the status.
-	// So dependent resources won't see READY before it has been updated with new values
-
-	// Now we can update the status
-	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		// When updated, object status is vanished.
-		// So we waste a copy for that,
-		// while the original object must already have all the fields updated in runtime
-		// Additionally, it gets the "latest version" to resolve optimistic concurrency control conflict
-		latest := obj.DeepCopyObject().(client.Object)
-		if err := r.Get(ctx, types.NamespacedName{Name: latest.GetName(), Namespace: latest.GetNamespace()}, latest); err != nil {
-			return err
+	// Capture annotation changes before Status().Update because it may mutate obj metadata.
+	annotations := map[string]any{}
+	for _, key := range []string{processedGenerationAnnotation, instanceIsRunningAnnotation} {
+		origValue, origOk := orig.GetAnnotations()[key]
+		value, ok := obj.GetAnnotations()[key]
+		if origOk == ok && origValue == value {
+			continue
 		}
 
-		updated := obj.DeepCopyObject().(client.Object)
-		updated.SetResourceVersion(latest.GetResourceVersion())
-		if err := r.Update(ctx, updated); err != nil {
-			return err
+		if ok {
+			annotations[key] = value
+		} else {
+			annotations[key] = nil
 		}
+	}
+
+	if err := r.Status().Update(ctx, obj); err != nil {
+		return err
+	}
+
+	if len(annotations) == 0 {
+		return nil
+	}
 
-		obj.SetResourceVersion(updated.GetResourceVersion())
-		return r.Status().Update(ctx, obj)
+	payload, err := json.Marshal(map[string]any{
+		"metadata": map[string]any{
+			"resourceVersion": obj.GetResourceVersion(),
+			"annotations":     annotations,
+		},
 	})
+	if err != nil {
+		return fmt.Errorf("marshalling managed annotations patch: %w", err)
+	}
+
+	if err := r.Patch(ctx, obj, client.RawPatch(types.MergePatchType, payload)); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (r *Reconciler[T]) newAivenClient(ctx context.Context, obj T) (avngen.Client, error) {