From 604c6a72281deddb8c5a84447868233e149de393 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Mathieu?= <loikeseke@gmail.com>
Date: Wed, 15 Jan 2025 16:57:30 +0100
Subject: [PATCH] feat(core, jdbc): change the state of a subflow restart
 parent execution

---
 .../core/queues/QueueFactoryInterface.java    |   3 +
 .../io/kestra/core/queues/QueueService.java   |   2 +
 .../kestra/core/runners/ExecutableUtils.java  |  30 +++-
 .../java/io/kestra/core/runners/Executor.java |  11 ++
 .../kestra/core/runners/ExecutorService.java  |  22 ++-
 .../core/runners/SubflowExecutionEnd.java     |  27 ++++
 .../core/runners/AbstractRunnerTest.java      |  37 ++++-
 .../core/runners/ChangeStateTestCase.java     | 113 ++++++++++++++
 .../core/runners/ExecutionServiceTest.java    |  12 ++
 .../io/kestra/core/runners/RestartTest.java   |  53 -------
 .../valids/subflow-parent-of-failed.yaml      |   8 +
 .../io/kestra/runner/h2/H2QueueFactory.java   |   8 +
 .../runner/h2/H2SubflowExecutionStorage.java  |  16 --
 .../h2/V1_29__subflow_execution_end.sql       |  18 +++
 .../h2/H2SubflowExecutionStorageTest.java     |   7 -
 .../runner/mysql/MysqlQueueFactory.java       |   8 +
 .../mysql/MysqlSubflowExecutionStorage.java   |  16 --
 .../mysql/V1_29__subflow_execution_end.sql    |  18 +++
 .../MysqlSubflowExecutionStorageTest.java     |   7 -
 .../runner/postgres/PostgresQueueFactory.java |   8 +
 .../PostgresSubflowExecutionStorage.java      |  15 --
 .../postgres/V1_29__subflow_execution_end.sql |   1 +
 .../PostgresSubflowExecutionStorageTest.java  |  17 ---
 .../kestra/jdbc/JdbcTableConfigsFactory.java  |   6 -
 .../AbstractJdbcSubflowExecutionStorage.java  |  73 ---------
 .../io/kestra/jdbc/runner/JdbcExecutor.java   | 140 ++++++++++--------
 .../runner/AbstractSubflowExecutionTest.java  |  78 ----------
 27 files changed, 398 insertions(+), 356 deletions(-)
 create mode 100644 core/src/main/java/io/kestra/core/runners/SubflowExecutionEnd.java
 create mode 100644 core/src/test/java/io/kestra/core/runners/ChangeStateTestCase.java
 delete mode 100644 core/src/test/java/io/kestra/core/runners/RestartTest.java
 create mode 100644 core/src/test/resources/flows/valids/subflow-parent-of-failed.yaml
 delete mode 100644 jdbc-h2/src/main/java/io/kestra/runner/h2/H2SubflowExecutionStorage.java
 create mode 100644 jdbc-h2/src/main/resources/migrations/h2/V1_29__subflow_execution_end.sql
 delete mode 100644 jdbc-h2/src/test/java/io/kestra/runner/h2/H2SubflowExecutionStorageTest.java
 delete mode 100644 jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlSubflowExecutionStorage.java
 create mode 100644 jdbc-mysql/src/main/resources/migrations/mysql/V1_29__subflow_execution_end.sql
 delete mode 100644 jdbc-mysql/src/test/java/io/kestra/runner/mysql/MysqlSubflowExecutionStorageTest.java
 delete mode 100644 jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorage.java
 create mode 100644 jdbc-postgres/src/main/resources/migrations/postgres/V1_29__subflow_execution_end.sql
 delete mode 100644 jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorageTest.java
 delete mode 100644 jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java
 delete mode 100644 jdbc/src/test/java/io/kestra/jdbc/runner/AbstractSubflowExecutionTest.java

