fix: retain recently written external resources missing from a stale update

An update of the whole resource set of a primary (a poll result or a received
event) might have been created before the reconciler wrote a resource, thus not
containing it yet. Since such updates are handled as the full actual state, the
write was lost from the cache, and the next reconciliation created a duplicate
of an already created resource or repeated an already executed update.

Writes are now marked as unconfirmed and retained for the next update if it
either does not contain the resource at all - the expected case for a create -
or still contains the state that the write replaced. Any other state is treated
as a change made outside of the reconciler and accepted as actual. Marks are
dropped on the first update, so a resource really deleted or changed meanwhile
is not retained indefinitely.

Also guards handleRecentResourceUpdate against a missing cache entry, and
resolves the actual resources from the state resources in the external state
bulk dependent integration test, which is the recommended approach for
resources that take longer to become visible.

Author: csviri

docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md

@@ -480,6 +480,17 @@ also be created, one per dependent resource.
 See [integration test](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent)
 as a sample.
 
+Note that an external resource and the state resource referencing it cannot be created atomically:
+the external resource has to be created first, since its identifier is what gets stored in the
+state. If the resources are fetched based on the state - which is usually the case, since the
+identifier is only known from the state - a poll happening in between the two steps cannot see the
+new external resource yet. JOSDK keeps such a recently created resource in the cache for the next
+update to avoid creating a duplicate of it, but for a resource that takes longer to become visible,
+it is recommended to resolve the actual resources from the state resources in
+`BulkDependentResource.getSecondaryResources`, as done in the integration test above. The state
+resources are managed by an `InformerEventSource`, thus are always up-to-date regarding the
+operator's own changes.
+
 ## GenericKubernetesResource based Dependent Resources
 
 In rare circumstances resource handling where there is no class representation or just typeless handling might be

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

@@ -67,6 +67,28 @@ public abstract class ExternalResourceCachingEventSource<R, P extends HasMetadat
 
   protected Map<ResourceID, Map<ID, R>> cache = new ConcurrentHashMap<>();
 
