fix: invalidate cluster cache after sync operations

pkg/cache/cluster.go

@@ -127,6 +127,8 @@ type ClusterCache interface {
 	GetGVKParser() *managedfields.GvkParser
 	// Invalidate cache and executes callback that optionally might update cache settings
 	Invalidate(opts ...UpdateSettingsFunc)
+	// InvalidateResources invalidates specific resources in the cache
+	InvalidateResources(keys []kube.ResourceKey)
 	// FindResources returns resources that matches given list of predicates from specified namespace or everywhere if specified namespace is empty
 	FindResources(namespace string, predicates ...func(r *Resource) bool) map[kube.ResourceKey]*Resource
 	// IterateHierarchyV2 iterates resource tree starting from the specified top level resources and executes callback for each resource in the tree.
@@ -490,6 +492,29 @@ func (c *clusterCache) Invalidate(opts ...UpdateSettingsFunc) {
 	c.log.Info("Invalidated cluster")
 }
 
+// InvalidateResources invalidates specific resources in the cache by removing them.
+// This forces the resources to be refreshed from the cluster on the next access.
+func (c *clusterCache) InvalidateResources(keys []kube.ResourceKey) {
+	if len(keys) == 0 {
+		return
+	}
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	for _, key := range keys {
+		if _, exists := c.resources[key]; exists {
+			// Remove the resource from cache - this will force it to be refreshed on next access
+			c.onNodeRemoved(key)
+			c.log.Info("Invalidated resource from cache", "key", key.String())
+		} else {
+			c.log.Info("Resource not found in cache for invalidation", "key", key.String())
+		}
+	}
+
+	c.log.Info("Invalidated specific resources from cache", "count", len(keys))
+}
+
 // clusterCacheSync's lock should be held before calling this method
 func (syncStatus *clusterCacheSync) synced(clusterRetryTimeout time.Duration) bool {
 	syncTime := syncStatus.syncTime

pkg/engine/engine.go

@@ -74,6 +74,13 @@ func (e *gitOpsEngine) Sync(ctx context.Context,
 	namespace string,
 	opts ...sync.SyncOpt,
 ) ([]common.ResourceSyncResult, error) {
+	// Ensure cache is synced before getting managed live objects
+	// This forces a refresh if the cache was invalidated
+	err := e.cache.EnsureSynced()
+	if err != nil {
+		return nil, fmt.Errorf("error during sync: failed to ensure cache is synced: %w", err)
+	}
+
 	managedResources, err := e.cache.GetManagedLiveObjs(resources, isManaged)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get managed live objects: %w", err)
@@ -84,6 +91,15 @@ func (e *gitOpsEngine) Sync(ctx context.Context,
 		return nil, fmt.Errorf("failed to diff objects: %w", err)
 	}
 	opts = append(opts, sync.WithSkipHooks(!diffRes.Modified))
+
+	// Add cache invalidation callback to invalidate cache for modified resources after sync
+	opts = append(opts, sync.WithCacheInvalidationCallback(func(modifiedResources []kube.ResourceKey) {
+		// Only invalidate the specific resources that were modified
+		if len(modifiedResources) > 0 {
+			e.cache.InvalidateResources(modifiedResources)
+		}
+	}))
+
 	syncCtx, cleanup, err := sync.NewSyncContext(revision, result, e.config, e.config, e.kubectl, namespace, e.cache.GetOpenAPISchema(), opts...)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create sync context: %w", err)

pkg/sync/sync_context.go

@@ -218,6 +218,14 @@ func WithClientSideApplyMigration(enabled bool, manager string) SyncOpt {
 	}
 }
 
+// WithCacheInvalidationCallback sets a callback that will be invoked after successful sync operations
+// to invalidate the cache
+func WithCacheInvalidationCallback(callback func([]kubeutil.ResourceKey)) SyncOpt {
+	return func(ctx *syncContext) {
+		ctx.cacheInvalidationCallback = callback
+	}
+}
+
 // NewSyncContext creates new instance of a SyncContext
 func NewSyncContext(
 	revision string,
@@ -389,6 +397,10 @@ type syncContext struct {
 	applyOutOfSyncOnly bool
 	// stores whether the resource is modified or not
 	modificationResult map[kubeutil.ResourceKey]bool
+
+	// cacheInvalidationCallback is a callback that will be invoked after successful sync operations
+	// to invalidate the cache
+	cacheInvalidationCallback func([]kubeutil.ResourceKey)
 }
 
 func (sc *syncContext) setRunningPhase(tasks []*syncTask, isPendingDeletion bool) {
@@ -557,6 +569,15 @@ func (sc *syncContext) Sync() {
 		// delete all completed hooks which have appropriate delete policy
 		sc.deleteHooks(hooksPendingDeletionSuccessful)
 		sc.setOperationPhase(common.OperationSucceeded, "successfully synced (no more tasks)")
+
+		// Invalidate cache after successful sync
+		if sc.cacheInvalidationCallback != nil {
+			modifiedResources := make([]kubeutil.ResourceKey, 0, len(sc.syncRes))
+			for _, result := range sc.syncRes {
+				modifiedResources = append(modifiedResources, result.ResourceKey)
+			}
+			sc.cacheInvalidationCallback(modifiedResources)
+		}
 		return
 	}
 
@@ -593,11 +614,34 @@ func (sc *syncContext) Sync() {
 		syncFailedTasks, _ := tasks.Split(func(t *syncTask) bool { return t.syncStatus == common.ResultCodeSyncFailed })
 		sc.deleteHooks(hooksPendingDeletionFailed)
 		sc.setOperationFailed(syncFailTasks, syncFailedTasks, "one or more objects failed to apply")
+
+		// Invalidate cache for successfully synced resources even if overall operation failed
+		if sc.cacheInvalidationCallback != nil {
+			var modifiedResources []kubeutil.ResourceKey
+			for _, result := range sc.syncRes {
+				// Only invalidate resources that were successfully synced
+				if result.Status == common.ResultCodeSynced {
+					modifiedResources = append(modifiedResources, result.ResourceKey)
+				}
+			}
+			if len(modifiedResources) > 0 {
+				sc.cacheInvalidationCallback(modifiedResources)
+			}
+		}
 	case successful:
 		if remainingTasks.Len() == 0 {
 			// delete all completed hooks which have appropriate delete policy
 			sc.deleteHooks(hooksPendingDeletionSuccessful)
 			sc.setOperationPhase(common.OperationSucceeded, "successfully synced (all tasks run)")
+
+			// Invalidate cache after successful sync
+			if sc.cacheInvalidationCallback != nil {
+				modifiedResources := make([]kubeutil.ResourceKey, 0, len(sc.syncRes))
+				for _, result := range sc.syncRes {
+					modifiedResources = append(modifiedResources, result.ResourceKey)
+				}
+				sc.cacheInvalidationCallback(modifiedResources)
+			}
 		} else {
 			sc.setRunningPhase(remainingTasks, false)
 		}