From 4116437eb64a31b6693c498e7b213967794bc267 Mon Sep 17 00:00:00 2001 From: Diego Braga Date: Thu, 30 Jul 2026 16:13:28 +0200 Subject: [PATCH 1/2] fix(status): retry UpdateStatus on optimistic-concurrency conflict MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tools.UpdateStatus submits the composition CR 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 the new one briefly contend on the same CR's status, making that 409 routine — the healthy Ready/Synced condition then never lands and the composition stays wedged Ready=False until a manual controller restart. Add updateStatusWithRetry (a drop-in for tools.UpdateStatus) that retries on conflict by re-fetching the latest object and re-applying the computed status; last-writer-wins is correct since this controller owns the status subresource. Wire all 7 status-write sites in composition.go through it, with unit tests covering conflict-recovery, non-conflict propagation, and the happy path. Refs braghettos/krateo-core-provider#57 (fix #1 of the layered plan; #2-#4 tracked on the issue). Co-Authored-By: Claude Opus 4.8 --- internal/composition/composition.go | 14 +-- internal/composition/status_update.go | 65 ++++++++++ internal/composition/status_update_test.go | 133 +++++++++++++++++++++ 3 files changed, 205 insertions(+), 7 deletions(-) create mode 100644 internal/composition/status_update.go create mode 100644 internal/composition/status_update_test.go diff --git a/internal/composition/composition.go b/internal/composition/composition.go index 0199141..cc93f51 100644 --- a/internal/composition/composition.go +++ b/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, }) diff --git a/internal/composition/status_update.go b/internal/composition/status_update.go new file mode 100644 index 0000000..963ddbb --- /dev/null +++ b/internal/composition/status_update.go @@ -0,0 +1,65 @@ +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 by re-fetching the latest object and re-applying the computed status. 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-fetching the latest +// object and re-applying our computed status is safe because this controller owns the status +// subresource, so last-writer-wins is the intended semantics. +func updateStatusWithRetry(ctx context.Context, mg *unstructured.Unstructured, opts tools.UpdateOptions) (*unstructured.Unstructured, error) { + // Snapshot the status this reconcile computed, so we can re-apply it onto a freshly-read object + // on conflict rather than resubmitting a stale resourceVersion forever. + desiredStatus, _, err := unstructured.NestedFieldCopy(mg.Object, "status") + if err != nil { + return mg, err + } + + 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-fetch for a fresh resourceVersion, re-apply our computed status, 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 + } + if desiredStatus != nil { + if sErr := unstructured.SetNestedField(latest.Object, desiredStatus, "status"); sErr != nil { + return sErr + } + } + *mg = *latest + return uErr + }) + return result, err +} diff --git a/internal/composition/status_update_test.go b/internal/composition/status_update_test.go new file mode 100644 index 0000000..bc7aac3 --- /dev/null +++ b/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) + } +} From 2071f2b8b4dc84403464434175ee7ed5191c4398 Mon Sep 17 00:00:00 2001 From: Diego Braga Date: Thu, 30 Jul 2026 16:33:26 +0200 Subject: [PATCH 2/2] fix(status): drop the status deep-copy (avoid DeepCopyJSONValue panic) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first cut snapshotted mg's status via unstructured.NestedFieldCopy to re-apply it on conflict, but NestedFieldCopy runs runtime.DeepCopyJSONValue, which panics ("cannot deep copy composition.ManagedResource") — the composition CR's in-memory status can hold typed values, not pure JSON. The integration TestController/Setup/Create exercised that path. Carry mg's status as-is and, on a 409, refresh only its resourceVersion before retrying — no status deep-copy. On the no-conflict path the behavior is now identical to the original tools.UpdateStatus call. Co-Authored-By: Claude Opus 4.8 --- internal/composition/status_update.go | 36 +++++++++++---------------- 1 file changed, 14 insertions(+), 22 deletions(-) diff --git a/internal/composition/status_update.go b/internal/composition/status_update.go index 963ddbb..750f26a 100644 --- a/internal/composition/status_update.go +++ b/internal/composition/status_update.go @@ -11,27 +11,24 @@ import ( ) // updateStatusWithRetry writes mg's status subresource, retrying on optimistic-concurrency -// conflicts by re-fetching the latest object and re-applying the computed status. It is a drop-in -// replacement for tools.UpdateStatus. +// 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-fetching the latest -// object and re-applying our computed status is safe because this controller owns the status -// subresource, so last-writer-wins is the intended semantics. +// 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) { - // Snapshot the status this reconcile computed, so we can re-apply it onto a freshly-read object - // on conflict rather than resubmitting a stale resourceVersion forever. - desiredStatus, _, err := unstructured.NestedFieldCopy(mg.Object, "status") - if err != nil { - return mg, err - } - var result *unstructured.Unstructured - err = retry.RetryOnConflict(retry.DefaultRetry, func() error { + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { updated, uErr := tools.UpdateStatus(ctx, mg, opts) if uErr == nil { result = updated @@ -40,9 +37,9 @@ func updateStatusWithRetry(ctx context.Context, mg *unstructured.Unstructured, o if !apierrors.IsConflict(uErr) { return uErr } - // Conflict: re-fetch for a fresh resourceVersion, re-apply our computed status, and let - // RetryOnConflict re-invoke. Pluralizer/DynamicClient are non-nil here (tools.UpdateStatus - // validates them and would have returned a non-conflict error otherwise). + // 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 @@ -53,12 +50,7 @@ func updateStatusWithRetry(ctx context.Context, mg *unstructured.Unstructured, o if gErr != nil { return gErr } - if desiredStatus != nil { - if sErr := unstructured.SetNestedField(latest.Object, desiredStatus, "status"); sErr != nil { - return sErr - } - } - *mg = *latest + mg.SetResourceVersion(latest.GetResourceVersion()) return uErr }) return result, err