Repository: kestra-io/kestra
Issue: Infinite loop with subflow
Labels: bug

### Expected Behavior

Created a recursive flow create an infinite loop :

```
id: "flow-a"
namespace: "dev"
tasks:
  - id: log
    type: io.kestra.core.tasks.log.Log
    message: hello
  - id: subflow
    type: io.kestra.core.tasks.flows.Flow
    flowId: flow-a
    namespace: dev
```

### Actual Behaviour

_No response_

### Steps To Reproduce

_No response_

### Environment Information

- Kestra Version: 0.10.0-snapshot

### Example flow

_No response_

Issue comments:
I'm not sure it is our responsability to prevent this kind of behaviour ? I don't think it is a good thing to cut off the ability to do recursive flows since there could be a stop condition or something. I see this as a while loop in Java and they are not preventing us from doing an infinite loop. WDYT ?
Or maybe we need to provide a way to stop any child of a given execution recursively instead ?
Right now, we don't the tooling to support flow loop (aka: prevent infinite loop) .
We need at first to block to protect breaking the production env and think later if there is usecase (I almost sure no real use case). 

Pull request review comments:
You also need to check the namespace.
If we cannot create a revursive flow, no need to check this at runtime.
Moreover it seems that this is always true!
Dont we care about possibly existing recursive flow ? 
I would say no, let's keep it simple for now

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
index ef1221806f..bcea3e58ed 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
@@ -15,11 +15,11 @@
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 
+import javax.annotation.Nullable;
+import javax.validation.constraints.NotNull;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
-import javax.annotation.Nullable;
-import javax.validation.constraints.NotNull;
 
 @SuperBuilder
 @ToString
diff --git a/core/src/main/java/io/kestra/core/validations/ValidationFactory.java b/core/src/main/java/io/kestra/core/validations/ValidationFactory.java
index 54218041a1..813bf43d6d 100644
--- a/core/src/main/java/io/kestra/core/validations/ValidationFactory.java
+++ b/core/src/main/java/io/kestra/core/validations/ValidationFactory.java
@@ -15,7 +15,6 @@
 import jakarta.inject.Named;
 import jakarta.inject.Singleton;
 
-import javax.validation.ConstraintViolation;
 import java.io.IOException;
 import java.text.SimpleDateFormat;
 import java.util.*;
@@ -193,6 +192,18 @@ ConstraintValidator<FlowValidation, Flow> flowValidation() {
                 violations.add("Duplicate task id with name [" + String.join(", ", taskDuplicates) + "]");
             }
 
+            value.allTasksWithChilds()
+                .stream()
+                .forEach(
+                    task -> {
+                        if (task instanceof io.kestra.core.tasks.flows.Flow taskFlow
+                                && taskFlow.getFlowId().equals(value.getId())
+                                && taskFlow.getNamespace().equals(value.getNamespace())) {
+                            violations.add("Recursive call to flow [" + value.getId() + "]");
+                        }
+                    }
+                );
+
             // input unique name
             if (value.getInputs() != null) {
                 List<String> inputNames = value.getInputs()


Test patch:
diff --git a/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java b/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java
new file mode 100644
index 0000000000..12747c8668
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java
@@ -0,0 +1,44 @@
+package io.kestra.core.validations;
+
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.validations.ModelValidator;
+import io.kestra.core.serializers.YamlFlowParser;
+import io.kestra.core.utils.TestsUtils;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+import javax.validation.ConstraintViolationException;
+import java.io.File;
+import java.net.URL;
+import java.util.Optional;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+
+@MicronautTest
+class FlowValidationTest {
+    @Inject
+    private ModelValidator modelValidator;
+    @Inject
+    YamlFlowParser yamlFlowParser = new YamlFlowParser();
+
+    @Test
+    void invalidRecursiveFlow() {
+        Flow flow = this.parse("flows/invalids/recursive-flow.yaml");
+        Optional<ConstraintViolationException> validate = modelValidator.isValid(flow);
+
+        assertThat(validate.isPresent(), is(true));
+        assertThat(validate.get().getMessage(), containsString(": Invalid Flow: Recursive call to flow [recursive-flow]"));
+    }
+
+    private Flow parse(String path) {
+        URL resource = TestsUtils.class.getClassLoader().getResource(path);
+        assert resource != null;
+
+        File file = new File(resource.getFile());
+
+        return yamlFlowParser.parse(file, Flow.class);
+    }
+}
diff --git a/core/src/test/resources/flows/invalids/recursive-flow.yaml b/core/src/test/resources/flows/invalids/recursive-flow.yaml
new file mode 100644
index 0000000000..47c80ae769
--- /dev/null
+++ b/core/src/test/resources/flows/invalids/recursive-flow.yaml
@@ -0,0 +1,8 @@
+id: recursive-flow
+namespace: io.kestra.tests
+
+tasks:
+  - id: taskInvalid
+    type: io.kestra.core.tasks.flows.Flow
+    flowId: recursive-flow
+    namespace: io.kestra.tests
\ No newline at end of file