Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fjtirado merged 4 commits into from
Oct 8, 2026
+515 −72
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
098ced2
[Fix #1749] Publish onWorkflowCancelled before cancelling pending fut…
edeandrea Oct 7, 2026
0ebc701
[Fix #1749] Keep cancellation final and clean up after it is published
edeandrea Oct 7, 2026
8085c7b
[Fix #1749] Fail terminal transitions of a cancelled instance
edeandrea Oct 7, 2026
aa1ebc0
[Fix #1749] Alternative approach
fjtirado Oct 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions impl/core/src/main/java/io/serverlessworkflow/impl/LifecycleEventsUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
178 changes: 106 additions & 72 deletions impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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;
Expand Down Expand Up @@ -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(
__ ->
Expand All @@ -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(
__ ->
Expand All @@ -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) {
Expand Down Expand Up @@ -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) {
Expand All @@ -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
Expand All @@ -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() {
Expand All @@ -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() {
Expand All @@ -343,13 +377,12 @@ private WorkflowStatus internalResume() {
return result;
}

public CompletableFuture<TaskContext> cancelCheck(TaskContext t) {
public <T> CompletableFuture<T> cancelCheck(T t) {
Comment thread
fjtirado marked this conversation as resolved.
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 {
Expand Down Expand Up @@ -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);
}
Comment thread
fjtirado marked this conversation as resolved.
return future;
}

public void addCancelable(CompletableFuture<?> cancelable) {
statusLock.lock();
if (status.get() == WorkflowStatus.CANCELLED) {
Expand Down
Loading
Loading