Skip to content

Commit 8085c7b

Browse files
committed
[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>
1 parent 0ebc701 commit 8085c7b

2 files changed

Lines changed: 107 additions & 6 deletions

File tree

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

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -238,9 +238,13 @@ public CompletableFuture<Boolean> status(WorkflowStatus state) {
238238
statusLock.lock();
239239
prevState = this.status.get();
240240
if (prevState == WorkflowStatus.CANCELLED) {
241-
// cancellation is final, late callbacks (e.g. an event received by a listen task) must
242-
// not move the instance out of it
243-
return CompletableFuture.completedFuture(false);
241+
// cancellation is final: late callbacks (e.g. an event received by a listen task) must
242+
// not move the instance out of it, and a terminal transition must fail so that neither
243+
// onWorkflowCompleted nor onWorkflowFailed is published for a cancelled instance
244+
return isTerminal(state)
245+
? CompletableFuture.failedFuture(
246+
new CancellationException("Workflow instance " + id + " has been cancelled"))
247+
: CompletableFuture.completedFuture(false);
244248
}
245249
this.status.set(state);
246250
} finally {
@@ -249,6 +253,10 @@ public CompletableFuture<Boolean> status(WorkflowStatus state) {
249253
return publishStatusChange(prevState, state);
250254
}
251255

256+
private static boolean isTerminal(WorkflowStatus state) {
257+
return state == WorkflowStatus.COMPLETED || state == WorkflowStatus.FAULTED;
258+
}
259+
252260
protected final void setStatus(WorkflowStatus state) {
253261
this.status.set(state);
254262
}

‎impl/test/src/test/java/io/serverlessworkflow/impl/test/CancelMetadataTest.java‎

Lines changed: 96 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,17 +24,24 @@
2424
import io.serverlessworkflow.impl.WorkflowInstance;
2525
import io.serverlessworkflow.impl.WorkflowModel;
2626
import io.serverlessworkflow.impl.WorkflowStatus;
27+
import io.serverlessworkflow.impl.lifecycle.EventType;
2728
import io.serverlessworkflow.impl.lifecycle.TaskCancelledEvent;
29+
import io.serverlessworkflow.impl.lifecycle.TaskCompletedEvent;
30+
import io.serverlessworkflow.impl.lifecycle.TaskFailedEvent;
2831
import io.serverlessworkflow.impl.lifecycle.WorkflowCancelledEvent;
32+
import io.serverlessworkflow.impl.lifecycle.WorkflowCompletedEvent;
33+
import io.serverlessworkflow.impl.lifecycle.WorkflowEvent;
2934
import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionCompletableListener;
3035
import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener;
36+
import io.serverlessworkflow.impl.lifecycle.WorkflowFailedEvent;
3137
import io.serverlessworkflow.impl.lifecycle.WorkflowStartedEvent;
3238
import io.serverlessworkflow.impl.lifecycle.WorkflowStatusEvent;
3339
import java.io.IOException;
3440
import java.time.Duration;
3541
import java.util.Map;
3642
import java.util.Optional;
3743
import java.util.Set;
44+
import java.util.concurrent.CancellationException;
3845
import java.util.concurrent.CompletableFuture;
3946
import java.util.concurrent.ConcurrentHashMap;
4047
import java.util.concurrent.ExecutionException;
@@ -180,6 +187,43 @@ private static Stream<Arguments> pipelineEndsWhileCancellationIsPublishedParamet
180187
Arguments.of("workflows-samples/wait-set.yaml", Map.of(), null));
181188
}
182189

