Skip to content

Fix #1354 - Isolate variables in ForExecutor to avoid racing condition to overwrite loop variables #1363

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

Open
ricardozanini wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Fix #1354 - Isolate variables in ForExecutor to avoid racing condition to overwrite loop variables #1363

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
73f122e
Isolate for loop variables in completable future
ricardozanini May 1, 2026
e2e3746
Fix #1354 - Isolate variables in ForExecutor to avoid racing conditio…
ricardozanini May 2, 2026
7c2549b
Minor formatting
ricardozanini May 2, 2026
2e95a6c
Apply suggestions from code review
ricardozanini May 4, 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
13 changes: 7 additions & 6 deletions experimental/test/src/test/java/io/serverlessworkflow/fluent/test/ForEachFuncTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
import io.serverlessworkflow.impl.WorkflowApplication;
import io.serverlessworkflow.impl.WorkflowModel;
import io.serverlessworkflow.impl.events.InMemoryEvents;
import io.serverlessworkflow.impl.lifecycle.TraceExecutionListener;
import java.time.Duration;
import java.util.Collection;
import java.util.List;
Expand All @@ -35,13 +35,13 @@

public class ForEachFuncTest {

private static record Order(String id) {}
private record Order(String id) {}

private static record EnhancedOrder(String id, int salary) {}
private record EnhancedOrder(String id, int salary) {}

private static record OrdersPayload(List<Order> orders) {}
private record OrdersPayload(List<Order> orders) {}

private static record OrderName(String id, String name) {}
private record OrderName(String id, String name) {}

@Test
void testForEachIteration() {
Expand Down Expand Up @@ -75,13 +75,14 @@ void testForEachEmit() {
.build();

List<CloudEvent> publishedEvents = new CopyOnWriteArrayList<>();
InMemoryEvents eventBroker = new InMemoryEvents();
LaggedInMemoryEvents eventBroker = new LaggedInMemoryEvents();
eventBroker.register(eventType, ce -> publishedEvents.add(ce));

try (WorkflowApplication app =
WorkflowApplication.builder()
.withEventConsumer(eventBroker)
.withEventPublisher(eventBroker)
.withListener(new TraceExecutionListener())
.build()) {
app.workflowDefinition(workflow)
.instance(
Expand Down
51 changes: 51 additions & 0 deletions experimental/test/src/test/java/io/serverlessworkflow/fluent/test/LaggedInMemoryEvents.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* 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.fluent.test;

import io.cloudevents.CloudEvent;
import io.serverlessworkflow.impl.events.InMemoryEvents;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class LaggedInMemoryEvents extends InMemoryEvents {

private static final Logger logger = LoggerFactory.getLogger(LaggedInMemoryEvents.class);

@Override
public CompletableFuture<Void> publish(CloudEvent ce) {
return CompletableFuture.runAsync(
() -> {
Consumer<CloudEvent> allConsumer = allConsumerRef.get();
if (allConsumer != null) {
allConsumer.accept(ce);
}
Consumer<CloudEvent> consumer = topicMap.get(ce.getType());
if (consumer != null) {
consumer.accept(ce);
}
try {
Thread.sleep(10);
} catch (InterruptedException e) {
Comment thread
ricardozanini marked this conversation as resolved.
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
logger.info("Accepted event {} for topic {}", ce.getId(), ce.getType());
},
serviceFactory.get());
}
}
10 changes: 4 additions & 6 deletions impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,10 @@
*/
public class InMemoryEvents extends AbstractTypeConsumer implements EventPublisher {

protected final ExecutorServiceFactory serviceFactory;
protected final Map<String, Consumer<CloudEvent>> topicMap = new ConcurrentHashMap<>();
protected final AtomicReference<Consumer<CloudEvent>> allConsumerRef = new AtomicReference<>();

public InMemoryEvents() {
this(new DefaultExecutorServiceFactory());
}
Expand All @@ -38,12 +42,6 @@ public InMemoryEvents(ExecutorServiceFactory serviceFactory) {
this.serviceFactory = serviceFactory;
}

private ExecutorServiceFactory serviceFactory;

private Map<String, Consumer<CloudEvent>> topicMap = new ConcurrentHashMap<>();

private AtomicReference<Consumer<CloudEvent>> allConsumerRef = new AtomicReference<>();

@Override
public void register(String topicName, Consumer<CloudEvent> consumer) {
topicMap.put(topicName, consumer);
Expand Down
16 changes: 11 additions & 5 deletions impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,14 +80,20 @@ protected CompletableFuture<WorkflowModel> internalExecute(
CompletableFuture<WorkflowModel> future =
CompletableFuture.completedFuture(taskContext.input());
while (iter.hasNext()) {
taskContext.variables().put(task.getFor().getEach(), iter.next());
taskContext.variables().put(task.getFor().getAt(), i++);
final Object currentItem = iter.next();
final int currentIndex = i++;
taskContext.variables().put(task.getFor().getEach(), currentItem);
taskContext.variables().put(task.getFor().getAt(), currentIndex);

Comment on lines +83 to +87
Copy link
Copy Markdown
Member Author

@ricardozanini ricardozanini May 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is safe because thenCompose chains the futures sequentially - each iteration's future completes only after processTaskList and all its subtasks (including parallel branches) finish. The next iteration can't start until the previous one completes, so even though different threads may write to the HashMap, they never overlap. The final locals ensure each lambda writes its own captured value right before spawning subtasks, establishing a happens-before relationship that guarantees visibility.

if (whileExpr.map(w -> w.test(workflow, taskContext, taskContext.input())).orElse(true)) {
future =
future.thenCompose(
input ->
TaskExecutorHelper.processTaskList(
taskExecutor, workflow, Optional.of(taskContext), input));
input -> {
taskContext.variables().put(task.getFor().getEach(), currentItem);
taskContext.variables().put(task.getFor().getAt(), currentIndex);
return TaskExecutorHelper.processTaskList(
taskExecutor, workflow, Optional.of(taskContext), input);
});
} else {
break;
}
Expand Down
13 changes: 1 addition & 12 deletions ...low/impl/test/TraceExecutionListener.java → ...mpl/lifecycle/TraceExecutionListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,19 +13,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.serverlessworkflow.impl.test;
package io.serverlessworkflow.impl.lifecycle;

import io.serverlessworkflow.impl.lifecycle.TaskCompletedEvent;
import io.serverlessworkflow.impl.lifecycle.TaskFailedEvent;
import io.serverlessworkflow.impl.lifecycle.TaskRetriedEvent;
import io.serverlessworkflow.impl.lifecycle.TaskStartedEvent;
import io.serverlessworkflow.impl.lifecycle.WorkflowCompletedEvent;
import io.serverlessworkflow.impl.lifecycle.WorkflowExecutionListener;
import io.serverlessworkflow.impl.lifecycle.WorkflowFailedEvent;
import io.serverlessworkflow.impl.lifecycle.WorkflowResumedEvent;
import io.serverlessworkflow.impl.lifecycle.WorkflowStartedEvent;
import io.serverlessworkflow.impl.lifecycle.WorkflowStatusEvent;
import io.serverlessworkflow.impl.lifecycle.WorkflowSuspendedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down
1 change: 1 addition & 0 deletions impl/test/src/test/java/io/serverlessworkflow/impl/test/DBGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import io.serverlessworkflow.impl.WorkflowApplication;
import io.serverlessworkflow.impl.WorkflowDefinition;
import io.serverlessworkflow.impl.WorkflowInstance;
import io.serverlessworkflow.impl.lifecycle.TraceExecutionListener;
import io.serverlessworkflow.impl.persistence.DefaultPersistenceInstanceHandlers;
import io.serverlessworkflow.impl.persistence.PersistenceApplicationBuilder;
import io.serverlessworkflow.impl.persistence.PersistenceInstanceHandlers;
Expand Down
1 change: 1 addition & 0 deletions impl/test/src/test/java/io/serverlessworkflow/impl/test/MvStorePersistenceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import io.serverlessworkflow.impl.WorkflowDefinition;
import io.serverlessworkflow.impl.WorkflowInstance;
import io.serverlessworkflow.impl.WorkflowStatus;
import io.serverlessworkflow.impl.lifecycle.TraceExecutionListener;
import io.serverlessworkflow.impl.persistence.DefaultPersistenceInstanceHandlers;
import io.serverlessworkflow.impl.persistence.PersistenceApplicationBuilder;
import io.serverlessworkflow.impl.persistence.PersistenceInstanceHandlers;
Expand Down
Loading