queue item fix

Signed-off-by: Per Goncalves da Silva <[email protected]>

Author: Per Goncalves da Silva

pkg/controller/registry/resolver/source_csvs.go

@@ -93,19 +93,19 @@ func (s *csvSource) Snapshot(ctx context.Context) (*cache.Snapshot, error) {
 			continue
 		}
 
-		//if cachedSubscription, ok := csvSubscriptions[csv]; !ok || cachedSubscription == nil {
-		//	// we might be in an incoherent state, so let's check with live clients to make sure
-		//	realSubscriptions, err := s.listSubscriptions(ctx)
-		//	if err != nil {
-		//		return nil, fmt.Errorf("failed to list subscriptions: %w", err)
-		//	}
-		//	for _, realSubscription := range realSubscriptions.Items {
-		//		if realSubscription.Status.InstalledCSV == csv.Name {
-		//			// oops, live cluster state is coherent
-		//			return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s - found owning Subscription %s/%s", csv.Namespace, csv.Name, realSubscription.Namespace, realSubscription.Name)
-		//		}
-		//	}
-		//}
+		if cachedSubscription, ok := csvSubscriptions[csv]; !ok || cachedSubscription == nil {
+			// we might be in an incoherent state, so let's check with live clients to make sure
+			realSubscriptions, err := s.listSubscriptions(ctx)
+			if err != nil {
+				return nil, fmt.Errorf("failed to list subscriptions: %w", err)
+			}
+			for _, realSubscription := range realSubscriptions.Items {
+				if realSubscription.Status.InstalledCSV == csv.Name {
+					// oops, live cluster state is coherent
+					return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s - found owning Subscription %s/%s", csv.Namespace, csv.Name, realSubscription.Namespace, realSubscription.Name)
+				}
+			}
+		}
 
 		if failForwardEnabled {
 			replacementChainEndsInFailure, err := isReplacementChainThatEndsInFailure(csv, ReplacementMapping(csvs))

pkg/lib/queueinformer/queueinformer_operator.go

@@ -282,9 +282,19 @@ func (o *operator) processNextWorkItem(ctx context.Context, loop *QueueInformer)
 			queue.Forget(item)
 			return true
 		}
-
+		
 		logger = logger.WithField("cache-key", key)
 
+		if !ok {
+			// we should really only (sync) resource event items
+			// otherwise we run the risk of processing two the same objects concurrently.
+			// As is, at least, the case for the namespace resolution informer,
+			// the namespace name gets added to the queue (e.g. nsResolveQueue.Add(obj.GetNamespace())),
+			// but the QueueInformer's Enqueue method always adds ResourceEvents.
+			queue.Add(kubestate.NewResourceEvent(kubestate.ResourceUpdated, key))
+			return true
+		}
+
 		var resource interface{}
 		if loop.indexer == nil {
 			resource = event.Resource()
@@ -305,11 +315,7 @@ func (o *operator) processNextWorkItem(ctx context.Context, loop *QueueInformer)
 			}
 		}
 
-		if !ok {
-			event = kubestate.NewResourceEvent(kubestate.ResourceUpdated, resource)
-		} else {
-			event = kubestate.NewResourceEvent(event.Type(), resource)
-		}
+		event = kubestate.NewResourceEvent(event.Type(), resource)
 	}
 
 	// Sync and requeue on error (throw out failed deletion syncs)