fix: evaluate activation conditions outside the workflow executor's monitor

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java

@@ -92,8 +92,8 @@ protected boolean alreadyVisited(DependentResourceNode<?, P> dependentResourceNo
     return getResultFlagFor(dependentResourceNode, BaseWorkflowResult.DetailBuilder::isVisited);
   }
 
-  protected boolean postDeleteConditionNotMet(DependentResourceNode<?, P> drn) {
-    return getResultFlagFor(drn, BaseWorkflowResult.DetailBuilder::hasPostDeleteConditionNotMet);
+  protected boolean postDeleteConditionMet(DependentResourceNode<?, P> drn) {
+    return !getResultFlagFor(drn, BaseWorkflowResult.DetailBuilder::hasPostDeleteConditionNotMet);
   }
 
   protected boolean isMarkedForDelete(DependentResourceNode<?, P> drn) {
@@ -132,6 +132,20 @@ protected synchronized void handleExceptionInExecutor(
     createOrGetResultFor(dependentResourceNode).withError(e);
   }
 
+  /**
+   * Called with the monitor held, after the node's execution mark is cleared and before the
+   * workflow may observe that no execution is left. Anything scheduled from here keeps its own
+   * execution mark, so the workflow does not complete while it is still running.
+   *
+   * <p>Implementations must not block, as the monitor is held, and must schedule any follow up work
+   * synchronously before returning. Scheduling it from another thread would allow that work to
+   * finish before its execution mark is set, leaving behind a mark that is never cleared.
+   *
+   * <p>An exception thrown here is recorded as an error on the node instead of being propagated, so
+   * that the workflow waiting on the monitor is notified in any case.
+   */
+  protected void onNodeExecutionFinished(DependentResourceNode<?, P> dependentResourceNode) {}
+
   protected boolean isReady(DependentResourceNode<?, P> dependentResourceNode) {
     return getResultFlagFor(dependentResourceNode, BaseWorkflowResult.DetailBuilder::isReady);
   }
@@ -144,8 +158,17 @@ protected synchronized void handleNodeExecutionFinish(
       DependentResourceNode<?, P> dependentResourceNode) {
     logger().trace("Finished execution for: {} primary: {}", dependentResourceNode, primaryID);
     actualExecutions.remove(dependentResourceNode);
-    if (noMoreExecutionsScheduled()) {
-      this.notifyAll();
+    try {
+      onNodeExecutionFinished(dependentResourceNode);
+    } catch (Exception e) {
+      handleExceptionInExecutor(dependentResourceNode, e);
+    } catch (Error e) {
+      logger().error("java.lang.Error during execution finish", e);
+      throw e;
+    } finally {
+      if (noMoreExecutionsScheduled()) {
+        this.notifyAll();
+      }
     }
   }
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java

@@ -137,7 +137,7 @@ private synchronized void handleDependentCleaned(
   private boolean allDependentsCleaned(DependentResourceNode dependentResourceNode) {
     List<DependentResourceNode> parents = dependentResourceNode.getParents();
     return parents.isEmpty()
-        || parents.stream().allMatch(d -> alreadyVisited(d) && !postDeleteConditionNotMet(d));
+        || parents.stream().allMatch(d -> alreadyVisited(d) && postDeleteConditionMet(d));
   }
 
   @SuppressWarnings("unchecked")

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java

@@ -18,6 +18,7 @@
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,6 +35,9 @@ class WorkflowReconcileExecutor<P extends HasMetadata> extends AbstractWorkflowE
   private static final String RECONCILE = "reconcile";
   private static final String DELETE = "delete";
 
+  private final Set<DependentResourceNode<?, P>> conditionNotMetNodes =
+      ConcurrentHashMap.newKeySet();
+
   public WorkflowReconcileExecutor(DefaultWorkflow<P> workflow, P primary, Context<P> context) {
     super(workflow, primary, context);
   }
@@ -51,6 +55,13 @@ protected Logger logger() {
     return log;
   }
 
+  @Override
+  protected void onNodeExecutionFinished(DependentResourceNode<?, P> dependentResourceNode) {
+    if (conditionNotMetNodes.remove(dependentResourceNode)) {
+      handleReconcileOrActivationConditionNotMet(dependentResourceNode);
+    }
+  }
+
   private synchronized <R> void handleReconcile(DependentResourceNode<R, P> dependentResourceNode) {
     log.debug("Considering for reconcile: {} primaryID: {}", dependentResourceNode, primaryID);
 
@@ -90,20 +101,7 @@ private synchronized <R> void handleReconcile(DependentResourceNode<R, P> depend
       return;
     }
 
-    boolean activationConditionMet =
-        isConditionMet(dependentResourceNode.getActivationCondition(), dependentResourceNode);
-    registerOrDeregisterEventSourceBasedOnActivation(activationConditionMet, dependentResourceNode);
-
-    boolean reconcileConditionMet = true;
-    if (activationConditionMet) {
-      reconcileConditionMet =
-          isConditionMet(dependentResourceNode.getReconcilePrecondition(), dependentResourceNode);
-    }
-    if (!reconcileConditionMet || !activationConditionMet) {
-      handleReconcileOrActivationConditionNotMet(dependentResourceNode);
-    } else {
-      submit(dependentResourceNode, new NodeReconcileExecutor<>(dependentResourceNode), RECONCILE);
-    }
+    submit(dependentResourceNode, new NodeReconcileExecutor<>(dependentResourceNode), RECONCILE);
   }
 
   private synchronized void handleDelete(DependentResourceNode dependentResourceNode) {
@@ -144,7 +142,7 @@ private boolean allDependentsDeletedAlready(DependentResourceNode<?, P> dependen
     var dependents = dependentResourceNode.getParents();
     return dependents.stream()
         .allMatch(
-            d -> alreadyVisited(d) && isReady(d) && !isInError(d) && !postDeleteConditionNotMet(d));
+            d -> alreadyVisited(d) && isReady(d) && !isInError(d) && postDeleteConditionMet(d));
   }
 
   private class NodeReconcileExecutor<R> extends NodeExecutor<R, P> {
@@ -155,6 +153,17 @@ private NodeReconcileExecutor(DependentResourceNode<R, P> dependentResourceNode)
 
     @Override
     protected void doRun(DependentResourceNode<R, P> dependentResourceNode) {
+      final var activationConditionMet =
+          isConditionMet(dependentResourceNode.getActivationCondition(), dependentResourceNode);
+      registerOrDeregisterEventSourceBasedOnActivation(
+          activationConditionMet, dependentResourceNode);
+      if (!activationConditionMet
+          || !isConditionMet(
+              dependentResourceNode.getReconcilePrecondition(), dependentResourceNode)) {
+        conditionNotMetNodes.add(dependentResourceNode);
+        return;
+      }
+
       final var dependentResource = dependentResourceNode.getDependentResource();
       log.debug("Reconciling for primary: {} node: {} ", primaryID, dependentResourceNode);
       ReconcileResult reconcileResult = dependentResource.reconcile(primary, context);
@@ -238,6 +247,7 @@ private void handleReconcileOrActivationConditionNotMet(
     Set<DependentResourceNode> bottomNodes = new HashSet<>();
     markDependentsForDelete(dependentResourceNode, bottomNodes);
     bottomNodes.forEach(this::handleDelete);
+    handleDelete(dependentResourceNode);
   }
 
   private void markDependentsForDelete(

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java

@@ -15,13 +15,21 @@
  */
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.api.Timeout;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,13 +53,12 @@ class WorkflowReconcileExecutorTest extends AbstractWorkflowExecutorTest {
   Context<TestCustomResource> mockContext = spy(Context.class);
 
   ExecutorService executorService = Executors.newCachedThreadPool();
-  EventSourceRetriever eventSourceRetriever = mock(EventSourceRetriever.class);
+  EventSourceRetriever<TestCustomResource> eventSourceRetriever = mock();
 
   TestDependent dr3 = new TestDependent("DR_3");
   TestDependent dr4 = new TestDependent("DR_4");
 
   @BeforeEach
-  @SuppressWarnings("unchecked")
   void setup(TestInfo testInfo) {
     log.debug("==> Starting test {}", testInfo.getDisplayName());
     when(mockContext.managedWorkflowAndDependentResourceContext())
@@ -673,6 +680,161 @@ void deletesDependentsOfNonActiveDependentButNotTheNonActive() {
     assertThat(executionHistory).deleted(drDeleter, drDeleter3).notReconciled(dr1, drDeleter2);
   }
 
+  @Test
+  @Timeout(10)
+  void deleteScheduledByUnmetPreconditionFinishesBeforeReconcileReturns() {
+    // the delete is scheduled from the completion hook, while the monitor is still held
+    var deleteFinished = new AtomicBoolean(false);
+    var slowDeleter =
+        new TestDeleterDependent("SLOW_DELETER") {
+          @Override
+          public void delete(TestCustomResource primary, Context<TestCustomResource> context) {
+            try {
+              Thread.sleep(200);
+            } catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
+            }
+            deleteFinished.set(true);
+            super.delete(primary, context);
+          }
+        };
+
+    var workflow =
+        new WorkflowBuilder<TestCustomResource>()
+            .addDependentResourceAndConfigure(slowDeleter)
+            .withReconcilePrecondition(notMetCondition)
+            .build();
+
+    workflow.reconcile(new TestCustomResource(), mockContext);
+
+    assertTrue(deleteFinished.get(), "reconcile() returned while the delete was still running");
+  }
+
+  @Test
+  @Timeout(10)
+  void deletedIfReconcilePreconditionNotMetWhileAnotherDeleteCascadeIsRunning() {
+    // the shared dependent's cascade reaches this one while it is still evaluating its precondition
+    var sharedDependentDeleted = new CountDownLatch(1);
+    TestDeleterDependent drDeleter2 = new TestDeleterDependent("DR_DELETER_2");
+    var drDeleter3 =
+        new TestDeleterDependent("DR_DELETER_3") {
+          @Override
+          public void delete(TestCustomResource primary, Context<TestCustomResource> context) {
+            super.delete(primary, context);
+            sharedDependentDeleted.countDown();
+          }
+        };
+    Condition<?, TestCustomResource> blockedNotMetCondition =
+        (dependentResource, primary, context) -> {
+          try {
+            if (!sharedDependentDeleted.await(5, TimeUnit.SECONDS)) {
+              throw new IllegalStateException("the shared dependent was never deleted");
+            }
+          } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException(e);
+          }
+          return false;
+        };
+
+    var workflow =
+        new WorkflowBuilder<TestCustomResource>()
+            .addDependentResourceAndConfigure(drDeleter)
+            .withReconcilePrecondition(notMetCondition)
+            .addDependentResourceAndConfigure(drDeleter2)
+            .withReconcilePrecondition(blockedNotMetCondition)
+            .addDependentResourceAndConfigure(drDeleter3)
+            .dependsOn(drDeleter, drDeleter2)
+            .build();
+
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
+
+    assertThat(executionHistory).deleted(drDeleter3, drDeleter, drDeleter2);
+    Assertions.assertThat(res.getErroredDependents()).isEmpty();
+  }
+
+  @Test
+  @Timeout(10)
+  void completesIfSchedulingDeleteForUnmetPreconditionIsRejected() {
+    // the delete is scheduled from a hook that already runs in the node executor's finally block
+    ExecutorService rejectingSecondSubmit = mock();
+    when(rejectingSecondSubmit.submit(any(Runnable.class)))
+        .thenAnswer(invocation -> executorService.submit((Runnable) invocation.getArgument(0)))
+        .thenThrow(new RejectedExecutionException("executor is shut down"));
+    when(mockContext.getWorkflowExecutorService()).thenReturn(rejectingSecondSubmit);
+
+    var workflow =
+        new WorkflowBuilder<TestCustomResource>()
+            .addDependentResourceAndConfigure(drDeleter)
+            .withReconcilePrecondition(notMetCondition)
+            .build();
+
+    var exception =
+        assertThrows(
+            AggregatedOperatorException.class,
+            () -> workflow.reconcile(new TestCustomResource(), mockContext));
+
+    Assertions.assertThat(exception.getAggregatedExceptions())
+        .containsOnlyKeys(drDeleter.name())
+        .extractingByKey(drDeleter.name())
+        .isInstanceOf(RejectedExecutionException.class);
+  }
+
+  @Test
+  @Timeout(10)
+  void activationConditionsEvaluatedConcurrently() {
+    // they can only meet at the barrier if neither of them holds the executor's monitor
+    var rendezvousCondition = rendezvousCondition(new CyclicBarrier(2));
+
+    var workflow =
+        new WorkflowBuilder<TestCustomResource>()
+            .addDependentResourceAndConfigure(dr1)
+            .withActivationCondition(rendezvousCondition)
+            .addDependentResourceAndConfigure(dr2)
+            .withActivationCondition(rendezvousCondition)
+            .build();
+
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
+
+    assertThat(executionHistory).reconciled(dr1, dr2);
+    Assertions.assertThat(res.getErroredDependents()).isEmpty();
+  }
+
+  @Test
+  void activationConditionErrorAttributedToItsOwnDependent() {
+    var workflow =
+        new WorkflowBuilder<TestCustomResource>()
+            .addDependentResource(dr1)
+            .addDependentResourceAndConfigure(dr2)
+            .dependsOn(dr1)
+            .withActivationCondition(throwingCondition())
+            .addDependentResourceAndConfigure(dr3)
+            .dependsOn(dr1)
+            .withThrowExceptionFurther(false)
+            .build();
+
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
+
+    Assertions.assertThat(res.getErroredDependents()).containsOnlyKeys(dr2);
+    assertThat(executionHistory).reconciled(dr1, dr3).notReconciled(dr2);
+  }
+
+  @Test
+  void activationConditionErrorOnTopLevelDependentDoesNotStopOthers() {
+    var workflow =
+        new WorkflowBuilder<TestCustomResource>()
+            .addDependentResourceAndConfigure(dr1)
+            .withActivationCondition(throwingCondition())
+            .addDependentResource(dr2)
+            .withThrowExceptionFurther(false)
+            .build();
+
+    var res = workflow.reconcile(new TestCustomResource(), mockContext);
+
+    Assertions.assertThat(res.getErroredDependents()).containsOnlyKeys(dr1);
+    assertThat(executionHistory).reconciled(dr2).notReconciled(dr1);
+  }
+
   @Test
   @SuppressWarnings("unchecked")
   void activationConditionOnlyCalledOnceOnDeleteDependents() {
@@ -695,7 +857,6 @@ void activationConditionOnlyCalledOnceOnDeleteDependents() {
   }
 
   @Test
-  @SuppressWarnings("unchecked")
   void activationConditionEventSourceRegistrationWithParentWithFalsePrecondition() {
     var workflow =
         new WorkflowBuilder<TestCustomResource>()
@@ -713,7 +874,6 @@ void activationConditionEventSourceRegistrationWithParentWithFalsePrecondition()
   }
 
   @Test
-  @SuppressWarnings("unchecked")
   void activationConditionEventSourceRegistration() {
     var workflow =
         new WorkflowBuilder<TestCustomResource>()
@@ -826,4 +986,24 @@ void shouldReturnEmptyIfNoConditionResultExists() {
     final var reconcileResult = workflow.reconcile(new TestCustomResource(), mockContext);
     assertTrue(reconcileResult.getNotReadyDependentResult(dr1, Integer.class).isEmpty());
   }
+
+  private Condition<?, TestCustomResource> rendezvousCondition(CyclicBarrier barrier) {
+    return (dependentResource, primary, context) -> {
+      try {
+        barrier.await(5, TimeUnit.SECONDS);
+        return true;
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw new IllegalStateException(e);
+      } catch (BrokenBarrierException | TimeoutException e) {
+        throw new IllegalStateException("activation conditions were serialized", e);
+      }
+    };
+  }
+
+  private Condition<?, TestCustomResource> throwingCondition() {
+    return (dependentResource, primary, context) -> {
+      throw new IllegalStateException("Test exception");
+    };
+  }
 }