Use local resource version management in agent client

This commit introduces local resource version management to prevent race
conditions in the CloudEvents agent client, particularly in the manifestwork
Patch operations.

Key changes:
- Add local versioner in AgentInformerWatcherStore to track resource versions
  independently from source versions
- Clear resource version when publishing status updates to source, allowing
  source to manage its own version tracking
- Refactor Patch method to handle status updates and deletions consistently,
  extracting version comparison logic into dedicated function
- Support empty resource version strings throughout the codebase (codec, agent
  client, resync operations) to enable separate local and remote versioning
- Add SetResourceVersion to ResourceObject interface for version management
- Add comprehensive tests for new functionality including race condition tests

The local versioner increments on each Add/Update operation and is cleared on
Delete, providing monotonic version tracking for conflict detection without
depending on source resource versions.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Jian Qiu <jqiu@redhat.com>

Author: qiujian16

pkg/cloudevents/clients/work/agent/client/manifestwork.go

@@ -3,12 +3,12 @@ package client
 import (
 	"context"
 	"fmt"
+	"k8s.io/apimachinery/pkg/api/meta"
 	"net/http"
 	"strconv"
 	"sync"
 
 	"k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kubetypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
@@ -41,7 +41,7 @@ type ManifestWorkAgentClient struct {
 var _ workv1client.ManifestWorkInterface = &ManifestWorkAgentClient{}
 
 func NewManifestWorkAgentClient(
-	clusterName string,
+	_ string,
 	watcherStore store.ClientWatcherStore[*workv1.ManifestWork],
 	cloudEventsClient generic.CloudEventsClient[*workv1.ManifestWork],
 ) *ManifestWorkAgentClient {
@@ -126,22 +126,39 @@ func (c *ManifestWorkAgentClient) Watch(ctx context.Context, opts metav1.ListOpt
 
 func (c *ManifestWorkAgentClient) Patch(ctx context.Context, name string, pt kubetypes.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (result *workv1.ManifestWork, err error) {
 	klog.V(4).Infof("patching manifestwork %s/%s", c.namespace, name)
+
+	// avoid race conditions among the agent's go routines
+	c.Lock()
+	defer c.Unlock()
+
+	var returnErr *errors.StatusError
+	defer func() {
+		if returnErr != nil {
+			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		} else {
+			metrics.IncreaseWorkProcessedCounter("patch", metav1.StatusSuccess)
+		}
+	}()
+
+	if len(subresources) != 0 && !utils.IsStatusPatch(subresources) {
+		msg := fmt.Sprintf("unsupported subresources %v", subresources)
+		returnErr = errors.NewGenericServerResponse(http.StatusMethodNotAllowed, "patch", common.ManifestWorkGR, name, msg, 0, false)
+		return nil, returnErr
+	}
+
 	lastWork, exists, err := c.watcherStore.Get(c.namespace, name)
 	if err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
 	if !exists {
-		returnErr := errors.NewNotFound(common.ManifestWorkGR, name)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewNotFound(common.ManifestWorkGR, name)
 		return nil, returnErr
 	}
 
 	patchedWork, err := utils.Patch(pt, lastWork, data)
 	if err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
 
@@ -155,93 +172,43 @@ func (c *ManifestWorkAgentClient) Patch(ctx context.Context, name string, pt kub
 	eventType := types.CloudEventsType{
 		CloudEventsDataType: *eventDataType,
 		SubResource:         types.SubResourceStatus,
+		Action:              types.UpdateRequestAction,
+	}
+
+	// we first compare local resource version
+	if returnErr = versoinCompare(patchedWork, lastWork); returnErr != nil {
+		return nil, returnErr
 	}
 
 	newWork := patchedWork.DeepCopy()
 
-	if utils.IsStatusPatch(subresources) {
-		// avoid race conditions among the agent's go routines
-		c.Lock()
-		defer c.Unlock()
+	isDeleted := !newWork.DeletionTimestamp.IsZero() && len(newWork.Finalizers) == 0
 
-		eventType.Action = types.UpdateRequestAction
-		// publish the status update event to source, source will check the resource version
-		// and reject the update if it's status update is outdated.
-		if err := c.cloudEventsClient.Publish(ctx, eventType, newWork); err != nil {
-			returnErr := cloudeventserrors.ToStatusError(common.ManifestWorkGR, name, err)
-			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
-			return nil, returnErr
+	if utils.IsStatusPatch(subresources) || isDeleted {
+		if isDeleted {
+			meta.SetStatusCondition(&newWork.Status.Conditions, metav1.Condition{
+				Type:    common.ResourceDeleted,
+				Status:  metav1.ConditionTrue,
+				Reason:  "ManifestsDeleted",
+				Message: fmt.Sprintf("The manifests are deleted from the cluster %s", newWork.Namespace),
+			})
 		}
 
-		// Fetch the latest work from the store and verify the resource version to avoid updating the store
-		// with outdated work. Return a conflict error if the resource version is outdated.
-		// Due to the lack of read-modify-write guarantees in the store, race conditions may occur between
-		// this update operation and one from the agent informer after receiving the event from the source.
-		latestWork, exists, err := c.watcherStore.Get(c.namespace, name)
-		if err != nil {
-			returnErr := errors.NewInternalError(err)
-			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
-			return nil, returnErr
-		}
-		if !exists {
-			returnErr := errors.NewNotFound(common.ManifestWorkGR, name)
-			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
-			return nil, returnErr
-		}
-		lastResourceVersion, err := strconv.ParseInt(latestWork.GetResourceVersion(), 10, 64)
-		if err != nil {
-			returnErr := errors.NewInternalError(err)
-			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
-			return nil, returnErr
-		}
-		newResourceVersion, err := strconv.ParseInt(newWork.GetResourceVersion(), 10, 64)
-		if err != nil {
-			returnErr := errors.NewInternalError(err)
-			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
-			return nil, returnErr
-		}
-		// ensure the resource version of the work is not outdated
-		if newResourceVersion < lastResourceVersion {
-			// It's safe to return a conflict error here, even if the status update event
-			// has already been sent. The source may reject the update due to an outdated resource version.
-			returnErr := errors.NewConflict(common.ManifestWorkGR, name, fmt.Errorf("the resource version of the work is outdated"))
-			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
-			return nil, returnErr
-		}
-		if err := c.watcherStore.Update(newWork); err != nil {
-			returnErr := errors.NewInternalError(err)
-			metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		// Set work's resource version to remote resource version for publishing
+		workToPublish := newWork.DeepCopy()
+		workToPublish.ResourceVersion = ""
+
+		// publish the status update event to source, source will check the resource version
+		// and reject the update if it's status update is outdated.
+		if err := c.cloudEventsClient.Publish(ctx, eventType, workToPublish); err != nil {
+			returnErr = cloudeventserrors.ToStatusError(common.ManifestWorkGR, name, err)
 			return nil, returnErr
 		}
-
-		metrics.IncreaseWorkProcessedCounter("patch", metav1.StatusSuccess)
-		return newWork, nil
-	}
-
-	if len(subresources) != 0 {
-		msg := fmt.Sprintf("unsupported subresources %v", subresources)
-		returnErr := errors.NewGenericServerResponse(http.StatusMethodNotAllowed, "patch", common.ManifestWorkGR, name, msg, 0, false)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
-		return nil, returnErr
 	}
 
 	// the finalizers of a deleting manifestwork are removed, marking the manifestwork status to deleted and sending
 	// it back to source
-	if !newWork.DeletionTimestamp.IsZero() && len(newWork.Finalizers) == 0 {
-		meta.SetStatusCondition(&newWork.Status.Conditions, metav1.Condition{
-			Type:    common.ResourceDeleted,
-			Status:  metav1.ConditionTrue,
-			Reason:  "ManifestsDeleted",
-			Message: fmt.Sprintf("The manifests are deleted from the cluster %s", newWork.Namespace),
-		})
-
-		eventType.Action = types.UpdateRequestAction
-		if err := c.cloudEventsClient.Publish(ctx, eventType, newWork); err != nil {
-			returnErr := cloudeventserrors.ToStatusError(common.ManifestWorkGR, name, err)
-			metrics.IncreaseWorkProcessedCounter("delete", string(returnErr.ErrStatus.Reason))
-			return nil, returnErr
-		}
-
+	if isDeleted {
 		if err := c.watcherStore.Delete(newWork); err != nil {
 			returnErr := errors.NewInternalError(err)
 			metrics.IncreaseWorkProcessedCounter("delete", string(returnErr.ErrStatus.Reason))
@@ -252,12 +219,49 @@ func (c *ManifestWorkAgentClient) Patch(ctx context.Context, name string, pt kub
 		return newWork, nil
 	}
 
+	// Fetch the latest work from the store and verify the resource version to avoid updating the store
+	// with outdated work. Return a conflict error if the resource version is outdated.
+	// Due to the lack of read-modify-write guarantees in the store, race conditions may occur between
+	// this update operation and one from the agent informer after receiving the event from the source.
+	latestWork, exists, err := c.watcherStore.Get(c.namespace, name)
+	if err != nil {
+		returnErr = errors.NewInternalError(err)
+		return nil, returnErr
+	}
+	if !exists {
+		returnErr = errors.NewNotFound(common.ManifestWorkGR, name)
+		return nil, returnErr
+	}
+	if returnErr = versoinCompare(patchedWork, latestWork); returnErr != nil {
+		return nil, returnErr
+	}
 	if err := c.watcherStore.Update(newWork); err != nil {
-		returnErr := errors.NewInternalError(err)
-		metrics.IncreaseWorkProcessedCounter("patch", string(returnErr.ErrStatus.Reason))
+		returnErr = errors.NewInternalError(err)
 		return nil, returnErr
 	}
-
-	metrics.IncreaseWorkProcessedCounter("patch", metav1.StatusSuccess)
 	return newWork, nil
 }
+
+func versoinCompare(new, old *workv1.ManifestWork) *errors.StatusError {
+	// If resource version is empty or 0, skip the comparison (e.g., when publishing status updates)
+	if new.GetResourceVersion() == "" || new.GetResourceVersion() == "0" {
+		return nil
+	}
+
+	lastResourceVersion, err := strconv.ParseInt(old.GetResourceVersion(), 10, 64)
+	if err != nil {
+		return errors.NewInternalError(err)
+	}
+	newResourceVersion, err := strconv.ParseInt(new.GetResourceVersion(), 10, 64)
+	if err != nil {
+		return errors.NewInternalError(err)
+	}
+	// ensure the resource version of the work is not outdated
+	if newResourceVersion < lastResourceVersion {
+		// It's safe to return a conflict error here, even if the status update event
+		// has already been sent. The source may reject the update due to an outdated resource version.
+		return errors.NewConflict(common.ManifestWorkGR, new.Name, fmt.Errorf(
+			"the resource version of the work is outdated, new %d, old %d", newResourceVersion, lastResourceVersion))
+	}
+	return nil
+}