Repository: kestra-io/kestra
Issue: Multiple condition on flow trigger does not handle consistently the flow state
Labels: bug

### Expected Behavior

I have a flow that need to have two flows dependencies in success, to run.
More precisely: 
- Given flow A and B in success state during the last time window, flow C will be trigger and run
- Given flow A in error and flow B in success state during the last time window, flow C will **not** be trigger and will not run
- Given flow A and flow B in error state during the last minute, flow C will **not** be trigger and will not run

### Actual Behaviour

When flow A is in error state and flow B is in success state, flow C will run when it should not.

### Steps To Reproduce

In the same time window configured in flow C:
1. Run flow A that will fail
2. Run flow B that will succeed
3. Check that flow C has been executed when it should not

### Environment Information

- Kestra Version: 0.9.8

dep graph:
![image](https://github.com/kestra-io/kestra/assets/113910656/f387eba5-bdfd-4b87-8602-6f2a57dfc92a)


### Example flow

Time window for testing: 1 minute
```yaml
id: trigger_test
namespace: my.namespace

tasks:
  - id: test
    type: io.kestra.core.tasks.log.Log
    message: "OK"

triggers:
  - id: multiple-listen-flow
    type: io.kestra.core.models.triggers.types.Flow
    conditions:
      - type: io.kestra.core.models.conditions.types.ExecutionStatusCondition
        in:
          - SUCCESS
      - id: multiple
        type: io.kestra.core.models.conditions.types.MultipleCondition
        window: PT1M
        windowAdvance: PT0S
        conditions:
          flow_a:
            type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
            namespace: my.namespace
            flowId: flow_a
          flow_b:
            type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
            namespace: my.namespace
            flowId: flow_b

```

flow A:
```yaml
id: flow_a
namespace: my.namespace

tasks:
  - id: test
    type: io.kestra.core.tasks.log.Log
    message: "OK"
  - id: fail
    type: io.kestra.core.tasks.executions.Fail
```

flow B:
```yaml
id: flow_b
namespace: my.namespace

tasks:
  - id: test
    type: io.kestra.core.tasks.log.Log
    message: "OK"
  #- id: fail
  #  type: io.kestra.core.tasks.executions.Fail
```

Issue comments:

Pull request review comments:
I don't see the point of having a type parameter here, we usually inject the interface and depending on the runner and implementation is choosed.
Please choose between field injection or constructor injection but don't use both.
Shouldn't this on be moved to the new AbstractFlowTriggerService?
I think this method is only used on the new AbstractFlowTriggerService, if so, move it there
It seems that all runner have a multiple condition storage so why is it optional?
That's something I forgot to remove as before the save & delete methods weren't in the interface so I had to have access to the impl directly to be able to do such action and that was preventing the need to cast
Then I'll have to stick with constructor as Kafka provides its multipleConditionStorage based on a State Store which is a runtime thing :'(
If both JdbcFlowTriggerService and MemoryFlowTriggerService are the same maybe you can create a DefaultFlowTriggerService (or just a FlowTrigerServiceImpl) that will be the default and replaces inside the Kafka runner.

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionStorageInterface.java b/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionStorageInterface.java
index f5672eebb6..3e774f9747 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionStorageInterface.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionStorageInterface.java
@@ -48,4 +48,8 @@ default MultipleConditionWindow getOrCreate(Flow flow, MultipleCondition multipl
                 .build()
             );
     }
+
+    void save(List<MultipleConditionWindow> multipleConditionWindows);
+
+    void delete(MultipleConditionWindow multipleConditionWindow);
 }
