Description
The ForExecutor has a race condition bug when iterating over collections with >2 items and executing async tasks (like emit). The shared TaskContext causes variables to be overwritten before async tasks execute, resulting in wrong items being processed.
Evidence
Observed Behavior
When using forEach to iterate over a collection of 3 items and emit events:
Input:
List.of("lint", "test", "build")Expected: 3 events emitted for: lint, test, build
Actual: 3 events emitted for: lint, build, build (skips "test", duplicates "build")
Code Analysis
Location: ForExecutor.java lines 82-90
CompletableFuture<WorkflowModel> future = CompletableFuture.completedFuture(taskContext.input());
while (iter.hasNext()) {
taskContext.variables().put(task.getFor().getEach(), iter.next()); // Line 83: Sets $item
taskContext.variables().put(task.getFor().getAt(), i++); // Line 84: Sets $index
if (whileExpr.map(w -> w.test(workflow, taskContext, taskContext.input())).orElse(true)) {
future = future.thenCompose(
input -> TaskExecutorHelper.processTaskList(
taskExecutor, workflow, Optional.of(taskContext), input)); // Line 90: Uses SAME taskContext
}
}Root Cause:
- The
taskContext object is shared across all loop iterations
- Line 83 sets the loop variable (
$item) in the shared context
- Line 90 schedules an async task that captures a reference to
taskContext
- The while loop continues and overwrites the variable with the next item before the async task reads it
- Result: Async tasks may read the wrong item value (typically the last item in the collection)
Reproduction
Workflow Definition
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.*;
import java.util.Collection;
import java.util.List;
public Workflow reproducer() {
return workflow("forEach-bug-reproducer")
.tasks(
// Create a collection of 3 items
function("createCollection", () -> List.of(
Map.of("id", "item-1", "name", "first"),
Map.of("id", "item-2", "name", "second"),
Map.of("id", "item-3", "name", "third")
)),
// ForEach should emit 3 events, one per item
forEach(
(Collection<Map<String, String>> items) -> items,
emitJson("test.item.emitted", Map.class)
.inputFrom("$item")
)
)
.build();
}Expected Result
3 CloudEvents emitted with data:
{"id": "item-1", "name": "first"}{"id": "item-2", "name": "second"}{"id": "item-3", "name": "third"}
Actual Result
3 CloudEvents emitted with data:
{"id": "item-1", "name": "first"} ✓{"id": "item-3", "name": "third"} ✗ (should be item-2){"id": "item-3", "name": "third"} ✗ (duplicate)
Why Existing Tests Don't Catch This
The existing test in ForTaskFunctionRegressionTest.java uses synchronous CallJava operations:
.withDo(List.of(new TaskItem("javaCall",
new Task().withCallTask(
new CallTaskJava(CallJava.loopFunction(CallTest::sum, forConfig.getEach()))))))These complete before the loop advances, so the race condition doesn't manifest. The bug only appears with async tasks like emit, listen, or other operations that schedule work via CompletableFuture.
Suggested Fix
The TaskContext should be cloned or isolated per iteration:
while (iter.hasNext()) {
Object currentItem = iter.next();
int currentIndex = i++;
// Create isolated context for this iteration
TaskContext iterationContext = taskContext.copy(); // or new TaskContext(...)
iterationContext.variables().put(task.getFor().getEach(), currentItem);
iterationContext.variables().put(task.getFor().getAt(), currentIndex);
if (whileExpr.map(w -> w.test(workflow, iterationContext, taskContext.input())).orElse(true)) {
future = future.thenCompose(
input -> TaskExecutorHelper.processTaskList(
taskExecutor, workflow, Optional.of(iterationContext), input));
}
}Impact
This bug affects any workflow using:
forEach with >2 items- Async tasks within the loop body (emit, listen, etc.)
- Results in data corruption, skipped items, and duplicates
Environment
- SDK Version: Latest main branch
- Observed in: Quarkus Flow integration with Reactive Messaging (async emit)
- File:
impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForExecutor.java
Description
The
ForExecutorhas a race condition bug when iterating over collections with >2 items and executing async tasks (likeemit). The sharedTaskContextcauses variables to be overwritten before async tasks execute, resulting in wrong items being processed.Evidence
Observed Behavior
When using
forEachto iterate over a collection of 3 items and emit events:Input:
Expected: 3 events emitted for:
lint,test,buildActual: 3 events emitted for:
lint,build,build(skips "test", duplicates "build")Code Analysis
Location:
ForExecutor.javalines 82-90Root Cause:
taskContextobject is shared across all loop iterations$item) in the shared contexttaskContextReproduction
Workflow Definition
Expected Result
3 CloudEvents emitted with data:
{"id": "item-1", "name": "first"}{"id": "item-2", "name": "second"}{"id": "item-3", "name": "third"}Actual Result
3 CloudEvents emitted with data:
{"id": "item-1", "name": "first"}✓{"id": "item-3", "name": "third"}✗ (should be item-2){"id": "item-3", "name": "third"}✗ (duplicate)Why Existing Tests Don't Catch This
The existing test in
ForTaskFunctionRegressionTest.javauses synchronousCallJavaoperations:These complete before the loop advances, so the race condition doesn't manifest. The bug only appears with async tasks like
emit,listen, or other operations that schedule work viaCompletableFuture.Suggested Fix
The
TaskContextshould be cloned or isolated per iteration:Impact
This bug affects any workflow using:
forEachwith >2 itemsEnvironment
impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForExecutor.java