wip

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java

@@ -141,7 +141,7 @@ private void handleOnAddOrUpdate(
       ResourceAction action, T oldCustomResource, T newCustomResource) {
     var handling =
         temporaryResourceCache.onAddOrUpdateEvent(action, newCustomResource, oldCustomResource);
-    if (handling == EventHandling.NEW || handling == EventHandling.IN_BETWEEN) {
+    if (handling == EventHandling.NEW || handling == EventHandling.INTERMEDIATE) {
       handleEvent(action, newCustomResource, oldCustomResource, null);
     } else if (log.isDebugEnabled()) {
       log.debug("{} event propagation for action: {}", handling, action);

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java

@@ -154,7 +154,7 @@ private synchronized void onAddOrUpdate(ResourceAction action, R newObject, R ol
 
     var eventHandling = temporaryResourceCache.onAddOrUpdateEvent(action, newObject, oldObject);
 
-    if (eventHandling != EventHandling.NEW && eventHandling != EventHandling.IN_BETWEEN) {
+    if (eventHandling != EventHandling.NEW && eventHandling != EventHandling.INTERMEDIATE) {
       log.debug(
           "{} event propagation", eventHandling == EventHandling.DEFER ? "Deferring" : "Skipping");
     } else if (eventAcceptedByFilter(action, newObject, oldObject)) {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCache.java

@@ -65,7 +65,7 @@ public class TemporaryResourceCache<T extends HasMetadata> {
   public enum EventHandling {
     DEFER,
     OBSOLETE,
-    IN_BETWEEN,
+    INTERMEDIATE,
     NEW
   }
 
@@ -149,16 +149,16 @@ private synchronized EventHandling onEvent(
         // in this case we received and event that might be in some edge case that was
         // already used in reconciler or after that, but before our updated resource version.
         // That would be hard to distinguish, so for those we are propagating the event further.
-        log.debug("Received in between event.");
-        result = EventHandling.IN_BETWEEN;
+        log.debug("Received intermediate event.");
+        result = EventHandling.INTERMEDIATE;
       }
     }
     var au = activeUpdates.get(resourceId);
     if (au != null) {
-      if (result == EventHandling.IN_BETWEEN) {
+      if (result == EventHandling.INTERMEDIATE) {
         return au.isOwnResourceVersions(resource.getMetadata().getResourceVersion())
             ? EventHandling.DEFER
-            : EventHandling.IN_BETWEEN;
+            : EventHandling.INTERMEDIATE;
       }
       if (result == EventHandling.NEW) {
         log.debug("Setting last event for id: {} delete: {}", resourceId, delete);

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java

@@ -149,6 +149,15 @@ void processEventPropagationWithIncorrectAnnotation() {
     verify(eventHandlerMock, times(1)).handleEvent(any());
   }
 
+  @Test
+  void propagatesIntermediateEventHandling() {
+    when(temporaryResourceCache.onAddOrUpdateEvent(any(), any(), any()))
+        .thenReturn(EventHandling.INTERMEDIATE);
+    informerEventSource.onUpdate(testDeployment(), testDeployment());
+
+    verify(eventHandlerMock, times(1)).handleEvent(any());
+  }
+
   @Test
   void propagateEventAndRemoveResourceFromTempCacheIfResourceVersionMismatch() {
     withRealTemporaryResourceCache();
@@ -439,6 +448,56 @@ void filteringUpdateAndGhostCheckWithNamespaceChange() {
     assertThat(temporaryResourceCache.getResourceFromCache(resourceId)).isEmpty();
   }
 
+  @Test
+  void propagatesIntermediateEventForExternalUpdateDuringFiltering() {
+    // Causal-dependency fix: another controller updated the resource between our read
+    // and our write. The informer delivers that update during our active filter; since
+    // its resource version is NOT one of our own writes, it must be propagated.
+    var realCache = realCacheWithWatchedNamespace();
+    var resourceId = ResourceID.fromResource(testDeployment());
+
+    realCache.startEventFilteringModify(resourceId);
+    realCache.putResource(deploymentWithResourceVersion(4));
+
+    informerEventSource.onUpdate(
+        deploymentWithResourceVersion(2), deploymentWithResourceVersion(3));
+
+    verify(eventHandlerMock, times(1)).handleEvent(any());
+
+    realCache.doneEventFilterModify(resourceId, "4");
+  }
+
+  @Test
+  void doesNotPropagateIntermediateEventForOurOwnIntermediateUpdate() {
+    // Two consecutive own writes within a single filter window: the older one's event
+    // arrives after the newer one has been cached. Because the version is recorded as
+    // our own, the event must be deferred (not propagated).
+    var realCache = realCacheWithWatchedNamespace();
+    var resourceId = ResourceID.fromResource(testDeployment());
+
+    realCache.startEventFilteringModify(resourceId);
+    realCache.putResource(deploymentWithResourceVersion(3));
+    realCache.putResource(deploymentWithResourceVersion(4));
+
+    informerEventSource.onUpdate(
+        deploymentWithResourceVersion(2), deploymentWithResourceVersion(3));
+
+    verify(eventHandlerMock, never()).handleEvent(any());
+
+    realCache.doneEventFilterModify(resourceId, "4");
+  }
+
+  private TemporaryResourceCache<Deployment> realCacheWithWatchedNamespace() {
+    var mes = mock(ManagedInformerEventSource.class);
+    var mim = mock(InformerManager.class);
+    when(mes.manager()).thenReturn(mim);
+    when(mim.isWatchingNamespace(any())).thenReturn(true);
+    when(mim.lastSyncResourceVersion(any())).thenReturn("1");
+    temporaryResourceCache = spy(new TemporaryResourceCache<>(true, mes));
+    informerEventSource.setTemporalResourceCache(temporaryResourceCache);
+    return temporaryResourceCache;
+  }
+
   private void assertNoEventProduced() {
     await()
         .pollDelay(Duration.ofMillis(50))

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/TemporaryResourceCacheTest.java

@@ -294,6 +294,68 @@ void putAfterEventWithEventFilteringWithPost() {
     assertTrue(postEvent.isPresent());
   }
 
+  @Test
+  void intermediateEventPropagatedWhenNoActiveUpdate() {
+    // Cache holds a newer version from a prior own write; no active filter is in progress.
+    // An older event arriving used to be OBSOLETE; now it must be propagated as INTERMEDIATE
+    // so callers can react to changes that happened between read and write.
+    var olderEvent = testResource();
+    var newer = testResource();
+    newer.getMetadata().setResourceVersion("3");
+
+    temporaryResourceCache.putResource(newer);
+    assertThat(temporaryResourceCache.getResourceFromCache(ResourceID.fromResource(olderEvent)))
+        .isPresent();
+
+    var result =
+        temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, olderEvent, null);
+
+    assertThat(result).isEqualTo(EventHandling.INTERMEDIATE);
+  }
+
+  @Test
+  void intermediateEventPropagatedWhenNotOurOwnUpdate() {
+    // Causal-dependency scenario: a third party updated the resource between our read and
+    // our write. Its version arrives as an event but is NOT in our own resource versions,
+    // so it must be propagated (INTERMEDIATE), not deferred.
+    var external = testResource(); // rv=2 — written by another controller
+    var resourceId = ResourceID.fromResource(external);
+
+    temporaryResourceCache.startEventFilteringModify(resourceId);
+
+    var ourUpdate = testResource();
+    ourUpdate.getMetadata().setResourceVersion("3");
+    temporaryResourceCache.putResource(ourUpdate);
+
+    var result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, external, null);
+
+    assertThat(result).isEqualTo(EventHandling.INTERMEDIATE);
+  }
+
+  @Test
+  void intermediateEventDeferredWhenItIsOurOwnIntermediateUpdate() {
+    // Two consecutive own writes within the same filter window: the older one's event
+    // arrives after the newer one is cached. Because the version is recorded as our own,
+    // the event must be DEFERred rather than propagated.
+    var testResource = testResource();
+    var resourceId = ResourceID.fromResource(testResource);
+
+    temporaryResourceCache.startEventFilteringModify(resourceId);
+
+    var ourFirst = testResource(); // rv=2
+    temporaryResourceCache.putResource(ourFirst);
+
+    var ourSecond = testResource();
+    ourSecond.getMetadata().setResourceVersion("3");
+
+    temporaryResourceCache.startEventFilteringModify(resourceId);
+    temporaryResourceCache.putResource(ourSecond);
+
+    var result = temporaryResourceCache.onAddOrUpdateEvent(ResourceAction.UPDATED, ourFirst, null);
+
+    assertThat(result).isEqualTo(EventHandling.DEFER);
+  }
+
   @Test
   void rapidDeletion() {
     var testResource = testResource();