diff --git a/core/src/main/java/io/kestra/core/services/AbstractFlowTriggerService.java b/core/src/main/java/io/kestra/core/services/AbstractFlowTriggerService.java
new file mode 100644
index 0000000000..80ae142b6d
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/services/AbstractFlowTriggerService.java
@@ -0,0 +1,156 @@
+package io.kestra.core.services;
+
+import io.kestra.core.models.conditions.types.MultipleCondition;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
+import io.kestra.core.models.triggers.multipleflows.MultipleConditionWindow;
+import io.kestra.core.runners.RunContextFactory;
+import jakarta.inject.Inject;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.ToString;
+
+import java.util.*;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public abstract class AbstractFlowTriggerService {
+    @Inject
+    private ConditionService conditionService;
+
+    @Inject
+    private RunContextFactory runContextFactory;
+
+    @Inject
+    private FlowService flowService;
+
+    public Stream<FlowWithFlowTrigger> withFlowTriggersOnly(Stream<Flow> allFlows) {
+        return allFlows
+            .filter(flow -> !flow.isDisabled())
+            .filter(flow -> flow.getTriggers() != null && !flow.getTriggers().isEmpty())
+            .flatMap(flow -> flowTriggers(flow).map(trigger -> new FlowWithFlowTrigger(flow, trigger)));
+    }
+
+    public Stream<io.kestra.core.models.triggers.types.Flow> flowTriggers(Flow flow) {
+        return flow.getTriggers()
+            .stream()
+            .filter(Predicate.not(AbstractTrigger::isDisabled))
+            .filter(io.kestra.core.models.triggers.types.Flow.class::isInstance)
+            .map(io.kestra.core.models.triggers.types.Flow.class::cast);
+    }
+
+    public List<Execution> computeExecutionsFromFlowTriggers(Execution execution, List<Flow> allFlows, Optional<MultipleConditionStorageInterface> multipleConditionStorage) {
+        List<FlowWithFlowTrigger> validTriggersBeforeMultipleConditionEval = allFlows.stream()
+            // prevent recursive flow triggers
+            .filter(flow -> flowService.removeUnwanted(flow, execution))
+            // ensure flow & triggers are enabled
+            .filter(flow -> !flow.isDisabled())
+            .filter(flow -> flow.getTriggers() != null && !flow.getTriggers().isEmpty())
+            // validate flow triggers conditions excluding multiple conditions
+            .flatMap(flow -> flowTriggers(flow).map(trigger -> new FlowWithFlowTrigger(flow, trigger)))
+            .filter(flowWithFlowTrigger -> conditionService.valid(
+                flowWithFlowTrigger.getFlow(),
+                flowWithFlowTrigger.getTrigger().getConditions().stream()
+                    .filter(Predicate.not(MultipleCondition.class::isInstance))
+                    .toList(),
+                conditionService.conditionContext(
+                    runContextFactory.of(flowWithFlowTrigger.getFlow(), execution),
+                    flowWithFlowTrigger.getFlow(),
+                    execution
+                )
+            )).toList();
+
+        Map<FlowWithFlowTriggerAndMultipleCondition, MultipleConditionWindow> multipleConditionWindowsByFlow = null;
+        if (multipleConditionStorage.isPresent()) {
+            List<FlowWithFlowTriggerAndMultipleCondition> flowWithMultipleConditionsToEvaluate = validTriggersBeforeMultipleConditionEval.stream()
+                .flatMap(flowWithFlowTrigger ->
+                    flowWithFlowTrigger.getTrigger().getConditions().stream()
+                        .filter(MultipleCondition.class::isInstance)
+                        .map(MultipleCondition.class::cast)
+                        .map(multipleCondition -> new FlowWithFlowTriggerAndMultipleCondition(
+                                flowWithFlowTrigger.getFlow(),
+                                multipleConditionStorage.get().getOrCreate(flowWithFlowTrigger.getFlow(), multipleCondition),
+                                flowWithFlowTrigger.getTrigger(),
+                                multipleCondition
+                            )
+                        )
+                ).toList();
+
+            // evaluate multiple conditions
+            multipleConditionWindowsByFlow = flowWithMultipleConditionsToEvaluate.stream().map(f -> {
+                    Map<String, Boolean> results = f.getMultipleCondition()
+                        .getConditions()
+                        .entrySet()
+                        .stream()
+                        .map(e -> new AbstractMap.SimpleEntry<>(
+                            e.getKey(),
+                            conditionService.isValid(e.getValue(), f.getFlow(), execution)
+                        ))
+                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+                    return Map.entry(f, f.getMultipleConditionWindow().with(results));
+                })
+                .filter(e -> !e.getValue().getResults().isEmpty())
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+            // persist results
+            multipleConditionStorage.get().save(new ArrayList<>(multipleConditionWindowsByFlow.values()));
+        }
+
+        // compute all executions to create from flow triggers now that multiple conditions storage is populated
+        List<Execution> executions = validTriggersBeforeMultipleConditionEval.stream().filter(flowWithFlowTrigger ->
+                conditionService.isValid(
+                    flowWithFlowTrigger.getTrigger(),
+                    flowWithFlowTrigger.getFlow(),
+                    execution,
+                    multipleConditionStorage.orElse(null)
+                )
+            ).map(f -> f.getTrigger().evaluate(
+                runContextFactory.of(f.getFlow(), execution),
+                f.getFlow(),
+                execution
+            ))
+            .filter(Optional::isPresent)
+            .map(Optional::get)
+            .toList();
+
+        if(multipleConditionStorage.isPresent() && multipleConditionWindowsByFlow != null) {
+            // purge fulfilled or expired multiple condition windows
+            Stream.concat(
+                multipleConditionWindowsByFlow.keySet().stream()
+                    .map(f -> Map.entry(
+                        f.getMultipleCondition().getConditions(),
+                        multipleConditionStorage.get().getOrCreate(f.getFlow(), f.getMultipleCondition())
+                    ))
+                    .filter(e -> e.getKey().size() == Optional.ofNullable(e.getValue().getResults())
+                        .map(Map::size)
+                        .orElse(0))
+                    .map(Map.Entry::getValue),
+                multipleConditionStorage.get().expired().stream()
+            ).forEach(multipleConditionStorage.get()::delete);
+        }
+
+        return executions;
+    }
+
+    @AllArgsConstructor
+    @Getter
+    @ToString
+    protected static class FlowWithFlowTriggerAndMultipleCondition {
+        private final Flow flow;
+        private final MultipleConditionWindow multipleConditionWindow;
+        private final io.kestra.core.models.triggers.types.Flow trigger;
+        private final MultipleCondition multipleCondition;
+    }
+
+    @AllArgsConstructor
+    @Getter
+    @ToString
+    public static class FlowWithFlowTrigger {
+        private final Flow flow;
+        private final io.kestra.core.models.triggers.types.Flow trigger;
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/services/DefaultFlowTriggerService.java b/core/src/main/java/io/kestra/core/services/DefaultFlowTriggerService.java
new file mode 100644
index 0000000000..ff90610246
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/services/DefaultFlowTriggerService.java
@@ -0,0 +1,7 @@
+package io.kestra.core.services;
+
+import jakarta.inject.Singleton;
+
+@Singleton
+public class DefaultFlowTriggerService extends AbstractFlowTriggerService {
+}
\ No newline at end of file
diff --git a/core/src/main/java/io/kestra/core/services/FlowService.java b/core/src/main/java/io/kestra/core/services/FlowService.java
index f0c080f9af..18cfc3b975 100644
--- a/core/src/main/java/io/kestra/core/services/FlowService.java
+++ b/core/src/main/java/io/kestra/core/services/FlowService.java
@@ -2,12 +2,9 @@
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.core.JsonProcessingException;
-import io.kestra.core.models.conditions.types.MultipleCondition;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.triggers.AbstractTrigger;
-import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
-import io.kestra.core.models.triggers.multipleflows.MultipleConditionWindow;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.utils.ListUtils;
@@ -15,10 +12,7 @@
 import io.micronaut.core.annotation.Nullable;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
-import lombok.AllArgsConstructor;
-import lombok.Getter;
 import lombok.SneakyThrows;
-import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.*;
@@ -83,131 +77,11 @@ private Stream<Flow> keepLastVersionCollector(Stream<Flow> stream) {
             .filter(Objects::nonNull);
     }
 
-    public List<FlowWithFlowTrigger> flowWithFlowTrigger(Stream<Flow> flowStream) {
-        return flowStream
-            .filter(flow -> flow.getTriggers() != null && flow.getTriggers().size() > 0)
-            .filter(flow -> !flow.isDisabled())
-            .flatMap(flow -> flow.getTriggers()
-                .stream()
-                .filter(abstractTrigger -> !abstractTrigger.isDisabled())
-                .map(trigger -> new FlowWithTrigger(flow, trigger))
-            )
-            .filter(f -> f.getTrigger() instanceof io.kestra.core.models.triggers.types.Flow)
-            .map(f -> new FlowWithFlowTrigger(
-                    f.getFlow(),
-                    (io.kestra.core.models.triggers.types.Flow) f.getTrigger()
-                )
-            )
-            .collect(Collectors.toList());
-    }
-
     protected boolean removeUnwanted(Flow f, Execution execution) {
         // we don't allow recursive
         return !f.uidWithoutRevision().equals(Flow.uidWithoutRevision(execution));
     }
 
-    public List<Execution> flowTriggerExecution(Stream<Flow> flowStream, Execution execution, @Nullable MultipleConditionStorageInterface multipleConditionStorage) {
-        return flowStream
-            .filter(flow -> flow.getTriggers() != null && flow.getTriggers().size() > 0)
-            .filter(flow -> !flow.isDisabled())
-            .flatMap(flow -> flow.getTriggers()
-                .stream()
-                .filter(abstractTrigger -> !abstractTrigger.isDisabled())
-                .map(trigger -> new FlowWithTrigger(flow, trigger))
-            )
-            .filter(f -> conditionService.isValid(
-                f.getTrigger(),
-                f.getFlow(),
-                execution,
-                multipleConditionStorage
-            ))
-            .filter(f -> f.getTrigger() instanceof io.kestra.core.models.triggers.types.Flow)
-            .map(f -> new FlowWithFlowTrigger(
-                    f.getFlow(),
-                    (io.kestra.core.models.triggers.types.Flow) f.getTrigger()
-                )
-            )
-            .filter(f -> this.removeUnwanted(f.getFlow(), execution))
-            .map(f -> f.getTrigger().evaluate(
-                runContextFactory.of(f.getFlow(), execution),
-                f.getFlow(),
-                execution
-            ))
-            .filter(Optional::isPresent)
-            .map(Optional::get)
-            .collect(Collectors.toList());
-    }
-
-    private Stream<FlowWithFlowTriggerAndMultipleCondition> multipleFlowStream(
-        Stream<Flow> flowStream,
-        MultipleConditionStorageInterface multipleConditionStorage
-    ) {
-        return flowWithFlowTrigger(flowStream)
-            .stream()
-            .flatMap(e -> e.getTrigger()
-                .getConditions()
-                .stream()
-                .filter(condition -> condition instanceof MultipleCondition)
-                .map(condition -> {
-                    MultipleCondition multipleCondition = (MultipleCondition) condition;
-
-                    return new FlowWithFlowTriggerAndMultipleCondition(
-                        e.getFlow(),
-                        multipleConditionStorage.getOrCreate(e.getFlow(), multipleCondition),
-                        e.getTrigger(),
-                        multipleCondition
-                    );
-                })
-            );
-    }
-
-    public List<MultipleConditionWindow> multipleFlowTrigger(
-        Stream<Flow> flowStream,
-        Flow flow,
-        Execution execution,
-        MultipleConditionStorageInterface multipleConditionStorage
-    ) {
-        return multipleFlowStream(flowStream, multipleConditionStorage)
-            .map(f -> {
-                Map<String, Boolean> results = f.getMultipleCondition()
-                    .getConditions()
-                    .entrySet()
-                    .stream()
-                    .map(e -> new AbstractMap.SimpleEntry<>(
-                        e.getKey(),
-                        conditionService.isValid(e.getValue(), flow, execution)
-                    ))
-                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-
-                return f.getMultipleConditionWindow().with(results);
-            })
-            .filter(multipleConditionWindow -> multipleConditionWindow.getResults().size() > 0)
-            .collect(Collectors.toList());
-    }
-
-    public List<MultipleConditionWindow> multipleFlowToDelete(
-        Stream<Flow> flowStream,
-        MultipleConditionStorageInterface multipleConditionStorage
-    ) {
-        return Stream
-            .concat(
-                multipleFlowStream(flowStream, multipleConditionStorage)
-                    .filter(f -> f.getMultipleCondition().getConditions().size() ==
-                        (f.getMultipleConditionWindow().getResults() == null ? 0 :
-                            f.getMultipleConditionWindow()
-                                .getResults()
-                                .entrySet()
-                                .stream()
-                                .filter(Map.Entry::getValue)
-                                .count()
-                        )
-                    )
-                    .map(FlowWithFlowTriggerAndMultipleCondition::getMultipleConditionWindow),
-                multipleConditionStorage.expired().stream()
-            )
-            .collect(Collectors.toList());
-    }
-
     public static List<AbstractTrigger> findRemovedTrigger(Flow flow, Flow previous) {
         return ListUtils.emptyOnNull(previous.getTriggers())
             .stream()
@@ -219,7 +93,7 @@ public static List<AbstractTrigger> findRemovedTrigger(Flow flow, Flow previous)
     }
 
     public static String cleanupSource(String source) {
-        return source.replaceFirst("(?m)^revision: \\d+\n?","");
+        return source.replaceFirst("(?m)^revision: \\d+\n?", "");
     }
 
     public static String injectDisabled(String source, Boolean disabled) {
@@ -306,29 +180,4 @@ private static Object fixSnakeYaml(Object object) {
 
         return object;
     }
-
-    @AllArgsConstructor
-    @Getter
-    private static class FlowWithTrigger {
-        private final Flow flow;
-        private final AbstractTrigger trigger;
-    }
-
-    @AllArgsConstructor
-    @Getter
-    @ToString
-    public static class FlowWithFlowTrigger {
-        private final Flow flow;
-        private final io.kestra.core.models.triggers.types.Flow trigger;
-    }
-
-    @AllArgsConstructor
-    @Getter
-    @ToString
-    private static class FlowWithFlowTriggerAndMultipleCondition {
-        private final Flow flow;
-        private final MultipleConditionWindow multipleConditionWindow;
-        private final io.kestra.core.models.triggers.types.Flow trigger;
-        private final MultipleCondition multipleCondition;
-    }
 }
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcMultipleConditionStorage.java b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcMultipleConditionStorage.java
index 702d807b13..24d0eaa2ad 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcMultipleConditionStorage.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcMultipleConditionStorage.java
@@ -61,6 +61,7 @@ public List<MultipleConditionWindow> expired() {
             });
     }
 
