fix: retain recently written external resources missing from a stale update (#3565)

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 a state that a write replaced. Every state replaced since the
last update is kept, since the reconciler might write the same resource multiple
times in between, and an update created before any of those writes is stale. 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,30 @@ 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<>();
+
+  /**
+   * The last state of a resource written by the reconciler and every state it replaced since the
+   * last update. There can be multiple replaced states if the reconciler wrote the resource more
+   * than once without an update in between; an update created before any of those writes is stale.
+   * The set is empty if the resource was created, since there was no previous state then.
+   */
+  private record RecentWrite<R>(R written, Set<R> replaced) {}
+
   protected ExternalResourceCachingEventSource(
       Class<R> resourceClass, ResourceIDMapper<R, ID> resourceIDMapper) {
     this(null, resourceClass, resourceIDMapper);
@@ -86,6 +110,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 +130,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 +163,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 +189,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 +198,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 a state that a 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 || write.replaced().contains(newResource)) {
+            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);
@@ -229,6 +299,7 @@ public synchronized void handleRecentResourceCreate(ResourceID primaryID, R reso
     } else {
       actualValues.computeIfAbsent(resourceId, r -> resource);
     }
+    markUnconfirmedWrite(primaryID, resourceId, resource, null);
   }
 
   @Override
@@ -240,10 +311,34 @@ public synchronized void handleRecentResourceUpdate(
       R actualResource = actualValues.get(resourceId);
       if (actualResource != null && actualResource.equals(previousVersionOfResource)) {
         actualValues.put(resourceId, resource);
+        markUnconfirmedWrite(primaryID, resourceId, resource, previousVersionOfResource);
       }
     }
   }
 
+  /**
+   * Marks the written resource as not confirmed yet by an update, keeping the states replaced by
+   * previous writes of the same resource. Without those, an update created before an earlier write
+   * would not be recognized as stale, and the last write would be lost from the cache.
+   *
+   * @param replaced the state the write replaced, {@code null} if the resource was created
+   * @see #unconfirmedWrites
+   */
+  private void markUnconfirmedWrite(ResourceID primaryID, ID resourceId, R written, R replaced) {
+    unconfirmedWrites
+        .computeIfAbsent(primaryID, id -> new HashMap<>())
+        .compute(
+            resourceId,
+            (id, previousWrite) -> {
+              Set<R> replacedStates =
+                  previousWrite == null ? new HashSet<>() : new HashSet<>(previousWrite.replaced());
+              if (replaced != null) {
+                replacedStates.add(replaced);
+              }
+              return new RecentWrite<>(written, replacedStates);
+            });
+  }
+
   @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,167 @@ 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());
+  }
+
+  @Test
+  void retainsResourceUpdatedTwiceIfUpdateContainsTheStateBeforeBothWrites() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+    source.handleRecentResourceUpdate(
+        primaryID1(), changedTwiceTestResource1(), changedTestResource1());
+
+    // the update was created before both writes, thus contains the state before the first one
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1()))
+        .containsExactly(changedTwiceTestResource1());
+    // no event for the retained resource, only the initial add event
+    verify(eventHandler, times(1)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void retainsResourceUpdatedTwiceIfUpdateContainsTheStateBetweenTheWrites() {
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+    source.handleRecentResourceUpdate(
+        primaryID1(), changedTwiceTestResource1(), changedTestResource1());
+
+    // the update was created between the two writes, thus contains the intermediate state
+    source.handleResources(primaryID1(), Set.of(changedTestResource1()));
+
+    assertThat(source.getSecondaryResources(primaryID1()))
+        .containsExactly(changedTwiceTestResource1());
+    verify(eventHandler, times(1)).handleEvent(new Event(primaryID1()));
+  }
+
+  @Test
+  void retainsRecentlyCreatedAndThenUpdatedResourceMissingFromUpdate() {
+    source.handleRecentResourceCreate(primaryID1(), testResource1());
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+
+    // the update was created before both writes, thus does not contain the resource yet
+    source.handleResources(primaryID1(), Set.of(testResource2()));
+
+    assertThat(source.getSecondaryResources(primaryID1()))
+        .containsExactlyInAnyOrder(changedTestResource1(), testResource2());
+  }
+
+  @Test
+  void doesNotRetainResourceUpdatedTwiceIfChangedOutsideOfTheReconciler() {
+    var externallyChanged = testResource1().setValue("externallyChangedValue");
+    source.handleResources(primaryID1(), Set.of(testResource1()));
+    source.handleRecentResourceUpdate(primaryID1(), changedTestResource1(), testResource1());
+    source.handleRecentResourceUpdate(
+        primaryID1(), changedTwiceTestResource1(), changedTestResource1());
+
+    source.handleResources(primaryID1(), Set.of(externallyChanged));
+
+    assertThat(source.getSecondaryResources(primaryID1())).containsExactly(externallyChanged);
+  }
+
+  private static SampleExternalResource changedTestResource1() {
+    return testResource1().setValue("changedValue");
+  }
+
+  private static SampleExternalResource changedTwiceTestResource1() {
+    return testResource1().setValue("changedValueAgain");
+  }
+
   @Test
   void getSecondaryResourcesReturnsASnapshotNotALiveView() {
     source.handleResources(primaryID1(), Set.of(testResource1()));

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