190+
@ParameterizedTest(name = "{0}")
191+
@MethodSource("terminalTransitionWhileCancellingParameters")
192+
void cancellationWhileLastTaskEventIsPendingWins(String workflowPath, EventType heldTaskEvent)
193+
throws IOException {
194+
WorkflowDefinition definition =
195+
appl.workflowDefinition(WorkflowReader.readWorkflowFromClasspath(workflowPath));
196+
WorkflowInstance instance = definition.instance(Map.of());
197+
CompletableFuture<Void> gate = cancelGate.hold(instance.id(), heldTaskEvent);
198+
CompletableFuture<WorkflowModel> future = instance.start();
199+
assertThat(future).isNotDone();
200+
201+
assertThat(instance.cancel()).isTrue();
202+
await().atMost(Duration.ofSeconds(5)).until(() -> listener.cancelledSeen(instance.id()));
203+
// lets the last task end and the workflow reach its terminal transition
204+
gate.complete(null);
205+
206+
assertThat(future)
207+
.failsWithin(Duration.ofSeconds(5))
208+
.withThrowableOfType(ExecutionException.class)
209+
.withCauseInstanceOf(CancellationException.class);
210+
assertThat(instance.status()).isEqualTo(WorkflowStatus.CANCELLED);
211+
assertThat(listener.workflowCompletedSeen(instance.id()))
212+
.as("onWorkflowCompleted published for a cancelled instance")
213+
.isFalse();
214+
assertThat(listener.workflowFailedSeen(instance.id()))
215+
.as("onWorkflowFailed published for a cancelled instance")
216+
.isFalse();
217+
}
218+
219+
private static Stream<Arguments> terminalTransitionWhileCancellingParameters() {
220+
return Stream.of(
221+
// completes normally: COMPLETED transition
222+
Arguments.of("workflows-samples/simple-expression.yaml", EventType.TASK_COMPLETED),
223+
// raises an error: FAULTED transition
224+
Arguments.of("workflows-samples/raise-inline.yaml", EventType.TASK_FAULTED));
225+
}
226+
183227
private static void assertMetadataSeenOnCancel(WorkflowInstance instance) {
184228
await().atMost(Duration.ofSeconds(5)).until(() -> listener.cancelledSeen(instance.id()));
185229
assertThat(listener.metadataOnCancel(instance.id()))
@@ -222,21 +266,48 @@ static class CancelGate implements WorkflowExecutionCompletableListener {
222266

223267
/** Keeps the CANCELLED status change of the given instance pending until completed. */
224268
CompletableFuture<Void> hold(String instanceId) {
225-
return gates.computeIfAbsent(instanceId, k -> new CompletableFuture<>());
269+
return hold(instanceId, EventType.WORKFLOW_STATUS_CHANGED);
270+
}
271+
272+
/**
273+
* Keeps the given lifecycle event of the given instance pending until completed. For {@link
274+
* EventType#WORKFLOW_STATUS_CHANGED} only the change to CANCELLED is held.
275+
*/
276+
CompletableFuture<Void> hold(String instanceId, EventType type) {
277+
return gates.computeIfAbsent(key(instanceId, type), k -> new CompletableFuture<>());
226278
}
227279

228280
void reset() {
229281
gates.values().forEach(g -> g.complete(null));
230282
gates.clear();
231283
}
232284

285+
private static String key(String instanceId, EventType type) {
286+
return instanceId + '/' + type;
287+
}
288+
289+
private CompletableFuture<?> gate(WorkflowEvent ev) {
290+
return gates.getOrDefault(
291+
key(ev.workflowContext().instanceData().id(), ev.type()),
292+
CompletableFuture.completedFuture(null));
293+
}
294+
233295
@Override
234296
public CompletableFuture<?> onWorkflowStatusChanged(WorkflowStatusEvent ev) {
235297
return ev.status() == WorkflowStatus.CANCELLED
236-
? gates.getOrDefault(
237-
ev.workflowContext().instanceData().id(), CompletableFuture.completedFuture(null))
298+
? gate(ev)
238299
: CompletableFuture.completedFuture(null);
239300
}
301+
302+
@Override
303+
public CompletableFuture<?> onTaskCompleted(TaskCompletedEvent ev) {
304+
return gate(ev);
305+
}
306+
307+
@Override
308+
public CompletableFuture<?> onTaskFailed(TaskFailedEvent ev) {
309+
return gate(ev);
310+
}
240311
}
241312

242313
static class TrackedMetadata implements AutoCloseable {
@@ -253,11 +324,23 @@ static class MetadataListener implements WorkflowExecutionListener {
253324
private final Map<String, Boolean> metadataOnCancel = new ConcurrentHashMap<>();
254325
private final Map<String, Boolean> closedBeforeCancel = new ConcurrentHashMap<>();
255326
private final Set<String> taskCancelled = ConcurrentHashMap.newKeySet();
327+
private final Set<String> workflowCompleted = ConcurrentHashMap.newKeySet();
328+
private final Set<String> workflowFailed = ConcurrentHashMap.newKeySet();
256329

257330
void reset() {
258331
metadataOnCancel.clear();
259332
closedBeforeCancel.clear();
260333
taskCancelled.clear();
334+
workflowCompleted.clear();
335+
workflowFailed.clear();
336+
}
337+
338+
boolean workflowCompletedSeen(String id) {
339+
return workflowCompleted.contains(id);
340+
}
341+
342+
boolean workflowFailedSeen(String id) {
343+
return workflowFailed.contains(id);
261344
}
262345

263346
boolean taskCancelledSeen(String id) {
@@ -287,6 +370,16 @@ public void onTaskCancelled(TaskCancelledEvent ev) {
287370
taskCancelled.add(ev.workflowContext().instanceData().id());
288371
}
289372

373+
@Override
374+
public void onWorkflowCompleted(WorkflowCompletedEvent ev) {
375+
workflowCompleted.add(ev.workflowContext().instanceData().id());
376+
}
377+
378+
@Override
379+
public void onWorkflowFailed(WorkflowFailedEvent ev) {
380+
workflowFailed.add(ev.workflowContext().instanceData().id());
381+
}
382+
290383
@Override
291384
public void onWorkflowCancelled(WorkflowCancelledEvent ev) {
292385
Optional<TrackedMetadata> metadata =

0 commit comments

Comments
 (0)