[Fix #1749] Publish onWorkflowCancelled before cancelling pending futures (#1751)

* [Fix #1749] Publish onWorkflowCancelled before cancelling pending futures

Cancelling the futures registered through addCancelable (e.g. by a
listen task) completes the execution pipeline synchronously, which runs
cleanUp and clears the instance metadata. cancel()/cancelFuture() used
to do that before publishing the status change and onWorkflowCancelled,
so listeners could not find their per-instance metadata when the
cancelled event arrived.

internalCancel() now returns the futures to cancel, and they are
cancelled only after the cancelled status change and
onWorkflowCancelled have been published.

Signed-off-by: Eric Deandrea <eric@ericdeandrea.dev>

* [Fix #1749] Keep cancellation final and clean up after it is published

Address review feedback on the cancellation ordering:

- status(WorkflowStatus) now checks and sets the status under
  statusLock and ignores changes once the instance is CANCELLED, so a
  listen task receiving an event while the cancellation is being
  published can no longer move the instance back to WAITING and let it
  complete normally.
- The execution pipeline waits for the cancellation to be published
  before running cleanUp, so instance metadata is still available to
  onWorkflowCancelled listeners even when the pipeline ends on its own
  (an event completes the listen task, or a wait elapses) while an
  asynchronous status change listener is still pending.

Adds regressions that hold the CANCELLED status change pending and
deliver a late event or let the pipeline finish in the meantime.

Signed-off-by: Eric Deandrea <eric@ericdeandrea.dev>

* [Fix #1749] Fail terminal transitions of a cancelled instance

status(WorkflowStatus) ignored every change once the instance was
CANCELLED by returning a normally completed future, but publishEvents()
and handleException() ignore that value. If the cancellation happened
while the last task's onTaskCompleted/onTaskFailed listener was still
pending, releasing it published onWorkflowCompleted (and completed
start() normally) or onWorkflowFailed for a cancelled instance.

A rejected COMPLETED or FAULTED transition now fails with a
CancellationException, so the pipeline ends as cancelled without
publishing completion or failure events. Late non-terminal changes
(e.g. WAITING from a listen task) are still ignored.

Signed-off-by: Eric Deandrea <eric@ericdeandrea.dev>

* [Fix #1749] Alternative approach

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

---------

Signed-off-by: Eric Deandrea <eric@ericdeandrea.dev>
Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>
Co-authored-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

Author: edeandrea

impl/core/src/main/java/io/serverlessworkflow/impl/LifecycleEventsUtils.java

@@ -28,6 +28,15 @@ public class LifecycleEventsUtils {
 
   private LifecycleEventsUtils() {}
 
+  public static CompletableFuture<Boolean> publishEvent(
+      boolean statusChanged,
+      WorkflowContext workflowContext,
+      Function<WorkflowExecutionCompletableListener, CompletableFuture<?>> function) {
+    return statusChanged
+        ? publishEvent(workflowContext, function).thenApply(__ -> true)
+        : CompletableFuture.completedFuture(false);
+  }
+
   public static CompletableFuture<?> publishEvent(
       WorkflowContext workflowContext,
       Function<WorkflowExecutionCompletableListener, CompletableFuture<?>> function) {

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java

@@ -39,6 +39,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
 import java.util.function.Supplier;
 
 public class WorkflowMutableInstance implements WorkflowInstance {
@@ -60,7 +61,11 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   private Lock statusLock = new ReentrantLock();
   private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
 
-  private Collection<CompletableFuture<?>> cancelables = new ArrayList<>();
+  private Collection<CompletableFuture<?>> cancelables =
+      Collections.synchronizedList(new ArrayList<>());
+
+  private Collection<CompletableFuture<Boolean>> outOfOrderListeners =
+      Collections.synchronizedList(new ArrayList<>());
 
   protected WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) {
     this.id = id;
@@ -104,15 +109,16 @@ protected final CompletableFuture<WorkflowModel> startExecution(
                                   .orElse(input))
                           .whenComplete(this::setCompleteDate)
                           .thenApply(this::filterAndValidate)
-                          .thenCompose(this::publishEvents)
+                          .thenCompose(this::publishCompletionEvents)
                           .exceptionallyCompose(this::handleException))
-              .whenComplete(this::cleanUp);
+              .handle(this::cleanUpWaitingListeners)
+              .thenCompose(Function.identity());
       futureRef.set(future);
     }
     return future;
   }
 
-  private CompletableFuture<WorkflowModel> publishEvents(WorkflowModel model) {
+  private CompletableFuture<WorkflowModel> publishCompletionEvents(WorkflowModel model) {
     return status(WorkflowStatus.COMPLETED)
         .thenCompose(
             __ ->
@@ -126,19 +132,37 @@ private void setCompleteDate(WorkflowModel result, Throwable ex) {
     completedAt = Instant.now();
   }
 
-  private void cleanUp(WorkflowModel result, Throwable ex) {
+  private CompletableFuture<WorkflowModel> cleanUpWaitingListeners(
+      WorkflowModel model, Throwable ex) {
+    return CompletableFuture.allOf(
+            outOfOrderListeners.toArray(new CompletableFuture[outOfOrderListeners.size()]))
+        .handle(this::cleanUp)
+        .thenCompose(__ -> fromResult(model, ex));
+  }
+
+  private Object cleanUp(Object ignored, Throwable ex) {
     additionalObjects.values().stream()
         .filter(AutoCloseable.class::isInstance)
         .map(AutoCloseable.class::cast)
         .forEach(WorkflowUtils::safeClose);
     additionalObjects.clear();
     workflowContext.definition().removeInstance(this);
+    outOfOrderListeners.clear();
+    return ignored;
+  }
+
+  private CompletableFuture<WorkflowModel> fromResult(WorkflowModel model, Throwable ex) {
+    return ex == null
+        ? CompletableFuture.completedFuture(model)
+        : CompletableFuture.failedFuture(ex);
   }
 
   private CompletableFuture<WorkflowModel> handleException(Throwable exception) {
     final Throwable cause =
         exception instanceof CompletionException ? exception.getCause() : exception;
-    if (!(cause instanceof CancellationException)) {
+    if (cause instanceof CancellationException) {
+      return CompletableFuture.failedFuture(cause);
+    } else {
       return status(WorkflowStatus.FAULTED)
           .thenCompose(
               __ ->
@@ -147,7 +171,6 @@ private CompletableFuture<WorkflowModel> handleException(Throwable exception) {
                       l -> l.onWorkflowFailed(new WorkflowFailedEvent(workflowContext, cause))))
           .thenCompose(__ -> CompletableFuture.failedFuture(exception));
     }
-    return CompletableFuture.failedFuture(exception);
   }
 
   private WorkflowModel filterAndValidate(WorkflowModel model) {
@@ -213,9 +236,21 @@ public <T> T outputAs(Class<T> clazz) {
         : null;
   }
 
-  public CompletableFuture<Boolean> status(WorkflowStatus state) {
-    WorkflowStatus prevState = this.status.getAndSet(state);
-    return publishStatusChange(prevState, state);
+  public CompletableFuture<Boolean> status(WorkflowStatus newState) {
+    WorkflowStatus prevState;
+    statusLock.lock();
+    try {
+      prevState = status.get();
+      if (prevState == WorkflowStatus.CANCELLED && newState != WorkflowStatus.CANCELLED) {
+        return CompletableFuture.failedFuture(
+            new CancellationException("Workflow has been cancelled"));
+      } else {
+        status.set(newState);
+      }
+    } finally {
+      statusLock.unlock();
+    }
+    return publishStatusChange(prevState, newState);
   }
 
   protected final void setStatus(WorkflowStatus state) {
@@ -224,14 +259,10 @@ protected final void setStatus(WorkflowStatus state) {
 
   private CompletableFuture<Boolean> publishStatusChange(
       WorkflowStatus prevState, WorkflowStatus state) {
-    return prevState != state
-        ? publishEvent(
-                workflowContext,
-                l ->
-                    l.onWorkflowStatusChanged(
-                        new WorkflowStatusEvent(workflowContext, prevState, state)))
-            .thenApply(__ -> true)
-        : CompletableFuture.completedFuture(false);
+    return publishEvent(
+        prevState != state,
+        workflowContext,
+        l -> l.onWorkflowStatusChanged(new WorkflowStatusEvent(workflowContext, prevState, state)));
   }
 
   @Override
@@ -252,28 +283,30 @@ public boolean suspend() {
     WorkflowStatus prevState = internalSuspend();
     boolean result = prevState != WorkflowStatus.SUSPENDED;
     if (result) {
-      publishStatusChange(prevState, WorkflowStatus.SUSPENDED)
-          .thenCompose(
-              __ ->
-                  publishEvent(
-                      workflowContext,
-                      l -> l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext))));
+      outOfOrder(
+          publishStatusChange(prevState, WorkflowStatus.SUSPENDED)
+              .thenCompose(
+                  changed ->
+                      publishEvent(
+                          changed,
+                          workflowContext,
+                          l ->
+                              l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext)))));
     }
     return result;
   }
 
   @Override
   public CompletableFuture<Boolean> suspendFuture() {
     WorkflowStatus prevState = internalSuspend();
-    return prevState != WorkflowStatus.SUSPENDED
-        ? publishStatusChange(prevState, WorkflowStatus.SUSPENDED)
+    return outOfOrder(
+        publishStatusChange(prevState, WorkflowStatus.SUSPENDED)
             .thenCompose(
-                __ ->
+                changed ->
                     publishEvent(
+                        changed,
                         workflowContext,
-                        l -> l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext))))
-            .thenApply(__ -> true)
-        : CompletableFuture.completedFuture(false);
+                        l -> l.onWorkflowSuspended(new WorkflowSuspendedEvent(workflowContext)))));
   }
 
   private WorkflowStatus internalSuspend() {
@@ -299,28 +332,29 @@ public boolean resume() {
     WorkflowStatus prevStatus = internalResume();
     boolean result = prevStatus != WorkflowStatus.RUNNING;
     if (result) {
-      publishStatusChange(prevStatus, WorkflowStatus.RUNNING)
-          .thenCompose(
-              __ ->
-                  publishEvent(
-                      workflowContext,
-                      l -> l.onWorkflowResumed(new WorkflowResumedEvent(workflowContext))));
+      outOfOrder(
+          publishStatusChange(prevStatus, WorkflowStatus.RUNNING)
+              .thenCompose(
+                  changed ->
+                      publishEvent(
+                          changed,
+                          workflowContext,
+                          l -> l.onWorkflowResumed(new WorkflowResumedEvent(workflowContext)))));
     }
     return result;
   }
 
   @Override
   public CompletableFuture<Boolean> resumeFuture() {
     WorkflowStatus prevStatus = internalResume();
-    return prevStatus != WorkflowStatus.RUNNING
-        ? publishStatusChange(prevStatus, WorkflowStatus.RUNNING)
+    return outOfOrder(
+        publishStatusChange(prevStatus, WorkflowStatus.RUNNING)
             .thenCompose(
-                __ ->
+                change ->
                     publishEvent(
+                        change,
                         workflowContext,
-                        l -> l.onWorkflowResumed(new WorkflowResumedEvent(workflowContext))))
-            .thenApply(__ -> true)
-        : CompletableFuture.completedFuture(false);
+                        l -> l.onWorkflowResumed(new WorkflowResumedEvent(workflowContext)))));
   }
 
   private WorkflowStatus internalResume() {
@@ -343,13 +377,12 @@ private WorkflowStatus internalResume() {
     return result;
   }
 
-  public CompletableFuture<TaskContext> cancelCheck(TaskContext t) {
+  public <T> CompletableFuture<T> cancelCheck(T t) {
     try {
       statusLock.lock();
       if (status.get() == WorkflowStatus.CANCELLED) {
-        CompletableFuture<TaskContext> cancelled = new CompletableFuture<TaskContext>();
-        cancelled.completeExceptionally(
-            new CancellationException("Task " + t.taskName() + " has been cancelled"));
+        CompletableFuture<T> cancelled = new CompletableFuture<>();
+        cancelled.completeExceptionally(new CancellationException(t + " has been cancelled"));
         return cancelled;
       }
     } finally {
@@ -380,54 +413,55 @@ public CompletableFuture<TaskContext> suspendedCheck(TaskContext t) {
 
   @Override
   public boolean cancel() {
-    WorkflowStatus prevStatus = internalCancel();
-    boolean result = prevStatus != WorkflowStatus.CANCELLED;
-    if (result) {
-      publishStatusChange(prevStatus, WorkflowStatus.CANCELLED)
-          .thenCompose(
-              __ ->
-                  publishEvent(
-                      workflowContext,
-                      l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext))));
-    }
-    return result;
+    CompletableFuture<Boolean> result = internalCancel();
+    return !result.isDone() || result.join();
   }
 
   @Override
   public CompletableFuture<Boolean> cancelFuture() {
-    WorkflowStatus prevState = internalCancel();
-    return prevState != WorkflowStatus.CANCELLED
-        ? publishStatusChange(prevState, WorkflowStatus.CANCELLED)
-            .thenCompose(
-                __ ->
-                    publishEvent(
-                        workflowContext,
-                        l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext))))
-            .thenApply(__ -> true)
-        : CompletableFuture.completedFuture(false);
+    return internalCancel();
   }
 
-  private WorkflowStatus internalCancel() {
-    WorkflowStatus result;
+  private CompletableFuture<Boolean> internalCancel() {
+    WorkflowStatus prevState;
     Collection<CompletableFuture<?>> toCancel = null;
     try {
       statusLock.lock();
       if (TaskExecutorHelper.isActive(status.get())) {
         toCancel = new ArrayList<>(cancelables);
         cancelables.clear();
-        result = status.getAndSet(WorkflowStatus.CANCELLED);
+        prevState = status.getAndSet(WorkflowStatus.CANCELLED);
       } else {
-        result = WorkflowStatus.CANCELLED;
+        prevState = WorkflowStatus.CANCELLED;
       }
     } finally {
       statusLock.unlock();
     }
-    if (result != WorkflowStatus.CANCELLED && toCancel != null) {
+    CompletableFuture<Boolean> result =
+        outOfOrder(
+            publishStatusChange(prevState, WorkflowStatus.CANCELLED)
+                .thenCompose(
+                    changed ->
+                        publishEvent(
+                            changed,
+                            workflowContext,
+                            l ->
+                                l.onWorkflowCancelled(
+                                    new WorkflowCancelledEvent(workflowContext)))));
+
+    if (prevState != WorkflowStatus.CANCELLED && toCancel != null) {
       toCancel.forEach(t -> t.cancel(true));
     }
     return result;
   }
 
+  private CompletableFuture<Boolean> outOfOrder(CompletableFuture<Boolean> future) {
+    if (!future.isDone()) {
+      outOfOrderListeners.add(future);
+    }
+    return future;
+  }
+
   public void addCancelable(CompletableFuture<?> cancelable) {
     statusLock.lock();
     if (status.get() == WorkflowStatus.CANCELLED) {