use a queue to avoid overwriting events

chetan-rns · chetan-rns · commit 91002a9de991 · 2025-11-25T13:05:00.000+05:30
Assisted-by: Cursor
Signed-off-by: Chetan Banavikalmutt &lt;chetanrns1997@gmail.com&gt;
diff --git a/internal/event/event.go b/internal/event/event.go
diff --git a/internal/event/event_test.go b/internal/event/event_test.go
diff --git a/internal/kube/client.go b/internal/kube/client.go
@@ -122,6 +122,13 @@ func IsRetryableError(err error) bool {
 	if err == nil {
 		return false
 	}
+
+	// Errors about immutable fields (especially deletionTimestamp) are NOT retryable
+	// Once a resource is being deleted, these fields cannot be changed no matter how many times we retry
+	if isDeletionImmutableError(err) {
+		return false
+	}
+
 	return kerrors.IsInternalError(err) ||
 		kerrors.IsInvalid(err) ||
 		kerrors.IsTooManyRequests(err) ||
@@ -137,6 +144,14 @@ func IsRetryableError(err error) bool {
 		errors.Is(err, syscall.ECONNRESET)
 }
 
+func isDeletionImmutableError(err error) bool {
+	// Detect errors about immutable deletionTimestamp or deletionGracePeriodSeconds
+	// These occur when trying to update a resource that's being deleted
+	errMsg := err.Error()
+	return (strings.Contains(errMsg, "deletionTimestamp") || strings.Contains(errMsg, "deletionGracePeriodSeconds")) &&
+		strings.Contains(errMsg, "field is immutable")
+}
+
 func isHTTP2GoawayErr(err error) bool {
 	return strings.Contains(err.Error(), "http2: server sent GOAWAY and closed the connection")
 }
diff --git a/internal/manager/application/application.go b/internal/manager/application/application.go
@@ -205,18 +205,7 @@ func (m *ApplicationManager) UpdateManagedApp(ctx context.Context, incoming *v1a
 		existing.Labels = incoming.Labels
 		existing.Finalizers = incoming.Finalizers
 		existing.Spec = *incoming.Spec.DeepCopy()
-		// Don't update Operation if the app is terminating
-		if incoming.Operation != nil {
-			allowMirrorOp := true
-			if existing.Status.OperationState != nil {
-				if existing.Status.OperationState.Phase == synccommon.OperationTerminating {
-					allowMirrorOp = false
-				}
-			}
-			if allowMirrorOp {
-				existing.Operation = incoming.Operation.DeepCopy()
-			}
-		}
+		existing.Operation = operationToUse(existing, incoming)
 
 		if incoming.DeletionTimestamp != nil && existing.DeletionTimestamp == nil {
 			deletionTimestampChanged = true
@@ -242,31 +231,14 @@ func (m *ApplicationManager) UpdateManagedApp(ctx context.Context, incoming *v1a
 			deletionTimestampChanged = true
 		}
 
-		var desiredOp *v1alpha1.Operation
-		if incoming.Operation != nil {
-			allowMirrorOp := true
-			if existing.Status.OperationState != nil {
-				if existing.Status.OperationState.Phase == synccommon.OperationTerminating {
-					allowMirrorOp = false
-				}
-			}
-			if allowMirrorOp {
-				desiredOp = incoming.Operation
-			} else {
-				desiredOp = existing.Operation
-			}
-		} else {
-			desiredOp = existing.Operation
-		}
-
 		target := &v1alpha1.Application{
 			ObjectMeta: v1.ObjectMeta{
 				Annotations: incoming.Annotations,
 				Labels:      incoming.Labels,
 				Finalizers:  incoming.Finalizers,
 			},
 			Spec:      incoming.Spec,
-			Operation: desiredOp,
+			Operation: operationToUse(existing, incoming),
 		}
 		source := &v1alpha1.Application{
 			ObjectMeta: v1.ObjectMeta{
@@ -363,8 +335,12 @@ func (m *ApplicationManager) UpdateAutonomousApp(ctx context.Context, namespace
 
 		existing.Annotations = incoming.Annotations
 		existing.Labels = incoming.Labels
-		existing.DeletionTimestamp = incoming.DeletionTimestamp
-		existing.DeletionGracePeriodSeconds = incoming.DeletionGracePeriodSeconds
+		if existing.DeletionTimestamp == nil {
+			existing.DeletionTimestamp = incoming.DeletionTimestamp
+		}
+		if existing.DeletionGracePeriodSeconds == nil {
+			existing.DeletionGracePeriodSeconds = incoming.DeletionGracePeriodSeconds
+		}
 		existing.Finalizers = incoming.Finalizers
 		existing.Spec = incoming.Spec
 		existing.Status = *incoming.Status.DeepCopy()
@@ -380,16 +356,18 @@ func (m *ApplicationManager) UpdateAutonomousApp(ctx context.Context, namespace
 
 		target := &v1alpha1.Application{
 			ObjectMeta: v1.ObjectMeta{
-				Labels:                     incoming.Labels,
-				Annotations:                incoming.Annotations,
-				DeletionTimestamp:          incoming.DeletionTimestamp,
-				DeletionGracePeriodSeconds: incoming.DeletionGracePeriodSeconds,
-				Finalizers:                 incoming.Finalizers,
+				Labels:      incoming.Labels,
+				Annotations: incoming.Annotations,
+				Finalizers:  incoming.Finalizers,
 			},
 			Spec:      incoming.Spec,
 			Status:    incoming.Status,
 			Operation: incoming.Operation,
 		}
+		if existing.DeletionTimestamp == nil && incoming.DeletionTimestamp != nil {
+			target.DeletionTimestamp = incoming.DeletionTimestamp
+			target.DeletionGracePeriodSeconds = incoming.DeletionGracePeriodSeconds
+		}
 		source := &v1alpha1.Application{
 			ObjectMeta: v1.ObjectMeta{
 				DeletionTimestamp:          existing.DeletionTimestamp,
@@ -511,6 +489,7 @@ func (m *ApplicationManager) UpdateOperation(ctx context.Context, incoming *v1al
 	updated, err = m.update(ctx, false, incoming, func(existing, incoming *v1alpha1.Application) {
 		existing.Annotations = incoming.Annotations
 		existing.Labels = incoming.Labels
+		existing.Operation = operationToUse(existing, incoming)
 		existing.Status = *incoming.Status.DeepCopy()
 	}, func(existing, incoming *v1alpha1.Application) (jsondiff.Patch, error) {
 		annotations := make(map[string]string)
@@ -523,7 +502,7 @@ func (m *ApplicationManager) UpdateOperation(ctx context.Context, incoming *v1al
 			ObjectMeta: v1.ObjectMeta{
 				Annotations: incoming.Annotations,
 			},
-			Operation: incoming.Operation,
+			Operation: operationToUse(existing, incoming),
 		}
 		source := &v1alpha1.Application{
 			ObjectMeta: v1.ObjectMeta{
@@ -543,6 +522,22 @@ func (m *ApplicationManager) UpdateOperation(ctx context.Context, incoming *v1al
 	return updated, err
 }
 
+func operationToUse(existing, incoming *v1alpha1.Application) *v1alpha1.Operation {
+	// Preserve existing operation if the incoming operation is nil
+	if incoming.Operation == nil {
+		return existing.Operation
+	}
+
+	// Don't update Operation if the app is terminating
+	if existing.Status.OperationState != nil {
+		if existing.Status.OperationState.Phase == synccommon.OperationTerminating {
+			return existing.Operation
+		}
+	}
+
+	return incoming.Operation.DeepCopy()
+}
+
 // TerminateOperation aborts a running sync operation by setting .status.operationState.phase to Terminating.
 func (m *ApplicationManager) TerminateOperation(ctx context.Context, incoming *v1alpha1.Application) (*v1alpha1.Application, error) {
 	incoming.SetNamespace(m.namespace)
diff --git a/principal/callbacks.go b/principal/callbacks.go
@@ -15,8 +15,6 @@
 package principal
 
 import (
-	"reflect"
-
 	"github.com/argoproj-labs/argocd-agent/internal/event"
 	"github.com/argoproj-labs/argocd-agent/internal/manager"
 	"github.com/argoproj-labs/argocd-agent/internal/manager/appproject"
@@ -117,12 +115,12 @@ func (s *Server) updateAppCallback(old *v1alpha1.Application, new *v1alpha1.Appl
 	if isTerminateOperation(old, new) {
 		ev = s.events.ApplicationEvent(event.TerminateOperation, new)
 	} else {
-		// For managed agents: prevent sending operation back on regular spec updates.
+		// Prevent sending operation back on regular spec updates.
 		// Allow only nil->non-nil transitions to carry operation (i.e. principal-initiated sync).
 
 		// DeepCopy to avoid mutating the informer object
 		out := new.DeepCopy()
-		if reflect.DeepEqual(old.Operation, new.Operation) && !isResourceFromAutonomousAgent(new) {
+		if !(old.Operation == nil && new.Operation != nil) {
 			out.Operation = nil
 		}
 
diff --git a/principal/callbacks_test.go b/principal/callbacks_test.go
@@ -1367,7 +1367,7 @@ func TestServer_updateAppCallback(t *testing.T) {
 		assert.Equal(t, 1, sendQ.Len())
 	})
 
-	t.Run("include operation in event if it is different from the old one", func(t *testing.T) {
+	t.Run("include operation in event if it is initiated for the first time", func(t *testing.T) {
 		mockBackend := &mocks.Application{}
 
 		appManager, err := application.NewApplicationManager(mockBackend, "argocd")
@@ -1386,11 +1386,6 @@ func TestServer_updateAppCallback(t *testing.T) {
 
 		oldApp := &v1alpha1.Application{
 			ObjectMeta: metav1.ObjectMeta{Name: "test-app", Namespace: "managed-agent", ResourceVersion: "1"},
-			Operation: &v1alpha1.Operation{
-				Sync: &v1alpha1.SyncOperation{
-					Revision: "main",
-				},
-			},
 		}
 		newApp := &v1alpha1.Application{
 			ObjectMeta: metav1.ObjectMeta{Name: "test-app", Namespace: "managed-agent", ResourceVersion: "2"},
diff --git a/principal/event.go b/principal/event.go
@@ -187,8 +187,19 @@ func (s *Server) processApplicationEvent(ctx context.Context, agentName string,
 				return fmt.Errorf("could not create application %s: %w", incoming.QualifiedName(), err)
 			}
 
+			// Check if the existing application is marked for deletion before attempting update
+			existing, err := s.appManager.Get(ctx, incoming.Name, agentName)
+			if err != nil {
+				logCtx.WithError(err).Errorf("Application should exist before attempting to update")
+				return nil
+			}
+			if existing.DeletionTimestamp != nil {
+				logCtx.Trace("Application is marked for deletion, skipping update to avoid immutable field error")
+				return nil
+			}
+
 			// Update the application if it already exists
-			_, err := s.appManager.UpdateAutonomousApp(ctx, agentName, incoming)
+			_, err = s.appManager.UpdateAutonomousApp(ctx, agentName, incoming)
 			if err != nil {
 				return fmt.Errorf("could not update application spec for %s: %w", incoming.QualifiedName(), err)
 			}
@@ -200,9 +211,20 @@ func (s *Server) processApplicationEvent(ctx context.Context, agentName string,
 			return event.NewEventNotAllowedErr("event type not allowed when mode is not autonomous")
 		}
 
+		// Check if the existing application is marked for deletion before attempting update
+		existing, err := s.appManager.Get(ctx, incoming.Name, agentName)
+		if err != nil {
+			logCtx.WithError(err).Errorf("Application should exist before attempting to update")
+			return nil
+		}
+		if existing.DeletionTimestamp != nil {
+			logCtx.Trace("Application is marked for deletion, skipping update to avoid immutable field error")
+			return nil
+		}
+
 		s.sourceCache.Application.Set(incoming.UID, incoming.Spec)
 
-		_, err := s.appManager.UpdateAutonomousApp(ctx, agentName, incoming)
+		_, err = s.appManager.UpdateAutonomousApp(ctx, agentName, incoming)
 		if err != nil {
 			return fmt.Errorf("could not update application spec for %s: %w", incoming.QualifiedName(), err)
 		}
diff --git a/test/e2e/sync_test.go b/test/e2e/sync_test.go
@@ -340,15 +340,15 @@ func (suite *SyncTestSuite) Test_TerminateOperationManaged() {
 	requires.Eventually(func() bool {
 		app := argoapp.Application{}
 		err := suite.ManagedAgentClient.Get(suite.Ctx, agentKey, &app, metav1.GetOptions{})
-		return err == nil && app.Operation != nil && app.Status.OperationState != nil &&
+		return err == nil && app.Status.OperationState != nil &&
 			app.Status.OperationState.Phase == synccommon.OperationRunning
 	}, 60*time.Second, 1*time.Second)
 
 	// Wait for the sync operation to start on the principal
 	requires.Eventually(func() bool {
 		app := argoapp.Application{}
 		err := suite.PrincipalClient.Get(suite.Ctx, principalKey, &app, metav1.GetOptions{})
-		return err == nil && app.Operation != nil && app.Status.OperationState != nil &&
+		return err == nil && app.Status.OperationState != nil &&
 			app.Status.OperationState.Phase == synccommon.OperationRunning
 	}, 60*time.Second, 1*time.Second)
 
@@ -401,7 +401,7 @@ func (suite *SyncTestSuite) Test_TerminateOperationManaged() {
 	requires.Eventually(func() bool {
 		app := argoapp.Application{}
 		err := suite.ManagedAgentClient.Get(suite.Ctx, agentKey, &app, metav1.GetOptions{})
-		if err == nil && app.Operation == nil && app.Status.OperationState != nil {
+		if err == nil && app.Status.OperationState != nil {
 			suite.T().Logf("Operation phase: %v", app.Status.OperationState.Phase)
 		}
 		return err == nil && app.Status.OperationState != nil && app.Status.OperationState.Phase == synccommon.OperationFailed &&
@@ -457,17 +457,17 @@ func (suite *SyncTestSuite) Test_TerminateOperationAutonomous() {
 	requires.Eventually(func() bool {
 		app := argoapp.Application{}
 		err := suite.PrincipalClient.Get(suite.Ctx, principalKey, &app, metav1.GetOptions{})
-		return err == nil && app.Operation != nil && app.Status.OperationState != nil &&
+		return err == nil && app.Status.OperationState != nil &&
 			app.Status.OperationState.Phase == synccommon.OperationRunning
-	}, 60*time.Second, 1*time.Second)
+	}, 90*time.Second, 1*time.Second)
 
 	// Wait for the sync operation to start on the autonomous-agent as well
 	requires.Eventually(func() bool {
 		app := argoapp.Application{}
 		err := suite.AutonomousAgentClient.Get(suite.Ctx, agentKey, &app, metav1.GetOptions{})
-		return err == nil && app.Operation != nil && app.Status.OperationState != nil &&
+		return err == nil && app.Status.OperationState != nil &&
 			app.Status.OperationState.Phase == synccommon.OperationRunning
-	}, 60*time.Second, 1*time.Second)
+	}, 90*time.Second, 1*time.Second)
 
 	// Get the Argo server endpoint to use
 	argoEndpoint, err := fixture.GetArgoCDServerEndpoint(suite.PrincipalClient)
@@ -517,7 +517,7 @@ func (suite *SyncTestSuite) Test_TerminateOperationAutonomous() {
 	requires.Eventually(func() bool {
 		app := argoapp.Application{}
 		err := suite.AutonomousAgentClient.Get(suite.Ctx, agentKey, &app, metav1.GetOptions{})
-		return err == nil && app.Operation == nil && app.Status.OperationState != nil &&
+		return err == nil && app.Status.OperationState != nil &&
 			app.Status.OperationState.Phase == synccommon.OperationFailed &&
 			app.Status.OperationState.Message == "Operation terminated"
 	}, 90*time.Second, 1*time.Second)