+  /**
+   * The resources written by the reconciler ({@link #handleRecentResourceCreate(ResourceID,
+   * Object)} and {@link #handleRecentResourceUpdate(ResourceID, Object, Object)}) that were not
+   * seen yet in a subsequent update of the whole resource set of a primary. Such an update might
+   * have been created (polled or received) before the resource was actually written, thus not
+   * containing the new state yet. Since these updates are handled as the full actual state, the
+   * write would be lost from the cache; the next reconciliation would then create a duplicate of an
+   * already created resource, or repeat an already executed update. Note that a mark is dropped on
+   * the first update, so a resource really deleted or changed in the meantime is not retained
+   * indefinitely.
+   *
+   * @see #retainUnconfirmedWrites(ResourceID, Map)
+   */
+  private final Map<ResourceID, Map<ID, RecentWrite<R>>> unconfirmedWrites =
+      new ConcurrentHashMap<>();
+
+  /**
+   * A resource written by the reconciler and the state it replaced, which is {@code null} in case
+   * the resource was created.
+   */
+  private record RecentWrite<R>(R written, R replaced) {}
+
   protected ExternalResourceCachingEventSource(
       Class<R> resourceClass, ResourceIDMapper<R, ID> resourceIDMapper) {
     this(null, resourceClass, resourceIDMapper);
@@ -86,6 +108,7 @@ protected ExternalResourceCachingEventSource(
   }
 
   protected synchronized void handleDelete(ResourceID primaryID) {
+    unconfirmedWrites.remove(primaryID);
     var res = cache.remove(primaryID);
     if (res != null && deleteAcceptedByFilter(res.values())) {
       getEventHandler().handleEvent(new Event(primaryID));
@@ -105,6 +128,13 @@ protected synchronized void handleDelete(ResourceID primaryID, Set<ID> resourceI
     if (!isRunning()) {
       return;
     }
+    var unconfirmed = unconfirmedWrites.get(primaryID);
+    if (unconfirmed != null) {
+      unconfirmed.keySet().removeAll(resourceIDs);
+      if (unconfirmed.isEmpty()) {
+        unconfirmedWrites.remove(primaryID);
+      }
+    }
     var cachedValues = cache.get(primaryID);
     List<R> removedResources =
         cachedValues == null
@@ -131,7 +161,16 @@ protected synchronized void handleResources(ResourceID primaryID, Set<R> newReso
 
   protected synchronized void handleResources(Map<ResourceID, Set<R>> allNewResources) {
     var toDelete = cache.keySet().stream().filter(k -> !allNewResources.containsKey(k)).toList();
-    toDelete.forEach(this::handleDelete);
+    toDelete.forEach(
+        primaryID -> {
+          if (unconfirmedWrites.containsKey(primaryID)) {
+            // handled as an empty update, so that a recently written resource, that this update
+            // could not see yet, is not removed from the cache
+            handleResources(primaryID, Collections.emptySet());
+          } else {
+            handleDelete(primaryID);
+          }
+        });
     allNewResources.forEach(this::handleResources);
   }
 
@@ -148,6 +187,7 @@ protected synchronized void handleResources(
     }
     var newResourcesMap =
         newResources.stream().collect(Collectors.toMap(resourceIDMapper::idFor, r -> r));
+    retainUnconfirmedWrites(primaryID, newResourcesMap);
     cache.put(primaryID, newResourcesMap);
     if (propagateEvent
         && !newResourcesMap.equals(cachedResources)
@@ -156,6 +196,34 @@ && acceptedByFiler(cachedResources, newResourcesMap)) {
     }
   }
 
+  /**
+   * Keeps the resources written since the received update was created, thus missing from it. An
+   * update is considered stale for a written resource if it does not contain it at all - which is
+   * the expected case for a create - or if it still contains the state that the write replaced. Any
+   * other state is a change that happened outside of the reconciler, so it is accepted as the
+   * actual state.
+   *
+   * @see #unconfirmedWrites
+   */
+  private void retainUnconfirmedWrites(ResourceID primaryID, Map<ID, R> newResourcesMap) {
+    var unconfirmed = unconfirmedWrites.remove(primaryID);
+    if (unconfirmed == null) {
+      return;
+    }
+    unconfirmed.forEach(
+        (id, write) -> {
+          var newResource = newResourcesMap.get(id);
+          if (newResource == null || newResource.equals(write.replaced())) {
+            log.debug(
+                "Retaining recently written resource missing from the update. Primary ID: {},"
+                    + " resource ID: {}",
+                primaryID,
+                id);
+            newResourcesMap.put(id, write.written());
+          }
+        });
+  }
+
   private boolean acceptedByFiler(Map<ID, R> cachedResourceMap, Map<ID, R> newResourcesMap) {
 
     var addedResources = new HashMap<>(newResourcesMap);
@@ -217,6 +285,7 @@ public synchronized void handleRecentResourceCreate(ResourceID primaryID, R reso
     } else {
       actualValues.computeIfAbsent(resourceId, r -> resource);
     }
+    markUnconfirmedWrite(primaryID, resourceId, new RecentWrite<>(resource, null));
   }
 
   @Override
@@ -226,12 +295,18 @@ public synchronized void handleRecentResourceUpdate(
     if (actualValues != null) {
       var resourceId = resourceIDMapper.idFor(resource);
       R actualResource = actualValues.get(resourceId);
-      if (actualResource.equals(previousVersionOfResource)) {
+      if (actualResource != null && actualResource.equals(previousVersionOfResource)) {
         actualValues.put(resourceId, resource);
+        markUnconfirmedWrite(
+            primaryID, resourceId, new RecentWrite<>(resource, previousVersionOfResource));
       }
     }
   }
 
+  private void markUnconfirmedWrite(ResourceID primaryID, ID resourceId, RecentWrite<R> write) {
+    unconfirmedWrites.computeIfAbsent(primaryID, id -> new HashMap<>()).put(resourceId, write);
+  }
+
   @Override
   public Set<R> getSecondaryResources(P primary) {
     return getSecondaryResources(ResourceID.fromResource(primary));

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

@@ -15,6 +15,7 @@
  */
 package io.javaoperatorsdk.operator.processing.event.source;
 
+import java.util.Map;
 import java.util.Set;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -211,6 +212,107 @@ void genericFilteringEvents() {
     verify(eventHandler, times(0)).handleEvent(any());
   }
 
+  @Test
+  void retainsRecentlyCreatedResourceMissingFromUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceCreate(primaryID1(), testResource2());
+
+    // the update was created before the resource, thus does not contain it yet
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1()))
+        .containsExactlyInAnyOrder(testResource1(), testResource2());
+    // no event for the retained resource, only the initial add event
+    verify(eventHandler, times(1)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void retainsRecentlyCreatedResourceOnlyForASingleUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceCreate(primaryID1(), testResource2());
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    // this update is created after the resource, so it is really deleted meanwhile
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1());
+    verify(eventHandler, times(2)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void doesNotRetainRecentlyCreatedResourceDeletedBeforeTheUpdate() {
+    source.handleRecentResourceCreate(primaryID1(), testResource2());
+    source.handleDelete(primaryID1(), testResource2());
+
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1());
+  }
+
+  @Test
+  void retainsRecentlyCreatedResourceMissingFromWholeCacheUpdate() {
+    source.handleRecentResourceCreate(primaryID1(), testResource1());
+
+    source.handleResources(Map.of());
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1());
+
+    source.handleResources(Map.of());
+
+    assertThat(source.getSecondaryResources(primaryID1())).isEmpty();
+  }
+
+  @Test
+  void retainsRecentlyUpdatedResourceMissingFromUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+
+    // the update was created before the resource was updated, thus still contains the old state
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(changedTestResource1());
+    // no event for the retained resource, only the initial add event
+    verify(eventHandler, times(1)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void retainsRecentlyUpdatedResourceOnlyForASingleUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    // this update is created after the resource was updated, so it was really changed meanwhile
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(testResource1());
+    verify(eventHandler, times(2)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void doesNotRetainRecentlyUpdatedResourceChangedOutsideOfTheReconciler() {
+    var externallyChanged = testResource1().setValue("externallyChangedValue");
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+
+    source.handleResources(primaryID1(), Set.of(externallyChanged));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(externallyChanged);
+  }
+
+  @Test
+  void retainsRecentlyUpdatedResourceInWholeCacheUpdate() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+
+    source.handleResources(Map.of(primaryID1(), Set.of(testResource1())));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(changedTestResource1());
+  }
+
+  private static SampleExternalResource changedTestResource1() {
+    return testResource1().setValue("changedValue");
+  }
+
   public static class TestExternalCachingEventSource
       extends ExternalResourceCachingEventSource<SampleExternalResource, HasMetadata, String> {
     public TestExternalCachingEventSource() {

operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/externalstate/externalstatebulkdependent/BulkDependentResourceExternalWithState.java

@@ -135,12 +135,21 @@ public Map<String, ExternalResource> desiredResources(
     return res;
   }
 
+  /**
+   * Resolves the actual resources from the persisted state instead of the polled cache. An external
+   * resource and the state referencing it cannot be created atomically, so a poll happening in
+   * between replaces the cached resources with the ones it can already see, dropping the freshly
+   * created one. The next reconciliation would then create a duplicate external resource that no
+   * state references anymore, thus is leaked. The state itself is read-after-write consistent,
+   * since it is managed through an {@link
+   * io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource}.
+   */
   @Override
   public Map<String, ExternalResource> getSecondaryResources(
       ExternalStateBulkDependentCustomResource primary,
       Context<ExternalStateBulkDependentCustomResource> context) {
-    var resources = context.getSecondaryResources(ExternalResource.class);
-    return resources.stream().collect(Collectors.toMap(this::externalResourceIndex, r -> r));
+    return fetchResources(primary).stream()
+        .collect(Collectors.toMap(this::externalResourceIndex, r -> r));
   }
 
   @Override