Skip to content

Commit ebbcd9c

Browse files
mdproctorclaude
andcommitted
fix: populate instanceData().output() before onWorkflowCompleted fires
For synchronous workflows the CompletableFuture chain inside startExecution() completes inline on the calling thread. futureRef is assigned only after startExecution() returns, so instanceData().output() — which routes through futureRef.get().join() — returned null when called from inside the onWorkflowCompleted hook. Fix: store the computed output in a volatile field inside whenSuccess(), which runs before the completion event is published. output() now reads that field first, falling back to futureRef for callers who retrieve output after start() returns. Adds WorkflowExecutionListenerOutputTest with two assertions: - instanceData().output() is non-null inside the hook - instanceData().output() equals event.output() (both carry same value) Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
1 parent 3dfd814 commit ebbcd9c

2 files changed

Lines changed: 112 additions & 0 deletions

File tree

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,8 @@ public class WorkflowMutableInstance implements WorkflowInstance {
5656
private Lock statusLock = new ReentrantLock();
5757
private Map<CompletableFuture<TaskContext>, TaskContext> suspended;
5858

59+
private volatile WorkflowModel completedOutput;
60+
5961
private Collection<CompletableFuture<?>> cancelables = new ArrayList<>();
6062

6163
protected WorkflowMutableInstance(WorkflowDefinition definition, String id, WorkflowModel input) {
@@ -140,6 +142,10 @@ private WorkflowModel whenSuccess(WorkflowModel node) {
140142
.map(f -> f.apply(workflowContext, null, node))
141143
.orElse(node);
142144
workflowContext.definition().outputSchemaValidator().ifPresent(v -> v.validate(output));
145+
// Store before onWorkflowCompleted fires so instanceData().output() is available in the hook.
146+
// futureRef is set after startExecution() returns, which is too late for synchronous
147+
// workflows where the CompletableFuture chain completes inline on the calling thread.
148+
this.completedOutput = output;
143149
status(WorkflowStatus.COMPLETED);
144150
return output;
145151
}
@@ -176,6 +182,9 @@ public WorkflowModel context() {
176182

177183
@Override
178184
public WorkflowModel output() {
185+
if (completedOutput != null) {
186+
return completedOutput;
187+
}
179188
CompletableFuture<WorkflowModel> future = futureRef.get();
180189
return future != null ? future.join() : null;
181190
}
Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
/*
2+
* Copyright 2020-Present The Serverless Workflow Specification Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.serverlessworkflow.impl.test;
17+
18+
import static io.serverlessworkflow.api.WorkflowReader.readWorkflowFromClasspath;
19+
import static org.assertj.core.api.Assertions.assertThat;
20+
21+
import io.serverlessworkflow.impl.WorkflowApplication;
22+
import io.serverlessworkflow.impl.WorkflowModel;
23+
import io.serverlessworkflow.impl.lifecycle.WorkflowCompletedEvent;
24+
import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener;
25+
import java.io.IOException;
26+
import java.util.Map;
27+
import java.util.concurrent.atomic.AtomicReference;
28+
import org.junit.jupiter.api.Test;
29+
30+
/**
31+
* Verifies that {@code instanceData().output()} is populated inside {@code onWorkflowCompleted}.
32+
*
33+
* <p>For synchronous workflows the {@code CompletableFuture} chain in {@code
34+
* WorkflowMutableInstance.startExecution()} completes inline on the calling thread. {@code
35+
* futureRef} is set only after the chain returns, so {@code instanceData().output()} — which routes
36+
* through {@code futureRef.get().join()} — was returning {@code null} when called from inside the
37+
* hook. The fix stores the output in {@code whenSuccess()} before the completion event fires.
38+
*/
39+
class WorkflowExecutionListenerOutputTest {
40+
41+
@Test
42+
void instanceDataOutputIsAvailableInsideOnWorkflowCompleted() throws IOException {
43+
AtomicReference<WorkflowModel> capturedOutput = new AtomicReference<>();
44+
45+
WorkflowExecutionListener listener =
46+
new WorkflowExecutionListener() {
47+
@Override
48+
public void onWorkflowCompleted(WorkflowCompletedEvent event) {
49+
capturedOutput.set(event.workflowContext().instanceData().output());
50+
}
51+
};
52+
53+
try (WorkflowApplication app = WorkflowApplication.builder().withListener(listener).build()) {
54+
WorkflowModel result =
55+
app.workflowDefinition(
56+
readWorkflowFromClasspath("workflows-samples/simple-expression.yaml"))
57+
.instance(Map.of())
58+
.start()
59+
.join();
60+
61+
assertThat(result).isNotNull();
62+
assertThat(capturedOutput.get())
63+
.as(
64+
"instanceData().output() must be non-null inside onWorkflowCompleted — "
65+
+ "it was null because futureRef is set after startExecution() returns, "
66+
+ "which is after the synchronous CompletableFuture chain completes")
67+
.isNotNull();
68+
assertThat(capturedOutput.get().asMap())
69+
.as("instanceData().output() inside the hook must equal the workflow's final output")
70+
.isEqualTo(result.asMap());
71+
}
72+
}
73+
74+
@Test
75+
void eventOutputMatchesInstanceDataOutput() throws IOException {
76+
AtomicReference<WorkflowModel> eventOutput = new AtomicReference<>();
77+
AtomicReference<WorkflowModel> instanceOutput = new AtomicReference<>();
78+
79+
WorkflowExecutionListener listener =
80+
new WorkflowExecutionListener() {
81+
@Override
82+
public void onWorkflowCompleted(WorkflowCompletedEvent event) {
83+
eventOutput.set(event.output());
84+
instanceOutput.set(event.workflowContext().instanceData().output());
85+
}
86+
};
87+
88+
try (WorkflowApplication app = WorkflowApplication.builder().withListener(listener).build()) {
89+
app.workflowDefinition(readWorkflowFromClasspath("workflows-samples/simple-expression.yaml"))
90+
.instance(Map.of())
91+
.start()
92+
.join();
93+
94+
assertThat(eventOutput.get())
95+
.as("event.output() must be non-null — it is passed directly to WorkflowCompletedEvent")
96+
.isNotNull();
97+
assertThat(instanceOutput.get())
98+
.as(
99+
"instanceData().output() must equal event.output() — both should carry the same value")
100+
.isEqualTo(eventOutput.get());
101+
}
102+
}
103+
}

0 commit comments

Comments
 (0)