+    @Override
     public synchronized void save(List<MultipleConditionWindow> multipleConditionWindows) {
         this.jdbcRepository
             .getDslContextWrapper()
@@ -75,6 +76,7 @@ public synchronized void save(List<MultipleConditionWindow> multipleConditionWin
             });
     }
 
+    @Override
     public void delete(MultipleConditionWindow multipleConditionWindow) {
         this.jdbcRepository.delete(multipleConditionWindow);
     }
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
index 3f3c25c3a5..c6ad983e9c 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -9,6 +9,7 @@
 import io.kestra.core.models.flows.FlowWithException;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.topologies.FlowTopology;
+import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
@@ -75,9 +76,6 @@ public class JdbcExecutor implements ExecutorInterface {
     @Inject
     private RunContextFactory runContextFactory;
 
-    @Inject
-    private FlowService flowService;
-
     @Inject
     private TaskDefaultService taskDefaultService;
 
@@ -91,13 +89,16 @@ public class JdbcExecutor implements ExecutorInterface {
     private ConditionService conditionService;
 
     @Inject
-    private MetricRegistry metricRegistry;
+    private MultipleConditionStorageInterface multipleConditionStorage;
 
     @Inject
-    protected FlowListenersInterface flowListeners;
+    private AbstractFlowTriggerService flowTriggerService;
 
     @Inject
-    private AbstractJdbcMultipleConditionStorage multipleConditionStorage;
+    private MetricRegistry metricRegistry;
+
+    @Inject
+    protected FlowListenersInterface flowListeners;
 
     @Inject
     private AbstractJdbcWorkerTaskExecutionStorage workerTaskExecutionStorage;
@@ -295,21 +296,8 @@ private void executionQueue(Execution message) {
                 conditionService.isTerminatedWithListeners(flow, execution) &&
                 this.deduplicateFlowTrigger(execution, executorState)
             ) {
-                // multiple conditions storage
-                multipleConditionStorage.save(
-                    flowService
-                        .multipleFlowTrigger(allFlows.stream(), flow, execution, multipleConditionStorage)
-                );
-
-                // Flow Trigger
-                flowService
-                    .flowTriggerExecution(allFlows.stream(), execution, multipleConditionStorage)
+                flowTriggerService.computeExecutionsFromFlowTriggers(execution, allFlows, Optional.of(multipleConditionStorage))
                     .forEach(this.executionQueue::emit);
-
-                // Trigger is done, remove matching multiple condition
-                flowService
-                    .multipleFlowToDelete(allFlows.stream(), multipleConditionStorage)
-                    .forEach(multipleConditionStorage::delete);
             }
 
             // worker task execution
diff --git a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
index bcfd8f3e8e..a27104637d 100644
--- a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
+++ b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
@@ -7,6 +7,7 @@
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
@@ -35,7 +36,6 @@
 @MemoryQueueEnabled
 @Slf4j
 public class MemoryExecutor implements ExecutorInterface {
-    private static final MemoryMultipleConditionStorage multipleConditionStorage = new MemoryMultipleConditionStorage();
     private static final ConcurrentHashMap<String, ExecutionState> EXECUTIONS = new ConcurrentHashMap<>();
     private static final ConcurrentHashMap<String, WorkerTaskExecution> WORKERTASKEXECUTIONS_WATCHER = new ConcurrentHashMap<>();
     private List<Flow> allFlows;
@@ -93,6 +93,11 @@ public class MemoryExecutor implements ExecutorInterface {
     @Inject
     private SkipExecutionService skipExecutionService;
 
+    @Inject
+    private AbstractFlowTriggerService flowTriggerService;
+
+    private final MultipleConditionStorageInterface multipleConditionStorage = new MemoryMultipleConditionStorage();
+
     @Override
     public void run() {
         flowListeners.run();
@@ -124,7 +129,7 @@ private Flow transform(Flow flow, Execution execution) {
                     (namespace, id) -> templateExecutorInterface.get().findById(namespace, id).orElse(null)
                 );
             } catch (InternalException e) {
-                log.debug("Failed to inject template",  e);
+                log.debug("Failed to inject template", e);
             }
         }
 
@@ -231,21 +236,8 @@ private void handleExecution(ExecutionState state) {
 
             // multiple condition
             if (conditionService.isTerminatedWithListeners(flow, execution)) {
-                // multiple conditions storage
-                multipleConditionStorage.save(
-                    flowService
-                        .multipleFlowTrigger(allFlows.stream(), flow, execution, multipleConditionStorage)
-                );
-
-                // Flow Trigger
-                flowService
-                    .flowTriggerExecution(allFlows.stream(), execution, multipleConditionStorage)
+                flowTriggerService.computeExecutionsFromFlowTriggers(execution, allFlows, Optional.of(multipleConditionStorage))
                     .forEach(this.executionQueue::emit);
-
-                // Trigger is done, remove matching multiple condition
-                flowService
-                    .multipleFlowToDelete(allFlows.stream(), multipleConditionStorage)
-                    .forEach(multipleConditionStorage::delete);
             }
 
             // worker task execution
@@ -385,7 +377,6 @@ private boolean deduplicateNexts(Execution execution, List<TaskRun> taskRuns) {
     }
 
 
-
     private static class ExecutionState {
         private final Execution execution;
         private Map<String, TaskRun> taskRuns = new ConcurrentHashMap<>();
diff --git a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryMultipleConditionStorage.java b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryMultipleConditionStorage.java
index b59d1d1744..ce3d10d966 100644
--- a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryMultipleConditionStorage.java
+++ b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryMultipleConditionStorage.java
@@ -51,6 +51,7 @@ public List<MultipleConditionWindow> expired() {
             .collect(Collectors.toList());
     }
 
+    @Override
     public synchronized void save(List<MultipleConditionWindow> multipleConditionWindows) {
         multipleConditionWindows
             .forEach(window -> {
@@ -70,6 +71,7 @@ public synchronized void save(List<MultipleConditionWindow> multipleConditionWin
             });
     }
 
+    @Override
     public void delete(MultipleConditionWindow multipleConditionWindow) {
         find(multipleConditionWindow.getNamespace(), multipleConditionWindow.getFlowId())
             .ifPresent(byCondition -> {


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java b/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
index d11b82c9b6..bd92823eaf 100644
--- a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
+++ b/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
@@ -101,18 +101,18 @@ public void failed() throws InterruptedException, TimeoutException {
         });
 
         // first one
-        Execution execution = runnerUtils.runOne("io.kestra.tests", "trigger-multiplecondition-flow-d", Duration.ofSeconds(60));
+        Execution execution = runnerUtils.runOne("io.kestra.tests", "trigger-multiplecondition-flow-c", Duration.ofSeconds(60));
         assertThat(execution.getTaskRunList().size(), is(1));
-        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
 
         // wait a little to be sure that the trigger is not launching execution
         countDownLatch.await(1, TimeUnit.SECONDS);
         assertThat(ended.size(), is(1));
 
         // second one
-        execution = runnerUtils.runOne("io.kestra.tests", "trigger-multiplecondition-flow-c", Duration.ofSeconds(60));
+        execution = runnerUtils.runOne("io.kestra.tests", "trigger-multiplecondition-flow-d", Duration.ofSeconds(60));
         assertThat(execution.getTaskRunList().size(), is(1));
-        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
         // trigger was not done
         countDownLatch.await(10, TimeUnit.SECONDS);
diff --git a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerFailedTest.java b/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerFailedTest.java
deleted file mode 100644
index 57232b1b68..0000000000
--- a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerFailedTest.java
+++ /dev/null
@@ -1,12 +0,0 @@
-package io.kestra.core.runners;
-
-import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
-import jakarta.inject.Inject;
-import org.junit.jupiter.api.Test;
-
-@MicronautTest
-public class MultipleConditionTriggerFailedTest extends AbstractMemoryRunnerTest {
-    @Inject
-    private MultipleConditionTriggerCaseTest runnerCaseTest;
-
-}