fix(status): retry UpdateStatus on optimistic-concurrency conflict (krateo-core-provider#57)

internal/composition/composition.go

@@ -266,7 +266,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c
 				"waiting for an in-flight helm operation to settle (status %s, pending for %s)",
 				string(rel.Status), pendingFor.Truncate(time.Second))
 			unstructuredtools.SetConditions(mg, pendingCond)
-			if _, uerr := tools.UpdateStatus(ctx, mg, updateOpts); uerr != nil {
+			if _, uerr := updateStatusWithRetry(ctx, mg, updateOpts); uerr != nil {
 				// Non-fatal: the wait itself is not an error, and failing the reconcile here would
 				// burn retry budget on a purely cosmetic write.
 				log.Debug("Could not record the pending-operation condition", "error", uerr)
@@ -310,7 +310,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c
 		condition := condition.Unavailable()
 		condition.Message = retErr.Error()
 		unstructuredtools.SetConditions(mg, condition)
-		_, err = tools.UpdateStatus(ctx, mg, updateOpts)
+		_, err = updateStatusWithRetry(ctx, mg, updateOpts)
 		if err != nil {
 			return controller.ExternalObservation{}, fmt.Errorf("updating status after failure: %w", err)
 		}
@@ -326,7 +326,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c
 		condition := condition.Unavailable()
 		condition.Message = retErr.Error()
 		unstructuredtools.SetConditions(mg, condition)
-		_, err = tools.UpdateStatus(ctx, mg, updateOpts)
+		_, err = updateStatusWithRetry(ctx, mg, updateOpts)
 		if err != nil {
 			return controller.ExternalObservation{}, fmt.Errorf("updating status after failure: %w", err)
 		}
@@ -398,7 +398,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c
 		condition := condition.Unavailable()
 		condition.Message = retErr.Error()
 		unstructuredtools.SetConditions(mg, condition)
-		_, err = tools.UpdateStatus(ctx, mg, updateOpts)
+		_, err = updateStatusWithRetry(ctx, mg, updateOpts)
 		if err != nil {
 			return controller.ExternalObservation{}, fmt.Errorf("updating status after failure: %w", err)
 		}
@@ -445,7 +445,7 @@ func (h *handler) Observe(ctx context.Context, mg *unstructured.Unstructured) (c
 		return controller.ExternalObservation{}, err
 	}
 
-	_, err = tools.UpdateStatus(ctx, mg, updateOpts)
+	_, err = updateStatusWithRetry(ctx, mg, updateOpts)
 	if err != nil {
 		return controller.ExternalObservation{}, err
 	}
@@ -621,7 +621,7 @@ func (h *handler) Create(ctx context.Context, mg *unstructured.Unstructured) err
 	log.Debug("Composition created.", "package", pkg.URL)
 
 	h.eventRecorder.Event(mg, event.Normal(reasonCreated, "Create", fmt.Sprintf("Composition created: %s", mg.GetName())))
-	mg, err = tools.UpdateStatus(ctx, mg, updateOpts)
+	mg, err = updateStatusWithRetry(ctx, mg, updateOpts)
 	if err != nil {
 		return fmt.Errorf("updating cr with values: %w", err)
 	}
@@ -765,7 +765,7 @@ func (h *handler) Update(ctx context.Context, mg *unstructured.Unstructured) err
 		return fmt.Errorf("setting status: %w", err)
 	}
 
-	mg, err = tools.UpdateStatus(ctx, mg, tools.UpdateOptions{
+	mg, err = updateStatusWithRetry(ctx, mg, tools.UpdateOptions{
 		Pluralizer:    h.pluralizer,
 		DynamicClient: dyn,
 	})

internal/composition/status_update.go

@@ -0,0 +1,57 @@
+package composition
+
+import (
+	"context"
+
+	"github.com/krateoplatformops/unstructured-runtime/pkg/tools"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/client-go/util/retry"
+)
+
+// updateStatusWithRetry writes mg's status subresource, retrying on optimistic-concurrency
+// conflicts. On a 409 it re-reads the object solely to pick up the current resourceVersion, then
+// re-submits the status this reconcile already computed. It is a drop-in replacement for
+// tools.UpdateStatus.
+//
+// Why this exists: tools.UpdateStatus submits mg with whatever resourceVersion it was read at and
+// performs no retry, so a benign "the object has been modified" 409 aborts the whole reconcile.
+// During a GVK-migration handover the retiring per-version controller and this one briefly contend
+// on the same composition CR's status, making that 409 routine; without a retry the healthy
+// Ready/Synced condition never lands and the composition stays wedged Ready=False until someone
+// manually restarts the controller (braghettos/krateo-core-provider#57). Re-submitting our status
+// is safe because this controller owns the status subresource (last-writer-wins).
+//
+// Note: we carry mg's status as-is and refresh only its resourceVersion — we deliberately do NOT
+// deep-copy the status subtree. mg.Object["status"] can hold typed values that
+// runtime.DeepCopyJSONValue (used by unstructured.NestedFieldCopy) panics on.
+func updateStatusWithRetry(ctx context.Context, mg *unstructured.Unstructured, opts tools.UpdateOptions) (*unstructured.Unstructured, error) {
+	var result *unstructured.Unstructured
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		updated, uErr := tools.UpdateStatus(ctx, mg, opts)
+		if uErr == nil {
+			result = updated
+			return nil
+		}
+		if !apierrors.IsConflict(uErr) {
+			return uErr
+		}
+		// Conflict: re-read for a fresh resourceVersion and let RetryOnConflict re-invoke.
+		// Pluralizer/DynamicClient are non-nil here — tools.UpdateStatus validates them and would
+		// have returned a non-conflict error otherwise.
+		gvr, gErr := opts.Pluralizer.GVKtoGVR(mg.GroupVersionKind())
+		if gErr != nil {
+			return gErr
+		}
+		latest, gErr := opts.DynamicClient.Resource(gvr).
+			Namespace(mg.GetNamespace()).
+			Get(ctx, mg.GetName(), metav1.GetOptions{})
+		if gErr != nil {
+			return gErr
+		}
+		mg.SetResourceVersion(latest.GetResourceVersion())
+		return uErr
+	})
+	return result, err
+}

internal/composition/status_update_test.go

@@ -0,0 +1,133 @@
+package composition
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/gobuffalo/flect"
+	"github.com/krateoplatformops/unstructured-runtime/pkg/pluralizer"
+	"github.com/krateoplatformops/unstructured-runtime/pkg/tools"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
+	clienttesting "k8s.io/client-go/testing"
+)
+
+type statusRetryPluralizer struct{}
+
+var _ pluralizer.PluralizerInterface = statusRetryPluralizer{}
+
+func (statusRetryPluralizer) GVKtoGVR(gvk schema.GroupVersionKind) (schema.GroupVersionResource, error) {
+	return schema.GroupVersionResource{
+		Group:    gvk.Group,
+		Version:  gvk.Version,
+		Resource: flect.Pluralize(strings.ToLower(gvk.Kind)),
+	}, nil
+}
+
+var srGVR = schema.GroupVersionResource{Group: "composition.krateo.io", Version: "v0-1-0", Resource: "portals"}
+
+func srPortal(rv, statusMarker string) *unstructured.Unstructured {
+	u := &unstructured.Unstructured{}
+	u.SetGroupVersionKind(schema.GroupVersionKind{Group: "composition.krateo.io", Version: "v0-1-0", Kind: "Portal"})
+	u.SetName("portal")
+	u.SetNamespace("krateo-system")
+	u.SetResourceVersion(rv)
+	_ = unstructured.SetNestedField(u.Object, statusMarker, "status", "marker")
+	return u
+}
+
+func srFakeDyn(objs ...runtime.Object) *dynamicfake.FakeDynamicClient {
+	scheme := runtime.NewScheme()
+	gvrToListKind := map[schema.GroupVersionResource]string{srGVR: "PortalList"}
+	return dynamicfake.NewSimpleDynamicClientWithCustomListKinds(scheme, gvrToListKind, objs...)
+}
+
+func srOpts(dyn *dynamicfake.FakeDynamicClient) tools.UpdateOptions {
+	return tools.UpdateOptions{Pluralizer: statusRetryPluralizer{}, DynamicClient: dyn}
+}
+
+// A 409 on the first status write (as happens during a GVK-migration handover) must be retried
+// after a re-fetch, and the status this reconcile computed must ultimately land. Regression guard
+// for braghettos/krateo-core-provider#57.
+func TestUpdateStatusWithRetry_RecoversFromConflict(t *testing.T) {
+	// The stored object was concurrently modified (rv=2) after we read it (rv=1).
+	dyn := srFakeDyn(srPortal("2", "concurrent"))
+
+	var statusUpdates int
+	dyn.PrependReactor("update", "portals", func(action clienttesting.Action) (bool, runtime.Object, error) {
+		ua, ok := action.(clienttesting.UpdateAction)
+		if !ok || ua.GetSubresource() != "status" {
+			return false, nil, nil
+		}
+		statusUpdates++
+		if statusUpdates == 1 {
+			return true, nil, apierrors.NewConflict(srGVR.GroupResource(), "portal", fmt.Errorf("the object has been modified"))
+		}
+		return false, nil, nil // fall through to the tracker (success)
+	})
+
+	// Our reconcile computed a "healthy" status against the now-stale rv=1.
+	mg := srPortal("1", "healthy")
+	got, err := updateStatusWithRetry(context.Background(), mg, srOpts(dyn))
+	if err != nil {
+		t.Fatalf("expected recovery from the conflict, got error: %v", err)
+	}
+	if statusUpdates < 2 {
+		t.Fatalf("expected a retry (>=2 status writes), got %d", statusUpdates)
+	}
+	if m, _, _ := unstructured.NestedString(got.Object, "status", "marker"); m != "healthy" {
+		t.Fatalf("computed status should win after re-fetch; got marker=%q", m)
+	}
+}
+
+// A non-conflict error must propagate immediately and never be retried.
+func TestUpdateStatusWithRetry_NonConflictNotRetried(t *testing.T) {
+	dyn := srFakeDyn(srPortal("1", "x"))
+
+	var statusUpdates int
+	dyn.PrependReactor("update", "portals", func(action clienttesting.Action) (bool, runtime.Object, error) {
+		ua, ok := action.(clienttesting.UpdateAction)
+		if !ok || ua.GetSubresource() != "status" {
+			return false, nil, nil
+		}
+		statusUpdates++
+		return true, nil, apierrors.NewInternalError(fmt.Errorf("boom"))
+	})
+
+	_, err := updateStatusWithRetry(context.Background(), srPortal("1", "healthy"), srOpts(dyn))
+	if err == nil {
+		t.Fatal("expected the non-conflict error to propagate")
+	}
+	if statusUpdates != 1 {
+		t.Fatalf("a non-conflict error must not be retried; got %d attempts", statusUpdates)
+	}
+}
+
+// The happy path: no conflict -> a single write, computed status returned.
+func TestUpdateStatusWithRetry_SuccessFirstTry(t *testing.T) {
+	dyn := srFakeDyn(srPortal("1", "x"))
+
+	var statusUpdates int
+	dyn.PrependReactor("update", "portals", func(action clienttesting.Action) (bool, runtime.Object, error) {
+		if ua, ok := action.(clienttesting.UpdateAction); ok && ua.GetSubresource() == "status" {
+			statusUpdates++
+		}
+		return false, nil, nil
+	})
+
+	got, err := updateStatusWithRetry(context.Background(), srPortal("1", "healthy"), srOpts(dyn))
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if statusUpdates != 1 {
+		t.Fatalf("expected exactly 1 status write, got %d", statusUpdates)
+	}
+	if m, _, _ := unstructured.NestedString(got.Object, "status", "marker"); m != "healthy" {
+		t.Fatalf("expected healthy marker, got %q", m)
+	}
+}