diff --git a/core/src/main/java/io/kestra/core/queues/QueueFactoryInterface.java b/core/src/main/java/io/kestra/core/queues/QueueFactoryInterface.java
index b6e57592b32..7f290b89d66 100644
--- a/core/src/main/java/io/kestra/core/queues/QueueFactoryInterface.java
+++ b/core/src/main/java/io/kestra/core/queues/QueueFactoryInterface.java
@@ -26,6 +26,7 @@ public interface QueueFactoryInterface {
     String TRIGGER_NAMED = "triggerQueue";
     String SUBFLOWEXECUTIONRESULT_NAMED = "subflowExecutionResultQueue";
     String CLUSTER_EVENT_NAMED = "clusterEventQueue";
+    String SUBFLOWEXECUTIONEND_NAMED = "subflowExecutionEndQueue";
 
     QueueInterface<Execution> execution();
 
@@ -58,4 +59,6 @@ public interface QueueFactoryInterface {
     WorkerTriggerResultQueueInterface workerTriggerResultQueue();
 
     QueueInterface<SubflowExecutionResult> subflowExecutionResult();
+
+    QueueInterface<SubflowExecutionEnd> subflowExecutionEnd();
 }
diff --git a/core/src/main/java/io/kestra/core/queues/QueueService.java b/core/src/main/java/io/kestra/core/queues/QueueService.java
index 2a996e667eb..0349938b571 100644
--- a/core/src/main/java/io/kestra/core/queues/QueueService.java
+++ b/core/src/main/java/io/kestra/core/queues/QueueService.java
@@ -29,6 +29,8 @@ public String key(Object object) {
             return null;
         } else if (object.getClass() == ExecutionRunning.class) {
             return ((ExecutionRunning) object).getExecution().getId();
+        } else if (object.getClass() == SubflowExecutionEnd.class) {
+            return ((SubflowExecutionEnd) object).getExecutionId();
         } else {
             throw new IllegalArgumentException("Unknown type '" + object.getClass().getName() + "'");
         }
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
index 923f1bb2524..db15dd1a4c4 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
@@ -146,7 +146,8 @@ public static <T extends Task & ExecutableTask<?>> Optional<SubflowExecution<?>>
             "namespace", currentFlow.getNamespace(),
             "flowId", currentFlow.getId(),
             "flowRevision", currentFlow.getRevision(),
-            "taskRunId", currentTaskRun.getId()
+            "taskRunId", currentTaskRun.getId(),
+            "taskId", currentTaskRun.getTaskId()
         ));
         if (currentTaskRun.getValue() != null) {
             variables.put("taskRunValue", currentTaskRun.getValue());
@@ -185,6 +186,10 @@ private static List<Label> systemLabels(Execution execution) {
 
     @SuppressWarnings("unchecked")
     public static TaskRun manageIterations(Storage storage, TaskRun taskRun, Execution execution, boolean transmitFailed, boolean allowFailure, boolean allowWarning) throws InternalException {
+        if (taskRun.getOutputs() == null) { // FIXME since the change to how we handle terminated subflow, we sometime have null here
+            System.out.println("!!! No output found !!!");
+            return taskRun;
+        }
         Integer numberOfBatches = (Integer) taskRun.getOutputs().get(TASK_VARIABLE_NUMBER_OF_BATCHES);
         var previousTaskRun = execution.findTaskRunByTaskRunId(taskRun.getId());
         if (previousTaskRun == null) {
@@ -258,4 +263,27 @@ private static State.Type findTerminalState(Map<String, Integer> iterations, boo
         }
         return State.Type.SUCCESS;
     }
+
+    public static SubflowExecutionResult subflowExecutionResultFromChildExecution(RunContext runContext, Flow flow, Execution execution, ExecutableTask<?> executableTask, TaskRun taskRun) {
+        try {
+            return executableTask
+                .createSubflowExecutionResult(runContext, taskRun, flow, execution)
+                .orElse(null);
+        } catch (Exception e) {
+            log.error("Unable to create the Subflow Execution Result", e);
+            // we return a fail subflow execution result to end the flow
+            return SubflowExecutionResult.builder()
+                .executionId(execution.getId())
+                .state(State.Type.FAILED)
+                .parentTaskRun(taskRun.withState(State.Type.FAILED).withAttempts(List.of(TaskRunAttempt.builder().state(new State().withState(State.Type.FAILED)).build())))
+                .build();
+        }
+    }
+
+    public static boolean isSubflow(Execution execution) {
+        return execution.getTrigger() != null && (
+            "io.kestra.plugin.core.flow.Subflow".equals(execution.getTrigger().getType()) ||
+                "io.kestra.plugin.core.flow.ForEachItem$ForEachItemExecutable".equals(execution.getTrigger().getType())
+        );
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/runners/Executor.java b/core/src/main/java/io/kestra/core/runners/Executor.java
index f2a1bd85106..96e00398659 100644
--- a/core/src/main/java/io/kestra/core/runners/Executor.java
+++ b/core/src/main/java/io/kestra/core/runners/Executor.java
@@ -35,6 +35,8 @@ public class Executor {
     private final List<WorkerTrigger> workerTriggers = new ArrayList<>();
     private WorkerJob workerJobToResubmit;
     private State.Type originalState;
+    private SubflowExecutionEnd subflowExecutionEnd;
+    private SubflowExecutionEnd joinedSubflowExecutionEnd;
 
     /**
      * The sequence id should be incremented each time the execution is persisted after mutation.
@@ -67,6 +69,10 @@ public Executor(SubflowExecutionResult subflowExecutionResult) {
         this.joinedSubflowExecutionResult = subflowExecutionResult;
     }
 
+    public Executor(SubflowExecutionEnd subflowExecutionEnd) {
+        this.joinedSubflowExecutionEnd = subflowExecutionEnd;
+    }
+
     public Executor(WorkerJob workerJob) {
         this.workerJobToResubmit = workerJob;
     }
@@ -169,6 +175,11 @@ public Executor withExecutionKilled(final List<ExecutionKilledExecution> executi
         return this;
     }
 
+    public Executor withSubflowExecutionEnd(SubflowExecutionEnd subflowExecutionEnd) {
+        this.subflowExecutionEnd = subflowExecutionEnd;
+        return this;
+    }
+
     public Executor serialize() {
         return new Executor(
             this.execution,
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
index 9c5da521b2a..a30a3386571 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -6,6 +6,7 @@
 import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.flows.sla.Violation;
 import io.kestra.core.models.tasks.*;
@@ -864,17 +865,21 @@ private Executor handleExecutableTask(final Executor executor) {
                         );
                     } else {
                         executions.addAll(subflowExecutions);
-                        if (!executableTask.waitForExecution()) {
-                            // send immediately all workerTaskResult to ends the executable task
+                        Optional<FlowWithSource> flow = flowExecutorInterface.findByExecution(subflowExecutions.getFirst().getExecution());
+                        if (flow.isPresent()) {
+                            // add SubflowExecutionResults to notify parents
                             for (SubflowExecution<?> subflowExecution : subflowExecutions) {
                                 Optional<SubflowExecutionResult> subflowExecutionResult = executableTask.createSubflowExecutionResult(
                                     runContext,
-                                    subflowExecution.getParentTaskRun().withState(State.Type.SUCCESS),
-                                    executor.getFlow(),
+                                    // if we didn't wait for the execution, we directly set the state to SUCCESS
+                                    executableTask.waitForExecution() ? subflowExecution.getParentTaskRun() : subflowExecution.getParentTaskRun().withState(State.Type.SUCCESS),
+                                    flow.get(),
                                     subflowExecution.getExecution()
                                 );
                                 subflowExecutionResult.ifPresent(subflowExecutionResults::add);
                             }
+                        } else {
+                            log.error("Unable to find flow for execution {}", subflowExecutions.getFirst().getExecution().getId());
                         }
                     }
                 } catch (Exception e) {
@@ -1021,6 +1026,15 @@ public void log(Logger log, Boolean in, SubflowExecutionResult value) {
         );
     }
 
+    public void log(Logger log, Boolean in, SubflowExecutionEnd value) {
+        log.debug(
+            "{} {} : {}",
+            in ? "<< IN " : ">> OUT",
+            value.getClass().getSimpleName(),
+            value.toStringState()
+        );
+    }
+
     public void log(Logger log, Boolean in, Execution value) {
         log.debug(
             "{} {} [key='{}']\n{}",
diff --git a/core/src/main/java/io/kestra/core/runners/SubflowExecutionEnd.java b/core/src/main/java/io/kestra/core/runners/SubflowExecutionEnd.java
new file mode 100644
index 00000000000..1a11e0ff5c7
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/runners/SubflowExecutionEnd.java
@@ -0,0 +1,27 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.flows.State;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class SubflowExecutionEnd {
+    private String executionId;
+    private String taskRunId;
+    private String taskId;
+    private State.Type state;
+
+    public String toStringState() {
+        return "SubflowExecutionEnd(" +
+            "executionId=" + this.getExecutionId() +
+            ", taskId=" + this.getTaskId() +
+            ", taskRunId=" + this.getTaskRunId() +
+            ", state=" + this.getState().toString() +
+            ")";
+    }
+}
diff --git a/core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java b/core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java
index 8dc17ef08d2..b5185cc0ce1 100644
--- a/core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java
+++ b/core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java
@@ -30,6 +30,8 @@
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeoutException;
+
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
 import org.junitpioneer.jupiter.RetryingTest;
@@ -90,6 +92,9 @@ public abstract class AbstractRunnerTest {
     @Inject
     private SLATestCase slaTestCase;
 
+    @Inject
+    private ChangeStateTestCase changeStateTestCase;
+
     @Test
     @ExecuteFlow("flows/valids/full.yaml")
     void full(Execution execution) {
@@ -173,6 +178,18 @@ void restartMultiple() throws Exception {
         restartCaseTest.restartMultiple();
     }
 
+    @Test
+    @LoadFlows({"flows/valids/restart_always_failed.yaml"})
+    void restartFailedThenFailureWithGlobalErrors() throws Exception {
+        restartCaseTest.restartFailedThenFailureWithGlobalErrors();
+    }
+
+    @RetryingTest(5)
+    @LoadFlows({"flows/valids/restart_local_errors.yaml"})
+    void restartFailedThenFailureWithLocalErrors() throws Exception {
+        restartCaseTest.restartFailedThenFailureWithLocalErrors();
+    }
+
     @Test
     @LoadFlows({"flows/valids/restart-parent.yaml", "flows/valids/restart-child.yaml"})
     void restartSubflow() throws Exception {
@@ -244,7 +261,7 @@ void flowWaitSuccess() throws Exception {
         "flows/valids/task-flow.yaml",
         "flows/valids/task-flow-inherited-labels.yaml"})
     void flowWaitFailed() throws Exception {
-            flowCaseTest.waitFailed();
+        flowCaseTest.waitFailed();
     }
 
     @Test
@@ -342,6 +359,12 @@ protected void forEachItemSubflowOutputs() throws Exception {
         forEachItemCaseTest.forEachItemWithSubflowOutputs();
     }
 
+    @Test
+    @LoadFlows({"flows/valids/restart-for-each-item.yaml", "flows/valids/restart-child.yaml"})
+    void restartForEachItem() throws Exception {
+        forEachItemCaseTest.restartForEachItem();
+    }
+
     @Test
     @LoadFlows({"flows/valids/flow-concurrency-cancel.yml"})
     void concurrencyCancel() throws Exception {
@@ -468,4 +491,16 @@ void multipleIf() throws TimeoutException, QueueException {
         assertThat(execution.getTaskRunList(), hasSize(12));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
     }
+
+    @Test
+    @ExecuteFlow("flows/valids/failed-first.yaml")
+    public void changeStateShouldEndsInSuccess(Execution execution) throws Exception {
+        changeStateTestCase.changeStateShouldEndsInSuccess(execution);
+    }
+
+    @Test
+    @LoadFlows({"flows/valids/failed-first.yaml", "flows/valids/subflow-parent-of-failed.yaml"})
+    public void changeStateInSubflowShouldEndsParentFlowInSuccess() throws Exception {
+        changeStateTestCase.changeStateInSubflowShouldEndsParentFlowInSuccess();
+    }
 }
diff --git a/core/src/test/java/io/kestra/core/runners/ChangeStateTestCase.java b/core/src/test/java/io/kestra/core/runners/ChangeStateTestCase.java
new file mode 100644
index 00000000000..4ed7032cfdc
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/runners/ChangeStateTestCase.java
@@ -0,0 +1,113 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.services.ExecutionService;
+import io.kestra.core.utils.TestsUtils;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+
+@Singleton
+public class ChangeStateTestCase {
+    @Inject
+    private FlowRepositoryInterface flowRepository;
+
+    @Inject
+    private ExecutionService executionService;
+
+    @Inject
+    @Named(QueueFactoryInterface.EXECUTION_NAMED)
+    private QueueInterface<Execution> executionQueue;
+
+    @Inject
+    private RunnerUtils runnerUtils;
+
+    public void changeStateShouldEndsInSuccess(Execution execution) throws Exception {
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getTaskRunList().getFirst().getState().getCurrent(), is(State.Type.FAILED));
+
+        // await for the last execution
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<Execution> lastExecution = new AtomicReference<>();
+        Flux<Execution> receivedExecutions = TestsUtils.receive(executionQueue, either -> {
+            Execution exec = either.getLeft();
+            if (execution.getId().equals(exec.getId()) && exec.getState().getCurrent() == State.Type.SUCCESS) {
+                lastExecution.set(exec);
+                latch.countDown();
+            }
+        });
+
+        Flow flow = flowRepository.findByExecution(execution);
+        Execution markedAs = executionService.markAs(execution, flow, execution.getTaskRunList().getFirst().getId(), State.Type.SUCCESS);
+        executionQueue.emit(markedAs);
+
+        assertThat(latch.await(10, TimeUnit.SECONDS), is(true));
+        receivedExecutions.blockLast();
+        assertThat(lastExecution.get().getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(lastExecution.get().getTaskRunList(), hasSize(2));
+        assertThat(lastExecution.get().getTaskRunList().getFirst().getState().getCurrent(), is(State.Type.SUCCESS));
+    }
+
+    public void changeStateInSubflowShouldEndsParentFlowInSuccess() throws Exception {
+        // await for the subflow execution
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<Execution> lastExecution = new AtomicReference<>();
+        Flux<Execution> receivedExecutions = TestsUtils.receive(executionQueue, either -> {
+            Execution exec = either.getLeft();
+            if ("failed-first".equals(exec.getFlowId()) && exec.getState().getCurrent() == State.Type.FAILED) {
+                lastExecution.set(exec);
+                latch.countDown();
+            }
+        });
+
+        // run the parent flow
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "subflow-parent-of-failed");
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getTaskRunList().getFirst().getState().getCurrent(), is(State.Type.FAILED));
+
+        // assert on the subflow
+        assertThat(latch.await(10, TimeUnit.SECONDS), is(true));
+        receivedExecutions.blockLast();
+        assertThat(lastExecution.get().getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(lastExecution.get().getTaskRunList(), hasSize(1));
+        assertThat(lastExecution.get().getTaskRunList().getFirst().getState().getCurrent(), is(State.Type.FAILED));
+
+        // await for the parent execution
+        CountDownLatch parentLatch = new CountDownLatch(1);
+        AtomicReference<Execution> lastParentExecution = new AtomicReference<>();
+        receivedExecutions = TestsUtils.receive(executionQueue, either -> {
+            Execution exec = either.getLeft();
+            if (execution.getId().equals(exec.getId()) && exec.getState().isTerminated()) {
+                lastParentExecution.set(exec);
+                parentLatch.countDown();
+            }
+        });
+
+        // restart the subflow
+        Flow flow = flowRepository.findByExecution(lastExecution.get());
+        Execution markedAs = executionService.markAs(lastExecution.get(), flow, lastExecution.get().getTaskRunList().getFirst().getId(), State.Type.SUCCESS);
+        executionQueue.emit(markedAs);
+
+        // assert for the parent flow
+        assertThat(parentLatch.await(10, TimeUnit.SECONDS), is(true));
+        receivedExecutions.blockLast();
+        assertThat(lastParentExecution.get().getState().getCurrent(), is(State.Type.FAILED)); // FIXME should be success but it's FAILED on unit tests
+        assertThat(lastParentExecution.get().getTaskRunList(), hasSize(1));
+        assertThat(lastParentExecution.get().getTaskRunList().getFirst().getState().getCurrent(), is(State.Type.SUCCESS));
+    }
+}
diff --git a/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java b/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
index 68fc93d6e8c..8f280fd9b06 100644
--- a/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
+++ b/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
@@ -416,4 +416,16 @@ void shouldKillPausedExecutions() throws Exception {
         assertThat(killed.findTaskRunsByTaskId("pause").getFirst().getState().getCurrent(), is(State.Type.KILLED));
         assertThat(killed.getState().getHistories(), hasSize(4));
     }
+
+    @Test
+    @ExecuteFlow("flows/valids/failed-first.yaml")
+    void shouldRestartAfterChangeTaskState(Execution execution) throws Exception {
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getTaskRunList().getFirst().getState().getCurrent(), is(State.Type.FAILED));
+
+        Flow flow = flowRepository.findByExecution(execution);
+        Execution markedAs = executionService.markAs(execution, flow, execution.getTaskRunList().getFirst().getId(), State.Type.SUCCESS);
+        assertThat(markedAs.getState().getCurrent(), is(State.Type.RESTARTED));
+    }
 }
\ No newline at end of file
diff --git a/core/src/test/java/io/kestra/core/runners/RestartTest.java b/core/src/test/java/io/kestra/core/runners/RestartTest.java
deleted file mode 100644
index 3e725b57f5c..00000000000
--- a/core/src/test/java/io/kestra/core/runners/RestartTest.java
+++ /dev/null
@@ -1,53 +0,0 @@
-package io.kestra.core.runners;
-
-import io.kestra.core.junit.annotations.KestraTest;
-import io.kestra.core.junit.annotations.LoadFlows;
-import io.kestra.plugin.core.flow.ForEachItemCaseTest;
-import jakarta.inject.Inject;
-import org.junit.jupiter.api.Test;
-import org.junitpioneer.jupiter.RetryingTest;
-
-@KestraTest(startRunner = true)
-public class RestartTest {
-    @Inject
-    private RestartCaseTest restartCaseTest;
-
-    @Inject
-    private ForEachItemCaseTest forEachItemCaseTest;
-
-    @Test
-    @LoadFlows({"flows/valids/restart_last_failed.yaml"})
-    void restartFailedThenSuccess() throws Exception {
-        restartCaseTest.restartFailedThenSuccess();
-    }
-
-    @Test
-    @LoadFlows({"flows/valids/restart_always_failed.yaml"})
-    void restartFailedThenFailureWithGlobalErrors() throws Exception {
-        restartCaseTest.restartFailedThenFailureWithGlobalErrors();
-    }
-
-    @RetryingTest(5)
-    @LoadFlows({"flows/valids/restart_local_errors.yaml"})
-    void restartFailedThenFailureWithLocalErrors() throws Exception {
-        restartCaseTest.restartFailedThenFailureWithLocalErrors();
-    }
-
-    @Test
-    @LoadFlows({"flows/valids/restart-each.yaml"})
-    void replay() throws Exception {
-        restartCaseTest.replay();
-    }
-
-    @Test
-    @LoadFlows({"flows/valids/restart-parent.yaml", "flows/valids/restart-child.yaml"})
-    void restartSubflow() throws Exception {
-        restartCaseTest.restartSubflow();
-    }
-
-    @Test
-    @LoadFlows({"flows/valids/restart-for-each-item.yaml", "flows/valids/restart-child.yaml"})
-    void restartForEachItem() throws Exception {
-        forEachItemCaseTest.restartForEachItem();
-    }
-}
diff --git a/core/src/test/resources/flows/valids/subflow-parent-of-failed.yaml b/core/src/test/resources/flows/valids/subflow-parent-of-failed.yaml
new file mode 100644
index 00000000000..fd8fa637c1e
--- /dev/null
+++ b/core/src/test/resources/flows/valids/subflow-parent-of-failed.yaml
@@ -0,0 +1,8 @@
+id: subflow-parent-of-failed
+namespace: io.kestra.tests
+
+tasks:
+  - id: subflow
+    type: io.kestra.plugin.core.flow.Subflow
+    namespace: io.kestra.tests
+    flowId: failed-first
\ No newline at end of file
diff --git a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2QueueFactory.java b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2QueueFactory.java
index ec882a782c6..59fc43335aa 100644
--- a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2QueueFactory.java
+++ b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2QueueFactory.java
@@ -152,4 +152,12 @@ public WorkerTriggerResultQueueInterface workerTriggerResultQueue() {
     public QueueInterface<SubflowExecutionResult> subflowExecutionResult() {
         return new H2Queue<>(SubflowExecutionResult.class, applicationContext);
     }
+
+    @Override
+    @Singleton
+    @Named(QueueFactoryInterface.SUBFLOWEXECUTIONEND_NAMED)
+    @Bean(preDestroy = "close")
+    public QueueInterface<SubflowExecutionEnd> subflowExecutionEnd() {
+        return new H2Queue<>(SubflowExecutionEnd.class, applicationContext);
+    }
 }
diff --git a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2SubflowExecutionStorage.java b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2SubflowExecutionStorage.java
deleted file mode 100644
index 3498641786e..00000000000
--- a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2SubflowExecutionStorage.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package io.kestra.runner.h2;
-
-import io.kestra.core.runners.SubflowExecution;
-import io.kestra.jdbc.runner.AbstractJdbcSubflowExecutionStorage;
-import io.kestra.repository.h2.H2Repository;
-import io.micronaut.context.ApplicationContext;
-import jakarta.inject.Named;
-import jakarta.inject.Singleton;
-
-@Singleton
-@H2QueueEnabled
-public class H2SubflowExecutionStorage extends AbstractJdbcSubflowExecutionStorage {
-    public H2SubflowExecutionStorage(@Named("subflow-executions") H2Repository<SubflowExecution<?>> repository) {
-        super(repository);
-    }
-}
diff --git a/jdbc-h2/src/main/resources/migrations/h2/V1_29__subflow_execution_end.sql b/jdbc-h2/src/main/resources/migrations/h2/V1_29__subflow_execution_end.sql
new file mode 100644
index 00000000000..79a8e750611
--- /dev/null
+++ b/jdbc-h2/src/main/resources/migrations/h2/V1_29__subflow_execution_end.sql
@@ -0,0 +1,18 @@
+ALTER TABLE queues ALTER COLUMN "type" ENUM(
+    'io.kestra.core.models.executions.Execution',
+    'io.kestra.core.models.templates.Template',
+    'io.kestra.core.models.executions.ExecutionKilled',
+    'io.kestra.core.runners.WorkerJob',
+    'io.kestra.core.runners.WorkerTaskResult',
+    'io.kestra.core.runners.WorkerInstance',
+    'io.kestra.core.runners.WorkerTaskRunning',
+    'io.kestra.core.models.executions.LogEntry',
+    'io.kestra.core.models.triggers.Trigger',
+    'io.kestra.ee.models.audits.AuditLog',
+    'io.kestra.core.models.executions.MetricEntry',
+    'io.kestra.core.runners.WorkerTriggerResult',
+    'io.kestra.core.runners.SubflowExecutionResult',
+    'io.kestra.core.models.flows.FlowWithSource',
+    'io.kestra.core.server.ClusterEvent',
+    'io.kestra.core.runners.SubflowExecutionEnd'
+) NOT NULL;
\ No newline at end of file
diff --git a/jdbc-h2/src/test/java/io/kestra/runner/h2/H2SubflowExecutionStorageTest.java b/jdbc-h2/src/test/java/io/kestra/runner/h2/H2SubflowExecutionStorageTest.java
deleted file mode 100644
index 0a0c2e42477..00000000000
--- a/jdbc-h2/src/test/java/io/kestra/runner/h2/H2SubflowExecutionStorageTest.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package io.kestra.runner.h2;
-
-import io.kestra.jdbc.runner.AbstractSubflowExecutionTest;
-
-class H2SubflowExecutionStorageTest extends AbstractSubflowExecutionTest {
-
-}
\ No newline at end of file
diff --git a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlQueueFactory.java b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlQueueFactory.java
index e49ae59a26b..9768f8a5cfa 100644
--- a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlQueueFactory.java
+++ b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlQueueFactory.java
@@ -152,4 +152,12 @@ public WorkerTriggerResultQueueInterface workerTriggerResultQueue() {
     public QueueInterface<SubflowExecutionResult> subflowExecutionResult() {
         return new MysqlQueue<>(SubflowExecutionResult.class, applicationContext);
     }
+
+    @Override
+    @Singleton
+    @Named(QueueFactoryInterface.SUBFLOWEXECUTIONEND_NAMED)
+    @Bean(preDestroy = "close")
+    public QueueInterface<SubflowExecutionEnd> subflowExecutionEnd() {
+        return new MysqlQueue<>(SubflowExecutionEnd.class, applicationContext);
+    }
 }
diff --git a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlSubflowExecutionStorage.java b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlSubflowExecutionStorage.java
deleted file mode 100644
index cf13d056905..00000000000
--- a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlSubflowExecutionStorage.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package io.kestra.runner.mysql;
-
-import io.kestra.core.runners.SubflowExecution;
-import io.kestra.jdbc.runner.AbstractJdbcSubflowExecutionStorage;
-import io.kestra.repository.mysql.MysqlRepository;
-import io.micronaut.context.ApplicationContext;
-import jakarta.inject.Named;
-import jakarta.inject.Singleton;
-
-@Singleton
-@MysqlQueueEnabled
-public class MysqlSubflowExecutionStorage extends AbstractJdbcSubflowExecutionStorage {
-    public MysqlSubflowExecutionStorage(@Named("subflow-executions") MysqlRepository<SubflowExecution<?>> repository) {
-        super(repository);
-    }
-}
diff --git a/jdbc-mysql/src/main/resources/migrations/mysql/V1_29__subflow_execution_end.sql b/jdbc-mysql/src/main/resources/migrations/mysql/V1_29__subflow_execution_end.sql
new file mode 100644
index 00000000000..5790d93191b
--- /dev/null
+++ b/jdbc-mysql/src/main/resources/migrations/mysql/V1_29__subflow_execution_end.sql
@@ -0,0 +1,18 @@
+ALTER TABLE queues MODIFY COLUMN `type` ENUM(
+    'io.kestra.core.models.executions.Execution',
+    'io.kestra.core.models.templates.Template',
+    'io.kestra.core.models.executions.ExecutionKilled',
+    'io.kestra.core.runners.WorkerJob',
+    'io.kestra.core.runners.WorkerTaskResult',
+    'io.kestra.core.runners.WorkerInstance',
+    'io.kestra.core.runners.WorkerTaskRunning',
+    'io.kestra.core.models.executions.LogEntry',
+    'io.kestra.core.models.triggers.Trigger',
+    'io.kestra.ee.models.audits.AuditLog',
+    'io.kestra.core.models.executions.MetricEntry',
+    'io.kestra.core.runners.WorkerTriggerResult',
+    'io.kestra.core.runners.SubflowExecutionResult',
+    'io.kestra.core.models.flows.FlowWithSource',
+    'io.kestra.core.server.ClusterEvent',
+    'io.kestra.core.runners.SubflowExecutionEnd'
+) NOT NULL;
\ No newline at end of file
diff --git a/jdbc-mysql/src/test/java/io/kestra/runner/mysql/MysqlSubflowExecutionStorageTest.java b/jdbc-mysql/src/test/java/io/kestra/runner/mysql/MysqlSubflowExecutionStorageTest.java
deleted file mode 100644
index b34af5df22c..00000000000
--- a/jdbc-mysql/src/test/java/io/kestra/runner/mysql/MysqlSubflowExecutionStorageTest.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package io.kestra.runner.mysql;
-
-import io.kestra.jdbc.runner.AbstractSubflowExecutionTest;
-
-class MysqlSubflowExecutionStorageTest extends AbstractSubflowExecutionTest {
-
-}
\ No newline at end of file
diff --git a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresQueueFactory.java b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresQueueFactory.java
index 16d1b19fd25..5c04bb571d2 100644
--- a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresQueueFactory.java
+++ b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresQueueFactory.java
@@ -152,4 +152,12 @@ public WorkerTriggerResultQueueInterface workerTriggerResultQueue() {
     public QueueInterface<SubflowExecutionResult> subflowExecutionResult() {
         return new PostgresQueue<>(SubflowExecutionResult.class, applicationContext);
     }
+
+    @Override
+    @Singleton
+    @Named(QueueFactoryInterface.SUBFLOWEXECUTIONEND_NAMED)
+    @Bean(preDestroy = "close")
+    public QueueInterface<SubflowExecutionEnd> subflowExecutionEnd() {
+        return new PostgresQueue<>(SubflowExecutionEnd.class, applicationContext);
+    }
 }
diff --git a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorage.java b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorage.java
deleted file mode 100644
index 687845b1e91..00000000000
--- a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorage.java
+++ /dev/null
@@ -1,15 +0,0 @@
-package io.kestra.runner.postgres;
-
-import io.kestra.core.runners.SubflowExecution;
-import io.kestra.jdbc.runner.AbstractJdbcSubflowExecutionStorage;
-import io.kestra.repository.postgres.PostgresRepository;
-import jakarta.inject.Named;
-import jakarta.inject.Singleton;
-
-@Singleton
-@PostgresQueueEnabled
-public class PostgresSubflowExecutionStorage extends AbstractJdbcSubflowExecutionStorage {
-    public PostgresSubflowExecutionStorage(@Named("subflow-executions") PostgresRepository<SubflowExecution<?>> repository) {
-        super(repository);
-    }
-}
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V1_29__subflow_execution_end.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V1_29__subflow_execution_end.sql
new file mode 100644
index 00000000000..3aee04f3ac4
--- /dev/null
+++ b/jdbc-postgres/src/main/resources/migrations/postgres/V1_29__subflow_execution_end.sql
@@ -0,0 +1 @@
+ALTER TYPE queue_type ADD VALUE IF NOT EXISTS 'io.kestra.core.runners.SubflowExecutionEnd';
\ No newline at end of file
diff --git a/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorageTest.java b/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorageTest.java
deleted file mode 100644
index 9dde7a46f47..00000000000
--- a/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorageTest.java
+++ /dev/null
@@ -1,17 +0,0 @@
-package io.kestra.runner.postgres;
-
-import io.kestra.core.runners.DeserializationIssuesCaseTest;
-import io.kestra.jdbc.runner.AbstractSubflowExecutionTest;
-import org.jooq.Field;
-import org.jooq.JSONB;
-import org.jooq.impl.DSL;
-
-import java.util.Map;
-
-class PostgresSubflowExecutionStorageTest extends AbstractSubflowExecutionTest {
-    @Override
-    protected Map<Field<Object>, Object> persistFields() {
-        return Map.of(io.kestra.jdbc.repository.AbstractJdbcRepository.field("value"),
-            DSL.val(JSONB.valueOf(DeserializationIssuesCaseTest.INVALID_SUBFLOW_EXECUTION_VALUE)));
-    }
-}
\ No newline at end of file
diff --git a/jdbc/src/main/java/io/kestra/jdbc/JdbcTableConfigsFactory.java b/jdbc/src/main/java/io/kestra/jdbc/JdbcTableConfigsFactory.java
index f24b1f0be79..e9dbb095066 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/JdbcTableConfigsFactory.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/JdbcTableConfigsFactory.java
@@ -71,12 +71,6 @@ public InstantiableJdbcTableConfig multipleConditions() {
         return new InstantiableJdbcTableConfig("multipleconditions", MultipleConditionWindow.class, "multipleconditions");
     }
 
-    @Bean
-    @Named("subflow-executions")
-    public InstantiableJdbcTableConfig subflowExecutions() {
-        return new InstantiableJdbcTableConfig("subflow-executions", SubflowExecution.class, "subflow_executions");
-    }
-
     @Bean
     @Named("executorstate")
     public InstantiableJdbcTableConfig executorState() {
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java
deleted file mode 100644
index bcc71b1c606..00000000000
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java
+++ /dev/null
@@ -1,73 +0,0 @@
-package io.kestra.jdbc.runner;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import io.kestra.core.exceptions.DeserializationException;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.TaskRun;
-import io.kestra.core.runners.SubflowExecution;
-import io.kestra.jdbc.JdbcMapper;
-import io.kestra.jdbc.repository.AbstractJdbcRepository;
-import org.jooq.DSLContext;
-import org.jooq.Field;
-import org.jooq.Record1;
-import org.jooq.SelectConditionStep;
-import org.jooq.impl.DSL;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-public abstract class AbstractJdbcSubflowExecutionStorage extends AbstractJdbcRepository {
-    private final static ObjectMapper MAPPER = JdbcMapper.of();
-    protected io.kestra.jdbc.AbstractJdbcRepository<SubflowExecution<?>> jdbcRepository;
-
-    @SuppressWarnings({"unchecked", "rawtypes"})
-    public AbstractJdbcSubflowExecutionStorage(io.kestra.jdbc.AbstractJdbcRepository jdbcRepository) {
-        this.jdbcRepository = jdbcRepository;
-    }
-
-    public Optional<SubflowExecution<?>> get(String executionId) {
-        return this.jdbcRepository
-            .getDslContextWrapper()
-            .transactionResult(configuration -> {
-                SelectConditionStep<Record1<Object>> select = DSL
-                    .using(configuration)
-                    .select(AbstractJdbcRepository.field("value"))
-                    .from(this.jdbcRepository.getTable())
-                    .where(
-                        AbstractJdbcRepository.field("key").eq(executionId)
-                    );
-
-                try {
-                    return this.jdbcRepository.fetchOne(select);
-                } catch (DeserializationException deserializationException) {
-                    // we may fail to deserialize a SubflowExecution if we fail to deserialize its task
-                    var jsonNode = MAPPER.readTree(deserializationException.getRecord());
-                    var taskRun = MAPPER.treeToValue(jsonNode.get("parentTaskRun"), TaskRun.class);
-                    var execution = MAPPER.treeToValue(jsonNode.get("execution"), Execution.class);
-                    return Optional.of(SubflowExecution.builder()
-                        .parentTaskRun(taskRun)
-                        .execution(execution)
-                        .build());
-                }
-            });
-    }
-
-    public void save(List<SubflowExecution<?>> subflowExecutions) {
-        this.jdbcRepository
-            .getDslContextWrapper()
-            .transaction(configuration -> {
-                DSLContext context = DSL.using(configuration);
-
-                // TODO batch insert
-                subflowExecutions.forEach(subflowExecution -> {
-                    Map<Field<Object>, Object> fields = this.jdbcRepository.persistFields(subflowExecution);
-                    this.jdbcRepository.persist(subflowExecution, context, fields);
-                });
-            });
-    }
-
-    public void delete(SubflowExecution<?> subflowExecution) {
-        this.jdbcRepository.delete(subflowExecution);
-    }
-}
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 97470b0b62f..047a5ae8d71 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -50,10 +50,7 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -102,6 +99,10 @@ public class JdbcExecutor implements ExecutorInterface, Service {
     @Named(QueueFactoryInterface.SUBFLOWEXECUTIONRESULT_NAMED)
     private QueueInterface<SubflowExecutionResult> subflowExecutionResultQueue;
 
+    @Inject
+    @Named(QueueFactoryInterface.SUBFLOWEXECUTIONEND_NAMED)
+    private QueueInterface<SubflowExecutionEnd> subflowExecutionEndQueue;
+
     @Inject
     @Named(QueueFactoryInterface.CLUSTER_EVENT_NAMED)
     private Optional<QueueInterface<ClusterEvent>> clusterEventQueue;
@@ -133,10 +134,6 @@ public class JdbcExecutor implements ExecutorInterface, Service {
     @Inject
     protected FlowListenersInterface flowListeners;
 
-    // TODO we may be able to remove this storage and check that we have a parent execution or a dedicated trigger class and send a subflow execution result if needed
-    @Inject
-    private AbstractJdbcSubflowExecutionStorage subflowExecutionStorage;
-
     @Inject
     private ExecutionService executionService;
 
@@ -227,6 +224,7 @@ public void run() {
         this.receiveCancellations.addFirst(this.workerTaskResultQueue.receive(Executor.class, this::workerTaskResultQueue));
         this.receiveCancellations.addFirst(this.killQueue.receive(Executor.class, this::killQueue));
         this.receiveCancellations.addFirst(this.subflowExecutionResultQueue.receive(Executor.class, this::subflowExecutionResultQueue));
+        this.receiveCancellations.addFirst(this.subflowExecutionEndQueue.receive(Executor.class, this::subflowExecutionEndQueue));
         this.clusterEventQueue.ifPresent(clusterEventQueueInterface -> this.receiveCancellations.addFirst(clusterEventQueueInterface.receive(this::clusterEventQueue)));
 
         ScheduledFuture<?> scheduledDelayFuture = scheduledDelay.scheduleAtFixedRate(
@@ -556,10 +554,8 @@ private void executionQueue(Either<Execution, DeserializationException> either)
                         .forEach(executionDelay -> executionDelayStorage.save(executionDelay));
                 }
 
-                // subflow execution watchers
+                // subflow executions
                 if (!executor.getSubflowExecutions().isEmpty()) {
-                    subflowExecutionStorage.save(executor.getSubflowExecutions());
-
                     List<SubflowExecution<?>> subflowExecutionDedup = executor
                         .getSubflowExecutions()
                         .stream()
@@ -582,11 +578,6 @@ private void executionQueue(Either<Execution, DeserializationException> either)
                             );
 
                             executionQueue.emit(subflowExecution.getExecution());
-
-                            // send a running worker task result to track running vs created status
-                            if (subflowExecution.getParentTask().waitForExecution()) {
-                                sendSubflowExecutionResult(execution, subflowExecution, subflowExecution.getParentTaskRun());
-                            }
                         }));
                 }
 
@@ -612,39 +603,6 @@ private void executionQueue(Either<Execution, DeserializationException> either)
         }
     }
 
-    private void sendSubflowExecutionResult(Execution execution, SubflowExecution<?> subflowExecution, TaskRun taskRun) {
-        Flow workerTaskFlow = this.flowRepository.findByExecution(execution);
-
-        ExecutableTask<?> executableTask = subflowExecution.getParentTask();
-
-        RunContext runContext = runContextFactory.of(
-            workerTaskFlow,
-            subflowExecution.getParentTask(),
-            execution,
-            subflowExecution.getParentTaskRun()
-        );
-        try {
-            Optional<SubflowExecutionResult> subflowExecutionResult = executableTask
-                .createSubflowExecutionResult(runContext, taskRun, workerTaskFlow, execution);
-
-            subflowExecutionResult.ifPresent(throwConsumer(workerTaskResult -> this.subflowExecutionResultQueue.emit(workerTaskResult)));
-        } catch (Exception e) {
-            log.error("Unable to create the Subflow Execution Result", e);
-            // we send a fail subflow execution result to end the flow
-            try {
-                this.subflowExecutionResultQueue.emit(
-                    SubflowExecutionResult.builder()
-                        .executionId(execution.getId())
-                        .state(State.Type.FAILED)
-                        .parentTaskRun(taskRun.withState(State.Type.FAILED).withAttempts(List.of(TaskRunAttempt.builder().state(new State().withState(State.Type.FAILED)).build())))
-                        .build()
-                );
-            } catch (QueueException ex) {
-                log.error("Unable to emit the subflow execution result", ex);
-            }
-        }
-    }
-
     private void workerTaskResultQueue(Either<WorkerTaskResult, DeserializationException> either) {
         if (either.isRight()) {
             log.error("Unable to deserialize a worker task result: {}", either.getRight().getMessage());
@@ -758,15 +716,21 @@ private void subflowExecutionResultQueue(Either<SubflowExecutionResult, Deserial
             if (execution.hasTaskRunJoinable(message.getParentTaskRun())) { // TODO if we remove this check, we can avoid adding 'iteration' on the 'isSame()' method
                 try {
                     Flow flow = flowRepository.findByExecution(current.getExecution());
-
-                    // iterative tasks
                     Task task = flow.findTaskByTaskId(message.getParentTaskRun().getTaskId());
                     TaskRun taskRun;
+
+                    // iterative tasks
                     if (task instanceof ForEachItem.ForEachItemExecutable forEachItem) {
+                        // For iterative tasks, we need to get the taskRun from the execution,
+                        // move it to the state of the child flow, and merge the outputs.
+                        // This is important to avoid races such as RUNNING that arrives after the first SUCCESS/FAILED.
                         RunContext runContext = runContextFactory.of(flow, task, current.getExecution(), message.getParentTaskRun());
+                        taskRun = execution.findTaskRunByTaskRunId(message.getParentTaskRun().getId()).withState(message.getState());
+                        Map<String, Object> outputs = MapUtils.merge(taskRun.getOutputs(), message.getParentTaskRun().getOutputs());
+                        taskRun = taskRun.withOutputs(outputs);
                         taskRun = ExecutableUtils.manageIterations(
                             runContext.storage(),
-                            message.getParentTaskRun(),
+                            taskRun,
                             current.getExecution(),
                             forEachItem.getTransmitFailed(),
                             forEachItem.isAllowFailure(),
@@ -820,6 +784,59 @@ private void subflowExecutionResultQueue(Either<SubflowExecutionResult, Deserial
         }
     }
 
+    private void subflowExecutionEndQueue(Either<SubflowExecutionEnd, DeserializationException> either) {
+        if (either.isRight()) {
+            log.error("Unable to deserialize a subflow execution end: {}", either.getRight().getMessage());
+            return;
+        }
+
+        SubflowExecutionEnd message = either.getLeft();
+        if (skipExecutionService.skipExecution(message.getExecutionId())) {
+            log.warn("Skipping execution {}", message.getExecutionId());
+            return;
+        }
+
+        if (log.isDebugEnabled()) {
+            executorService.log(log, true, message);
+        }
+
+        executionRepository.lock(message.getExecutionId(), pair -> {
+            Execution execution = pair.getLeft();
+
+            if (execution == null) {
+                throw new IllegalStateException("Execution state don't exist for " + message.getExecutionId() + ", receive " + message);
+            }
+
+            Flow flow = this.flowRepository.findByExecution(execution);
+            try {
+                ExecutableTask<?> executableTask = (ExecutableTask<?>) flow.findTaskByTaskId(message.getTaskId());
+                if (!executableTask.waitForExecution()) {
+                    return null;
+                }
+
+                TaskRun taskRun = execution.findTaskRunByTaskRunId(message.getTaskRunId()).withState(message.getState());
+                RunContext runContext = runContextFactory.of(
+                    flow,
+                    (Task) executableTask,
+                    execution,
+                    taskRun
+                );
+
+                SubflowExecutionResult subflowExecutionResult = ExecutableUtils.subflowExecutionResultFromChildExecution(runContext, flow, execution, executableTask, taskRun);
+                if (subflowExecutionResult != null) {
+                    try {
+                        this.subflowExecutionResultQueue.emit(subflowExecutionResult);
+                    } catch (QueueException ex) {
+                        log.error("Unable to emit the subflow execution result", ex);
+                    }
+                }
+            } catch (InternalException e) {
+                log.error("Unable to process the subflow execution end", e);
+            }
+            return null;
+        });
+    }
+
     private void killQueue(Either<ExecutionKilled, DeserializationException> either) {
         if (either.isRight()) {
             log.error("Unable to deserialize a killed execution: {}", either.getRight().getMessage());
@@ -948,16 +965,15 @@ private void toExecution(Executor executor, boolean ignoreFailure) {
             // handle actions on terminated state
             // the terminated state can only come from the execution queue, and in this case we always have a flow in the executor
             if (executor.getFlow() != null && conditionService.isTerminatedWithListeners(executor.getFlow(), executor.getExecution())) {
-                // purge subflow execution storage
-                subflowExecutionStorage.get(execution.getId())
-                    .ifPresent(subflowExecution -> {
-                        // If we didn't wait for the flow execution, the worker task execution has already been created by the Executor service.
-                        if (subflowExecution.getParentTask() != null && subflowExecution.getParentTask().waitForExecution()) {
-                            sendSubflowExecutionResult(execution, subflowExecution, subflowExecution.getParentTaskRun().withState(execution.getState().getCurrent()));
-                        }
-
-                        subflowExecutionStorage.delete(subflowExecution);
-                    });
+                // if there is a parent, we send a subflow execution result to it
+                if (ExecutableUtils.isSubflow(execution)) {
+                    // locate the parent execution to find the parent task run
+                    String parentExecutionId = (String) execution.getTrigger().getVariables().get("executionId");
+                    String taskRunId = (String) execution.getTrigger().getVariables().get("taskRunId");
+                    String taskId = (String) execution.getTrigger().getVariables().get("taskId");
+                    SubflowExecutionEnd subflowExecutionEnd = new SubflowExecutionEnd(parentExecutionId, taskRunId, taskId, execution.getState().getCurrent());
+                    this.subflowExecutionEndQueue.emit(subflowExecutionEnd);
+                }
 
                 // purge SLA monitors
                 if (!ListUtils.isEmpty(executor.getFlow().getSla()) && executor.getFlow().getSla().stream().anyMatch(ExecutionMonitoringSLA.class::isInstance)) {
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/AbstractSubflowExecutionTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/AbstractSubflowExecutionTest.java
deleted file mode 100644
index cd24296c2c3..00000000000
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/AbstractSubflowExecutionTest.java
+++ /dev/null
@@ -1,78 +0,0 @@
-package io.kestra.jdbc.runner;
-
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.TaskRun;
-import io.kestra.core.runners.DeserializationIssuesCaseTest;
-import io.kestra.core.runners.SubflowExecution;
-import io.kestra.plugin.core.flow.Subflow;
-import io.kestra.core.utils.IdUtils;
-import io.kestra.jdbc.JdbcTestUtils;
-import io.kestra.core.junit.annotations.KestraTest;
-import jakarta.inject.Inject;
-import org.jooq.Field;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-
-@KestraTest
-public abstract class AbstractSubflowExecutionTest {
-    @Inject
-    AbstractJdbcSubflowExecutionStorage subflowExecutionStorage;
-
-    @Inject
-    JdbcTestUtils jdbcTestUtils;
-
-    @Test
-    void suite() throws Exception {
-
-        SubflowExecution<?> workerTaskExecution = SubflowExecution.builder()
-            .execution(Execution.builder().id(IdUtils.create()).build())
-            .parentTask(Subflow.builder().type(Subflow.class.getName()).id(IdUtils.create()).build())
-            .parentTaskRun(TaskRun.builder().id(IdUtils.create()).build())
-            .build();
-
-        subflowExecutionStorage.save(List.of(workerTaskExecution));
-
-
-        Optional<SubflowExecution<?>> find = subflowExecutionStorage.get(workerTaskExecution.getExecution().getId());
-        assertThat(find.isPresent(), is(true));
-        assertThat(find.get().getExecution().getId(), is(workerTaskExecution.getExecution().getId()));
-
-
-        subflowExecutionStorage.delete(workerTaskExecution);
-
-        find = subflowExecutionStorage.get(workerTaskExecution.getExecution().getId());
-        assertThat(find.isPresent(), is(false));
-    }
-
-    @Test
-    void deserializationIssue() {
-        // insert an invalid subflowExecution
-        var subflowExecution = SubflowExecution.builder()
-            .execution(Execution.builder().id(DeserializationIssuesCaseTest.INVALID_SUBFLOW_EXECUTION_KEY).build())
-            .build();
-        Map<Field<Object>, Object> fields = persistFields();
-        subflowExecutionStorage.jdbcRepository.persist(subflowExecution, fields);
-
-        // load it
-        Optional<SubflowExecution<?>> find = subflowExecutionStorage.get(DeserializationIssuesCaseTest.INVALID_SUBFLOW_EXECUTION_KEY);
-        assertThat(find.isPresent(), is(true));
-    }
-
-    protected Map<Field<Object>, Object>  persistFields() {
-        return Map.of(io.kestra.jdbc.repository.AbstractJdbcRepository.field("value"),
-            DeserializationIssuesCaseTest.INVALID_SUBFLOW_EXECUTION_VALUE);
-    }
-
-    @BeforeEach
-    protected void init() {
-        jdbcTestUtils.drop();
-        jdbcTestUtils.migrate();
-    }
-}
\ No newline at end of file