From 098ced2716c82972b360030cef5478bd20eefcf8 Mon Sep 17 00:00:00 2001 From: Eric Deandrea Date: Wed, 7 Oct 2026 13:55:00 -0400 Subject: [PATCH 1/4] [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 --- .../impl/WorkflowMutableInstance.java | 62 ++++--- .../impl/test/CancelMetadataTest.java | 173 ++++++++++++++++++ 2 files changed, 206 insertions(+), 29 deletions(-) create mode 100644 impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java index 281830dc9..5c320c5ee 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java @@ -380,52 +380,56 @@ public CompletableFuture 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)))); + CancelRequest request = internalCancel(); + if (request.accepted()) { + publishCancelled(request); } - return result; + return request.accepted(); } @Override public CompletableFuture cancelFuture() { - WorkflowStatus prevState = internalCancel(); - return prevState != WorkflowStatus.CANCELLED - ? publishStatusChange(prevState, WorkflowStatus.CANCELLED) - .thenCompose( - __ -> - publishEvent( - workflowContext, - l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext)))) - .thenApply(__ -> true) + CancelRequest request = internalCancel(); + return request.accepted() + ? publishCancelled(request).thenApply(__ -> true) : CompletableFuture.completedFuture(false); } - private WorkflowStatus internalCancel() { - WorkflowStatus result; - Collection> toCancel = null; + /** + * Publishes the cancellation and only then cancels the pending futures. Cancelling them may + * complete the execution pipeline synchronously, which runs {@link #cleanUp} and clears the + * instance metadata, so listeners must be notified first. + */ + private CompletableFuture publishCancelled(CancelRequest request) { + return publishStatusChange(request.prevStatus(), WorkflowStatus.CANCELLED) + .thenCompose( + __ -> + publishEvent( + workflowContext, + l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext)))) + .whenComplete((__, ex) -> request.toCancel().forEach(t -> t.cancel(true))); + } + + private record CancelRequest( + WorkflowStatus prevStatus, Collection> toCancel) { + boolean accepted() { + return prevStatus != WorkflowStatus.CANCELLED; + } + } + + private CancelRequest internalCancel() { try { statusLock.lock(); if (TaskExecutorHelper.isActive(status.get())) { - toCancel = new ArrayList<>(cancelables); + Collection> toCancel = new ArrayList<>(cancelables); cancelables.clear(); - result = status.getAndSet(WorkflowStatus.CANCELLED); + return new CancelRequest(status.getAndSet(WorkflowStatus.CANCELLED), toCancel); } else { - result = WorkflowStatus.CANCELLED; + return new CancelRequest(WorkflowStatus.CANCELLED, Collections.emptyList()); } } finally { statusLock.unlock(); } - if (result != WorkflowStatus.CANCELLED && toCancel != null) { - toCancel.forEach(t -> t.cancel(true)); - } - return result; } public void addCancelable(CompletableFuture cancelable) { diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java new file mode 100644 index 000000000..a64a80d09 --- /dev/null +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java @@ -0,0 +1,173 @@ +/* + * Copyright 2020-Present The Serverless Workflow Specification Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.serverlessworkflow.impl.test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import io.serverlessworkflow.api.WorkflowReader; +import io.serverlessworkflow.impl.WorkflowApplication; +import io.serverlessworkflow.impl.WorkflowDefinition; +import io.serverlessworkflow.impl.WorkflowInstance; +import io.serverlessworkflow.impl.WorkflowModel; +import io.serverlessworkflow.impl.WorkflowStatus; +import io.serverlessworkflow.impl.lifecycle.WorkflowCancelledEvent; +import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener; +import io.serverlessworkflow.impl.lifecycle.WorkflowStartedEvent; +import java.io.IOException; +import java.time.Duration; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.stream.Stream; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Instance metadata must still be available to listeners when {@code onWorkflowCancelled} is + * published, whatever the instance was doing when it was cancelled. + */ +class CancelMetadataTest { + + private static final String KEY = "cancel-metadata-test"; + + private static WorkflowApplication appl; + private static MetadataListener listener; + + @BeforeAll + static void init() { + listener = new MetadataListener(); + appl = + WorkflowApplication.builder().disableLifeCycleCEPublishing().withListener(listener).build(); + } + + @AfterAll + static void cleanup() { + appl.close(); + } + + @BeforeEach + void reset() { + listener.reset(); + } + + @ParameterizedTest(name = "{0} / {1}") + @MethodSource("cancelParameters") + void metadataIsAvailableWhenCancelledEventIsPublished( + String workflowPath, String cancelMethod, WorkflowStatus statusBeforeCancel) + throws IOException { + WorkflowDefinition definition = + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(workflowPath)); + WorkflowInstance instance = definition.instance(Map.of("threshold", 1000)); + CompletableFuture future = instance.start(); + await() + .atMost(Duration.ofSeconds(5)) + .pollInterval(Duration.ofMillis(5)) + .until(() -> instance.status() == statusBeforeCancel); + + cancelMethods().get(cancelMethod).accept(instance); + + await().atMost(Duration.ofSeconds(5)).until(future::isDone); + await().atMost(Duration.ofSeconds(5)).until(() -> listener.cancelledSeen(instance.id())); + assertThat(instance.status()).isEqualTo(WorkflowStatus.CANCELLED); + assertThat(listener.metadataOnCancel(instance.id())) + .as("metadata seen by onWorkflowCancelled") + .isTrue(); + assertThat(listener.closedBeforeCancel(instance.id())) + .as("metadata closed before onWorkflowCancelled") + .isFalse(); + } + + private static Map> cancelMethods() { + return Map.of( + "cancel", + WorkflowInstance::cancel, + "cancelFuture", + i -> { + try { + assertThat(i.cancelFuture().get(5, TimeUnit.SECONDS)).isTrue(); + } catch (Exception e) { + throw new IllegalStateException(e); + } + }); + } + + private static Stream cancelParameters() { + // listen registers a cancelable future (cancel completes the pipeline synchronously); + // wait does not (the pipeline notices the cancellation on the next task). + return Stream.of( + Arguments.of( + "workflows-samples/listen-to-any-filter.yaml", "cancel", WorkflowStatus.WAITING), + Arguments.of( + "workflows-samples/listen-to-any-filter.yaml", "cancelFuture", WorkflowStatus.WAITING), + Arguments.of("workflows-samples/wait-set.yaml", "cancel", WorkflowStatus.WAITING), + Arguments.of("workflows-samples/wait-set.yaml", "cancelFuture", WorkflowStatus.WAITING)); + } + + static class TrackedMetadata implements AutoCloseable { + private volatile boolean closed; + + @Override + public void close() { + closed = true; + } + } + + static class MetadataListener implements WorkflowExecutionListener { + + private final Map metadataOnCancel = new ConcurrentHashMap<>(); + private final Map closedBeforeCancel = new ConcurrentHashMap<>(); + + void reset() { + metadataOnCancel.clear(); + closedBeforeCancel.clear(); + } + + boolean cancelledSeen(String id) { + return metadataOnCancel.containsKey(id); + } + + boolean metadataOnCancel(String id) { + return metadataOnCancel.get(id); + } + + boolean closedBeforeCancel(String id) { + return closedBeforeCancel.get(id); + } + + @Override + public void onWorkflowStarted(WorkflowStartedEvent ev) { + ((WorkflowInstance) ev.workflowContext().instanceData()) + .addMetadataIfAbsent(KEY, TrackedMetadata::new); + } + + @Override + public void onWorkflowCancelled(WorkflowCancelledEvent ev) { + Optional metadata = + ev.workflowContext().instanceData().findMetadata(KEY, TrackedMetadata.class); + String id = ev.workflowContext().instanceData().id(); + closedBeforeCancel.put(id, metadata.map(m -> m.closed).orElse(false)); + metadataOnCancel.put(id, metadata.isPresent()); + } + } +} From 0ebc701ffbfde025c77239a0fc01b99a07dc2e37 Mon Sep 17 00:00:00 2001 From: Eric Deandrea Date: Wed, 7 Oct 2026 14:31:16 -0400 Subject: [PATCH 2/4] [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 --- .../impl/WorkflowMutableInstance.java | 59 ++++++-- .../impl/test/CancelMetadataTest.java | 128 +++++++++++++++++- 2 files changed, 178 insertions(+), 9 deletions(-) diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java index 5c320c5ee..70919d132 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java +++ b/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 { @@ -62,6 +63,8 @@ public class WorkflowMutableInstance implements WorkflowInstance { private Collection> cancelables = new ArrayList<>(); + private volatile CompletableFuture cancelPublished = CompletableFuture.completedFuture(null); + protected WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) { this.id = id; this.input = input; @@ -106,12 +109,28 @@ protected final CompletableFuture startExecution( .thenApply(this::filterAndValidate) .thenCompose(this::publishEvents) .exceptionallyCompose(this::handleException)) - .whenComplete(this::cleanUp); + .handle(this::cleanUpAfterCancelPublished) + .thenCompose(Function.identity()); futureRef.set(future); } return future; } + private CompletableFuture cleanUpAfterCancelPublished( + WorkflowModel result, Throwable ex) { + return cancelPublished + .handle( + (__, ___) -> { + cleanUp(result, ex); + return null; + }) + .thenCompose( + __ -> + ex == null + ? CompletableFuture.completedFuture(result) + : CompletableFuture.failedFuture(ex)); + } + private CompletableFuture publishEvents(WorkflowModel model) { return status(WorkflowStatus.COMPLETED) .thenCompose( @@ -214,7 +233,19 @@ public T outputAs(Class clazz) { } public CompletableFuture status(WorkflowStatus state) { - WorkflowStatus prevState = this.status.getAndSet(state); + WorkflowStatus prevState; + try { + statusLock.lock(); + prevState = this.status.get(); + if (prevState == WorkflowStatus.CANCELLED) { + // cancellation is final, late callbacks (e.g. an event received by a listen task) must + // not move the instance out of it + return CompletableFuture.completedFuture(false); + } + this.status.set(state); + } finally { + statusLock.unlock(); + } return publishStatusChange(prevState, state); } @@ -397,8 +428,9 @@ public CompletableFuture cancelFuture() { /** * Publishes the cancellation and only then cancels the pending futures. Cancelling them may - * complete the execution pipeline synchronously, which runs {@link #cleanUp} and clears the - * instance metadata, so listeners must be notified first. + * complete the execution pipeline synchronously, and the pipeline may also end on its own while + * the cancellation is being published, so {@link #cleanUpAfterCancelPublished} waits for {@link + * CancelRequest#published()} before clearing the instance metadata. */ private CompletableFuture publishCancelled(CancelRequest request) { return publishStatusChange(request.prevStatus(), WorkflowStatus.CANCELLED) @@ -407,11 +439,17 @@ private CompletableFuture publishCancelled(CancelRequest request) { publishEvent( workflowContext, l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext)))) - .whenComplete((__, ex) -> request.toCancel().forEach(t -> t.cancel(true))); + .whenComplete( + (__, ex) -> { + request.published().complete(null); + request.toCancel().forEach(t -> t.cancel(true)); + }); } private record CancelRequest( - WorkflowStatus prevStatus, Collection> toCancel) { + WorkflowStatus prevStatus, + Collection> toCancel, + CompletableFuture published) { boolean accepted() { return prevStatus != WorkflowStatus.CANCELLED; } @@ -423,9 +461,14 @@ private CancelRequest internalCancel() { if (TaskExecutorHelper.isActive(status.get())) { Collection> toCancel = new ArrayList<>(cancelables); cancelables.clear(); - return new CancelRequest(status.getAndSet(WorkflowStatus.CANCELLED), toCancel); + CompletableFuture published = new CompletableFuture<>(); + cancelPublished = published; + return new CancelRequest(status.getAndSet(WorkflowStatus.CANCELLED), toCancel, published); } else { - return new CancelRequest(WorkflowStatus.CANCELLED, Collections.emptyList()); + return new CancelRequest( + WorkflowStatus.CANCELLED, + Collections.emptyList(), + CompletableFuture.completedFuture(null)); } } finally { statusLock.unlock(); diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java index a64a80d09..bfc262a3a 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java @@ -24,21 +24,27 @@ import io.serverlessworkflow.impl.WorkflowInstance; import io.serverlessworkflow.impl.WorkflowModel; import io.serverlessworkflow.impl.WorkflowStatus; +import io.serverlessworkflow.impl.lifecycle.TaskCancelledEvent; import io.serverlessworkflow.impl.lifecycle.WorkflowCancelledEvent; +import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionCompletableListener; import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener; import io.serverlessworkflow.impl.lifecycle.WorkflowStartedEvent; +import io.serverlessworkflow.impl.lifecycle.WorkflowStatusEvent; import java.io.IOException; import java.time.Duration; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.stream.Stream; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -53,12 +59,18 @@ class CancelMetadataTest { private static WorkflowApplication appl; private static MetadataListener listener; + private static CancelGate cancelGate; @BeforeAll static void init() { listener = new MetadataListener(); + cancelGate = new CancelGate(); appl = - WorkflowApplication.builder().disableLifeCycleCEPublishing().withListener(listener).build(); + WorkflowApplication.builder() + .disableLifeCycleCEPublishing() + .withListener(listener) + .withListener(cancelGate) + .build(); } @AfterAll @@ -69,6 +81,7 @@ static void cleanup() { @BeforeEach void reset() { listener.reset(); + cancelGate.reset(); } @ParameterizedTest(name = "{0} / {1}") @@ -98,6 +111,85 @@ void metadataIsAvailableWhenCancelledEventIsPublished( .isFalse(); } + @Test + void lateEventWhileCancellationIsPublishedDoesNotReviveInstance() throws IOException { + WorkflowDefinition listenDefinition = + appl.workflowDefinition( + WorkflowReader.readWorkflowFromClasspath( + "workflows-samples/listen-to-any-until-consumed.yaml")); + WorkflowDefinition emitDefinition = + appl.workflowDefinition( + WorkflowReader.readWorkflowFromClasspath("workflows-samples/emit-doctor.yaml")); + WorkflowInstance instance = listenDefinition.instance(Map.of()); + CompletableFuture future = instance.start(); + assertThat(instance.status()).isEqualTo(WorkflowStatus.WAITING); + CompletableFuture gate = cancelGate.hold(instance.id()); + + instance.cancel(); + // matches the listen filter but does not satisfy its event-based until + emitDefinition.instance(Map.of("temperature", 39)).start().join(); + gate.complete(null); + + assertThat(future) + .failsWithin(Duration.ofSeconds(5)) + .withThrowableOfType(ExecutionException.class); + assertThat(instance.status()).isEqualTo(WorkflowStatus.CANCELLED); + assertMetadataSeenOnCancel(instance); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("pipelineEndsWhileCancellationIsPublishedParameters") + void metadataIsAvailableWhenPipelineEndsWhileCancellationIsPublished( + String workflowPath, Map input, Map event) + throws IOException { + WorkflowDefinition definition = + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(workflowPath)); + WorkflowDefinition emitDefinition = + appl.workflowDefinition( + WorkflowReader.readWorkflowFromClasspath("workflows-samples/emit-doctor.yaml")); + WorkflowInstance instance = definition.instance(input); + CompletableFuture future = instance.start(); + await() + .atMost(Duration.ofSeconds(5)) + .pollInterval(Duration.ofMillis(5)) + .until(() -> instance.status() == WorkflowStatus.WAITING); + CompletableFuture gate = cancelGate.hold(instance.id()); + + instance.cancel(); + if (event != null) { + // completes the listen task + emitDefinition.instance(event).start().join(); + } + // the pipeline has seen the cancellation while onWorkflowCancelled is still pending + await().atMost(Duration.ofSeconds(5)).until(() -> listener.taskCancelledSeen(instance.id())); + gate.complete(null); + + assertThat(future) + .failsWithin(Duration.ofSeconds(5)) + .withThrowableOfType(ExecutionException.class); + assertThat(instance.status()).isEqualTo(WorkflowStatus.CANCELLED); + assertMetadataSeenOnCancel(instance); + } + + private static Stream pipelineEndsWhileCancellationIsPublishedParameters() { + return Stream.of( + Arguments.of( + "workflows-samples/listen-to-any-filter.yaml", + Map.of("threshold", 38), + Map.of("temperature", 39)), + Arguments.of("workflows-samples/wait-set.yaml", Map.of(), null)); + } + + private static void assertMetadataSeenOnCancel(WorkflowInstance instance) { + await().atMost(Duration.ofSeconds(5)).until(() -> listener.cancelledSeen(instance.id())); + assertThat(listener.metadataOnCancel(instance.id())) + .as("metadata seen by onWorkflowCancelled") + .isTrue(); + assertThat(listener.closedBeforeCancel(instance.id())) + .as("metadata closed before onWorkflowCancelled") + .isFalse(); + } + private static Map> cancelMethods() { return Map.of( "cancel", @@ -124,6 +216,29 @@ private static Stream cancelParameters() { Arguments.of("workflows-samples/wait-set.yaml", "cancelFuture", WorkflowStatus.WAITING)); } + static class CancelGate implements WorkflowExecutionCompletableListener { + + private final Map> gates = new ConcurrentHashMap<>(); + + /** Keeps the CANCELLED status change of the given instance pending until completed. */ + CompletableFuture hold(String instanceId) { + return gates.computeIfAbsent(instanceId, k -> new CompletableFuture<>()); + } + + void reset() { + gates.values().forEach(g -> g.complete(null)); + gates.clear(); + } + + @Override + public CompletableFuture onWorkflowStatusChanged(WorkflowStatusEvent ev) { + return ev.status() == WorkflowStatus.CANCELLED + ? gates.getOrDefault( + ev.workflowContext().instanceData().id(), CompletableFuture.completedFuture(null)) + : CompletableFuture.completedFuture(null); + } + } + static class TrackedMetadata implements AutoCloseable { private volatile boolean closed; @@ -137,10 +252,16 @@ static class MetadataListener implements WorkflowExecutionListener { private final Map metadataOnCancel = new ConcurrentHashMap<>(); private final Map closedBeforeCancel = new ConcurrentHashMap<>(); + private final Set taskCancelled = ConcurrentHashMap.newKeySet(); void reset() { metadataOnCancel.clear(); closedBeforeCancel.clear(); + taskCancelled.clear(); + } + + boolean taskCancelledSeen(String id) { + return taskCancelled.contains(id); } boolean cancelledSeen(String id) { @@ -161,6 +282,11 @@ public void onWorkflowStarted(WorkflowStartedEvent ev) { .addMetadataIfAbsent(KEY, TrackedMetadata::new); } + @Override + public void onTaskCancelled(TaskCancelledEvent ev) { + taskCancelled.add(ev.workflowContext().instanceData().id()); + } + @Override public void onWorkflowCancelled(WorkflowCancelledEvent ev) { Optional metadata = From 8085c7b7034b49c4c3f3c382b52b214b8c6d74eb Mon Sep 17 00:00:00 2001 From: Eric Deandrea Date: Wed, 7 Oct 2026 14:46:15 -0400 Subject: [PATCH 3/4] [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 --- .../impl/WorkflowMutableInstance.java | 14 ++- .../impl/test/CancelMetadataTest.java | 99 ++++++++++++++++++- 2 files changed, 107 insertions(+), 6 deletions(-) diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java index 70919d132..41b122dac 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java @@ -238,9 +238,13 @@ public CompletableFuture status(WorkflowStatus state) { statusLock.lock(); prevState = this.status.get(); if (prevState == WorkflowStatus.CANCELLED) { - // cancellation is final, late callbacks (e.g. an event received by a listen task) must - // not move the instance out of it - return CompletableFuture.completedFuture(false); + // cancellation is final: late callbacks (e.g. an event received by a listen task) must + // not move the instance out of it, and a terminal transition must fail so that neither + // onWorkflowCompleted nor onWorkflowFailed is published for a cancelled instance + return isTerminal(state) + ? CompletableFuture.failedFuture( + new CancellationException("Workflow instance " + id + " has been cancelled")) + : CompletableFuture.completedFuture(false); } this.status.set(state); } finally { @@ -249,6 +253,10 @@ public CompletableFuture status(WorkflowStatus state) { return publishStatusChange(prevState, state); } + private static boolean isTerminal(WorkflowStatus state) { + return state == WorkflowStatus.COMPLETED || state == WorkflowStatus.FAULTED; + } + protected final void setStatus(WorkflowStatus state) { this.status.set(state); } diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java index bfc262a3a..3db834bd6 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java @@ -24,10 +24,16 @@ import io.serverlessworkflow.impl.WorkflowInstance; import io.serverlessworkflow.impl.WorkflowModel; import io.serverlessworkflow.impl.WorkflowStatus; +import io.serverlessworkflow.impl.lifecycle.EventType; import io.serverlessworkflow.impl.lifecycle.TaskCancelledEvent; +import io.serverlessworkflow.impl.lifecycle.TaskCompletedEvent; +import io.serverlessworkflow.impl.lifecycle.TaskFailedEvent; import io.serverlessworkflow.impl.lifecycle.WorkflowCancelledEvent; +import io.serverlessworkflow.impl.lifecycle.WorkflowCompletedEvent; +import io.serverlessworkflow.impl.lifecycle.WorkflowEvent; import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionCompletableListener; import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener; +import io.serverlessworkflow.impl.lifecycle.WorkflowFailedEvent; import io.serverlessworkflow.impl.lifecycle.WorkflowStartedEvent; import io.serverlessworkflow.impl.lifecycle.WorkflowStatusEvent; import java.io.IOException; @@ -35,6 +41,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; @@ -180,6 +187,43 @@ private static Stream pipelineEndsWhileCancellationIsPublishedParamet Arguments.of("workflows-samples/wait-set.yaml", Map.of(), null)); } + @ParameterizedTest(name = "{0}") + @MethodSource("terminalTransitionWhileCancellingParameters") + void cancellationWhileLastTaskEventIsPendingWins(String workflowPath, EventType heldTaskEvent) + throws IOException { + WorkflowDefinition definition = + appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(workflowPath)); + WorkflowInstance instance = definition.instance(Map.of()); + CompletableFuture gate = cancelGate.hold(instance.id(), heldTaskEvent); + CompletableFuture future = instance.start(); + assertThat(future).isNotDone(); + + assertThat(instance.cancel()).isTrue(); + await().atMost(Duration.ofSeconds(5)).until(() -> listener.cancelledSeen(instance.id())); + // lets the last task end and the workflow reach its terminal transition + gate.complete(null); + + assertThat(future) + .failsWithin(Duration.ofSeconds(5)) + .withThrowableOfType(ExecutionException.class) + .withCauseInstanceOf(CancellationException.class); + assertThat(instance.status()).isEqualTo(WorkflowStatus.CANCELLED); + assertThat(listener.workflowCompletedSeen(instance.id())) + .as("onWorkflowCompleted published for a cancelled instance") + .isFalse(); + assertThat(listener.workflowFailedSeen(instance.id())) + .as("onWorkflowFailed published for a cancelled instance") + .isFalse(); + } + + private static Stream terminalTransitionWhileCancellingParameters() { + return Stream.of( + // completes normally: COMPLETED transition + Arguments.of("workflows-samples/simple-expression.yaml", EventType.TASK_COMPLETED), + // raises an error: FAULTED transition + Arguments.of("workflows-samples/raise-inline.yaml", EventType.TASK_FAULTED)); + } + private static void assertMetadataSeenOnCancel(WorkflowInstance instance) { await().atMost(Duration.ofSeconds(5)).until(() -> listener.cancelledSeen(instance.id())); assertThat(listener.metadataOnCancel(instance.id())) @@ -222,7 +266,15 @@ static class CancelGate implements WorkflowExecutionCompletableListener { /** Keeps the CANCELLED status change of the given instance pending until completed. */ CompletableFuture hold(String instanceId) { - return gates.computeIfAbsent(instanceId, k -> new CompletableFuture<>()); + return hold(instanceId, EventType.WORKFLOW_STATUS_CHANGED); + } + + /** + * Keeps the given lifecycle event of the given instance pending until completed. For {@link + * EventType#WORKFLOW_STATUS_CHANGED} only the change to CANCELLED is held. + */ + CompletableFuture hold(String instanceId, EventType type) { + return gates.computeIfAbsent(key(instanceId, type), k -> new CompletableFuture<>()); } void reset() { @@ -230,13 +282,32 @@ void reset() { gates.clear(); } + private static String key(String instanceId, EventType type) { + return instanceId + '/' + type; + } + + private CompletableFuture gate(WorkflowEvent ev) { + return gates.getOrDefault( + key(ev.workflowContext().instanceData().id(), ev.type()), + CompletableFuture.completedFuture(null)); + } + @Override public CompletableFuture onWorkflowStatusChanged(WorkflowStatusEvent ev) { return ev.status() == WorkflowStatus.CANCELLED - ? gates.getOrDefault( - ev.workflowContext().instanceData().id(), CompletableFuture.completedFuture(null)) + ? gate(ev) : CompletableFuture.completedFuture(null); } + + @Override + public CompletableFuture onTaskCompleted(TaskCompletedEvent ev) { + return gate(ev); + } + + @Override + public CompletableFuture onTaskFailed(TaskFailedEvent ev) { + return gate(ev); + } } static class TrackedMetadata implements AutoCloseable { @@ -253,11 +324,23 @@ static class MetadataListener implements WorkflowExecutionListener { private final Map metadataOnCancel = new ConcurrentHashMap<>(); private final Map closedBeforeCancel = new ConcurrentHashMap<>(); private final Set taskCancelled = ConcurrentHashMap.newKeySet(); + private final Set workflowCompleted = ConcurrentHashMap.newKeySet(); + private final Set workflowFailed = ConcurrentHashMap.newKeySet(); void reset() { metadataOnCancel.clear(); closedBeforeCancel.clear(); taskCancelled.clear(); + workflowCompleted.clear(); + workflowFailed.clear(); + } + + boolean workflowCompletedSeen(String id) { + return workflowCompleted.contains(id); + } + + boolean workflowFailedSeen(String id) { + return workflowFailed.contains(id); } boolean taskCancelledSeen(String id) { @@ -287,6 +370,16 @@ public void onTaskCancelled(TaskCancelledEvent ev) { taskCancelled.add(ev.workflowContext().instanceData().id()); } + @Override + public void onWorkflowCompleted(WorkflowCompletedEvent ev) { + workflowCompleted.add(ev.workflowContext().instanceData().id()); + } + + @Override + public void onWorkflowFailed(WorkflowFailedEvent ev) { + workflowFailed.add(ev.workflowContext().instanceData().id()); + } + @Override public void onWorkflowCancelled(WorkflowCancelledEvent ev) { Optional metadata = From aa1ebc0a61ec5d69043b20d8f7ed15e3ac27d549 Mon Sep 17 00:00:00 2001 From: Francisco Javier Tirado Sarti Date: Thu, 8 Oct 2026 14:09:58 +0200 Subject: [PATCH 4/4] [Fix #1749] Alternative approach Signed-off-by: Francisco Javier Tirado Sarti --- .../impl/LifecycleEventsUtils.java | 9 + .../impl/WorkflowMutableInstance.java | 225 ++++++++---------- .../impl/test/CancelMetadataTest.java | 12 +- 3 files changed, 121 insertions(+), 125 deletions(-) diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/LifecycleEventsUtils.java b/impl/core/src/main/java/io/serverlessworkflow/impl/LifecycleEventsUtils.java index 0cb97c1c1..b20509585 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/LifecycleEventsUtils.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/LifecycleEventsUtils.java @@ -28,6 +28,15 @@ public class LifecycleEventsUtils { private LifecycleEventsUtils() {} + public static CompletableFuture publishEvent( + boolean statusChanged, + WorkflowContext workflowContext, + Function> function) { + return statusChanged + ? publishEvent(workflowContext, function).thenApply(__ -> true) + : CompletableFuture.completedFuture(false); + } + public static CompletableFuture publishEvent( WorkflowContext workflowContext, Function> function) { diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java index 41b122dac..08780a89e 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java @@ -61,9 +61,11 @@ public class WorkflowMutableInstance implements WorkflowInstance { private Lock statusLock = new ReentrantLock(); private Map, TaskContext> suspended; - private Collection> cancelables = new ArrayList<>(); + private Collection> cancelables = + Collections.synchronizedList(new ArrayList<>()); - private volatile CompletableFuture cancelPublished = CompletableFuture.completedFuture(null); + private Collection> outOfOrderListeners = + Collections.synchronizedList(new ArrayList<>()); protected WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) { this.id = id; @@ -107,31 +109,16 @@ protected final CompletableFuture startExecution( .orElse(input)) .whenComplete(this::setCompleteDate) .thenApply(this::filterAndValidate) - .thenCompose(this::publishEvents) + .thenCompose(this::publishCompletionEvents) .exceptionallyCompose(this::handleException)) - .handle(this::cleanUpAfterCancelPublished) + .handle(this::cleanUpWaitingListeners) .thenCompose(Function.identity()); futureRef.set(future); } return future; } - private CompletableFuture cleanUpAfterCancelPublished( - WorkflowModel result, Throwable ex) { - return cancelPublished - .handle( - (__, ___) -> { - cleanUp(result, ex); - return null; - }) - .thenCompose( - __ -> - ex == null - ? CompletableFuture.completedFuture(result) - : CompletableFuture.failedFuture(ex)); - } - - private CompletableFuture publishEvents(WorkflowModel model) { + private CompletableFuture publishCompletionEvents(WorkflowModel model) { return status(WorkflowStatus.COMPLETED) .thenCompose( __ -> @@ -145,19 +132,37 @@ private void setCompleteDate(WorkflowModel result, Throwable ex) { completedAt = Instant.now(); } - private void cleanUp(WorkflowModel result, Throwable ex) { + private CompletableFuture 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 fromResult(WorkflowModel model, Throwable ex) { + return ex == null + ? CompletableFuture.completedFuture(model) + : CompletableFuture.failedFuture(ex); } private CompletableFuture 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( __ -> @@ -166,7 +171,6 @@ private CompletableFuture handleException(Throwable exception) { l -> l.onWorkflowFailed(new WorkflowFailedEvent(workflowContext, cause)))) .thenCompose(__ -> CompletableFuture.failedFuture(exception)); } - return CompletableFuture.failedFuture(exception); } private WorkflowModel filterAndValidate(WorkflowModel model) { @@ -232,29 +236,21 @@ public T outputAs(Class clazz) { : null; } - public CompletableFuture status(WorkflowStatus state) { + public CompletableFuture status(WorkflowStatus newState) { WorkflowStatus prevState; + statusLock.lock(); try { - statusLock.lock(); - prevState = this.status.get(); - if (prevState == WorkflowStatus.CANCELLED) { - // cancellation is final: late callbacks (e.g. an event received by a listen task) must - // not move the instance out of it, and a terminal transition must fail so that neither - // onWorkflowCompleted nor onWorkflowFailed is published for a cancelled instance - return isTerminal(state) - ? CompletableFuture.failedFuture( - new CancellationException("Workflow instance " + id + " has been cancelled")) - : CompletableFuture.completedFuture(false); + prevState = status.get(); + if (prevState == WorkflowStatus.CANCELLED && newState != WorkflowStatus.CANCELLED) { + return CompletableFuture.failedFuture( + new CancellationException("Workflow has been cancelled")); + } else { + status.set(newState); } - this.status.set(state); } finally { statusLock.unlock(); } - return publishStatusChange(prevState, state); - } - - private static boolean isTerminal(WorkflowStatus state) { - return state == WorkflowStatus.COMPLETED || state == WorkflowStatus.FAULTED; + return publishStatusChange(prevState, newState); } protected final void setStatus(WorkflowStatus state) { @@ -263,14 +259,10 @@ protected final void setStatus(WorkflowStatus state) { private CompletableFuture 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 @@ -291,12 +283,15 @@ 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; } @@ -304,15 +299,14 @@ public boolean suspend() { @Override public CompletableFuture 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() { @@ -338,12 +332,14 @@ 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; } @@ -351,15 +347,14 @@ public boolean resume() { @Override public CompletableFuture 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() { @@ -382,13 +377,12 @@ private WorkflowStatus internalResume() { return result; } - public CompletableFuture cancelCheck(TaskContext t) { + public CompletableFuture cancelCheck(T t) { try { statusLock.lock(); if (status.get() == WorkflowStatus.CANCELLED) { - CompletableFuture cancelled = new CompletableFuture(); - cancelled.completeExceptionally( - new CancellationException("Task " + t.taskName() + " has been cancelled")); + CompletableFuture cancelled = new CompletableFuture<>(); + cancelled.completeExceptionally(new CancellationException(t + " has been cancelled")); return cancelled; } } finally { @@ -419,68 +413,53 @@ public CompletableFuture suspendedCheck(TaskContext t) { @Override public boolean cancel() { - CancelRequest request = internalCancel(); - if (request.accepted()) { - publishCancelled(request); - } - return request.accepted(); + CompletableFuture result = internalCancel(); + return !result.isDone() || result.join(); } @Override public CompletableFuture cancelFuture() { - CancelRequest request = internalCancel(); - return request.accepted() - ? publishCancelled(request).thenApply(__ -> true) - : CompletableFuture.completedFuture(false); - } - - /** - * Publishes the cancellation and only then cancels the pending futures. Cancelling them may - * complete the execution pipeline synchronously, and the pipeline may also end on its own while - * the cancellation is being published, so {@link #cleanUpAfterCancelPublished} waits for {@link - * CancelRequest#published()} before clearing the instance metadata. - */ - private CompletableFuture publishCancelled(CancelRequest request) { - return publishStatusChange(request.prevStatus(), WorkflowStatus.CANCELLED) - .thenCompose( - __ -> - publishEvent( - workflowContext, - l -> l.onWorkflowCancelled(new WorkflowCancelledEvent(workflowContext)))) - .whenComplete( - (__, ex) -> { - request.published().complete(null); - request.toCancel().forEach(t -> t.cancel(true)); - }); - } - - private record CancelRequest( - WorkflowStatus prevStatus, - Collection> toCancel, - CompletableFuture published) { - boolean accepted() { - return prevStatus != WorkflowStatus.CANCELLED; - } + return internalCancel(); } - private CancelRequest internalCancel() { + private CompletableFuture internalCancel() { + WorkflowStatus prevState; + Collection> toCancel = null; try { statusLock.lock(); if (TaskExecutorHelper.isActive(status.get())) { - Collection> toCancel = new ArrayList<>(cancelables); + toCancel = new ArrayList<>(cancelables); cancelables.clear(); - CompletableFuture published = new CompletableFuture<>(); - cancelPublished = published; - return new CancelRequest(status.getAndSet(WorkflowStatus.CANCELLED), toCancel, published); + prevState = status.getAndSet(WorkflowStatus.CANCELLED); } else { - return new CancelRequest( - WorkflowStatus.CANCELLED, - Collections.emptyList(), - CompletableFuture.completedFuture(null)); + prevState = WorkflowStatus.CANCELLED; } } finally { statusLock.unlock(); } + CompletableFuture 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 outOfOrder(CompletableFuture future) { + if (!future.isDone()) { + outOfOrderListeners.add(future); + } + return future; } public void addCancelable(CompletableFuture cancelable) { diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java index 3db834bd6..737479a66 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java @@ -160,15 +160,23 @@ void metadataIsAvailableWhenPipelineEndsWhileCancellationIsPublished( .atMost(Duration.ofSeconds(5)) .pollInterval(Duration.ofMillis(5)) .until(() -> instance.status() == WorkflowStatus.WAITING); + + // Hold onWorkflowCancelled callback to delay its execution CompletableFuture gate = cancelGate.hold(instance.id()); + // Cancel: sets CANCELLED status and immediately cancels listen futures instance.cancel(); + + // For listen tasks: emit event but subscription is already cancelled (intentional) + // For wait tasks: no event to emit if (event != null) { - // completes the listen task emitDefinition.instance(event).start().join(); } - // the pipeline has seen the cancellation while onWorkflowCancelled is still pending + + // Pipeline ends due to forced cancellation while onWorkflowCancelled is pending await().atMost(Duration.ofSeconds(5)).until(() -> listener.taskCancelledSeen(instance.id())); + + // Release gate to execute onWorkflowCancelled callback gate.complete(null); assertThat(future)