fix(syncer): retry status patches on latest object

pkg/controllers/resources/persistentvolumeclaims/syncer.go

@@ -311,6 +311,7 @@ func (s *persistentVolumeClaimSyncer) Sync(ctx *synccontext.SyncContext, event *
 	if !maps.Equal(oldHostAnnotations, event.Host.GetAnnotations()) {
 		translateDataProtectionRestoreSourceAnnotationsToVirtual(ctx, event.Virtual)
 	}
+	ensureVirtualPVCBindCompletionAnnotations(event.Host, event.Virtual)
 
 	return ctrl.Result{}, nil
 }
@@ -328,6 +329,7 @@ func (s *persistentVolumeClaimSyncer) SyncToVirtual(ctx *synccontext.SyncContext
 		return ctrl.Result{}, err
 	}
 
+	ensureVirtualPVCBindCompletionAnnotations(event.Host, vPvc)
 	return patcher.CreateVirtualObject(ctx, event.Host, vPvc, s.EventRecorder(), true)
 }
 
@@ -759,6 +761,28 @@ func isVirtualPVCBound(pvc *corev1.PersistentVolumeClaim) bool {
 	return ok && !storage.IsZero()
 }
 
+func ensureVirtualPVCBindCompletionAnnotations(pObj, vObj *corev1.PersistentVolumeClaim) {
+	if pObj == nil || vObj == nil || !isVirtualPVCBound(vObj) || pObj.Spec.VolumeName == "" || pObj.Status.Phase != corev1.ClaimBound {
+		return
+	}
+
+	hostAnnotations := pObj.GetAnnotations()
+	bindCompleted, ok := hostAnnotations[bindCompletedAnnotation]
+	if !ok {
+		return
+	}
+
+	annotations := maps.Clone(vObj.GetAnnotations())
+	if annotations == nil {
+		annotations = map[string]string{}
+	}
+	annotations[bindCompletedAnnotation] = bindCompleted
+	if boundByController, ok := hostAnnotations[boundByControllerAnnotation]; ok {
+		annotations[boundByControllerAnnotation] = boundByController
+	}
+	vObj.SetAnnotations(annotations)
+}
+
 func isHostPVCWaitingForVolume(pvc *corev1.PersistentVolumeClaim) bool {
 	if pvc.Status.Phase == corev1.ClaimBound {
 		return false

pkg/controllers/resources/persistentvolumeclaims/syncer_test.go

@@ -301,8 +301,15 @@ func TestSync(t *testing.T) {
 			corev1.ResourceStorage: resource.MustParse("1Gi"),
 		},
 	}
+	dataProtectionNoDataHostBoundWithBackupSource.Annotations[bindCompletedAnnotation] = "yes"
+	dataProtectionNoDataHostBoundWithBackupSource.Annotations[boundByControllerAnnotation] = "yes"
 	dataProtectionNoDataRestorePvcWithHostBoundStatus := dataProtectionNoDataRestorePvc.DeepCopy()
 	dataProtectionNoDataRestorePvcWithHostBoundStatus.Status = *dataProtectionNoDataHostBoundWithBackupSource.Status.DeepCopy()
+	if dataProtectionNoDataRestorePvcWithHostBoundStatus.Annotations == nil {
+		dataProtectionNoDataRestorePvcWithHostBoundStatus.Annotations = map[string]string{}
+	}
+	dataProtectionNoDataRestorePvcWithHostBoundStatus.Annotations[bindCompletedAnnotation] = "yes"
+	dataProtectionNoDataRestorePvcWithHostBoundStatus.Annotations[boundByControllerAnnotation] = "yes"
 	dataProtectionNoDataHostPendingWithoutBackupSource := dataProtectionHostPendingPvcWithUID.DeepCopy()
 	dataProtectionNoDataHostPendingWithoutBackupSource.Spec = corev1.PersistentVolumeClaimSpec{}
 	dataProtectionNoDataHostDeletingWithoutBackupSource := dataProtectionNoDataHostPendingWithoutBackupSource.DeepCopy()

pkg/patcher/apply.go

@@ -15,6 +15,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/events"
+	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -224,12 +225,14 @@ func applyObjectWithPatch(ctx *synccontext.SyncContext, objPatch patch.Patch, ob
 		kubeClient = ctx.VirtualClient
 	}
 
-	// check if we should create or update the object
-	isUpdate := false
-	err := kubeClient.Get(ctx, types.NamespacedName{
+	key := types.NamespacedName{
 		Namespace: obj.GetNamespace(),
 		Name:      obj.GetName(),
-	}, obj.DeepCopyObject().(client.Object))
+	}
+
+	// check if we should create or update the object
+	isUpdate := false
+	err := kubeClient.Get(ctx, key, obj.DeepCopyObject().(client.Object))
 	if err != nil && !kerrors.IsNotFound(err) {
 		return fmt.Errorf("get object: %w", err)
 	} else if err == nil {
@@ -241,6 +244,61 @@ func applyObjectWithPatch(ctx *synccontext.SyncContext, objPatch patch.Patch, ob
 		return fmt.Errorf("cannot create status only object")
 	}
 
+	// Status updates should be based on the latest object because binders,
+	// populators, and other controllers commonly update status concurrently.
+	// Full-object updates keep the original event object semantics because some
+	// syncers intentionally overwrite fields rather than merge with newer state.
+	if isUpdate && isStatus {
+		err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+			latestObj := obj.DeepCopyObject().(client.Object)
+			err := kubeClient.Get(ctx, key, latestObj)
+			if err != nil {
+				return fmt.Errorf("get object: %w", err)
+			}
+
+			beforeObject := latestObj.DeepCopyObject().(client.Object)
+			err = objPatch.Apply(latestObj)
+			if err != nil {
+				return fmt.Errorf("apply patch: %w", err)
+			} else if apiequality.Semantic.DeepEqual(beforeObject, latestObj) {
+				// nothing to patch
+				return nil
+			}
+
+			logUpdate(ctx, isStatus, direction, beforeObject, latestObj)
+
+			// update
+			afterObj := latestObj.DeepCopyObject().(client.Object)
+			err = kubeClient.Status().Update(ctx, latestObj)
+			if err != nil {
+				return err
+			}
+
+			// set the fields correctly, but only if the update succeeds
+			afterObj.SetUID(latestObj.GetUID())
+			afterObj.SetGeneration(latestObj.GetGeneration())
+			afterObj.SetResourceVersion(latestObj.GetResourceVersion())
+			afterObj.SetCreationTimestamp(latestObj.GetCreationTimestamp())
+			afterObj.SetDeletionTimestamp(latestObj.GetDeletionTimestamp())
+			afterObj.SetManagedFields(latestObj.GetManagedFields())
+			afterObj.SetDeletionGracePeriodSeconds(latestObj.GetDeletionGracePeriodSeconds())
+			afterObj.SetGenerateName(latestObj.GetGenerateName())
+			afterObj.SetOwnerReferences(latestObj.GetOwnerReferences())
+			if ctx.ObjectCache != nil {
+				if direction == synccontext.SyncHostToVirtual {
+					ctx.ObjectCache.Virtual().Put(afterObj)
+				} else if direction == synccontext.SyncVirtualToHost {
+					ctx.ObjectCache.Host().Put(afterObj)
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			return fmt.Errorf("update object status: %w", err)
+		}
+		return nil
+	}
+
 	// apply the patch when it's an update, otherwise the patch is the create
 	if isUpdate {
 		beforeObject := obj.DeepCopyObject().(client.Object)

pkg/patcher/apply_test.go

@@ -1,14 +1,80 @@
 package patcher
 
 import (
+	"context"
 	"encoding/json"
 	"testing"
 
+	"github.com/loft-sh/vcluster/pkg/scheme"
+	"github.com/loft-sh/vcluster/pkg/syncer/synccontext"
+	testingutil "github.com/loft-sh/vcluster/pkg/util/testing"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+func TestApplyObjectStatusPatchUsesLatestObjectOnUpdate(t *testing.T) {
+	live := &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pvc",
+			Namespace: "ns",
+		},
+		Status: corev1.PersistentVolumeClaimStatus{
+			Phase: corev1.ClaimPending,
+			Conditions: []corev1.PersistentVolumeClaimCondition{
+				{
+					Type:   corev1.PersistentVolumeClaimFileSystemResizePending,
+					Status: corev1.ConditionTrue,
+				},
+			},
+		},
+	}
+	before := &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      live.Name,
+			Namespace: live.Namespace,
+		},
+		Status: corev1.PersistentVolumeClaimStatus{
+			Phase: corev1.ClaimPending,
+		},
+	}
+	after := before.DeepCopy()
+	after.Status = corev1.PersistentVolumeClaimStatus{
+		Phase:       corev1.ClaimBound,
+		AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+		Capacity: corev1.ResourceList{
+			corev1.ResourceStorage: resource.MustParse("1Gi"),
+		},
+	}
+
+	vClient := testingutil.NewFakeClient(scheme.Scheme, live)
+	ctx := &synccontext.SyncContext{
+		Context:       context.Background(),
+		HostClient:    testingutil.NewFakeClient(scheme.Scheme),
+		VirtualClient: vClient,
+	}
+
+	if err := ApplyObject(ctx, before, after, synccontext.SyncHostToVirtual, true); err != nil {
+		t.Fatalf("ApplyObject() error = %v", err)
+	}
+
+	got := &corev1.PersistentVolumeClaim{}
+	if err := vClient.Get(ctx, client.ObjectKeyFromObject(live), got); err != nil {
+		t.Fatalf("get live object: %v", err)
+	}
+	if got.Status.Phase != corev1.ClaimBound {
+		t.Fatalf("status phase = %s, want %s", got.Status.Phase, corev1.ClaimBound)
+	}
+	storage := got.Status.Capacity[corev1.ResourceStorage]
+	if storage.IsZero() {
+		t.Fatalf("status capacity missing after patch: %#v", got.Status.Capacity)
+	}
+	if len(got.Status.Conditions) != 1 || got.Status.Conditions[0].Type != corev1.PersistentVolumeClaimFileSystemResizePending {
+		t.Fatalf("live status condition was lost by stale status patch: %#v", got.Status.Conditions)
+	}
+}
+
 func TestSanitizePatchForLog(t *testing.T) {
 	secret := &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "s", Namespace: "ns"}}
 	configMap := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "cm", Namespace: "ns"}}