Repository: kestra-io/kestra
Pull request #2450 (2023-11-06 12:56:27)

feat(ui): Select first namespace by default in editor, if none exist and tutorial not done redirect to flow creation
Only redirect if tutorial has been done or skip (as the files editor now allow flow creation)


Patch:
diff --git a/cli/src/main/resources/application.yml b/cli/src/main/resources/application.yml
index 06fa1fad4..4d03906f4 100644
--- a/cli/src/main/resources/application.yml
+++ b/cli/src/main/resources/application.yml
@@ -140,6 +140,9 @@ kestra:
       workerjobrunning:
         cls: io.kestra.core.runners.WorkerJobRunning
         table: "worker_job_running"
+      executionqueued:
+        table: "execution_queued"
+        cls: io.kestra.core.runners.ExecutionQueued
 
     queues:
       min-poll-interval: 25ms
diff --git a/core/src/main/java/io/kestra/core/models/flows/Concurrency.java b/core/src/main/java/io/kestra/core/models/flows/Concurrency.java
new file mode 100644
index 000000000..9faf2e6ea
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/Concurrency.java
@@ -0,0 +1,28 @@
+package io.kestra.core.models.flows;
+
+import io.micronaut.core.annotation.Introspected;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import javax.validation.constraints.NotNull;
+import javax.validation.constraints.Positive;
+
+@SuperBuilder
+@Getter
+@NoArgsConstructor
+@Introspected
+public class Concurrency {
+    @Positive
+    @NotNull
+    private Integer limit;
+
+    @NotNull
+    @Builder.Default
+    private Behavior behavior = Behavior.QUEUE;
+
+    public enum Behavior {
+        QUEUE, CANCEL, FAIL;
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/models/flows/Flow.java b/core/src/main/java/io/kestra/core/models/flows/Flow.java
index df244c14f..3fb9487d1 100644
--- a/core/src/main/java/io/kestra/core/models/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/models/flows/Flow.java
@@ -98,6 +98,8 @@ public class Flow implements DeletedInterface, TenantInterface {
     @Valid
     List<AbstractTrigger> triggers;
 
+
+    @Valid
     List<TaskDefault> taskDefaults;
 
     @NotNull
@@ -108,6 +110,9 @@ public class Flow implements DeletedInterface, TenantInterface {
     @Builder.Default
     boolean deleted = false;
 
+    @Valid
+    Concurrency concurrency;
+
     public Logger logger() {
         return LoggerFactory.getLogger("flow." + this.id);
     }
diff --git a/core/src/main/java/io/kestra/core/models/flows/State.java b/core/src/main/java/io/kestra/core/models/flows/State.java
index e3e759e71..612b36c3a 100644
--- a/core/src/main/java/io/kestra/core/models/flows/State.java
+++ b/core/src/main/java/io/kestra/core/models/flows/State.java
@@ -156,10 +156,11 @@ public class State {
         SUCCESS,
         WARNING,
         FAILED,
-        KILLED;
+        KILLED,
+        CANCELLED;
 
         public boolean isTerminated() {
-            return this == Type.FAILED || this == Type.WARNING || this == Type.SUCCESS || this == Type.KILLED;
+            return this == Type.FAILED || this == Type.WARNING || this == Type.SUCCESS || this == Type.KILLED || this ==  Type.CANCELLED;
         }
 
         public boolean isCreated() {
diff --git a/core/src/main/java/io/kestra/core/models/flows/TaskDefault.java b/core/src/main/java/io/kestra/core/models/flows/TaskDefault.java
index a0c5e892f..fbeb418d8 100644
--- a/core/src/main/java/io/kestra/core/models/flows/TaskDefault.java
+++ b/core/src/main/java/io/kestra/core/models/flows/TaskDefault.java
@@ -1,5 +1,6 @@
 package io.kestra.core.models.flows;
 
+import io.micronaut.core.annotation.Introspected;
 import io.micronaut.core.convert.format.MapFormat;
 import io.micronaut.core.naming.conventions.StringConvention;
 import lombok.AllArgsConstructor;
@@ -11,6 +12,7 @@ import java.util.Map;
 @Getter
 @Builder(toBuilder = true)
 @AllArgsConstructor
+@Introspected
 public class TaskDefault {
     private final String type;
 
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 f11bc7c5c..6f753c29a 100644
--- a/core/src/main/java/io/kestra/core/queues/QueueService.java
+++ b/core/src/main/java/io/kestra/core/queues/QueueService.java
@@ -55,6 +55,8 @@ public class QueueService {
             return ((WorkerTriggerRunning) object).getTriggerContext().uid();
         } else if (object.getClass() == WorkerTriggerResult.class) {
             return ((WorkerTriggerResult) object).getTriggerContext().uid();
+        } else if (object.getClass() == ExecutionQueued.class) {
+            return ((ExecutionQueued) object).uid();
         } else {
             throw new IllegalArgumentException("Unknown type '" + object.getClass().getName() + "'");
         }
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutionQueued.java b/core/src/main/java/io/kestra/core/runners/ExecutionQueued.java
new file mode 100644
index 000000000..722434764
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/runners/ExecutionQueued.java
@@ -0,0 +1,33 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.utils.IdUtils;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Value;
+
+import java.time.Instant;
+import javax.validation.constraints.NotNull;
+
+@Value
+@AllArgsConstructor
+@Builder
+public class ExecutionQueued {
+    String tenantId;
+
+    @NotNull
+    String namespace;
+
+    @NotNull
+    String flowId;
+
+    @NotNull
+    Execution execution;
+
+    @NotNull
+    Instant date;
+
+    public String uid() {
+        return IdUtils.fromParts(this.tenantId, this.namespace, this.flowId, this.execution.getId());
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutionsRunning.java b/core/src/main/java/io/kestra/core/runners/ExecutionsRunning.java
new file mode 100644
index 000000000..5faf55f3f
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/runners/ExecutionsRunning.java
@@ -0,0 +1,5 @@
+package io.kestra.core.runners;
+
+import java.util.Set;
+
+public record ExecutionsRunning(String flowUid, Set<String> executionIds) { }
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 5970c2a40..f34e5d198 100644
--- a/core/src/main/java/io/kestra/core/runners/Executor.java
+++ b/core/src/main/java/io/kestra/core/runners/Executor.java
@@ -10,6 +10,7 @@ import lombok.Getter;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 
 @Getter
 @AllArgsConstructor
@@ -27,6 +28,8 @@ public class Executor {
     private final List<ExecutionDelay> executionDelays = new ArrayList<>();
     private WorkerTaskResult joined;
     private final List<WorkerTaskExecution> workerTaskExecutions = new ArrayList<>();
+    private ExecutionsRunning executionsRunning;
+    private ExecutionQueued executionQueued;
 
     public Executor(Execution execution, Long offset) {
         this.execution = execution;
@@ -98,6 +101,18 @@ public class Executor {
         return this;
     }
 
+    public Executor withExecutionQueued(ExecutionQueued executionQueued) {
+        this.executionQueued = executionQueued;
+
+        return this;
+    }
+
+    public Executor withExecutionsRunning(ExecutionsRunning executionsRunning) {
+        this.executionsRunning = executionsRunning;
+
+        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 fc5defbad..a511cbbb3 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -23,6 +23,7 @@ import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -58,6 +59,36 @@ public class ExecutorService {
         return this.flowExecutorInterface;
     }
 
+    public Executor checkConcurrencyLimit(Executor executor, Flow flow, Execution execution, long count) {
+        if (count >= flow.getConcurrency().getLimit()) {
+            return switch(flow.getConcurrency().getBehavior()) {
+                case QUEUE -> {
+                    ExecutionQueued executionQueued = ExecutionQueued.builder()
+                        .tenantId(flow.getTenantId())
+                        .namespace(flow.getNamespace())
+                        .flowId(flow.getId())
+                        .date(Instant.now())
+                        .execution(execution)
+                        .build();
+
+                    // when max concurrency is reached, we throttle the execution and stop processing
+                    flow.logger().info(
+                        "[namespace: {}] [flow: {}] [execution: {}] Flow is queued due to concurrency limit exceeded",
+                        execution.getNamespace(),
+                        execution.getFlowId(),
+                        execution.getId()
+                    );
+                    // return the execution queued
+                    yield executor.withExecutionQueued(executionQueued);
+                }
+                case CANCEL -> executor.withExecution(execution.withState(State.Type.CANCELLED), "checkConcurrencyLimit");
+                case FAIL -> executor.withException(new IllegalStateException("Flow is FAILED due to concurrency limit exceeded"), "checkConcurrencyLimit");
+            };
+        }
+
+        return executor;
+    }
+
     public Executor process(Executor executor) {
         // previous failed (flow join can fail), just forward
         if (!executor.canBeProcessed()) {
@@ -72,7 +103,7 @@ public class ExecutorService {
             // if all tasks are  killed or terminated, set the execution to killed
             executor = this.handleKilling(executor);
 
-            // killing, so no more nexts
+            // process next task if not killing or killed
             if (executor.getExecution().getState().getCurrent() != State.Type.KILLING && executor.getExecution().getState().getCurrent() != State.Type.KILLED) {
                 executor = this.handleNext(executor);
                 executor = this.handleChildNext(executor);
diff --git a/core/src/main/java/io/kestra/core/runners/RunnerUtils.java b/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
index 7aefa81fe..d9637e956 100644
--- a/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
@@ -338,6 +338,32 @@ public class RunnerUtils {
         }, duration);
     }
 
+    public Execution runOneUntilRunning(String tenantId, String namespace, String flowId) throws TimeoutException {
+        return this.runOneUntilRunning(tenantId, namespace, flowId, null, null, null);
+    }
+
+    public Execution runOneUntilRunning(String tenantId, String namespace, String flowId, Integer revision, BiFunction<Flow, Execution, Map<String, Object>> inputs, Duration duration) throws TimeoutException {
+        return this.runOneUntilRunning(
+            flowRepository
+                .findById(tenantId, namespace, flowId, revision != null ? Optional.of(revision) : Optional.empty())
+                .orElseThrow(() -> new IllegalArgumentException("Unable to find flow '" + flowId + "'")),
+            inputs,
+            duration
+        );
+    }
+
+    public Execution runOneUntilRunning(Flow flow, BiFunction<Flow, Execution, Map<String, Object>> inputs, Duration duration) throws TimeoutException {
+        if (duration == null) {
+            duration = Duration.ofSeconds(15);
+        }
+
+        Execution execution = this.newExecution(flow, inputs, null);
+
+        return this.awaitExecution(isRunningExecution(execution), () -> {
+            this.executionQueue.emit(execution);
+        }, duration);
+    }
+
     public Execution awaitExecution(Predicate<Execution> predicate, Runnable executionEmitter, Duration duration) throws TimeoutException {
         AtomicReference<Execution> receive = new AtomicReference<>();
 
@@ -376,6 +402,10 @@ public class RunnerUtils {
         return e -> e.getId().equals(execution.getId()) && e.getState().isPaused() && e.getTaskRunList().stream().anyMatch(t -> t.getState().isPaused());
     }
 
+    public Predicate<Execution> isRunningExecution(Execution execution) {
+        return e -> e.getId().equals(execution.getId()) && e.getState().isRunning() && e.getTaskRunList().stream().anyMatch(t -> t.getState().isRunning());
+    }
+
     private Predicate<Execution> isTerminatedChildExecution(Execution parentExecution, Flow flow) {
         return e -> e.getParentId() != null && e.getParentId().equals(parentExecution.getId()) && conditionService.isTerminatedWithListeners(flow, e);
     }
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java b/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
index 5e940fba4..28dcf9c2e 100644
--- a/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
+++ b/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
@@ -2,6 +2,7 @@ package io.kestra.core.runners.pebble.functions;
 
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.tenant.TenantService;
+import io.kestra.core.utils.Slugify;
 import io.pebbletemplates.pebble.error.PebbleException;
 import io.pebbletemplates.pebble.extension.Function;
 import io.pebbletemplates.pebble.template.EvaluationContext;
@@ -54,17 +55,33 @@ public class ReadFileFunction implements Function {
     private String readFromInternalStorageUri(EvaluationContext context, String path) throws IOException {
         Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
         Map<String, String> execution = (Map<String, String>) context.getVariable("execution");
-        validateFileUri(flow.get("namespace"), flow.get("id"), execution.get("id"), path);
+
+        // check if the file is from the current execution
+        if (!validateFileUri(flow.get("namespace"), flow.get("id"), execution.get("id"), path)) {
+            // if not, it can be from the parent execution, so we check if there is a trigger of type execution
+            if (context.getVariable("trigger") != null) {
+                // if there is a trigger of type execution, we also allow accessing a file from the parent execution
+                Map<String, String> trigger = (Map<String, String>) context.getVariable("trigger");
+                if (!validateFileUri(trigger.get("namespace"), trigger.get("flowId"), trigger.get("executionId"), path)) {
+                    throw new IllegalArgumentException("Unable to read the file '" + path + "' as it didn't belong to the current execution");
+                }
+            }
+            else {
+                throw new IllegalArgumentException("Unable to read the file '" + path + "' as it didn't belong to the current execution");
+            }
+        }
         URI internalStorageFile = URI.create(path);
         return new String(storageInterface.get(tenantService.resolveTenant(), internalStorageFile).readAllBytes(), StandardCharsets.UTF_8);
     }
 
-    private void validateFileUri(String namespace, String flowId, String executionId, String path) {
+    private boolean validateFileUri(String namespace, String flowId, String executionId, String path) {
         // Internal storage URI should be: kestra:///$namespace/$flowId/executions/$executionId/tasks/$taskName/$taskRunId/$random.ion or kestra:///$namespace/$flowId/executions/$executionId/trigger/$triggerName/$random.ion
         // We check that the file is for the given flow execution
-        String authorizedBasePath = KESTRA_SCHEME + namespace + "/" + flowId + "/executions/" + executionId + "/";
-        if (!path.startsWith(authorizedBasePath)) {
-            throw new IllegalArgumentException("Unable to read a file that didn't belong to the current execution");
+        if (namespace == null || flowId == null || executionId == null) {
+            return false;
         }
+        
+        String authorizedBasePath = KESTRA_SCHEME + Slugify.of(namespace) + "/" + Slugify.of(flowId) + "/executions/" + executionId + "/";
+        return path.startsWith(authorizedBasePath);
     }
 }
\ No newline at end of file
diff --git a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2ExecutionQueuedStorage.java b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2ExecutionQueuedStorage.java
new file mode 100644
index 000000000..2fb87a58d
--- /dev/null
+++ b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2ExecutionQueuedStorage.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.h2;
+
+import io.kestra.core.runners.ExecutionQueued;
+import io.kestra.jdbc.runner.AbstractJdbcExecutionQueuedStorage;
+import io.kestra.repository.h2.H2Repository;
+import io.micronaut.context.ApplicationContext;
+import jakarta.inject.Singleton;
+
+@Singleton
+@H2QueueEnabled
+public class H2ExecutionQueuedStorage extends AbstractJdbcExecutionQueuedStorage {
+    public H2ExecutionQueuedStorage(ApplicationContext applicationContext) {
+        super(new H2Repository<>(ExecutionQueued.class, applicationContext));
+    }
+}
diff --git a/jdbc-h2/src/main/resources/migrations/h2/V1_6__execution_queued.sql b/jdbc-h2/src/main/resources/migrations/h2/V1_6__execution_queued.sql
new file mode 100644
index 000000000..796ab8f1e
--- /dev/null
+++ b/jdbc-h2/src/main/resources/migrations/h2/V1_6__execution_queued.sql
@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS execution_queued (
+    "key" VARCHAR(250) NOT NULL PRIMARY KEY,
+    "value" TEXT NOT NULL,
+    "tenant_id" VARCHAR(250) GENERATED ALWAYS AS (JQ_STRING("value", '.tenantId')),
+    "namespace" VARCHAR(150) NOT NULL GENERATED ALWAYS AS (JQ_STRING("value", '.namespace')),
+    "flow_id" VARCHAR(150) NOT NULL GENERATED ALWAYS AS (JQ_STRING("value", '.flowId')),
+    "date" TIMESTAMP NOT NULL GENERATED ALWAYS AS (PARSEDATETIME(JQ_STRING("value", '.date'), 'yyyy-MM-dd''T''HH:mm:ss.SSSXXX'))
+);
+
+CREATE INDEX IF NOT EXISTS execution_queued__flow ON execution_queued ("tenant_id", "namespace", "flow_id");
\ No newline at end of file
diff --git a/jdbc-h2/src/main/resources/migrations/h2/V1_7__execution_cancelled.sql b/jdbc-h2/src/main/resources/migrations/h2/V1_7__execution_cancelled.sql
new file mode 100644
index 000000000..6cc74a691
--- /dev/null
+++ b/jdbc-h2/src/main/resources/migrations/h2/V1_7__execution_cancelled.sql
@@ -0,0 +1,12 @@
+ALTER TABLE executions ALTER COLUMN "state_current" ENUM (
+    'CREATED',
+    'RUNNING',
+    'PAUSED',
+    'RESTARTED',
+    'KILLING',
+    'SUCCESS',
+    'WARNING',
+    'FAILED',
+    'KILLED',
+    'CANCELLED'
+) NOT NULL GENERATED ALWAYS AS (JQ_STRING("value", '.state.current'));
\ No newline at end of file
diff --git a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlExecutionQueuedStorage.java b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlExecutionQueuedStorage.java
new file mode 100644
index 000000000..4f16615c9
--- /dev/null
+++ b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlExecutionQueuedStorage.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.mysql;
+
+import io.kestra.core.runners.ExecutionQueued;
+import io.kestra.jdbc.runner.AbstractJdbcExecutionQueuedStorage;
+import io.kestra.repository.mysql.MysqlRepository;
+import io.micronaut.context.ApplicationContext;
+import jakarta.inject.Singleton;
+
+@Singleton
+@MysqlQueueEnabled
+public class MysqlExecutionQueuedStorage extends AbstractJdbcExecutionQueuedStorage {
+    public MysqlExecutionQueuedStorage(ApplicationContext applicationContext) {
+        super(new MysqlRepository<>(ExecutionQueued.class, applicationContext));
+    }
+}
diff --git a/jdbc-mysql/src/main/resources/migrations/mysql/V1_7__execution_queued.sql b/jdbc-mysql/src/main/resources/migrations/mysql/V1_7__execution_queued.sql
new file mode 100644
index 000000000..c618bb0c0
--- /dev/null
+++ b/jdbc-mysql/src/main/resources/migrations/mysql/V1_7__execution_queued.sql
@@ -0,0 +1,9 @@
+CREATE TABLE IF NOT EXISTS execution_queued (
+    `key` VARCHAR(250) NOT NULL PRIMARY KEY,
+    `value` JSON NOT NULL,
+    `tenant_id` VARCHAR(250) GENERATED ALWAYS AS (value ->> '$.tenantId') STORED,
+    `namespace` VARCHAR(150) GENERATED ALWAYS AS (value ->> '$.namespace') STORED NOT NULL,
+    `flow_id` VARCHAR(150) GENERATED ALWAYS AS (value ->> '$.flowId') STORED NOT NULL,
+    `date` DATETIME(6) GENERATED ALWAYS AS (STR_TO_DATE(value ->> '$.date' , '%Y-%m-%dT%H:%i:%s.%fZ')) STORED NOT NULL,
+    INDEX ix_flow (tenant_id, namespace, flow_id)
+);
\ No newline at end of file
diff --git a/jdbc-mysql/src/main/resources/migrations/mysql/V1_8__execution_cancelled.sql b/jdbc-mysql/src/main/resources/migrations/mysql/V1_8__execution_cancelled.sql
new file mode 100644
index 000000000..9cd230f1a
--- /dev/null
+++ b/jdbc-mysql/src/main/resources/migrations/mysql/V1_8__execution_cancelled.sql
@@ -0,0 +1,12 @@
+ALTER TABLE executions MODIFY COLUMN `state_current` ENUM (
+    'CREATED',
+    'RUNNING',
+    'PAUSED',
+    'RESTARTED',
+    'KILLING',
+    'SUCCESS',
+    'WARNING',
+    'FAILED',
+    'KILLED',
+    'CANCELLED'
+) GENERATED ALWAYS AS (value ->> '$.state.current') STORED NOT NULL;
\ No newline at end of file
diff --git a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresExecutionQueuedStorage.java b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresExecutionQueuedStorage.java
new file mode 100644
index 000000000..c2a455783
--- /dev/null
+++ b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresExecutionQueuedStorage.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.postgres;
+
+import io.kestra.core.runners.ExecutionQueued;
+import io.kestra.jdbc.runner.AbstractJdbcExecutionQueuedStorage;
+import io.kestra.repository.postgres.PostgresRepository;
+import io.micronaut.context.ApplicationContext;
+import jakarta.inject.Singleton;
+
+@Singleton
+@PostgresQueueEnabled
+public class PostgresExecutionQueuedStorage extends AbstractJdbcExecutionQueuedStorage {
+    public PostgresExecutionQueuedStorage(ApplicationContext applicationContext) {
+        super(new PostgresRepository<>(ExecutionQueued.class, applicationContext));
+    }
+}
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V1_7__execution_queued.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V1_7__execution_queued.sql
new file mode 100644
index 000000000..36ab0f33d
--- /dev/null
+++ b/jdbc-postgres/src/main/resources/migrations/postgres/V1_7__execution_queued.sql
@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS execution_queued (
+    key VARCHAR(250) NOT NULL PRIMARY KEY,
+    value JSONB NOT NULL,
+    tenant_id VARCHAR(250) GENERATED ALWAYS AS (value ->> 'tenantId') STORED,
+    namespace VARCHAR(150) NOT NULL GENERATED ALWAYS AS (value ->> 'namespace') STORED,
+    flow_id VARCHAR(150) NOT NULL GENERATED ALWAYS AS (value ->> 'flowId') STORED,
+    date TIMESTAMPTZ NOT NULL GENERATED ALWAYS AS (PARSE_ISO8601_DATETIME(value ->> 'date')) STORED
+);
+
+CREATE INDEX IF NOT EXISTS execution_queued__flow ON execution_queued (tenant_id, namespace, flow_id);
\ No newline at end of file
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V1_8__execution_cancelled.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V1_8__execution_cancelled.sql
new file mode 100644
index 000000000..882ce4ef6
--- /dev/null
+++ b/jdbc-postgres/src/main/resources/migrations/postgres/V1_8__execution_cancelled.sql
@@ -0,0 +1,1 @@
+ALTER TYPE state_type ADD VALUE IF NOT EXISTS 'CANCELLED';
\ No newline at end of file
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
index 31402f030..f9aa7e0e2 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
@@ -575,13 +575,18 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
                         DSL.count().as("count")
                     ))
                     .from(this.jdbcRepository.getTable())
-                    .where(this.defaultFilter(tenantId))
-                    .and(field("start_date").greaterOrEqual(finalStartDate.toOffsetDateTime()))
-                    .and(field("end_date").lessOrEqual(finalEndDate.toOffsetDateTime()));
+                    .where(this.defaultFilter(tenantId));
+
+                if (startDate != null) {
+                    select = select.and(field("start_date").greaterOrEqual(finalStartDate.toOffsetDateTime()));
+                }
+
+                if (endDate != null) {
+                    select = select.and(field("end_date").lessOrEqual(finalEndDate.toOffsetDateTime()));
+                }
 
                 if (states != null) {
-                    select = select.and(field("state_current")
-                        .in(states.stream().map(Enum::name).collect(Collectors.toList())));
+                    select = select.and(this.statesFilter(states));
                 }
 
                 // add flow & namespace filters
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcExecutionQueuedStorage.java b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcExecutionQueuedStorage.java
new file mode 100644
index 000000000..434429531
--- /dev/null
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcExecutionQueuedStorage.java
@@ -0,0 +1,48 @@
+package io.kestra.jdbc.runner;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.runners.ExecutionQueued;
+import io.kestra.jdbc.repository.AbstractJdbcRepository;
+import org.jooq.Field;
+import org.jooq.impl.DSL;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Consumer;
+
+public abstract class AbstractJdbcExecutionQueuedStorage extends AbstractJdbcRepository {
+    protected io.kestra.jdbc.AbstractJdbcRepository<ExecutionQueued> jdbcRepository;
+
+    public AbstractJdbcExecutionQueuedStorage(io.kestra.jdbc.AbstractJdbcRepository<ExecutionQueued> jdbcRepository) {
+        this.jdbcRepository = jdbcRepository;
+    }
+
+    public void save(ExecutionQueued executionQueued) {
+        Map<Field<Object>, Object> fields = this.jdbcRepository.persistFields(executionQueued);
+        this.jdbcRepository.persist(executionQueued, fields);
+    }
+
+    public void pop(String tenantId, String namespace, String flowId, Consumer<Execution> consumer) {
+        this.jdbcRepository
+            .getDslContextWrapper()
+            .transaction(configuration -> {
+                var select = DSL
+                    .using(configuration)
+                    .select(AbstractJdbcRepository.field("value"))
+                    .from(this.jdbcRepository.getTable())
+                    .where(buildTenantCondition(tenantId))
+                    .and(field("namespace").eq(namespace))
+                    .and(field("flow_id").eq(flowId))
+                    .orderBy(field("date").asc())
+                    .limit(1)
+                    .forUpdate();
+
+                Optional<ExecutionQueued> maybeExecution = this.jdbcRepository.fetchOne(select);
+                if (maybeExecution.isPresent()) {
+                    consumer.accept(maybeExecution.get().getExecution());
+                    this.jdbcRepository.delete(maybeExecution.get());
+                }
+            });
+    }
+
+}
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 3b985c852..1e69b8979 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -6,6 +6,8 @@ import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.executions.statistics.ExecutionCount;
+import io.kestra.core.models.flows.Concurrency;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithException;
 import io.kestra.core.models.flows.State;
@@ -116,7 +118,10 @@ public class JdbcExecutor implements ExecutorInterface {
     private ExecutionService executionService;
 
     @Inject
-    private AbstractJdbcExecutionDelayStorage abstractExecutionDelayStorage;
+    private AbstractJdbcExecutionDelayStorage executionDelayStorage;
+
+    @Inject
+    private AbstractJdbcExecutionQueuedStorage executionQueuedStorage;
 
     @Inject
     private AbstractJdbcExecutorStateStorage executorStateStorage;
@@ -292,10 +297,38 @@ public class JdbcExecutor implements ExecutorInterface {
             final Flow flow = transform(this.flowRepository.findByExecution(execution), execution);
             Executor executor = new Executor(execution, null).withFlow(flow);
 
+            // queue execution if needed (limit concurrency)
+            if (execution.getState().isCreated() && flow.getConcurrency() != null) {
+                ExecutionCount count = executionRepository.executionCounts(
+                    flow.getTenantId(),
+                    List.of(new io.kestra.core.models.executions.statistics.Flow(flow.getNamespace(), flow.getId())),
+                    List.of(State.Type.RUNNING),
+                    null,
+                    null
+                ).get(0);
+
+                executor = executorService.checkConcurrencyLimit(executor, flow, execution, count.getCount());
+                if (executor.getExecutionQueued() != null) {
+                    // the execution has been queued, we save the queued execution and stops here
+                    executionQueuedStorage.save(executor.getExecutionQueued());
+                    return Pair.of(
+                        executor,
+                        executorState
+                    );
+                }
+                if (executor.getExecution().getState().isTerminated()) {
+                    // the execution has been moved to FAILED or CANCELLED, we stop here
+                    return Pair.of(
+                        executor,
+                        executorState
+                    );
+                }
+            }
+
+            // process the execution
             if (log.isDebugEnabled()) {
                 executorService.log(log, true, executor);
             }
-
             executor = executorService.process(executor);
 
             if (!executor.getNexts().isEmpty() && deduplicateNexts(execution, executorState, executor.getNexts())) {
@@ -336,7 +369,7 @@ public class JdbcExecutor implements ExecutorInterface {
             // schedulerDelay
             if (!executor.getExecutionDelays().isEmpty()) {
                 executor.getExecutionDelays()
-                    .forEach(executionDelay -> abstractExecutionDelayStorage.save(executionDelay));
+                    .forEach(executionDelay -> executionDelayStorage.save(executionDelay));
             }
 
             // worker task execution watchers
@@ -384,7 +417,7 @@ public class JdbcExecutor implements ExecutorInterface {
                     .forEach(this.executionQueue::emit);
             }
 
-            // worker task execution
+            // when terminated: handle worker task execution and execution throttled
             if (conditionService.isTerminatedWithListeners(flow, execution)) {
                 workerTaskExecutionStorage.get(execution.getId())
                     .ifPresent(workerTaskExecution -> {
@@ -413,6 +446,15 @@ public class JdbcExecutor implements ExecutorInterface {
 
                         workerTaskExecutionStorage.delete(workerTaskExecution);
                     });
+
+                if (flow.getConcurrency() != null && flow.getConcurrency().getBehavior() == Concurrency.Behavior.QUEUE) {
+                    // as the execution is terminated, we can check if there exist a queued execution and submit it to the execution queue
+                    executionQueuedStorage.pop(flow.getTenantId(),
+                        flow.getNamespace(),
+                        flow.getId(),
+                        queued -> executionQueue.emit(queued)
+                    );
+                }
             }
 
             return Pair.of(
@@ -573,7 +615,7 @@ public class JdbcExecutor implements ExecutorInterface {
             return;
         }
 
-        abstractExecutionDelayStorage.get(executionDelay -> {
+        executionDelayStorage.get(executionDelay -> {
             Executor result = executionRepository.lock(executionDelay.getExecutionId(), pair -> {
                 Executor executor = new Executor(pair.getLeft(), null);
 
diff --git a/ui/src/components/flows/FlowRun.vue b/ui/src/components/flows/FlowRun.vue
index e02647f56..b56996e48 100644
--- a/ui/src/components/flows/FlowRun.vue
+++ b/ui/src/components/flows/FlowRun.vue
@@ -129,6 +129,7 @@
     import LabelInput from "../../components/labels/LabelInput.vue";
     import Markdown from "../layout/Markdown.vue";
     import {pageFromRoute} from "../../utils/eventsRouter";
+    import {executeFlowBehaviours, storageKeys} from "../../utils/constants";
 
     export default {
         components: {Editor, LabelInput},
@@ -136,10 +137,6 @@
             redirect: {
                 type: Boolean,
                 default: true
-            },
-            newTab: {
-                type: Boolean,
-                default: false
             }
         },
         data() {
@@ -148,7 +145,8 @@
                 inputNewLabel: "",
                 executionLabels: [],
                 inputVisible: false,
-                collapseName: undefined
+                collapseName: undefined,
+                newTab: localStorage.getItem(storageKeys.EXECUTE_FLOW_BEHAVIOUR) === executeFlowBehaviours.NEW_TAB
             };
         },
         emits: ["executionTrigger"],
diff --git a/ui/src/components/flows/TriggerFlow.vue b/ui/src/components/flows/TriggerFlow.vue
index 7e48df33a..22124e887 100644
--- a/ui/src/components/flows/TriggerFlow.vue
+++ b/ui/src/components/flows/TriggerFlow.vue
@@ -1,18 +1,11 @@
 <template>
     <div class="trigger-flow-wrapper">
-        <el-dropdown class="edit-flow-trigger-button" split-button :icon="icon.Flash" :disabled=isDisabled() :type="type" @click="onClick(false)">
-            <Flash /> <span class="mx-2">{{ $t('execute') }}</span>
-            <template #dropdown>
-                <el-dropdown-menu>
-                    <el-button :icon="icon.FlashAlert" :type="type" @click="onClick(true)">{{ $t('execute new tab') }}</el-button>
-                </el-dropdown-menu>
-            </template>
-        </el-dropdown>
+        <el-button :icon="icon.Flash" :type="type" :disabled="isDisabled()" @click="onClick()">{{ $t("execute") }}</el-button>
         <el-dialog v-if="isOpen" v-model="isOpen" destroy-on-close :append-to-body="true">
             <template #header>
                 <span v-html="$t('execute the flow', {id: flowId})" />
             </template>
-            <flow-run @execution-trigger="closeModal" :redirect="true" :new-tab="newTab" />
+            <flow-run @execution-trigger="closeModal" :redirect="true" />
         </el-dialog>
     </div>
 </template>
@@ -22,14 +15,12 @@
     import FlowRun from "./FlowRun.vue";
     import {mapState} from "vuex";
     import Flash from "vue-material-design-icons/Flash.vue";
-    import FlashAlert from "vue-material-design-icons/FlashAlert.vue";
     import {shallowRef} from "vue";
     import {pageFromRoute} from "../../utils/eventsRouter";
 
     export default {
         components: {
-            FlowRun,
-            Flash
+            FlowRun
         },
         props: {
             flowId: {
@@ -52,9 +43,8 @@
         data() {
             return {
                 isOpen: false,
-                newTab: false,
                 icon: {
-                    FlashAlert: shallowRef(FlashAlert)
+                    Flash: shallowRef(Flash)
                 }
             };
         },
@@ -69,7 +59,7 @@
             }
         },
         methods: {
-            onClick(newTab) {
+            onClick() {
                 if (this.$tours["guidedTour"].isRunning.value && !this.guidedProperties.executeFlow) {
                     this.$store.dispatch("api/events", {
                         type: "ONBOARDING",
@@ -82,8 +72,7 @@
                     this.$tours["guidedTour"].nextStep();
                     return;
                 }
-                this.newTab = newTab;
-                this.isOpen = !this.isOpen
+                this.isOpen = !this.isOpen;
             },
             closeModal() {
                 this.isOpen = false;
diff --git a/ui/src/components/layout/Labels.vue b/ui/src/components/layout/Labels.vue
index 84c8c0001..fbe0fdb03 100644
--- a/ui/src/components/layout/Labels.vue
+++ b/ui/src/components/layout/Labels.vue
@@ -4,7 +4,7 @@
         <el-tag
             v-for="(value, key) in labelMap"
             :key="key"
-            :type="checked(key) ? 'info' : ''"
+            :type="checked(key, value) ? 'info' : ''"
             class="me-1 labels"
             size="small"
             disable-transitions
@@ -31,16 +31,13 @@
         },
         // this is needed as flows uses a Map and Execution a List of Labels.
         // if we align both of them this can be removed
-        mounted() {
-            if (Array.isArray(this.labels)) {
-                this.labelMap = Object.fromEntries(this.labels.map(label => [label.key, label.value]))
-            } else {
-                this.labelMap = this.labels;
-            }
-        },
-        data() {
-            return {
-                labelMap: {}
+        computed: {
+            labelMap() {
+                if (Array.isArray(this.labels)) {
+                    return Object.fromEntries(this.labels.map(label => [label.key, label.value]));
+                } else {
+                    return this.labels;
+                }
             }
         },
         methods: {
@@ -58,8 +55,8 @@
 
                 return labels;
             },
-            checked(key) {
-                return this.getLabelsFromQuery().has(key);
+            checked(key, value) {
+                return this.getLabelsFromQuery().has(key) && this.getLabelsFromQuery().get(key) === value;
             },
             link(key, value) {
                 const labels = this.getLabelsFromQuery();
diff --git a/ui/src/components/namespace/Editor.vue b/ui/src/components/namespace/Editor.vue
index b70bf3080..046c74829 100644
--- a/ui/src/components/namespace/Editor.vue
+++ b/ui/src/components/namespace/Editor.vue
@@ -40,6 +40,7 @@
     import RouteContext from "../../mixins/routeContext";
     import RestoreUrl from "../../mixins/restoreUrl";
     import {apiUrl} from "override/utils/route";
+    import {mapState} from "vuex";
 
     export default {
         mixins: [RouteContext, RestoreUrl],
@@ -65,12 +66,6 @@
                 tabsNotSaved: []
             }
         },
-        created() {
-            const namespace = localStorage.getItem("defaultNamespace");
-            if (namespace) {
-                this.namespaceUpdate(namespace);
-            }
-        },
         mounted() {
             window.addEventListener("message", (event) => {
                 const message = event.data;
@@ -85,8 +80,19 @@
                     this.handleTabsDirty(message.tabs);
                 }
             });
+
+            // Setup namespace
+            const namespace = localStorage.getItem("defaultNamespace");
+            if (namespace) {
+                this.namespaceUpdate(namespace);
+            } else if (this.namespaces?.length > 0) {
+                this.namespaceUpdate(this.namespaces[0]);
+            } else if (localStorage.getItem("tourDoneOrSkip") !== "true") {
+                this.$router.push({name: "flows/create"});
+            }
         },
         computed: {
+            ...mapState("namespace", ["namespaces"]),
             routeInfo() {
                 return {
                     title: this.$t("editor")
diff --git a/ui/src/components/settings/BasicSettings.vue b/ui/src/components/settings/BasicSettings.vue
index 844a59a39..262da3b9a 100644
--- a/ui/src/components/settings/BasicSettings.vue
+++ b/ui/src/components/settings/BasicSettings.vue
@@ -126,6 +126,16 @@
                     show-alpha
                 />
             </el-form-item>
+            <el-form-item :label="$t('execute flow behaviour')">
+                <el-select :model-value="executeFlowBehaviour" @update:model-value="onExecuteFlowBehaviourChange">
+                    <el-option
+                        v-for="item in Object.values(executeFlowBehaviours)"
+                        :key="item"
+                        :label="$t(`open in ${item}`)"
+                        :value="item"
+                    />
+                </el-select>
+            </el-form-item>
             <slot name="form-items"/>
         </el-form>
     </div>
@@ -133,6 +143,7 @@
 
 <script setup>
     import Download from "vue-material-design-icons/Download.vue";
+    import {executeFlowBehaviours} from "../../utils/constants";
 </script>
 
 <script>
@@ -144,7 +155,7 @@
     import {mapGetters, mapState, useStore} from "vuex";
     import permission from "../../models/permission";
     import action from "../../models/action";
-    import {logDisplayTypes} from "../../utils/constants";
+    import {logDisplayTypes, storageKeys} from "../../utils/constants";
 
     export const DATE_FORMAT_STORAGE_KEY = "dateFormat";
     export const TIMEZONE_STORAGE_KEY = "timezone";
@@ -177,6 +188,7 @@
                 logDisplay: undefined,
                 editorFontSize: undefined,
                 editorFontFamily: undefined,
+                executeFlowBehaviour: undefined,
                 now: this.$moment(),
                 envName: undefined,
                 envColor: undefined
@@ -198,6 +210,7 @@
             this.logDisplay = localStorage.getItem("logDisplay") || logDisplayTypes.DEFAULT;
             this.editorFontSize = localStorage.getItem("editorFontSize") || 12;
             this.editorFontFamily = localStorage.getItem("editorFontFamily") || "'Source Code Pro', monospace";
+            this.executeFlowBehaviour = localStorage.getItem("executeFlowBehaviour") || "same tab";
             this.envName = store.getters["layout/envName"] || this.configs?.environment?.name;
             this.envColor = store.getters["layout/envColor"] || this.configs?.environment?.color;
         },
@@ -295,6 +308,13 @@
                     this.$store.commit("layout/setEnvColor", value);
                 }
 
+                this.$toast().saved();
+            },
+            onExecuteFlowBehaviourChange(value) {
+                this.executeFlowBehaviour = value;
+
+                localStorage.setItem(storageKeys.EXECUTE_FLOW_BEHAVIOUR, value);
+
                 this.$toast().saved();
             }
         },
diff --git a/ui/src/translations.json b/ui/src/translations.json
index 17f069127..4383214c4 100644
--- a/ui/src/translations.json
+++ b/ui/src/translations.json
@@ -242,7 +242,9 @@
     "Total": "Total",
     "sub flow": "Sub-flow",
     "execute": "Execute",
-    "execute new tab": "Execute in new tab",
+    "execute flow behaviour": "Execute the flow",
+    "open in same tab": "In the same tab",
+    "open in new tab": "In a new tab",
     "execute the flow": "Execute the flow <code>{id}</code>",
     "execute flow now ?": "Do you want to execute this flow?",
     "Default namespace": "Default namespace",
@@ -760,7 +762,9 @@
     "Total": "Total",
     "sub flow": "Sous-flow",
     "execute": "Exécuter",
-    "execute new tab": "Exécuter dans un nouvel onglet",
+    "execute flow behaviour": "Exécuter le flow",
+    "open in same tab": "Dans le même onglet",
+    "open in new tab": "Dans un nouvel onglet",
     "execute the flow": "Exécuter le flow <code>{id}</code>",
     "execute flow now ?": "Voulez-vous exécuter ce flow ?",
     "Default namespace": "Espace de nom par défaut",
diff --git a/ui/src/utils/constants.js b/ui/src/utils/constants.js
index a19cd0a7a..d7657ef36 100644
--- a/ui/src/utils/constants.js
+++ b/ui/src/utils/constants.js
@@ -27,5 +27,11 @@ export const editorViewTypes = {
 export const storageKeys = {
     DISPLAY_EXECUTIONS_COLUMNS: "displayExecutionsColumns",
     DISPLAY_FLOW_EXECUTIONS_COLUMNS: "displayFlowExecutionsColumns",
-    SELECTED_TENANT: "selectedTenant"
+    SELECTED_TENANT: "selectedTenant",
+    EXECUTE_FLOW_BEHAVIOUR: "executeFlowBehaviour"
+}
+
+export const executeFlowBehaviours = {
+    SAME_TAB: "same tab",
+    NEW_TAB: "new tab"
 }
\ No newline at end of file
diff --git a/ui/src/utils/state.js b/ui/src/utils/state.js
index 6fb690b8c..87946e7f7 100644
--- a/ui/src/utils/state.js
+++ b/ui/src/utils/state.js
@@ -8,6 +8,7 @@ import SkipPreviousCircle from "vue-material-design-icons/SkipPreviousCircle.vue
 import AlertCircle from "vue-material-design-icons/AlertCircle.vue";
 import DotsVerticalCircle from "vue-material-design-icons/DotsVerticalCircle.vue";
 import {cssVariable} from "./global"
+import StopCircleOutline from "vue-material-design-icons/StopCircleOutline.vue";
 
 const STATE = Object.freeze({
     CREATED: {
@@ -81,6 +82,14 @@ const STATE = Object.freeze({
         isRunning: true,
         isKillable: true,
         isFailed: false,
+    },
+    CANCELLED: {
+        name: "CANCELLED",
+        colorClass: "gray",
+        icon: StopCircleOutline,
+        isRunning: false,
+        isKillable: false,
+        isFailed: true,
     }
 });
 
@@ -121,6 +130,10 @@ export default class State {
         return STATE.PAUSED.name;
     }
 
+    static get CANCELLED() {
+        return STATE.CANCELLED.name;
+    }
+
     static isRunning(state) {
         return STATE[state] && STATE[state].isRunning;
     }


Test patch:
diff --git a/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java b/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java
index 877d0a237..86389f1ff 100644
--- a/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java
+++ b/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java
@@ -348,20 +348,35 @@ public abstract class AbstractExecutionRepositoryTest {
         List<ExecutionCount> result = executionRepository.executionCounts(
             null,
             List.of(
-                new io.kestra.core.models.executions.statistics.Flow(NAMESPACE, "first"),
-                new io.kestra.core.models.executions.statistics.Flow(NAMESPACE, "second"),
-                new io.kestra.core.models.executions.statistics.Flow(NAMESPACE, "third"),
+                new Flow(NAMESPACE, "first"),
+                new Flow(NAMESPACE, "second"),
+                new Flow(NAMESPACE, "third"),
                 new Flow(NAMESPACE, "missing")
             ),
             null,
             ZonedDateTime.now().minusDays(10),
             ZonedDateTime.now()
         );
-
         assertThat(result.size(), is(4));
         assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("first")).findFirst().get().getCount(), is(2L));
         assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("second")).findFirst().get().getCount(), is(3L));
         assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("third")).findFirst().get().getCount(), is(9L));
         assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("missing")).findFirst().get().getCount(), is(0L));
+
+        result = executionRepository.executionCounts(
+            null,
+            List.of(
+                new Flow(NAMESPACE, "first"),
+                new Flow(NAMESPACE, "second"),
+                new Flow(NAMESPACE, "third")
+            ),
+            List.of(State.Type.SUCCESS),
+            null,
+            null
+        );
+        assertThat(result.size(), is(3));
+        assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("first")).findFirst().get().getCount(), is(2L));
+        assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("second")).findFirst().get().getCount(), is(3L));
+        assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("third")).findFirst().get().getCount(), is(9L));
     }
 }
diff --git a/core/src/test/java/io/kestra/core/runners/FlowConcurrencyCaseTest.java b/core/src/test/java/io/kestra/core/runners/FlowConcurrencyCaseTest.java
new file mode 100644
index 000000000..f7f87fbe8
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/runners/FlowConcurrencyCaseTest.java
@@ -0,0 +1,151 @@
+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 jakarta.inject.Inject;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+@Singleton
+public class FlowConcurrencyCaseTest {
+    @Inject
+    private RunnerUtils runnerUtils;
+
+    @Inject
+    private FlowRepositoryInterface flowRepository;
+
+    @Inject
+    @Named(QueueFactoryInterface.EXECUTION_NAMED)
+    protected QueueInterface<Execution> executionQueue;
+
+    public void flowConcurrencyCancel() throws TimeoutException, InterruptedException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(null, "io.kestra.tests", "flow-concurrency-cancel", null, null, Duration.ofSeconds(30));
+        Execution execution2 = runnerUtils.runOne(null, "io.kestra.tests", "flow-concurrency-cancel");
+
+        assertThat(execution1.getState().isRunning(), is(true));
+        assertThat(execution2.getState().getCurrent(), is(State.Type.CANCELLED));
+
+        var executionResult1  = new AtomicReference<Execution>();
+        var executionResult2  = new AtomicReference<Execution>();
+
+        CountDownLatch latch1 = new CountDownLatch(1);
+        CountDownLatch latch2 = new CountDownLatch(1);
+
+        executionQueue.receive(e -> {
+            if (e.getLeft().getId().equals(execution1.getId())) {
+                executionResult1.set(e.getLeft());
+                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
+                    latch1.countDown();
+                }
+            }
+
+            if (e.getLeft().getId().equals(execution2.getId())) {
+                executionResult2.set(e.getLeft());
+                if (e.getLeft().getState().getCurrent() == State.Type.CANCELLED) {
+                    latch2.countDown();
+                }
+            }
+        });
+
+        latch1.await(1, TimeUnit.MINUTES);
+        latch2.await(1, TimeUnit.MINUTES);
+
+        assertThat(executionResult1.get().getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(executionResult2.get().getState().getCurrent(), is(State.Type.CANCELLED));
+    }
+
+    public void flowConcurrencyFail() throws TimeoutException, InterruptedException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(null, "io.kestra.tests", "flow-concurrency-fail", null, null, Duration.ofSeconds(30));
+        Execution execution2 = runnerUtils.runOne(null, "io.kestra.tests", "flow-concurrency-fail");
+
+        assertThat(execution1.getState().isRunning(), is(true));
+        assertThat(execution2.getState().getCurrent(), is(State.Type.FAILED));
+
+        var executionResult1  = new AtomicReference<Execution>();
+        var executionResult2  = new AtomicReference<Execution>();
+
+        CountDownLatch latch1 = new CountDownLatch(1);
+        CountDownLatch latch2 = new CountDownLatch(1);
+
+        executionQueue.receive(e -> {
+            if (e.getLeft().getId().equals(execution1.getId())) {
+                executionResult1.set(e.getLeft());
+                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
+                    latch1.countDown();
+                }
+            }
+
+            if (e.getLeft().getId().equals(execution2.getId())) {
+                executionResult2.set(e.getLeft());
+                if (e.getLeft().getState().getCurrent() == State.Type.FAILED) {
+                    latch2.countDown();
+                }
+            }
+        });
+
+        latch1.await(1, TimeUnit.MINUTES);
+        latch2.await(1, TimeUnit.MINUTES);
+
+        assertThat(executionResult1.get().getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(executionResult2.get().getState().getCurrent(), is(State.Type.FAILED));
+    }
+
+    public void flowConcurrencyQueue() throws TimeoutException, InterruptedException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(null, "io.kestra.tests", "flow-concurrency-queue", null, null, Duration.ofSeconds(30));
+        Flow flow = flowRepository
+            .findById(null, "io.kestra.tests", "flow-concurrency-queue", Optional.empty())
+            .orElseThrow();
+        Execution execution2 = runnerUtils.newExecution(flow, null, null);
+        executionQueue.emit(execution2);
+
+        assertThat(execution1.getState().isRunning(), is(true));
+        assertThat(execution2.getState().getCurrent(), is(State.Type.CREATED));
+
+        var executionResult1  = new AtomicReference<Execution>();
+        var executionResult2  = new AtomicReference<Execution>();
+
+        CountDownLatch latch1 = new CountDownLatch(1);
+        CountDownLatch latch2 = new CountDownLatch(1);
+        CountDownLatch latch3 = new CountDownLatch(1);
+
+        executionQueue.receive(e -> {
+            if (e.getLeft().getId().equals(execution1.getId())) {
+                executionResult1.set(e.getLeft());
+                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
+                    latch1.countDown();
+                }
+            }
+
+            if (e.getLeft().getId().equals(execution2.getId())) {
+                executionResult2.set(e.getLeft());
+                if (e.getLeft().getState().getCurrent() == State.Type.RUNNING) {
+                    latch2.countDown();
+                }
+                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
+                    latch3.countDown();
+                }
+            }
+        });
+
+        latch1.await(1, TimeUnit.MINUTES);
+        latch2.await(1, TimeUnit.MINUTES);
+        latch3.await(1, TimeUnit.MINUTES);
+
+        assertThat(executionResult1.get().getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(executionResult2.get().getState().getCurrent(), is(State.Type.SUCCESS));
+    }
+}
diff --git a/core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java b/core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java
index 2d06c1bf3..4e52da151 100644
--- a/core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java
+++ b/core/src/test/java/io/kestra/core/runners/pebble/functions/ReadFileFunctionTest.java
@@ -52,6 +52,8 @@ class ReadFileFunctionTest {
         String executionId = IdUtils.create();
         URI internalStorageURI = URI.create("/" + namespace + "/" + flowId + "/executions/" + executionId + "/tasks/task/" + IdUtils.create() + "/123456.ion");
         URI internalStorageFile = storageInterface.put(null, internalStorageURI, new ByteArrayInputStream("Hello from a task output".getBytes()));
+
+        // test for an authorized execution
         Map<String, Object> variables = Map.of(
             "flow", Map.of(
                 "id", flowId,
@@ -61,15 +63,33 @@ class ReadFileFunctionTest {
 
         String render = variableRenderer.render("{{ read('" + internalStorageFile + "') }}", variables);
         assertThat(render, is("Hello from a task output"));
+
+        // test for an authorized parent execution (execution trigger)
+        variables = Map.of(
+            "flow", Map.of(
+                "id", "subflow",
+                "namespace", namespace),
+            "execution", Map.of("id", IdUtils.create()),
+            "trigger", Map.of(
+                "flowId", flowId,
+                "namespace", namespace,
+                "executionId", executionId
+            )
+        );
+
+        render = variableRenderer.render("{{ read('" + internalStorageFile + "') }}", variables);
+        assertThat(render, is("Hello from a task output"));
     }
 
     @Test
-    void readUnauthorizedInternalStorageFile() throws IOException, IllegalVariableEvaluationException {
+    void readUnauthorizedInternalStorageFile() throws IOException {
         String namespace = "namespace";
         String flowId = "flow";
         String executionId = IdUtils.create();
         URI internalStorageURI = URI.create("/" + namespace + "/" + flowId + "/executions/" + executionId + "/tasks/task/" + IdUtils.create() + "/123456.ion");
         URI internalStorageFile = storageInterface.put(null, internalStorageURI, new ByteArrayInputStream("Hello from a task output".getBytes()));
+
+        // test for an un-authorized execution with no trigger
         Map<String, Object> variables = Map.of(
             "flow", Map.of(
                 "id", "notme",
@@ -78,6 +98,37 @@ class ReadFileFunctionTest {
         );
 
         var exception = assertThrows(IllegalArgumentException.class, () -> variableRenderer.render("{{ read('" + internalStorageFile + "') }}", variables));
-        assertThat(exception.getMessage(), is("Unable to read a file that didn't belong to the current execution"));
+        assertThat(exception.getMessage(), is("Unable to read the file '" + internalStorageFile + "' as it didn't belong to the current execution"));
+
+        // test for an un-authorized execution with a trigger of type execution
+        Map<String, Object> executionTriggerVariables = Map.of(
+            "flow", Map.of(
+                "id", "notme",
+                "namespace", "notme"),
+            "execution", Map.of("id", "notme"),
+            "trigger", Map.of(
+                "flowId", "notme",
+                "namespace", "notme",
+                "executionId", "notme"
+            )
+        );
+
+        exception = assertThrows(IllegalArgumentException.class, () -> variableRenderer.render("{{ read('" + internalStorageFile + "') }}", executionTriggerVariables));
+        assertThat(exception.getMessage(), is("Unable to read the file '" + internalStorageFile + "' as it didn't belong to the current execution"));
+
+        // test for an un-authorized execution with a trigger of another type
+        Map<String, Object> triggerVariables = Map.of(
+            "flow", Map.of(
+                "id", "notme",
+                "namespace", "notme"),
+            "execution", Map.of("id", "notme"),
+            "trigger", Map.of(
+                "date", "somedate",
+                "row", "somerow"
+            )
+        );
+
+        exception = assertThrows(IllegalArgumentException.class, () -> variableRenderer.render("{{ read('" + internalStorageFile + "') }}", triggerVariables));
+        assertThat(exception.getMessage(), is("Unable to read the file '" + internalStorageFile + "' as it didn't belong to the current execution"));
     }
 }
diff --git a/core/src/test/resources/flows/valids/flow-concurrency-cancel.yml b/core/src/test/resources/flows/valids/flow-concurrency-cancel.yml
new file mode 100644
index 000000000..6643da26e
--- /dev/null
+++ b/core/src/test/resources/flows/valids/flow-concurrency-cancel.yml
@@ -0,0 +1,11 @@
+id: flow-concurrency-cancel
+namespace: io.kestra.tests
+
+concurrency:
+  behavior: CANCEL
+  limit: 1
+
+tasks:
+  - id: sleep
+    type: io.kestra.core.tasks.test.Sleep
+    duration: 2000
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/flow-concurrency-fail.yml b/core/src/test/resources/flows/valids/flow-concurrency-fail.yml
new file mode 100644
index 000000000..016138322
--- /dev/null
+++ b/core/src/test/resources/flows/valids/flow-concurrency-fail.yml
@@ -0,0 +1,11 @@
+id: flow-concurrency-fail
+namespace: io.kestra.tests
+
+concurrency:
+  behavior: FAIL
+  limit: 1
+
+tasks:
+  - id: sleep
+    type: io.kestra.core.tasks.test.Sleep
+    duration: 2000
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/flow-concurrency-queue.yml b/core/src/test/resources/flows/valids/flow-concurrency-queue.yml
new file mode 100644
index 000000000..ae29e8cfa
--- /dev/null
+++ b/core/src/test/resources/flows/valids/flow-concurrency-queue.yml
@@ -0,0 +1,11 @@
+id: flow-concurrency-queue
+namespace: io.kestra.tests
+
+concurrency:
+  behavior: QUEUE
+  limit: 1
+
+tasks:
+  - id: sleep
+    type: io.kestra.core.tasks.test.Sleep
+    duration: 2000
\ No newline at end of file
diff --git a/jdbc-h2/src/test/resources/application.yml b/jdbc-h2/src/test/resources/application.yml
index c143a3c62..fe628e433 100644
--- a/jdbc-h2/src/test/resources/application.yml
+++ b/jdbc-h2/src/test/resources/application.yml
@@ -74,6 +74,9 @@ kestra:
       workerjobrunning:
         cls: io.kestra.core.runners.WorkerJobRunning
         table: "worker_job_running"
+      executionqueued:
+        table: "execution_queued"
+        cls: io.kestra.core.runners.ExecutionQueued
 
     queues:
       min-poll-interval: 10ms
diff --git a/jdbc-mysql/src/test/resources/application.yml b/jdbc-mysql/src/test/resources/application.yml
index dc16a87da..b62405950 100644
--- a/jdbc-mysql/src/test/resources/application.yml
+++ b/jdbc-mysql/src/test/resources/application.yml
@@ -75,6 +75,9 @@ kestra:
       workerjobrunning:
         cls: io.kestra.core.runners.WorkerJobRunning
         table: "worker_job_running"
+      executionqueued:
+        table: "execution_queued"
+        cls: io.kestra.core.runners.ExecutionQueued
 
     queues:
       min-poll-interval: 10ms
diff --git a/jdbc-postgres/src/test/resources/application.yml b/jdbc-postgres/src/test/resources/application.yml
index 433fa312e..daa9834de 100644
--- a/jdbc-postgres/src/test/resources/application.yml
+++ b/jdbc-postgres/src/test/resources/application.yml
@@ -76,6 +76,9 @@ kestra:
       workerjobrunning:
         cls: io.kestra.core.runners.WorkerJobRunning
         table: "worker_job_running"
+      executionqueued:
+        table: "execution_queued"
+        cls: io.kestra.core.runners.ExecutionQueued
 
     queues:
       min-poll-interval: 10ms
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
index c31275faf..2126fa393 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -2,8 +2,6 @@ package io.kestra.jdbc.runner;
 
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
-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.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
@@ -81,6 +79,9 @@ public abstract class JdbcRunnerTest {
     @Inject
     private ForEachItemCaseTest forEachItemCaseTest;
 
+    @Inject
+    private FlowConcurrencyCaseTest flowConcurrencyCaseTest;
+
     @BeforeAll
     void init() throws IOException, URISyntaxException {
         jdbcTestUtils.drop();
@@ -276,4 +277,19 @@ public abstract class JdbcRunnerTest {
     void forEachItemNoWait() throws URISyntaxException, IOException, InterruptedException, TimeoutException {
         forEachItemCaseTest.forEachItemNoWait();
     }
+
+    @Test
+    void concurrencyCancel() throws TimeoutException, InterruptedException {
+        flowConcurrencyCaseTest.flowConcurrencyCancel();
+    }
+
+    @Test
+    void concurrencyFail() throws TimeoutException, InterruptedException  {
+        flowConcurrencyCaseTest.flowConcurrencyFail();
+    }
+
+    @Test
+    void concurrencyQueue() throws TimeoutException, InterruptedException  {
+        flowConcurrencyCaseTest.flowConcurrencyQueue();
+    }
 }
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
index 43a1af713..89ddb74d5 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
@@ -433,7 +433,7 @@ class ExecutionControllerTest extends JdbcH2ControllerTest {
             FileMetas.class
         ).blockingFirst();
 
-        assertThat(metas.getSize(), equalTo(2767L));
+        assertThat(metas.getSize(), equalTo(2876L));
 
         String newExecutionId = IdUtils.create();
 
diff --git a/webserver/src/test/resources/application.yml b/webserver/src/test/resources/application.yml
index 679640cec..953610538 100644
--- a/webserver/src/test/resources/application.yml
+++ b/webserver/src/test/resources/application.yml
@@ -91,6 +91,9 @@ kestra:
       workerjobrunning:
         cls: io.kestra.core.runners.WorkerJobRunning
         table: "worker_job_running"
+      executionqueued:
+        table: "execution_queued"
+        cls: io.kestra.core.runners.ExecutionQueued
 datasources:
   h2:
     url: jdbc:h2:mem:public;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE