Repository: karmada-io/karmada

Issue:
 The resource spec contents of pp and rb are not synchronized in time.
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!
-->

**What happened**:
1. During controller restart, the .spec.suspension fields of rb and pp are different

https://www.processon.com/v/6772523638625f07c0ec6273
![karmada 灰度发布流程](https://github.com/user-attachments/assets/87047c81-eed0-476a-a8e6-3e929a906b26)

During the restart, the resource detector controller processes the currently released workload later than the binding controller, causing the pause setting to become invalid; our custom controller extension (for grayscale release) does not update the op resource, and the number of new pods more than partition.

2. After  pp sets .spec.preserveResourcesOnDeletion=true, execute the deletion operation immediately, and the resources of the member clusters may be deleted in cascade

3. When updating the Dispatching field of pp to false, and then updating deployment, the Dispatching field of rb is first updated to false, and then updated to true again.

![Image](https://github.com/user-attachments/assets/1c9526e7-278a-4444-a139-1caa92ef8291)

![Image](https://github.com/user-attachments/assets/35e67544-f301-40b1-8173-48e375ab1b3d)
Add the following log in ApplyPolicy, and the cached data will be correct and meet expectations, but when rb is updated at the end, the order will be disrupted

```
	klog.Infof("Applying policy(%s/%s) for object: %s", policy.Namespace, policy.Name, objectKey)
	if policy.Spec.Suspension != nil && policy.Spec.Suspension.Dispatching != nil {
		klog.Infof("Applying policy(%s/%s) with dispatching suspension: %v", policy.Namespace, policy.Name, *policy.Spec.Suspension.Dispatching)
	} else {
		klog.Infof("Applying policy(%s/%s) without dispatching suspension", policy.Namespace, policy.Name)
	}
```

```
	switch operationResult {
	case controllerutil.OperationResultCreated:
		klog.Infof("Create ResourceBinding(%s/%s) successfully.", binding.GetNamespace(), binding.GetName())
	case controllerutil.OperationResultUpdated:
		if bindingCopy.Spec.Suspension != nil && bindingCopy.Spec.Suspension.Dispatching != nil {
			klog.Infof("Update ResourceBinding(%s/%s) successfully. dispatching: %v", binding.GetNamespace(), binding.GetName(), *bindingCopy.Spec.Suspension.Dispatching)
		} else {
			klog.Infof("Update ResourceBinding(%s/%s) successfully.", binding.GetNamespace(), binding.GetName())
		}
	default:
		klog.V(2).Infof("ResourceBinding(%s/%s) is up to date.", binding.GetNamespace(), binding.GetName())
	}
```
![Image](https://github.com/user-attachments/assets/6d3a1c61-bb83-48d7-bd63-643e81001e28)

![Image](https://github.com/user-attachments/assets/2e821b6f-3134-431e-b7a6-0dc00695f93b)
Add the following log after the CreateOrUpdate rb function to show that the original execution order changes after this function.

```
	if policy.Spec.Suspension != nil && policy.Spec.Suspension.Dispatching != nil {
		klog.Infof("CreateOrUpdate (%s/%s) with dispatching suspension: %v, cost: %v", policy.Namespace, policy.Name, *policy.Spec.Suspension.Dispatching, time.Since(start3))
	} else {
		klog.Infof("CreateOrUpdate (%s/%s) without dispatching suspension", policy.Namespace, policy.Name)
	}
```

![Image](https://github.com/user-attachments/assets/da28cedb-4336-445f-9ea9-1db0318ae4bc)


Add log in function `CreateOrUpdate`, Indicates that an optimistic lock conflict occurs when updating the rb object, and a retry occurs, causing the update to be delayed.
```
	startTime := time.Now()
	if NewErr := c.Update(ctx, obj); NewErr != nil {
		klog.Infof("Failed to update (%s/%s) used time: %v, err: %v", obj.GetNamespace(), obj.GetName(), time.Since(startTime), NewErr)
		return controllerutil.OperationResultNone, NewErr
	}
```

![Image](https://github.com/user-attachments/assets/89b1d3ca-a44f-4421-a6fd-920d81bfa2d4)


Q:  Why dispatching=true is triggered 3 times?

Deployment is released, it will be updated once, remove label `propagationpolicy.karmada.io/permanent-id` and Annonation 
`propagationpolicy.karmada.io/namespace`, `propagationpolicy.karmada.io/name`, after finding pp through `LookForMatchedPolicy` function, it will pass through `ClaimPolicyForObject` function, this will trigger twice, pp will be updated once more, a total of 3 times

Q: how the workers consume the worker queue?
https://github.com/karmada-io/karmada/blob/c5b2d3c65a7357f37ab6fc7d331b4ad53e887e43/pkg/util/fedinformer/keys/keys.go#L114-L121

![Image](https://github.com/user-attachments/assets/eb58aa4d-a0ee-435c-aaa6-e6dce8d7be5e)

set struct `ClusterWideKeyWithConfig`  field `ResourceChangeByKarmada=false`
![Image](https://github.com/user-attachments/assets/d2d03c29-b6f5-4e61-a65e-2b7dd95652a2)

Add log for Reconcile funciton:
```
rand.Seed(time.Now().UnixNano())
	randStr := randomString(20)
	startTime := time.Now()
	defer func() {
		klog.Infof("Finished Reconcile, (%s/%s), cost: %v, randStr: %s", clusterWideKeyWithConfig.ClusterWideKey.Namespace, clusterWideKeyWithConfig.ClusterWideKey.Name, time.Since(startTime), randStr)
	}()

	clusterWideKey := clusterWideKeyWithConfig.ClusterWideKey
	resourceChangeByKarmada := clusterWideKeyWithConfig.ResourceChangeByKarmada
	klog.Infof("Reconciling object: %s, resourceChangeByKarmada: %v, randStr: %s", clusterWideKey, resourceChangeByKarmada, randStr)
```

![Image](https://github.com/user-attachments/assets/ce88c094-b5be-46cf-867e-172863f65050)


![Image](https://github.com/user-attachments/assets/a54c04de-32fe-4ef3-a24f-ab776acbed7d)

Because the Resource Change By Karmada field has changed, different workers are triggered to execute the Reconcile function at the same time.

**What you expected to happen**:

The relevant fields in pp, rb, and work can be consistent. If the fields in rb and work are inconsistent with pp, then the subsequent operations related to rb and work do not need to be performed until the synchronization of the relevant fields is completed

**How to reproduce it (as minimally and precisely as possible)**:
1. have lost of resouce template And pp,  restart controller and set  .spec.suspension.dispatching=true, check rb field .spec.suspension.dispatching.
2. set pp field .spec.preserveResourcesOnDeletion=true, and deletion operation immediately, check whether the resource has been cascade deleted  in member cluster.

**Anything else we need to know?**:

**Environment**:
- Karmada version:
v1.12
- kubectl-karmada or karmadactl version (the result of `kubectl-karmada version` or `karmadactl version`):
- Others:


Discussion:
@CharlesQQ: I have a solution for reference only

When PP changes, the hash value of the current PP is calculated and added to the annotation By webhook; when the resoure detector updates RB, the hash value is also synchronized; before reconcile, the binding controller first detects whether the hash values ​​of rb and pp are consistent; if they are not same, then Return to wait; the path from rb to work is similar

![解决pp 和rb 字段存在gap 的问题](https://github.com/user-attachments/assets/e9c294bb-07c8-4bdf-ae59-a8dcb10722ff)

Reference for calculation method of hash value from openkruise.
https://github.com/openkruise/kruise/blob/e3e6d471a75737606e8cfe5338ad92bdddc72699/pkg/webhook/sidecarset/mutating/sidecarset_create_update_handler.go#L48-L66


Solution patch:
diff --git a/pkg/detector/detector.go b/pkg/detector/detector.go
index d4cefcce5..2586a4790 100644
--- a/pkg/detector/detector.go
+++ b/pkg/detector/detector.go
@@ -337,13 +337,26 @@ func (d *ResourceDetector) OnUpdate(oldObj, newObj interface{}) {
 		return
 	}
 
-	resourceChangeByKarmada := eventfilter.ResourceChangeByKarmada(unstructuredOldObj, unstructuredNewObj)
+	isLazyActivation, err := d.isClaimedByLazyPolicy(unstructuredNewObj)
+	if err != nil {
+		// should never come here
+		klog.Errorf("Failed to check if the object (kind=%s, %s/%s) is bound by lazy policy. err: %v", unstructuredNewObj.GetKind(), unstructuredNewObj.GetNamespace(), unstructuredNewObj.GetName(), err)
+	}
 
-	resourceItem := ResourceItem{
-		Obj:                     newRuntimeObj,
-		ResourceChangeByKarmada: resourceChangeByKarmada,
+	if isLazyActivation {
+		resourceItem := ResourceItem{
+			Obj:                     newRuntimeObj,
+			ResourceChangeByKarmada: eventfilter.ResourceChangeByKarmada(unstructuredOldObj, unstructuredNewObj),
+		}
+
+		d.Processor.Enqueue(resourceItem)
+		return
 	}
 
+	// For non-lazy policies, it is no need to distinguish whether the change is from Karmada or not.
+	resourceItem := ResourceItem{
+		Obj: newRuntimeObj,
+	}
 	d.Processor.Enqueue(resourceItem)
 }
 
@@ -1263,7 +1276,7 @@ func (d *ResourceDetector) HandlePropagationPolicyCreationOrUpdate(policy *polic
 		if err != nil {
 			return err
 		}
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: resourceKey, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(resourceKey, policy.Spec.ActivationPreference)
 	}
 
 	// check whether there are matched RT in waiting list, is so, add it to processor
@@ -1281,7 +1294,7 @@ func (d *ResourceDetector) HandlePropagationPolicyCreationOrUpdate(policy *polic
 
 	for _, key := range matchedKeys {
 		d.RemoveWaiting(key)
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: key, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(key, policy.Spec.ActivationPreference)
 	}
 
 	// If preemption is enabled, handle the preemption process.
@@ -1330,14 +1343,14 @@ func (d *ResourceDetector) HandleClusterPropagationPolicyCreationOrUpdate(policy
 		if err != nil {
 			return err
 		}
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: resourceKey, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(resourceKey, policy.Spec.ActivationPreference)
 	}
 	for _, crb := range clusterResourceBindings.Items {
 		resourceKey, err := helper.ConstructClusterWideKey(crb.Spec.Resource)
 		if err != nil {
 			return err
 		}
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: resourceKey, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(resourceKey, policy.Spec.ActivationPreference)
 	}
 
 	matchedKeys := d.GetMatching(policy.Spec.ResourceSelectors)
@@ -1354,7 +1367,7 @@ func (d *ResourceDetector) HandleClusterPropagationPolicyCreationOrUpdate(policy
 
 	for _, key := range matchedKeys {
 		d.RemoveWaiting(key)
-		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: key, ResourceChangeByKarmada: true})
+		d.enqueueResourceTemplateForPolicyChange(key, policy.Spec.ActivationPreference)
 	}
 
 	// If preemption is enabled, handle the preemption process.
@@ -1497,3 +1510,21 @@ func (d *ResourceDetector) CleanupClusterResourceBindingClaimMetadata(crbName st
 		return updateErr
 	})
 }
+
+// enqueueResourceTemplateForPolicyChange enqueues a resource template key for reconciliation in response to a
+// PropagationPolicy or ClusterPropagationPolicy change. If the policy's ActivationPreference is set to Lazy,
+// the ResourceChangeByKarmada flag is set to true, indicating that the resource template is being enqueued
+// due to a policy change and should not be propagated to member clusters. For non-lazy policies, this flag
+// is omitted as the distinction is unnecessary.
+//
+// Note: Setting ResourceChangeByKarmada changes the effective queue key. Mixing both true/false for the same
+// resource may result in two different queue keys being processed concurrently, which can cause race conditions.
+// Therefore, only set ResourceChangeByKarmada in lazy activation mode.
+// For more details, see: https://github.com/karmada-io/karmada/issues/5996.
+func (d *ResourceDetector) enqueueResourceTemplateForPolicyChange(key keys.ClusterWideKey, pref policyv1alpha1.ActivationPreference) {
+	if util.IsLazyActivationEnabled(pref) {
+		d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: key, ResourceChangeByKarmada: true})
+		return
+	}
+	d.Processor.Add(keys.ClusterWideKeyWithConfig{ClusterWideKey: key})
+}
diff --git a/pkg/detector/policy.go b/pkg/detector/policy.go
index 71c084be3..1c317c3cc 100644
--- a/pkg/detector/policy.go
+++ b/pkg/detector/policy.go
@@ -373,6 +373,51 @@ func (d *ResourceDetector) listCPPDerivedCRBs(policyID, policyName string) (*wor
 	return bindings, nil
 }
 
+func (d *ResourceDetector) isClaimedByLazyPolicy(obj *unstructured.Unstructured) (bool, error) {
+	policyAnnotations := obj.GetAnnotations()
+	policyLabels := obj.GetLabels()
+	policyNamespace := util.GetAnnotationValue(policyAnnotations, policyv1alpha1.PropagationPolicyNamespaceAnnotation)
+	policyName := util.GetAnnotationValue(policyAnnotations, policyv1alpha1.PropagationPolicyNameAnnotation)
+	claimedID := util.GetLabelValue(policyLabels, policyv1alpha1.PropagationPolicyPermanentIDLabel)
+	if policyNamespace != "" && policyName != "" && claimedID != "" {
+		policyObject, err := d.propagationPolicyLister.ByNamespace(policyNamespace).Get(policyName)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				return false, nil
+			}
+
+			return false, err
+		}
+		matchedPropagationPolicy := &policyv1alpha1.PropagationPolicy{}
+		if err = helper.ConvertToTypedObject(policyObject, matchedPropagationPolicy); err != nil {
+			return false, err
+		}
+
+		return util.IsLazyActivationEnabled(matchedPropagationPolicy.Spec.ActivationPreference), nil
+	}
+
+	policyName = util.GetAnnotationValue(policyAnnotations, policyv1alpha1.ClusterPropagationPolicyAnnotation)
+	claimedID = util.GetLabelValue(policyLabels, policyv1alpha1.ClusterPropagationPolicyPermanentIDLabel)
+	if policyName != "" && claimedID != "" {
+		policyObject, err := d.clusterPropagationPolicyLister.Get(policyName)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				return false, nil
+			}
+
+			return false, err
+		}
+		matchedClusterPropagationPolicy := &policyv1alpha1.ClusterPropagationPolicy{}
+		if err = helper.ConvertToTypedObject(policyObject, matchedClusterPropagationPolicy); err != nil {
+			return false, err
+		}
+
+		return util.IsLazyActivationEnabled(matchedClusterPropagationPolicy.Spec.ActivationPreference), nil
+	}
+
+	return false, nil
+}
+
 // excludeClusterPolicy excludes cluster propagation policy.
 // If propagation policy was claimed, cluster propagation policy should not exist.
 func excludeClusterPolicy(obj metav1.Object) (hasClaimedClusterPolicy bool) {



Test patch:
diff --git a/pkg/detector/detector_test.go b/pkg/detector/detector_test.go
index e1c79ff4d..ea025b514 100644
--- a/pkg/detector/detector_test.go
+++ b/pkg/detector/detector_test.go
@@ -431,7 +431,6 @@ func TestOnUpdate(t *testing.T) {
 		oldObj                    interface{}
 		newObj                    interface{}
 		expectedEnqueue           bool
-		expectedChangeByKarmada   bool
 		expectToUnstructuredError bool
 	}{
 		{
@@ -462,8 +461,7 @@ func TestOnUpdate(t *testing.T) {
 					},
 				},
 			},
-			expectedEnqueue:         true,
-			expectedChangeByKarmada: false,
+			expectedEnqueue: true,
 		},
 		{
 			name: "update without changes",
@@ -526,8 +524,7 @@ func TestOnUpdate(t *testing.T) {
 					},
 				},
 			},
-			expectedEnqueue:         true,
-			expectedChangeByKarmada: true,
+			expectedEnqueue: true,
 		},
 		{
 			name: "core v1 object",
@@ -575,7 +572,6 @@ func TestOnUpdate(t *testing.T) {
 				assert.IsType(t, ResourceItem{}, mockProcessor.lastEnqueued, "Enqueued item should be of type ResourceItem")
 				enqueued := mockProcessor.lastEnqueued.(ResourceItem)
 				assert.Equal(t, tt.newObj, enqueued.Obj, "Enqueued object should match the new object")
-				assert.Equal(t, tt.expectedChangeByKarmada, enqueued.ResourceChangeByKarmada, "ResourceChangeByKarmada flag should match expected value")
 			} else {
 				assert.Equal(t, 0, mockProcessor.enqueueCount, "Object should not be enqueued")
 			}
@@ -973,6 +969,71 @@ func TestApplyClusterPolicy(t *testing.T) {
 	}
 }
 
+func TestEnqueueResourceKeyWithActivationPref(t *testing.T) {
+	testClusterWideKey := keys.ClusterWideKey{
+		Group:     "foo",
+		Version:   "foo",
+		Kind:      "foo",
+		Namespace: "foo",
+		Name:      "foo",
+	}
+	tests := []struct {
+		name string
+		key  keys.ClusterWideKey
+		pref policyv1alpha1.ActivationPreference
+		want keys.ClusterWideKeyWithConfig
+	}{
+		{
+			name: "lazy pp and resourceChangeByKarmada is true",
+			key:  testClusterWideKey,
+			pref: policyv1alpha1.LazyActivation,
+			want: keys.ClusterWideKeyWithConfig{
+				ClusterWideKey:          testClusterWideKey,
+				ResourceChangeByKarmada: true,
+			},
+		},
+		{
+			name: "non-lazy ignores ResourceChangeByKarmada",
+			key:  testClusterWideKey,
+			pref: "",
+			want: keys.ClusterWideKeyWithConfig{
+				ClusterWideKey:          testClusterWideKey,
+				ResourceChangeByKarmada: false,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			detector := ResourceDetector{
+				Processor: util.NewAsyncWorker(util.Options{
+					Name:    "resource detector",
+					KeyFunc: ResourceItemKeyFunc,
+					ReconcileFunc: func(key util.QueueKey) (err error) {
+						defer cancel()
+						defer func() {
+							assert.NoError(t, err)
+						}()
+						clusterWideKeyWithConfig, ok := key.(keys.ClusterWideKeyWithConfig)
+						if !ok {
+							err = fmt.Errorf("invalid key")
+							return err
+						}
+						if clusterWideKeyWithConfig != tt.want {
+							err = fmt.Errorf("unexpected key. want:%+v, got:%+v", tt.want, clusterWideKeyWithConfig)
+							return err
+						}
+						return nil
+					},
+				}),
+			}
+			detector.Processor.Run(1, ctx.Done())
+			detector.enqueueResourceTemplateForPolicyChange(tt.key, tt.pref)
+			<-ctx.Done()
+		})
+	}
+}
+
 // Helper Functions
 
 // setupTestScheme creates a runtime scheme with necessary types for testing