Repository: kestra-io/kestra
Pull request #2793 (2024-01-02 21:08:02)

feat(): new execution date property
The objective of this PR is to always have the next date in the trigger, this way we don't have to evaluate every triggers every seconds

closes https://github.com/kestra-io/kestra-ee/issues/611 
closes https://github.com/kestra-io/kestra/issues/2922 
closes https://github.com/kestra-io/kestra/issues/2587 

sonarcloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) **Quality Gate failed**  
Failed conditions

[B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) (required ≥ A)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![SonarLint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'SonarLint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
sonarcloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) **Quality Gate failed**  
Failed conditions

[B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) (required ≥ A)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![SonarLint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'SonarLint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
sonarcloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) **Quality Gate failed**  
Failed conditions

[B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) (required ≥ A)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![SonarLint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'SonarLint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
sonarcloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) **Quality Gate failed**  
Failed conditions

[B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) (required ≥ A)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793)



![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![SonarLint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'SonarLint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
sonarcloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) **Quality Gate failed**  
Failed conditions  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/failed-16px.png '') [B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793) (required ≥ A)  
  
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=2793)

##   
![idea](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/light_bulb-16px.png 'idea') Catch issues before they fail your Quality Gate with our IDE extension ![SonarLint](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/sonarlint-16px.png 'SonarLint') [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)



Patch:
diff --git a/core/src/main/java/io/kestra/core/endpoints/SchedulerEndpoint.java b/core/src/main/java/io/kestra/core/endpoints/SchedulerEndpoint.java
index 572395712..2e09d52c6 100644
--- a/core/src/main/java/io/kestra/core/endpoints/SchedulerEndpoint.java
+++ b/core/src/main/java/io/kestra/core/endpoints/SchedulerEndpoint.java
@@ -1,20 +1,20 @@
 package io.kestra.core.endpoints;
 
+import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.models.triggers.Trigger;
+import io.kestra.core.schedulers.AbstractScheduler;
 import io.micronaut.context.annotation.Requires;
 import io.micronaut.management.endpoint.annotation.Endpoint;
 import io.micronaut.management.endpoint.annotation.Read;
+import jakarta.inject.Inject;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Getter;
-import io.kestra.core.models.triggers.AbstractTrigger;
-import io.kestra.core.models.triggers.Trigger;
-import io.kestra.core.schedulers.AbstractScheduler;
 
 import java.time.ZonedDateTime;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
-import jakarta.inject.Inject;
 
 @Endpoint(id = "scheduler", defaultSensitive = false)
 @Requires(property = "kestra.server-type", pattern = "(SCHEDULER|STANDALONE)")
@@ -29,13 +29,13 @@ public class SchedulerEndpoint {
         List<SchedulerEndpointSchedule> result = scheduler.getSchedulable()
             .stream()
             .map(flowWithTrigger -> {
-                String uid = Trigger.uid(flowWithTrigger.getFlow(), flowWithTrigger.getTrigger());
+                String uid = Trigger.uid(flowWithTrigger.getFlow(), flowWithTrigger.getAbstractTrigger());
 
                 return new SchedulerEndpointSchedule(
                     flowWithTrigger.getFlow().getId(),
                     flowWithTrigger.getFlow().getNamespace(),
                     flowWithTrigger.getFlow().getRevision(),
-                    flowWithTrigger.getTrigger(),
+                    flowWithTrigger.getAbstractTrigger(),
                     schedulableNextDate.containsKey(uid) ? schedulableNextDate.get(uid).getNext() : null
                 );
             })
diff --git a/core/src/main/java/io/kestra/core/models/executions/LogEntry.java b/core/src/main/java/io/kestra/core/models/executions/LogEntry.java
index 3396459dc..4131b54be 100644
--- a/core/src/main/java/io/kestra/core/models/executions/LogEntry.java
+++ b/core/src/main/java/io/kestra/core/models/executions/LogEntry.java
@@ -17,9 +17,9 @@ import org.slf4j.event.Level;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 import java.time.Instant;
-import java.util.Arrays;
-import java.util.List;
+import java.util.*;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 @Value
 @Builder(toBuilder = true)
@@ -122,4 +122,19 @@ public class LogEntry implements DeletedInterface, TenantInterface {
     public static String toPrettyString(LogEntry logEntry) {
         return logEntry.getTimestamp().toString() + " " + logEntry.getLevel() + " " + logEntry.getMessage();
     }
+
+    public Map<String, String> toMap() {
+        return Stream
+            .of(
+                new AbstractMap.SimpleEntry<>("tenantId", this.tenantId),
+                new AbstractMap.SimpleEntry<>("namespace", this.namespace),
+                new AbstractMap.SimpleEntry<>("flowId", this.flowId),
+                new AbstractMap.SimpleEntry<>("taskId", this.taskId),
+                new AbstractMap.SimpleEntry<>("executionId", this.executionId),
+                new AbstractMap.SimpleEntry<>("taskRunId", this.taskRunId),
+                new AbstractMap.SimpleEntry<>("triggerId", this.triggerId)
+            )
+            .filter(e -> e.getValue() != null)
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java b/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
index aa007770b..dd156d0d0 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
@@ -54,5 +54,14 @@ abstract public class AbstractTrigger {
     @Valid
     private WorkerGroup workerGroup;
 
-    private Level minLogLevel;
+    private Level logLevel;
+
+    /**
+     * For backward compatibility: we rename minLogLevel to logLevel.
+     * @deprecated use {@link #logLevel} instead
+     */
+    @Deprecated
+    public void setMinLogLevel(Level minLogLevel) {
+        this.logLevel = minLogLevel;
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/models/triggers/PollingTriggerInterface.java b/core/src/main/java/io/kestra/core/models/triggers/PollingTriggerInterface.java
index 8f42141a2..f03a80180 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/PollingTriggerInterface.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/PollingTriggerInterface.java
@@ -13,7 +13,11 @@ public interface PollingTriggerInterface {
     Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception;
 
     default ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optional<? extends TriggerContext> last) throws Exception {
-        return ZonedDateTime.now();
+        return ZonedDateTime.now().plus(this.getInterval());
+    }
+
+    default ZonedDateTime nextEvaluationDate() {
+        return ZonedDateTime.now().plus(this.getInterval());
     }
 
     @Schema(
diff --git a/core/src/main/java/io/kestra/core/models/triggers/Trigger.java b/core/src/main/java/io/kestra/core/models/triggers/Trigger.java
index b04729322..0231346a4 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/Trigger.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/Trigger.java
@@ -1,17 +1,21 @@
 package io.kestra.core.models.triggers;
 
+import io.kestra.core.models.conditions.ConditionContext;
+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.utils.IdUtils;
-import lombok.*;
+import io.micronaut.core.annotation.Nullable;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
 import lombok.experimental.SuperBuilder;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
 
 import java.time.Instant;
 import java.time.ZonedDateTime;
-
-import io.micronaut.core.annotation.Nullable;
-import jakarta.validation.constraints.NotNull;
+import java.time.temporal.ChronoUnit;
+import java.util.Optional;
 
 @SuperBuilder(toBuilder = true)
 @ToString
@@ -19,7 +23,7 @@ import jakarta.validation.constraints.NotNull;
 @Getter
 @NoArgsConstructor
 public class Trigger extends TriggerContext {
-    @NotNull
+    @Nullable
     private String executionId;
 
     @Nullable
@@ -89,6 +93,22 @@ public class Trigger extends TriggerContext {
             .build();
     }
 
+    /**
+     * Create a new Trigger from polling trigger with no execution information and no evaluation lock.
+     */
+    public static Trigger of(TriggerContext triggerContext, ZonedDateTime nextExecutionDate) {
+        return Trigger.builder()
+            .tenantId(triggerContext.getTenantId())
+            .namespace(triggerContext.getNamespace())
+            .flowId(triggerContext.getFlowId())
+            .flowRevision(triggerContext.getFlowRevision())
+            .triggerId(triggerContext.getTriggerId())
+            .date(triggerContext.getDate())
+            .nextExecutionDate(nextExecutionDate)
+            .build();
+    }
+
+
     /**
      * Create a new Trigger with execution information.
      *
@@ -104,6 +124,27 @@ public class Trigger extends TriggerContext {
             .date(triggerContext.getDate())
             .executionId(execution.getId())
             .updatedDate(Instant.now())
+            .nextExecutionDate(triggerContext.getNextExecutionDate())
+            .build();
+    }
+
+    /**
+     * Create a new Trigger with execution information and specific nextExecutionDate.
+     * This one is use when starting a schedule execution as the nextExecutionDate come from the execution variables
+     *
+     * This is used to lock the trigger while an execution is running, it will also erase the evaluation lock.
+     */
+    public static Trigger of(TriggerContext triggerContext, Execution execution, ZonedDateTime nextExecutionDate) {
+        return Trigger.builder()
+            .tenantId(triggerContext.getTenantId())
+            .namespace(triggerContext.getNamespace())
+            .flowId(triggerContext.getFlowId())
+            .flowRevision(triggerContext.getFlowRevision())
+            .triggerId(triggerContext.getTriggerId())
+            .date(triggerContext.getDate())
+            .executionId(execution.getId())
+            .updatedDate(Instant.now())
+            .nextExecutionDate(nextExecutionDate)
             .build();
     }
 
@@ -112,14 +153,15 @@ public class Trigger extends TriggerContext {
      *
      * This is used to update the trigger with the execution information, it will also erase the trigger date.
      */
-    public static Trigger of(Execution execution, ZonedDateTime date) {
+    public static Trigger of(Execution execution, Trigger trigger) {
         return Trigger.builder()
             .tenantId(execution.getTenantId())
             .namespace(execution.getNamespace())
             .flowId(execution.getFlowId())
             .flowRevision(execution.getFlowRevision())
             .triggerId(execution.getTrigger().getId())
-            .date(date)
+            .date(trigger.getDate())
+            .nextExecutionDate(trigger.getNextExecutionDate())
             .executionId(execution.getId())
             .executionCurrentState(execution.getState().getCurrent())
             .updatedDate(Instant.now())
@@ -131,19 +173,32 @@ public class Trigger extends TriggerContext {
      *
      * This is used to lock the trigger evaluation.
      */
-    public static Trigger of(TriggerContext triggerContext, ZonedDateTime evaluateRunningDate) {
+    public static Trigger of(Trigger trigger, ZonedDateTime evaluateRunningDate) {
         return Trigger.builder()
-            .tenantId(triggerContext.getTenantId())
-            .namespace(triggerContext.getNamespace())
-            .flowId(triggerContext.getFlowId())
-            .flowRevision(triggerContext.getFlowRevision())
-            .triggerId(triggerContext.getTriggerId())
-            .date(triggerContext.getDate())
+            .tenantId(trigger.getTenantId())
+            .namespace(trigger.getNamespace())
+            .flowId(trigger.getFlowId())
+            .flowRevision(trigger.getFlowRevision())
+            .triggerId(trigger.getTriggerId())
+            .date(trigger.getDate())
+            .nextExecutionDate(trigger.getNextExecutionDate())
             .evaluateRunningDate(evaluateRunningDate)
             .updatedDate(Instant.now())
             .build();
     }
 
+    public static Trigger of(Flow flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext, Optional<Trigger> lastTrigger) throws Exception {
+        return Trigger.builder()
+            .tenantId(flow.getTenantId())
+            .namespace(flow.getNamespace())
+            .flowId(flow.getId())
+            .flowRevision(flow.getRevision())
+            .triggerId(abstractTrigger.getId())
+            .date(ZonedDateTime.now().truncatedTo(ChronoUnit.SECONDS))
+            .nextExecutionDate(((PollingTriggerInterface) abstractTrigger).nextEvaluationDate(conditionContext, lastTrigger))
+            .build();
+    }
+
     public Trigger resetExecution() {
         return Trigger.builder()
             .tenantId(this.getTenantId())
@@ -152,6 +207,7 @@ public class Trigger extends TriggerContext {
             .flowRevision(this.getFlowRevision())
             .triggerId(this.getTriggerId())
             .date(this.getDate())
+            .nextExecutionDate(this.getNextExecutionDate())
             .build();
     }
 }
diff --git a/core/src/main/java/io/kestra/core/models/triggers/TriggerContext.java b/core/src/main/java/io/kestra/core/models/triggers/TriggerContext.java
index 46b06c5b3..82040166e 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/TriggerContext.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/TriggerContext.java
@@ -3,6 +3,7 @@ package io.kestra.core.models.triggers;
 import io.kestra.core.models.TenantInterface;
 import io.kestra.core.utils.IdUtils;
 import io.micronaut.core.annotation.Introspected;
+import io.micronaut.core.annotation.Nullable;
 import io.swagger.v3.oas.annotations.Hidden;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -38,6 +39,9 @@ public class TriggerContext implements TenantInterface {
     @NotNull
     private ZonedDateTime date;
 
+    @Nullable
+    private ZonedDateTime nextExecutionDate;
+
     public String uid() {
         return uid(this);
     }
diff --git a/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java b/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java
index fa671e9ec..b1f7cda6b 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java
@@ -25,6 +25,7 @@ import io.kestra.core.validations.TimezoneId;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
+import lombok.extern.slf4j.Slf4j;
 
 import java.time.Duration;
 import java.time.ZoneId;
@@ -38,6 +39,7 @@ import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Null;
 
+@Slf4j
 @SuperBuilder
 @ToString
 @EqualsAndHashCode
@@ -178,10 +180,9 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
     @Override
     public ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optional<? extends TriggerContext> last) throws Exception {
         ExecutionTime executionTime = this.executionTime();
-
+        ZonedDateTime nextDate;
         if (last.isPresent()) {
             ZonedDateTime lastDate = convertDateTime(last.get().getDate());
-
             // previous present & scheduleConditions
             if (this.scheduleConditions != null) {
                 Optional<ZonedDateTime> next = this.truePreviousNextDateWithCondition(
@@ -190,63 +191,65 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
                     lastDate,
                     true
                 );
-
                 if (next.isPresent()) {
+
                     return next.get().truncatedTo(ChronoUnit.SECONDS);
                 }
             }
-
             // previous present but no scheduleConditions
-            return computeNextEvaluationDate(executionTime, lastDate).orElse(null);
+            nextDate = computeNextEvaluationDate(executionTime, lastDate).orElse(null);
         }
-
-
         // no previous present but backfill
-        if (backfill != null && backfill.getStart() != null) {
-            return this.timezone != null ?
+        else if (backfill != null && backfill.getStart() != null) {
+            nextDate = this.timezone != null ?
                 backfill.getStart().withZoneSameLocal(ZoneId.of(this.timezone)) :
                 backfill.getStart();
         }
-
         // no previous present & no backfill, just provide now
-        return computeNextEvaluationDate(executionTime, ZonedDateTime.now()).orElse(null);
+        else {
+            nextDate = computeNextEvaluationDate(executionTime, ZonedDateTime.now()).orElse(null);
+        }
+        // if max delay reach, we calculate a new date
+        if (this.lateMaximumDelay != null && nextDate != null) {
+            Output scheduleDates = this.scheduleDates(executionTime, nextDate).orElse(null);
+            scheduleDates = this.handleMaxDelay(scheduleDates);
+            if (scheduleDates != null) {
+                nextDate = scheduleDates.getDate();
+            } else {
+                return null;
+            }
+        }
+
+        return nextDate;
     }
 
     @Override
     public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception {
         RunContext runContext = conditionContext.getRunContext();
         ExecutionTime executionTime = this.executionTime();
-        ZonedDateTime previousDate = convertDateTime(context.getDate());
-
-        Output output = this.output(executionTime, previousDate).orElse(null);
+        ZonedDateTime currentDateTimeExecution = convertDateTime(context.getDate());
 
-        // if max delay reach, we calculate a new date
-        output = this.handleMaxDelay(output);
+        Output scheduleDates = this.scheduleDates(executionTime, currentDateTimeExecution).orElse(null);
 
-        if (output == null || output.getDate() == null) {
+        if (scheduleDates == null || scheduleDates.getDate() == null) {
             return Optional.empty();
         }
 
-        ZonedDateTime next = output.getDate();
-
-        // we try at the exact time / standard behaviour
-        boolean isReady = next.compareTo(previousDate) == 0;
-
-        // in case on cron expression changed, the next date will never match, so we allow past operation to start
-        boolean isLate = next.compareTo(ZonedDateTime.now().minus(Duration.ofMinutes(1))) < 0;
-
-        if (!isReady && !isLate) {
-            return Optional.empty();
-        }
+        ZonedDateTime next = scheduleDates.getDate();
 
         // we are in the future don't allow
+        // No use case, just here for prevention but it should never happen
         if (next.compareTo(ZonedDateTime.now().plus(Duration.ofSeconds(1))) > 0) {
+            if (log.isTraceEnabled()) {
+                log.trace("Schedule is in the future, execution skipped, this behavior should never happen.");
+            }
             return Optional.empty();
         }
 
         // inject outputs variables for scheduleCondition
-        conditionContext = conditionContext(conditionContext, output);
+        conditionContext = conditionContext(conditionContext, scheduleDates);
 
+        // FIXME make scheduleConditions generic
         // control scheduleConditions
         if (scheduleConditions != null) {
             boolean conditionResults = this.validateScheduleCondition(conditionContext);
@@ -255,7 +258,7 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
             }
 
             // recalculate true output for previous and next based on conditions
-            output = this.trueOutputWithCondition(executionTime, conditionContext, output);
+            scheduleDates = this.trueOutputWithCondition(executionTime, conditionContext, scheduleDates);
         }
 
         Map<String, Object> inputs = new HashMap<>();
@@ -274,9 +277,9 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
 
         Map<String, Object> variables;
         if (this.timezone != null) {
-            variables = output.toMap(ZoneId.of(this.timezone));
+            variables = scheduleDates.toMap(ZoneId.of(this.timezone));
         } else {
-            variables = output.toMap();
+            variables = scheduleDates.toMap();
         }
 
         ExecutionTrigger executionTrigger = ExecutionTrigger.of(this, variables);
@@ -305,27 +308,27 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
         return Optional.of(execution);
     }
 
-    private Optional<Output> output(ExecutionTime executionTime, ZonedDateTime date) {
+    private Optional<Output> scheduleDates(ExecutionTime executionTime, ZonedDateTime date) {
         Optional<ZonedDateTime> next = executionTime.nextExecution(date.minus(Duration.ofSeconds(1)));
 
         if (next.isEmpty()) {
             return Optional.empty();
         }
 
-        Output.OutputBuilder<?, ?> outputBuilder = Output.builder()
+        Output.OutputBuilder<?, ?> outputDatesBuilder = Output.builder()
             .date(convertDateTime(next.get()));
 
         computeNextEvaluationDate(executionTime, next.get())
             .map(this::convertDateTime)
-            .ifPresent(outputBuilder::next);
+            .ifPresent(outputDatesBuilder::next);
 
         executionTime.lastExecution(date)
             .map(this::convertDateTime)
-            .ifPresent(outputBuilder::previous);
+            .ifPresent(outputDatesBuilder::previous);
 
-        Output output = outputBuilder.build();
+        Output scheduleDates = outputDatesBuilder.build();
 
-        return Optional.of(output);
+        return Optional.of(scheduleDates);
     }
 
     private ConditionContext conditionContext(ConditionContext conditionContext, Output output) {
@@ -383,7 +386,7 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
                 return currentDate;
             }
 
-            Optional<Output> currentOutput = this.output(executionTime, currentDate.get());
+            Optional<Output> currentOutput = this.scheduleDates(executionTime, currentDate.get());
 
             if (currentOutput.isEmpty()) {
                 return Optional.empty();
@@ -416,7 +419,7 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
                 (output.getDate().getYear() > ZonedDateTime.now().getYear() - 10)
         ) {
             if (output.getDate().plus(this.lateMaximumDelay).compareTo(ZonedDateTime.now()) < 0) {
-                output = this.output(executionTime, output.getNext()).orElse(null);
+                output = this.scheduleDates(executionTime, output.getNext()).orElse(null);
                 if (output == null) {
                     return null;
                 }
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 e16bf796d..6f3139ad3 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -373,7 +373,7 @@ public class ExecutorService {
         );
 
         if (logger.isTraceEnabled()) {
-            logger.debug(newExecution.toString(true));
+            logger.trace(newExecution.toString(true));
         }
 
         metricRegistry
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorState.java b/core/src/main/java/io/kestra/core/runners/ExecutorState.java
index b448c976d..1588dbd36 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorState.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorState.java
@@ -14,7 +14,6 @@ public class ExecutorState {
     private Map<String, State.Type> workerTaskDeduplication = new ConcurrentHashMap<>();
     private Map<String, String> childDeduplication = new ConcurrentHashMap<>();
     private Map<String, State.Type> subflowExecutionDeduplication = new ConcurrentHashMap<>();
-    private Boolean flowTriggerDeduplication = false;
 
     public ExecutorState(String executionId) {
         this.executionId = executionId;
diff --git a/core/src/main/java/io/kestra/core/runners/RunContext.java b/core/src/main/java/io/kestra/core/runners/RunContext.java
index 722e5f9af..fb4961b68 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContext.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContext.java
@@ -29,7 +29,6 @@ import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.utils.IdUtils;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.inject.qualifiers.Qualifiers;
-import jakarta.inject.Inject;
 import lombok.NoArgsConstructor;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.FilenameUtils;
@@ -43,6 +42,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.security.GeneralSecurityException;
 import java.util.*;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -199,7 +199,7 @@ public class RunContext {
                 Qualifiers.byName(QueueFactoryInterface.WORKERTASKLOG_NAMED)
             ).orElseThrow(),
             LogEntry.of(triggerContext, trigger),
-            trigger.getMinLogLevel()
+            trigger.getLogLevel()
         );
     }
 
@@ -211,7 +211,7 @@ public class RunContext {
                 Qualifiers.byName(QueueFactoryInterface.WORKERTASKLOG_NAMED)
             ).orElseThrow(),
             LogEntry.of(flow, trigger),
-            trigger.getMinLogLevel()
+            trigger.getLogLevel()
         );
     }
 
@@ -489,6 +489,8 @@ public class RunContext {
             clone.put("taskrun", taskrun);
         }
 
+        clone.put("addSecretConsumer", (Consumer<String>) s -> runContextLogger.usedSecret(s));
+
         this.variables = ImmutableMap.copyOf(clone);
         this.storage = new InternalStorage(logger(), StorageContext.forTask(taskRun), storageInterface);
         return this;
diff --git a/core/src/main/java/io/kestra/core/runners/RunContextLogger.java b/core/src/main/java/io/kestra/core/runners/RunContextLogger.java
index 068103096..3947d108e 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContextLogger.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContextLogger.java
@@ -5,7 +5,9 @@ import ch.qos.logback.classic.Logger;
 import ch.qos.logback.classic.LoggerContext;
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.classic.spi.IThrowableProxy;
+import ch.qos.logback.classic.spi.LoggingEvent;
 import ch.qos.logback.classic.spi.ThrowableProxy;
+import ch.qos.logback.classic.util.LogbackMDCAdapter;
 import ch.qos.logback.core.AppenderBase;
 import com.cronutils.utils.VisibleForTesting;
 import com.google.common.base.Splitter;
@@ -13,21 +15,23 @@ import com.google.common.base.Throwables;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.utils.IdUtils;
+import lombok.extern.slf4j.Slf4j;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
 
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 import java.util.stream.Collectors;
 
 public class RunContextLogger {
     private static final int MAX_MESSAGE_LENGTH = 1024*10;
+
     private final String loggerName;
     private Logger logger;
     private QueueInterface<LogEntry> logQueue;
     private LogEntry logEntry;
     private Level loglevel;
+    private List<String> useSecrets = new ArrayList<>();
 
     @VisibleForTesting
     public RunContextLogger() {
@@ -83,7 +87,6 @@ public class RunContextLogger {
         return result;
     }
 
-    @SuppressWarnings("UnstableApiUsage")
     public static List<LogEntry> logEntries(ILoggingEvent event, LogEntry logEntry) {
         Throwable throwable = throwable(event);
 
@@ -123,21 +126,32 @@ public class RunContextLogger {
         return result;
     }
 
+    public void usedSecret(String secret) {
+        this.useSecrets.add(secret);
+    }
+
     public org.slf4j.Logger logger() {
         if (this.logger == null) {
             LoggerContext loggerContext = new LoggerContext();
+            LogbackMDCAdapter mdcAdapter = new LogbackMDCAdapter();
+
+            loggerContext.setMDCAdapter(mdcAdapter);
+            loggerContext.start();
+
             this.logger = loggerContext.getLogger(this.loggerName);
 
             // unit test don't need the logqueue
             if (this.logQueue != null && this.logEntry != null) {
-                ContextAppender contextAppender = new ContextAppender(this.logQueue, this.logEntry);
+                ContextAppender contextAppender = new ContextAppender(this, this.logger, this.logQueue, this.logEntry);
                 contextAppender.setContext(loggerContext);
                 contextAppender.start();
 
                 this.logger.addAppender(contextAppender);
+
+                MDC.setContextMap(this.logEntry.toMap());
             }
 
-            ForwardAppender forwardAppender = new ForwardAppender();
+            ForwardAppender forwardAppender = new ForwardAppender(this, this.logger);
             forwardAppender.setContext(loggerContext);
             forwardAppender.start();
             this.logger.addAppender(forwardAppender);
@@ -149,24 +163,108 @@ public class RunContextLogger {
         return this.logger;
     }
 
-    public static class ContextAppender extends AppenderBase<ILoggingEvent> {
+    @Slf4j
+    public abstract static class BaseAppender extends AppenderBase<ILoggingEvent> {
+        protected RunContextLogger runContextLogger;
+        protected Logger logger;
+
+        protected BaseAppender(RunContextLogger runContextLogger, Logger logger) {
+            this.runContextLogger = runContextLogger;
+            this.logger = logger;
+        }
+
+        private String replaceSecret(String data) {
+            for (String s : runContextLogger.useSecrets) {
+                if (data.contains(s)) {
+                    data = data.replace(s, "*".repeat(s.length()));
+                    data = data.replaceFirst("[*]{9}", "**masked*");
+                }
+            }
+
+            return data;
+        }
+
+        private Object recursive(Object object) {
+            if (object instanceof Map<?, ?> value) {
+                return value
+                    .entrySet()
+                    .stream()
+                    .map(e -> new AbstractMap.SimpleEntry<>(
+                        recursive(e.getKey()),
+                        recursive(e.getValue())
+                    ))
+                    .collect(HashMap::new, (m, v) -> m.put(v.getKey(), v.getValue()), HashMap::putAll);
+            } else if (object instanceof Collection<?> value) {
+                return value
+                    .stream()
+                    .map(this::recursive)
+                    .collect(Collectors.toList());
+            } else if (object instanceof String string) {
+                return replaceSecret(string);
+            } else {
+                return object;
+            }
+        }
+
+        private Object[] replaceSecret(Object[] data) {
+            if (data == null) {
+                return data;
+            }
+
+            Object[] result = new Object[data.length];
+
+            for (int i = 0; i < data.length; i++) {
+                result[i] = recursive(data[i]);
+            }
+
+            return result;
+        }
+
+        protected ILoggingEvent transform(ILoggingEvent event) {
+            try {
+                String message = replaceSecret(event.getMessage());
+                Object[] argumentArray = replaceSecret(event.getArgumentArray());
+
+                return new LoggingEvent(
+                    "ch.qos.logback.classic.Logger",
+                    this.logger,
+                    event.getLevel(),
+                    message,
+                    event.getThrowableProxy() instanceof ThrowableProxy ? ((ThrowableProxy) event.getThrowableProxy()).getThrowable() : null,
+                    argumentArray
+                );
+            } catch (Throwable e) {
+                log.warn("Unable to replace secret", e);
+                return event;
+            }
+        }
+    }
+
+    public static class ContextAppender extends BaseAppender {
         private final QueueInterface<LogEntry> logQueue;
         private final LogEntry logEntry;
 
-        public ContextAppender(QueueInterface<LogEntry> logQueue, LogEntry logEntry) {
+        public ContextAppender(RunContextLogger runContextLogger, Logger logger, QueueInterface<LogEntry> logQueue, LogEntry logEntry) {
+            super(runContextLogger, logger);
             this.logQueue = logQueue;
             this.logEntry = logEntry;
         }
 
         @Override
         protected void append(ILoggingEvent e) {
+            e = this.transform(e);
+
             logEntries(e, logEntry)
                 .forEach(logQueue::emitAsync);
         }
     }
 
-    public static class ForwardAppender extends AppenderBase<ILoggingEvent> {
-        private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger("flow");
+    public static class ForwardAppender extends BaseAppender {
+        private static final ch.qos.logback.classic.Logger LOGGER = (ch.qos.logback.classic.Logger) LoggerFactory.getLogger("flow");
+
+        protected ForwardAppender(RunContextLogger runContextLogger, Logger logger) {
+            super(runContextLogger, logger);
+        }
 
         @Override
         public void start() {
@@ -180,9 +278,10 @@ public class RunContextLogger {
 
         @Override
         protected void append(ILoggingEvent e) {
-            var logger = ((ch.qos.logback.classic.Logger) LOGGER);
-            if (logger.isEnabledFor(e.getLevel())) {
-                ((ch.qos.logback.classic.Logger) LOGGER).callAppenders(e);
+            e = this.transform(e);
+
+            if (LOGGER.isEnabledFor(e.getLevel())) {
+                LOGGER.callAppenders(e);
             }
         }
     }
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index f1043761b..4a723c30d 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -343,10 +343,6 @@ public class Worker implements Runnable, AutoCloseable {
             workerTask.getTask().getClass().getSimpleName()
         );
 
-        if (workerTask.logger().isDebugEnabled()) {
-            workerTask.logger().debug("Variables\n{}", JacksonMapper.log(workerTask.getRunContext().getVariables()));
-        }
-
         workerTask = workerTask.withTaskRun(workerTask.getTaskRun().withState(State.Type.RUNNING));
         this.workerTaskResultQueue.emit(new WorkerTaskResult(workerTask));
 
@@ -560,15 +556,6 @@ public class Worker implements Runnable, AutoCloseable {
             .build()
             .withState(state);
 
-        // logs
-        if (workerThread.getTaskOutput() != null && log.isDebugEnabled()) {
-            log.debug("Outputs\n{}", JacksonMapper.log(workerThread.getTaskOutput()));
-        }
-
-        if (!runContext.metrics().isEmpty() && log.isTraceEnabled()) {
-            log.trace("Metrics\n{}", JacksonMapper.log(runContext.metrics()));
-        }
-
         // metrics
         runContext.metrics().forEach(metric -> this.metricEntryQueue.emit(MetricEntry.of(workerTask.getTaskRun(), metric)));
 
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/functions/SecretFunction.java b/core/src/main/java/io/kestra/core/runners/pebble/functions/SecretFunction.java
index 49bcb4585..960c1dcba 100644
--- a/core/src/main/java/io/kestra/core/runners/pebble/functions/SecretFunction.java
+++ b/core/src/main/java/io/kestra/core/runners/pebble/functions/SecretFunction.java
@@ -8,11 +8,14 @@ import io.pebbletemplates.pebble.template.EvaluationContext;
 import io.pebbletemplates.pebble.template.PebbleTemplate;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 
+@Slf4j
 @Singleton
 public class SecretFunction implements Function {
     @Inject
@@ -30,7 +33,16 @@ public class SecretFunction implements Function {
         Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
 
         try {
-            return secretService.findSecret(flow.get("tenantId"), flow.get("namespace"), key);
+            String secret = secretService.findSecret(flow.get("tenantId"), flow.get("namespace"), key);
+
+            try {
+                Consumer<String> addSecretConsumer = (Consumer<String>) context.getVariable("addSecretConsumer");
+                addSecretConsumer.accept(secret);
+            } catch (Exception e) {
+                log.warn("Unable to get secret consumer", e);
+            }
+
+            return secret;
         } catch (IllegalVariableEvaluationException | IOException e) {
             throw new PebbleException(e, e.getMessage(), lineNumber, self.getName());
         }
diff --git a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
index ca8fc8e05..36f7a3eed 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -10,7 +10,6 @@ import io.kestra.core.models.flows.FlowWithException;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.PollingTriggerInterface;
 import io.kestra.core.models.triggers.Trigger;
-import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.models.triggers.types.Schedule;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
@@ -25,6 +24,7 @@ import jakarta.annotation.PreDestroy;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
@@ -37,10 +37,9 @@ import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.function.BiConsumer;
 import java.util.stream.Stream;
 
-import static io.kestra.core.utils.Rethrow.throwSupplier;
-
 @Slf4j
 @Singleton
 public abstract class AbstractScheduler implements Scheduler {
@@ -58,16 +57,12 @@ public abstract class AbstractScheduler implements Scheduler {
     protected Boolean isReady = false;
 
     private final ScheduledExecutorService scheduleExecutor = Executors.newSingleThreadScheduledExecutor();
-    private final Map<String, ZonedDateTime> lastEvaluate = new ConcurrentHashMap<>();
 
-    // The triggerStateSavedLock must be used when accessing triggerStateSaved
-    protected final Object triggerStateSavedLock = new Object();
-    private final Map<String, Trigger> triggerStateSaved = new ConcurrentHashMap<>();
     protected SchedulerTriggerStateInterface triggerState;
 
     // schedulable and schedulableNextDate must be volatile and their access synchronized as they are updated and read by different threads.
     @Getter
-    private volatile List<FlowWithTrigger> schedulable = new ArrayList<>();
+    private volatile List<FlowWithTriggers> schedulable = new ArrayList<>();
     @Getter
     private volatile Map<String, FlowWithPollingTriggerNextDate> schedulableNextDate = new ConcurrentHashMap<>();
 
@@ -92,7 +87,8 @@ public abstract class AbstractScheduler implements Scheduler {
 
     @Override
     public void run() {
-        flowListeners.run();
+        this.flowListeners.run();
+        this.flowListeners.listen(this::initializedTriggers);
 
         ScheduledFuture<?> handle = scheduleExecutor.scheduleAtFixedRate(
             this::handle,
@@ -101,8 +97,6 @@ public abstract class AbstractScheduler implements Scheduler {
             TimeUnit.SECONDS
         );
 
-        flowListeners.listen(this::computeSchedulable);
-
         // look at exception on the main thread
         Thread thread = new Thread(
             () -> {
@@ -122,27 +116,29 @@ public abstract class AbstractScheduler implements Scheduler {
         );
         thread.start();
 
-        // remove trigger on flow update
+        // remove trigger on flow update & update local triggers store
         this.flowListeners.listen((flow, previous) -> {
-            synchronized (triggerStateSavedLock) {
-                if (flow.isDeleted()) {
-                    ListUtils.emptyOnNull(flow.getTriggers())
-                        .forEach(abstractTrigger -> {
-                            Trigger trigger = Trigger.of(flow, abstractTrigger);
-
-                            triggerStateSaved.remove(trigger.uid());
-                            triggerQueue.delete(trigger);
-                        });
-                } else if (previous != null) {
-                    FlowService
-                        .findRemovedTrigger(flow, previous)
-                        .forEach(abstractTrigger -> {
-                            Trigger trigger = Trigger.of(flow, abstractTrigger);
-
-                            triggerStateSaved.remove(trigger.uid());
-                            triggerQueue.delete(trigger);
-                        });
-                }
+            if (flow.isDeleted() || previous != null) {
+                List<AbstractTrigger> triggersDeleted = flow.isDeleted() ? ListUtils.emptyOnNull(flow.getTriggers()) : FlowService
+                    .findRemovedTrigger(flow, previous);
+                triggersDeleted.forEach(abstractTrigger -> {
+                    Trigger trigger = Trigger.of(flow, abstractTrigger);
+                    this.triggerQueue.delete(trigger);
+                });
+            }
+            if (previous != null) {
+                FlowService.findUpdatedTrigger(flow, previous)
+                    .forEach(abstractTrigger -> {
+                        if (abstractTrigger instanceof PollingTriggerInterface) {
+                            RunContext runContext = runContextFactory.of(flow, abstractTrigger);
+                            ConditionContext conditionContext = conditionService.conditionContext(runContext, flow, null);
+                            try {
+                                this.triggerState.update(flow, abstractTrigger, conditionContext);
+                            } catch (Exception e) {
+                                logError(conditionContext, flow, abstractTrigger, e);
+                            }
+                        }
+                    });
             }
         });
 
@@ -153,18 +149,18 @@ public abstract class AbstractScheduler implements Scheduler {
             either -> {
                 if (either.isRight()) {
                     log.error("Unable to deserialize a worker trigger result: {}", either.getRight().getMessage());
+
                     return;
                 }
 
                 WorkerTriggerResult workerTriggerResult = either.getLeft();
                 if (workerTriggerResult.getSuccess() && workerTriggerResult.getExecution().isPresent()) {
-                    var triggerExecution = new SchedulerExecutionWithTrigger(
+                    SchedulerExecutionWithTrigger triggerExecution = new SchedulerExecutionWithTrigger(
                         workerTriggerResult.getExecution().get(),
                         workerTriggerResult.getTriggerContext()
                     );
                     this.handleEvaluatePollingTriggerResult(triggerExecution);
-                }
-                else {
+                } else {
                     // previously, if no interval the trigger was executed immediately. I think only the Schedule trigger has no interval
                     // now that all triggers are sent to the worker, we need to do this to avoid issues with backfills
                     // as the same trigger will be evaluated multiple-time which is not supported by the 'addToRunning' method
@@ -172,92 +168,148 @@ public abstract class AbstractScheduler implements Scheduler {
                     // TODO now that Schedule is executed by the Scheduler this could be relaxed
                     if (workerTriggerResult.getTrigger() instanceof PollingTriggerInterface &&
                         ((PollingTriggerInterface) workerTriggerResult.getTrigger()).getInterval() != null) {
-                        var triggerNotRunning = Trigger.of(workerTriggerResult.getTriggerContext());
-                        triggerState.save(triggerNotRunning);
+
+                        ZonedDateTime nextExecutionDate = ((PollingTriggerInterface) workerTriggerResult.getTrigger()).nextEvaluationDate();
+                        this.triggerState.update(Trigger.of(workerTriggerResult.getTriggerContext(), nextExecutionDate));
                     }
                 }
             }
         );
     }
 
-    // must be synchronized as it update schedulableNextDate and schedulable, and will be executed on the flow listener thread
-    private synchronized void computeSchedulable(List<Flow> flows) {
-        this.schedulableNextDate = new HashMap<>();
+    // Initialized local trigger state
+    // and if some flows were created outside the box, for example from CLI
+    // then we may have some triggers that are not created yet
+    private void initializedTriggers(List<Flow> flows) {
+        List<Trigger> triggers = triggerState.findAllForAllTenants();
+
+        flows.forEach(flow -> {
+            ListUtils.emptyOnNull(flow.getTriggers()).forEach(abstractTrigger -> {
+                if (triggers.stream().noneMatch(trigger -> trigger.uid().equals(Trigger.uid(flow, abstractTrigger))) && abstractTrigger instanceof PollingTriggerInterface pollingAbstractTrigger) {
+                    RunContext runContext = runContextFactory.of(flow, abstractTrigger);
+                    ConditionContext conditionContext = conditionService.conditionContext(runContext, flow, null);
+                    try {
+                        Trigger newTrigger = Trigger.builder()
+                            .tenantId(flow.getTenantId())
+                            .namespace(flow.getNamespace())
+                            .flowId(flow.getId())
+                            .flowRevision(flow.getRevision())
+                            .triggerId(abstractTrigger.getId())
+                            .date(now())
+                            .nextExecutionDate(pollingAbstractTrigger.nextEvaluationDate(conditionContext, Optional.empty()))
+                            .build();
+                        this.triggerState.create(newTrigger);
+                    } catch (Exception e) {
+                        logError(conditionContext, flow, abstractTrigger, e);
+                    }
+                }
+            });
+        });
+    }
+
+    private List<FlowWithTriggers> computeSchedulable(List<Flow> flows, List<Trigger> triggerContextsToEvaluate, ScheduleContextInterface scheduleContext) {
+        List<String> flowToKeep = triggerContextsToEvaluate.stream().map(Trigger::getFlowId).toList();
 
-        this.schedulable = flows
+        return flows
             .stream()
+            .filter(flow -> flowToKeep.contains(flow.getId()))
             .filter(flow -> flow.getTriggers() != null && !flow.getTriggers().isEmpty())
             .filter(flow -> !flow.isDisabled() && !(flow instanceof FlowWithException))
             .flatMap(flow -> flow.getTriggers()
                 .stream()
                 .filter(abstractTrigger -> !abstractTrigger.isDisabled() && abstractTrigger instanceof PollingTriggerInterface)
-                .map(trigger -> {
-                    RunContext runContext = runContextFactory.of(flow, trigger);
-
-                    return new FlowWithTrigger(
+                .map(abstractTrigger -> {
+                    RunContext runContext = runContextFactory.of(flow, abstractTrigger);
+                    ConditionContext conditionContext = conditionService.conditionContext(runContext, flow, null);
+                    Trigger triggerContext = null;
+                    try {
+                        Trigger lastTrigger = triggerContextsToEvaluate
+                            .stream()
+                            .filter(triggerContextToFind -> triggerContextToFind.uid().equals(Trigger.uid(flow, abstractTrigger)))
+                            .findFirst()
+                            .orElse(null);
+                        // If trigger is not found in triggers to evaluate, then we ignore it
+                        if (lastTrigger == null) {
+                            return null;
+                            // Backwards compatibility
+                        } else if (lastTrigger.getNextExecutionDate() == null) {
+                            triggerContext = lastTrigger.toBuilder()
+                                .nextExecutionDate(((PollingTriggerInterface) abstractTrigger).nextEvaluationDate(conditionContext, Optional.of(lastTrigger)))
+                                .build();
+                            this.triggerState.save(triggerContext, scheduleContext);
+                        } else {
+                            triggerContext = lastTrigger;
+                        }
+                    } catch (Exception e) {
+                        logError(conditionContext, flow, abstractTrigger, e);
+                        return null;
+                    }
+                    return new FlowWithTriggers(
                         flow,
-                        trigger,
+                        abstractTrigger,
+                        triggerContext,
                         runContext,
                         conditionService.conditionContext(runContext, flow, null)
                     );
                 })
             )
-            .toList();
+            .filter(Objects::nonNull).toList();
     }
 
+    abstract public void handleNext(List<Flow> flows, ZonedDateTime now, BiConsumer<List<Trigger>, ScheduleContextInterface> consumer);
+
     private void handle() {
         if (!this.isReady) {
             log.warn("Scheduler is not ready, waiting");
         }
 
-        metricRegistry
-            .counter(MetricRegistry.SCHEDULER_LOOP_COUNT)
-            .increment();
-
         ZonedDateTime now = now();
 
-        if (log.isTraceEnabled()) {
-            log.trace(
-                "Scheduler next iteration for {} with {} schedulables of {} flows",
-                now,
-                schedulable.size(),
-                this.flowListeners.flows().size()
-            );
-        }
+        this.handleNext(this.flowListeners.flows(), now, (triggers, scheduleContext) -> {
+
+            triggers.forEach(trigger -> schedulableNextDate.remove(trigger.uid()));
 
-        synchronized (this) {
-            // get all triggers that are ready from evaluation
+            List<FlowWithTriggers> schedulable = this.computeSchedulable(flowListeners.flows(), triggers, scheduleContext);
+
+            metricRegistry
+                .counter(MetricRegistry.SCHEDULER_LOOP_COUNT)
+                .increment();
+
+            if (log.isTraceEnabled()) {
+                log.trace(
+                    "Scheduler next iteration for {} with {} schedulables of {} flows",
+                    now,
+                    schedulable.size(),
+                    this.flowListeners.flows().size()
+                );
+            }
+
+            // Get all triggers that are ready for evaluation
             List<FlowWithPollingTriggerNextDate> readyForEvaluate = schedulable
                 .stream()
-                .filter(f -> conditionService.isValid(f.getFlow(), f.getTrigger(), f.getConditionContext()))
-                .map(flowWithTrigger -> FlowWithPollingTrigger.builder()
-                    .flow(flowWithTrigger.getFlow())
-                    .trigger(flowWithTrigger.getTrigger())
-                    .pollingTrigger((PollingTriggerInterface) flowWithTrigger.getTrigger())
-                    .conditionContext(flowWithTrigger.getConditionContext())
-                    .triggerContext(TriggerContext
+                .filter(f -> conditionService.isValid(f.getFlow(), f.getAbstractTrigger(), f.getConditionContext()))
+                .map(flowWithTriggers -> FlowWithPollingTrigger.builder()
+                    .flow(flowWithTriggers.getFlow())
+                    .abstractTrigger(flowWithTriggers.getAbstractTrigger())
+                    .pollingTrigger((PollingTriggerInterface) flowWithTriggers.getAbstractTrigger())
+                    .conditionContext(flowWithTriggers.getConditionContext())
+                    .triggerContext(Trigger
                         .builder()
-                        .tenantId(flowWithTrigger.getFlow().getTenantId())
-                        .namespace(flowWithTrigger.getFlow().getNamespace())
-                        .flowId(flowWithTrigger.getFlow().getId())
-                        .flowRevision(flowWithTrigger.getFlow().getRevision())
-                        .triggerId(flowWithTrigger.getTrigger().getId())
+                        .tenantId(flowWithTriggers.getFlow().getTenantId())
+                        .namespace(flowWithTriggers.getFlow().getNamespace())
+                        .flowId(flowWithTriggers.getFlow().getId())
+                        .flowRevision(flowWithTriggers.getFlow().getRevision())
+                        .triggerId(flowWithTriggers.getAbstractTrigger().getId())
                         .date(now())
+                        .nextExecutionDate(flowWithTriggers.getTriggerContext().getNextExecutionDate())
                         .build()
                     )
-                    .build()
-                )
-                .filter(f -> this.isEvaluationInterval(f, now))
-                .filter(f -> !this.isTriggerRunning(f))
-                .filter(f -> this.isExecutionNotRunning(f, now))
+                    .build())
+                .filter(f -> f.getTriggerContext().getEvaluateRunningDate() == null)
+                .filter(this::isExecutionNotRunning)
                 .map(f -> {
                     try {
-                        Trigger lastTrigger = this.getLastTrigger(f, now);
-
-                        return FlowWithPollingTriggerNextDate.of(
-                            f,
-                            f.getPollingTrigger().nextEvaluationDate(f.getConditionContext(), Optional.of(lastTrigger))
-                        );
+                        return FlowWithPollingTriggerNextDate.of(f);
                     } catch (Exception e) {
                         logError(f, e);
 
@@ -280,7 +332,6 @@ public abstract class AbstractScheduler implements Scheduler {
                 .counter(MetricRegistry.SCHEDULER_EVALUATE_COUNT)
                 .increment(readyForEvaluate.size());
 
-
             // submit ready one to the worker
             readyForEvaluate
                 .forEach(f -> {
@@ -288,35 +339,35 @@ public abstract class AbstractScheduler implements Scheduler {
                     Logger logger = f.getConditionContext().getRunContext().logger();
 
                     if (f.getPollingTrigger().getInterval() != null) {
-                        // If and interval the trigger is executed by the Worker.
+                        // If have an interval, the trigger is executed by the Worker.
                         // Normally, only the Schedule trigger has no interval.
-                        var triggerRunning = Trigger.of(f.getTriggerContext(), now);
-                        triggerState.save(triggerRunning);
+                        Trigger triggerRunning = Trigger.of(f.getTriggerContext(), now);
 
                         try {
+                            this.triggerState.save(triggerRunning, scheduleContext);
                             this.sendPollingTriggerToWorker(f);
                         } catch (InternalException e) {
                             logger.error(
                                 "[namespace: {}] [flow: {}] [trigger: {}] Unable to send polling trigger to worker",
                                 f.getFlow().getNamespace(),
                                 f.getFlow().getId(),
-                                f.getTrigger().getId(),
+                                f.getAbstractTrigger().getId(),
                                 e
                             );
                         }
-                    } else if(f.getPollingTrigger() instanceof Schedule) {
-                        // This is the Schedule, all others trigger should have an interval.
+                    } else if (f.getPollingTrigger() instanceof Schedule) {
+                        // This is the Schedule, all other triggers should have an interval.
                         // So we evaluate it now as there is no need to send it to the worker.
                         // Schedule didn't use the triggerState to allow backfill.
                         try {
-                            var schedulerExecutionWithTrigger = evaluateScheduleTrigger(f);
-                            this.handleEvaluatePollingTriggerResult(schedulerExecutionWithTrigger);
+                            SchedulerExecutionWithTrigger schedulerExecutionWithTrigger = evaluateScheduleTrigger(f);
+                            this.handleEvaluateSchedulingTriggerResult(schedulerExecutionWithTrigger, scheduleContext);
                         } catch (Exception e) {
                             logger.error(
                                 "[namespace: {}] [flow: {}] [trigger: {}] Evaluate schedule trigger failed",
                                 f.getFlow().getNamespace(),
                                 f.getFlow().getId(),
-                                f.getTrigger().getId(),
+                                f.getAbstractTrigger().getId(),
                                 e
                             );
                         }
@@ -325,28 +376,61 @@ public abstract class AbstractScheduler implements Scheduler {
                             "[namespace: {}] [flow: {}] [trigger: {}] Polling trigger must have an interval (except the Schedule)",
                             f.getFlow().getNamespace(),
                             f.getFlow().getId(),
-                            f.getTrigger().getId()
+                            f.getAbstractTrigger().getId()
                         );
                     }
                 });
-        }
+
+        });
     }
 
+    // Polling triggers result is evaluated in another thread
+    // with the workerTriggerResultQueue,
+    // so we can't save them now
     private void handleEvaluatePollingTriggerResult(SchedulerExecutionWithTrigger result) {
         Stream.of(result)
             .filter(Objects::nonNull)
             .peek(this::log)
-            .forEach(this::saveLastTriggerAndEmitExecution);
+            .forEach(executionWithTrigger -> {
+                    Trigger trigger = Trigger.of(
+                        executionWithTrigger.getTriggerContext(),
+                        executionWithTrigger.getExecution()
+                    );
+                    // Check if the localTriggerState contains it
+                    // however, its mean it has been deleted during the execution time
+                    this.triggerState.update(trigger);
+                    this.saveLastTriggerAndEmitExecution(executionWithTrigger, trigger);
+                }
+            );
     }
 
-    private boolean isExecutionNotRunning(FlowWithPollingTrigger f, ZonedDateTime now) {
-        Trigger lastTrigger = null;
-        try {
-            lastTrigger = this.getLastTrigger(f, now);
-        } catch (Exception e) {
-            logError(f, e);
-            return false;
-        }
+    // Schedule triggers are being executed directly from
+    // the handle method within the context where triggers are locked,
+    // so we can save it now by pass the context
+    private void handleEvaluateSchedulingTriggerResult(SchedulerExecutionWithTrigger result, ScheduleContextInterface scheduleContext) {
+        Stream.of(result)
+            .filter(Objects::nonNull)
+            .peek(this::log)
+            .forEach(executionWithTrigger -> {
+                    Trigger trigger = Trigger.of(
+                        executionWithTrigger.getTriggerContext(),
+                        executionWithTrigger.getExecution(),
+                        ZonedDateTime.parse((String) executionWithTrigger.getExecution().getTrigger().getVariables().get("next"))
+                    );
+                    this.triggerState.save(trigger, scheduleContext);
+                    this.saveLastTriggerAndEmitExecution(executionWithTrigger, trigger);
+                }
+            );
+    }
+
+    protected void saveLastTriggerAndEmitExecution(SchedulerExecutionWithTrigger executionWithTrigger, Trigger trigger) {
+        // we need to be sure that the tenantId is propagated from the trigger to the execution
+        var execution = executionWithTrigger.getExecution().withTenantId(executionWithTrigger.getTriggerContext().getTenantId());
+        this.executionQueue.emit(execution);
+    }
+
+    private boolean isExecutionNotRunning(FlowWithPollingTrigger f) {
+        Trigger lastTrigger = f.getTriggerContext();
 
         if (lastTrigger.getExecutionId() == null) {
             return true;
@@ -432,91 +516,6 @@ public abstract class AbstractScheduler implements Scheduler {
         );
     }
 
-    private Trigger getLastTrigger(FlowWithPollingTrigger f, ZonedDateTime now) throws Exception {
-        return triggerState
-            .findLast(f.getTriggerContext())
-            .orElseGet(throwSupplier(() -> {
-                ZonedDateTime nextDate = f.getPollingTrigger().nextEvaluationDate(f.getConditionContext(), Optional.empty());
-
-                Trigger build = Trigger.builder()
-                    .tenantId(f.getTriggerContext().getTenantId())
-                    .date(nextDate.compareTo(now) < 0 ? nextDate : now)
-                    .flowId(f.getFlow().getId())
-                    .flowRevision(f.getFlow().getRevision())
-                    .namespace(f.getFlow().getNamespace())
-                    .triggerId(f.getTriggerContext().getTriggerId())
-                    .updatedDate(Instant.now())
-                    .build();
-
-                // We don't find a trigger, so the flow was never started.
-                // We create a trigger context with previous date in the past.
-                // This fix an edge case when the evaluation loop of the scheduler didn't catch up so new triggers was detected but not stored.
-                synchronized (triggerStateSavedLock) {
-                    if (triggerStateSaved.containsKey(build.uid())) {
-                        Trigger cachedTrigger = triggerStateSaved.get(build.uid());
-
-                        triggerState.save(build);
-                        triggerStateSaved.remove(build.uid());
-
-                        return cachedTrigger;
-                    } else {
-                        triggerStateSaved.put(build.uid(), build);
-                    }
-                }
-
-                return build;
-            }));
-    }
-
-    private boolean isEvaluationInterval(FlowWithPollingTrigger flowWithPollingTrigger, ZonedDateTime now) {
-        if (flowWithPollingTrigger.getPollingTrigger().getInterval() == null) {
-            return true;
-        }
-
-        String key = flowWithPollingTrigger.getTriggerContext().uid();
-
-        if (!this.lastEvaluate.containsKey(key)) {
-            this.lastEvaluate.put(key, now);
-            return true;
-        }
-
-        boolean result = this.lastEvaluate.get(key)
-            .plus(flowWithPollingTrigger.getPollingTrigger().getInterval())
-            .compareTo(now) < 0;
-
-        if (result) {
-            this.lastEvaluate.put(key, now);
-        }
-
-        return result;
-    }
-
-    private boolean isTriggerRunning(FlowWithPollingTrigger flowWithPollingTrigger) {
-        // We don't want to check if a trigger is running for Schedule trigger which didn't use the triggerState store
-        if (flowWithPollingTrigger.getPollingTrigger().getInterval() == null) {
-            return false;
-        }
-
-        var runningTrigger = this.triggerState
-            .findLast(flowWithPollingTrigger.getTriggerContext())
-            .filter(trigger -> trigger.getEvaluateRunningDate() != null);
-        return runningTrigger.isPresent();
-    }
-
-    protected void saveLastTriggerAndEmitExecution(SchedulerExecutionWithTrigger executionWithTrigger) {
-        Trigger trigger = Trigger.of(
-            executionWithTrigger.getTriggerContext(),
-            executionWithTrigger.getExecution()
-        );
-
-        synchronized (triggerStateSavedLock) {
-            this.triggerState.save(trigger);
-            // we need to be sure that the tenantId is propagated from the trigger to the execution
-            var execution = executionWithTrigger.getExecution().withTenantId(executionWithTrigger.getTriggerContext().getTenantId());
-            this.executionQueue.emit(execution);
-        }
-    }
-
     private static ZonedDateTime now() {
         return ZonedDateTime.now().truncatedTo(ChronoUnit.SECONDS);
     }
@@ -531,7 +530,7 @@ public abstract class AbstractScheduler implements Scheduler {
             // @TODO: mutability dirty that force creation of a new triggerExecutionId
             flowWithPollingTrigger.getConditionContext().getRunContext().forScheduler(
                 flowWithPollingTrigger.getTriggerContext(),
-                flowWithTrigger.getTrigger()
+                flowWithTrigger.getAbstractTrigger()
             );
 
             Optional<Execution> evaluate = flowWithPollingTrigger.getPollingTrigger().evaluate(
@@ -545,8 +544,8 @@ public abstract class AbstractScheduler implements Scheduler {
                         "[namespace: {}] [flow: {}] [trigger: {}] [type: {}] {}",
                         flowWithPollingTrigger.getFlow().getNamespace(),
                         flowWithPollingTrigger.getFlow().getId(),
-                        flowWithPollingTrigger.getTrigger().getId(),
-                        flowWithPollingTrigger.getTrigger().getType(),
+                        flowWithPollingTrigger.getAbstractTrigger().getId(),
+                        flowWithPollingTrigger.getAbstractTrigger().getType(),
                         evaluate.map(execution -> "New execution '" + execution.getId() + "'").orElse("Empty evaluation")
                     );
                 }
@@ -582,6 +581,20 @@ public abstract class AbstractScheduler implements Scheduler {
         }
     }
 
+    private void logError(ConditionContext conditionContext, Flow flow, AbstractTrigger trigger, Throwable e) {
+        Logger logger = conditionContext.getRunContext().logger();
+
+        logger.error(
+            "[namespace: {}] [flow: {}] [trigger: {}] [date: {}] Evaluate Failed with error '{}'",
+            flow.getNamespace(),
+            flow.getId(),
+            trigger.getId(),
+            now(),
+            e.getMessage(),
+            e
+        );
+    }
+
     private void sendPollingTriggerToWorker(FlowWithPollingTrigger flowWithTrigger) throws InternalException {
         FlowWithPollingTrigger flowWithTriggerWithDefault = flowWithTrigger.from(
             taskDefaultService.injectDefaults(flowWithTrigger.getFlow(),
@@ -600,7 +613,7 @@ public abstract class AbstractScheduler implements Scheduler {
 
         var workerTrigger = WorkerTrigger
             .builder()
-            .trigger(flowWithTriggerWithDefault.trigger)
+            .trigger(flowWithTriggerWithDefault.abstractTrigger)
             .triggerContext(flowWithTriggerWithDefault.triggerContext)
             .conditionContext(flowWithTriggerWithDefault.conditionContext)
             .build();
@@ -618,21 +631,21 @@ public abstract class AbstractScheduler implements Scheduler {
     @NoArgsConstructor
     private static class FlowWithPollingTrigger {
         private Flow flow;
-        private AbstractTrigger trigger;
+        private AbstractTrigger abstractTrigger;
         private PollingTriggerInterface pollingTrigger;
-        private TriggerContext triggerContext;
+        private Trigger triggerContext;
         private ConditionContext conditionContext;
 
         public FlowWithPollingTrigger from(Flow flow) throws InternalException {
             AbstractTrigger abstractTrigger = flow.getTriggers()
                 .stream()
-                .filter(a -> a.getId().equals(this.trigger.getId()) && a instanceof PollingTriggerInterface)
+                .filter(a -> a.getId().equals(this.abstractTrigger.getId()) && a instanceof PollingTriggerInterface)
                 .findFirst()
-                .orElseThrow(() -> new InternalException("Couldn't find the trigger '" + this.trigger.getId() + "' on flow '" + flow.uid() + "'"));
+                .orElseThrow(() -> new InternalException("Couldn't find the trigger '" + this.abstractTrigger.getId() + "' on flow '" + flow.uid() + "'"));
 
             return this.toBuilder()
                 .flow(flow)
-                .trigger(abstractTrigger)
+                .abstractTrigger(abstractTrigger)
                 .pollingTrigger((PollingTriggerInterface) abstractTrigger)
                 .build();
         }
@@ -644,32 +657,39 @@ public abstract class AbstractScheduler implements Scheduler {
     public static class FlowWithPollingTriggerNextDate extends FlowWithPollingTrigger {
         private ZonedDateTime next;
 
-        public static FlowWithPollingTriggerNextDate of(FlowWithPollingTrigger f, ZonedDateTime next) {
+        public static FlowWithPollingTriggerNextDate of(FlowWithPollingTrigger f) {
             return FlowWithPollingTriggerNextDate.builder()
                 .flow(f.getFlow())
-                .trigger(f.getTrigger())
+                .abstractTrigger(f.getAbstractTrigger())
                 .pollingTrigger(f.getPollingTrigger())
                 .conditionContext(f.getConditionContext())
-                .triggerContext(TriggerContext.builder()
+                .triggerContext(Trigger.builder()
                     .tenantId(f.getTriggerContext().getTenantId())
                     .namespace(f.getTriggerContext().getNamespace())
                     .flowId(f.getTriggerContext().getFlowId())
                     .flowRevision(f.getTriggerContext().getFlowRevision())
                     .triggerId(f.getTriggerContext().getTriggerId())
-                    .date(next)
+                    .date(f.getTriggerContext().getNextExecutionDate())
+                    .nextExecutionDate(f.getTriggerContext().getNextExecutionDate())
                     .build()
                 )
-                .next(next)
+                .next(f.getTriggerContext().getNextExecutionDate())
                 .build();
         }
     }
 
     @AllArgsConstructor
     @Getter
-    public static class FlowWithTrigger {
+    @Builder(toBuilder = true)
+    public static class FlowWithTriggers {
         private final Flow flow;
-        private final AbstractTrigger trigger;
+        private final AbstractTrigger AbstractTrigger;
+        private final Trigger TriggerContext;
         private final RunContext runContext;
         private final ConditionContext conditionContext;
+
+        public String uid() {
+            return Trigger.uid(flow, AbstractTrigger);
+        }
     }
 }
diff --git a/core/src/main/java/io/kestra/core/schedulers/DefaultScheduleContext.java b/core/src/main/java/io/kestra/core/schedulers/DefaultScheduleContext.java
new file mode 100644
index 000000000..7e0dd7838
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/schedulers/DefaultScheduleContext.java
@@ -0,0 +1,4 @@
+package io.kestra.core.schedulers;
+
+// For tests purpose
+public class DefaultScheduleContext implements ScheduleContextInterface {}
diff --git a/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java b/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
index dcbdb7571..11257313e 100644
--- a/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
@@ -1,6 +1,7 @@
 package io.kestra.core.schedulers;
 
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
@@ -10,13 +11,16 @@ import io.kestra.core.services.FlowListenersInterface;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.inject.qualifiers.Qualifiers;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
 
 import java.time.Duration;
+import java.time.ZonedDateTime;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
+import java.util.function.BiConsumer;
 
 @Slf4j
 @Singleton
@@ -59,10 +63,10 @@ public class DefaultScheduler extends AbstractScheduler {
                 Trigger trigger = Await.until(()  -> watchingTrigger.get(execution.getId()), Duration.ofSeconds(5));
                 var flow = flowRepository.findById(execution.getTenantId(), execution.getNamespace(), execution.getFlowId()).orElse(null);
                 if (execution.isDeleted() || conditionService.isTerminatedWithListeners(flow, execution)) {
-                    triggerState.save(trigger.resetExecution());
+                    triggerState.update(trigger.resetExecution());
                     watchingTrigger.remove(execution.getId());
                 } else {
-                    triggerState.save(Trigger.of(execution, trigger.getDate()));
+                    triggerState.update(Trigger.of(execution, trigger));
                 }
             }
         });
@@ -81,4 +85,11 @@ public class DefaultScheduler extends AbstractScheduler {
 
         super.run();
     }
+
+    @Override
+    public void handleNext(List<Flow> flows, ZonedDateTime now, BiConsumer<List<Trigger>, ScheduleContextInterface> consumer) {
+        List<Trigger> triggers =  triggerState.findAllForAllTenants().stream().filter(trigger -> trigger.getNextExecutionDate() == null || trigger.getNextExecutionDate().isBefore(now)).toList();
+        DefaultScheduleContext schedulerContext = new DefaultScheduleContext();
+        consumer.accept(triggers, schedulerContext);
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/schedulers/ScheduleContextInterface.java b/core/src/main/java/io/kestra/core/schedulers/ScheduleContextInterface.java
new file mode 100644
index 000000000..d58e0f864
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/schedulers/ScheduleContextInterface.java
@@ -0,0 +1,4 @@
+package io.kestra.core.schedulers;
+
+public interface ScheduleContextInterface {
+}
diff --git a/core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java b/core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java
index 9bc79d987..c23f2a7de 100644
--- a/core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java
+++ b/core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java
@@ -1,13 +1,31 @@
 package io.kestra.core.schedulers;
 
+import io.kestra.core.models.conditions.ConditionContext;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.TriggerContext;
+import jakarta.validation.ConstraintViolationException;
 
+import java.time.ZonedDateTime;
+import java.util.List;
 import java.util.Optional;
-import jakarta.validation.ConstraintViolationException;
 
 public interface SchedulerTriggerStateInterface {
     Optional<Trigger> findLast(TriggerContext trigger);
 
-    Trigger save(Trigger trigger) throws ConstraintViolationException;
+    List<Trigger> findAllForAllTenants();
+
+    Trigger save(Trigger trigger, ScheduleContextInterface scheduleContext) throws ConstraintViolationException;
+
+    Trigger create(Trigger trigger) throws ConstraintViolationException;
+
+    Trigger update(Trigger trigger);
+
+    Trigger update(Flow flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext) throws Exception;
+
+    List<Trigger> findByNextExecutionDateReadyForAllTenants(ZonedDateTime now, ScheduleContextInterface scheduleContext);
+
+    // Required for Kafka
+    List<Trigger> findByNextExecutionDateReadyForGivenFlows(List<Flow> flows, ZonedDateTime now, ScheduleContextInterface scheduleContext);
 }
diff --git a/core/src/main/java/io/kestra/core/services/FlowService.java b/core/src/main/java/io/kestra/core/services/FlowService.java
index ad28bb433..ab0cdc0ea 100644
--- a/core/src/main/java/io/kestra/core/services/FlowService.java
+++ b/core/src/main/java/io/kestra/core/services/FlowService.java
@@ -172,6 +172,16 @@ public class FlowService {
             .collect(Collectors.toList());
     }
 
+    public static List<AbstractTrigger> findUpdatedTrigger(Flow flow, Flow previous) {
+        return ListUtils.emptyOnNull(previous.getTriggers())
+            .stream()
+            .filter(oldTrigger -> ListUtils.emptyOnNull(flow.getTriggers())
+                .stream()
+                .anyMatch(trigger -> trigger.getId().equals(oldTrigger.getId()) && !trigger.equals(oldTrigger))
+            )
+            .toList();
+    }
+
     public static String cleanupSource(String source) {
         return source.replaceFirst("(?m)^revision: \\d+\n?", "");
     }
diff --git a/core/src/main/java/io/kestra/core/validations/validator/WorkerGroupValidator.java b/core/src/main/java/io/kestra/core/validations/validator/WorkerGroupValidator.java
index 859358420..239ef5cce 100644
--- a/core/src/main/java/io/kestra/core/validations/validator/WorkerGroupValidator.java
+++ b/core/src/main/java/io/kestra/core/validations/validator/WorkerGroupValidator.java
@@ -18,7 +18,7 @@ public class WorkerGroupValidator  implements ConstraintValidator<WorkerGroupVal
         @Nullable WorkerGroup value,
         @NonNull AnnotationValue<WorkerGroupValidation> annotationMetadata,
         @NonNull ConstraintValidatorContext context) {
-        if (value == null) {
+        if (value == null|| value.getKey() == null) {
             return true;
         }
 
diff --git a/jdbc-h2/src/main/resources/migrations/h2/V1_15__trigger_store_next_date.sql b/jdbc-h2/src/main/resources/migrations/h2/V1_15__trigger_store_next_date.sql
new file mode 100644
index 000000000..34660fdd3
--- /dev/null
+++ b/jdbc-h2/src/main/resources/migrations/h2/V1_15__trigger_store_next_date.sql
@@ -0,0 +1,1 @@
+ALTER TABLE triggers ADD COLUMN "next_execution_date" TIMESTAMP GENERATED ALWAYS AS (PARSEDATETIME(JQ_STRING("value", '.nextExecutionDate'), 'yyyy-MM-dd''T''HH:mm:ss.SSSXXX'));
\ No newline at end of file
diff --git a/jdbc-mysql/src/main/resources/migrations/mysql/V1_15__trigger_store_next_date.sql b/jdbc-mysql/src/main/resources/migrations/mysql/V1_15__trigger_store_next_date.sql
new file mode 100644
index 000000000..315200017
--- /dev/null
+++ b/jdbc-mysql/src/main/resources/migrations/mysql/V1_15__trigger_store_next_date.sql
@@ -0,0 +1,11 @@
+ALTER TABLE triggers ADD COLUMN `next_execution_date` DATETIME(6) GENERATED ALWAYS AS (
+        IF(
+            SUBSTRING(value ->> '$.nextExecutionDate', LENGTH(value ->> '$.nextExecutionDate'), LENGTH(value ->> '$.nextExecutionDate')) = 'Z',
+            STR_TO_DATE(value ->> '$.nextExecutionDate', '%Y-%m-%dT%H:%i:%s.%fZ'),
+            CONVERT_TZ(
+                STR_TO_DATE(SUBSTRING(value ->> '$.nextExecutionDate', 1, LENGTH(value ->> '$.nextExecutionDate') - 6), '%Y-%m-%dT%H:%i:%s.%f'),
+                SUBSTRING(value ->> '$.nextExecutionDate', LENGTH(value ->> '$.nextExecutionDate') - 5, 5),
+                'UTC'
+                )
+        )
+    ) STORED;
\ No newline at end of file
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V1_15__trigger_store_next_date.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V1_15__trigger_store_next_date.sql
new file mode 100644
index 000000000..c12209792
--- /dev/null
+++ b/jdbc-postgres/src/main/resources/migrations/postgres/V1_15__trigger_store_next_date.sql
@@ -0,0 +1,1 @@
+ALTER TABLE triggers ADD COLUMN "next_execution_date" TIMESTAMPTZ GENERATED ALWAYS AS (PARSE_ISO8601_DATETIME(value ->> 'nextExecutionDate')) STORED;
\ No newline at end of file
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
index 9cf7392f9..18232954f 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
@@ -1,16 +1,22 @@
 package io.kestra.jdbc.repository;
 
+import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.repositories.ArrayListTotal;
 import io.kestra.core.repositories.TriggerRepositoryInterface;
+import io.kestra.core.schedulers.ScheduleContextInterface;
 import io.kestra.jdbc.runner.JdbcIndexerInterface;
+import io.kestra.jdbc.runner.JdbcSchedulerContext;
 import io.micronaut.data.model.Pageable;
 import jakarta.inject.Singleton;
 import org.jooq.*;
 import org.jooq.impl.DSL;
 
+import java.time.ZonedDateTime;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -70,6 +76,32 @@ public abstract class AbstractJdbcTriggerRepository extends AbstractJdbcReposito
             });
     }
 
+    public List<Trigger> findByNextExecutionDateReadyForAllTenants(ZonedDateTime now, ScheduleContextInterface scheduleContextInterface) {
+        JdbcSchedulerContext jdbcSchedulerContext = (JdbcSchedulerContext) scheduleContextInterface;
+
+        return jdbcSchedulerContext.getContext()
+            .select(field("value"))
+            .from(this.jdbcRepository.getTable())
+            .where(
+                field("next_execution_date").lessThan(now.toOffsetDateTime())
+                    // we check for null for backwards compatibility
+                    .or(field("next_execution_date").isNull())
+            )
+            .orderBy(field("next_execution_date").asc())
+            .forUpdate()
+            .fetch()
+            .map(r -> this.jdbcRepository.deserialize(r.get("value").toString()));
+    }
+
+    public Trigger save(Trigger trigger, ScheduleContextInterface scheduleContextInterface) {
+        JdbcSchedulerContext jdbcSchedulerContext = (JdbcSchedulerContext) scheduleContextInterface;
+
+        Map<Field<Object>, Object> fields = this.jdbcRepository.persistFields(trigger);
+        this.jdbcRepository.persist(trigger, jdbcSchedulerContext.getContext(), fields);
+
+        return trigger;
+    }
+
     @Override
     public Trigger save(Trigger trigger) {
         Map<Field<Object>, Object> fields = this.jdbcRepository.persistFields(trigger);
@@ -86,11 +118,119 @@ public abstract class AbstractJdbcTriggerRepository extends AbstractJdbcReposito
         return trigger;
     }
 
+    public Trigger create(Trigger trigger) {
+        return this.jdbcRepository
+            .getDslContextWrapper()
+            .transactionResult(configuration -> {
+                DSL.using(configuration)
+                    .insertInto(this.jdbcRepository.getTable())
+                    .set(AbstractJdbcRepository.field("key"), this.jdbcRepository.key(trigger))
+                    .set(this.jdbcRepository.persistFields(trigger))
+                    .execute();
+
+                return trigger;
+            });
+    }
+
     @Override
     public void delete(Trigger trigger) {
         this.jdbcRepository.delete(trigger);
     }
 
+    public Trigger update(Trigger trigger) {
+        return this.jdbcRepository
+            .getDslContextWrapper()
+            .transactionResult(configuration -> {
+                DSL.using(configuration)
+                    .update(this.jdbcRepository.getTable())
+                    .set(this.jdbcRepository.persistFields((trigger)))
+                    .where(field("key").eq(trigger.uid()))
+                    .execute();
+
+                return trigger;
+            });
+    }
+
+    // update/reset execution need to be done in a transaction
+    // to be sure we get the correct date/nextDate when updating
+    public Trigger updateExecution(Trigger trigger) {
+        return this.jdbcRepository
+            .getDslContextWrapper()
+            .transactionResult(configuration -> {
+                DSLContext context = DSL.using(configuration);
+                Optional<Trigger> optionalTrigger = this.jdbcRepository.fetchOne(context.select(field("value"))
+                    .from(this.jdbcRepository.getTable())
+                    .where(
+                        field("key").eq(trigger.uid())
+                    ).forUpdate());
+
+                if (optionalTrigger.isPresent()) {
+                    Trigger current = optionalTrigger.get();
+                    current = current.toBuilder()
+                        .executionId(trigger.getExecutionId())
+                        .executionCurrentState(trigger.getExecutionCurrentState())
+                        .updatedDate(trigger.getUpdatedDate())
+                        .build();
+                    this.save(context, current);
+
+                    return current;
+                }
+
+                return null;
+            });
+    }
+
+    // update/reset execution need to be done in a transaction
+    // to be sure we get the correct date/nextDate when updating
+    public Trigger resetExecution(Trigger trigger) {
+        return this.jdbcRepository
+            .getDslContextWrapper()
+            .transactionResult(configuration -> {
+                DSLContext context = DSL.using(configuration);
+                Optional<Trigger> optionalTrigger = this.jdbcRepository.fetchOne(context.select(field("value"))
+                    .from(this.jdbcRepository.getTable())
+                    .where(
+                        field("key").eq(trigger.uid())
+                    ).forUpdate());
+
+                if (optionalTrigger.isPresent()) {
+                    Trigger current = optionalTrigger.get();
+                    this.save(context, current.resetExecution());
+
+                    return current;
+                }
+
+                return null;
+            });
+    }
+
+    // Allow to update a trigger from a flow & an abstract trigger
+    // using forUpdate to avoid the lastTrigger to be updated by another thread
+    // before doing the update
+    public Trigger update(Flow flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext) {
+        return this.jdbcRepository
+            .getDslContextWrapper()
+            .transactionResult(configuration -> {
+                Optional<Trigger> lastTrigger = this.jdbcRepository.fetchOne(DSL
+                    .using(configuration)
+                    .select(field("value"))
+                    .from(this.jdbcRepository.getTable())
+                    .where(field("key").eq(Trigger.uid(flow, abstractTrigger)))
+                    .forUpdate()
+                );
+
+                Trigger updatedTrigger = Trigger.of(flow, abstractTrigger, conditionContext, lastTrigger);
+
+                DSL.using(configuration)
+                    .update(this.jdbcRepository.getTable())
+                    .set(this.jdbcRepository.persistFields(updatedTrigger))
+                    .where(field("key").eq(updatedTrigger.uid()))
+                    .execute();
+
+                return updatedTrigger;
+            });
+    }
+
     @Override
     public ArrayListTotal<Trigger> find(Pageable pageable, String query, String tenantId, String namespace) {
         return this.jdbcRepository
@@ -120,7 +260,7 @@ public abstract class AbstractJdbcTriggerRepository extends AbstractJdbcReposito
     }
 
     protected Condition defaultFilter(String tenantId) {
-        return buildTenantCondition(tenantId) ;
+        return buildTenantCondition(tenantId);
     }
 
     @Override
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 ad0172a05..a2b3dcaaa 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -502,37 +502,6 @@ public class JdbcExecutor implements ExecutorInterface {
                     });
             }
 
-            // multiple condition
-            if (
-                conditionService.isTerminatedWithListeners(flow, execution) &&
-                    this.deduplicateFlowTrigger(execution, executorState)
-            ) {
-                flowTriggerService.computeExecutionsFromFlowTriggers(execution, allFlows, Optional.of(multipleConditionStorage))
-                    .forEach(this.executionQueue::emit);
-            }
-
-            // when terminated: handle queued executions and subflow executions
-            if (conditionService.isTerminatedWithListeners(flow, execution)) {
-                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 (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.withState(State.Type.RUNNING))
-                    );
-                }
-            }
-
             return Pair.of(
                 executor,
                 executorState
@@ -842,6 +811,35 @@ public class JdbcExecutor implements ExecutorInterface {
         if (executorService.canBePurged(executor)) {
             executorStateStorage.delete(executor.getExecution());
         }
+
+        // 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())) {
+            Execution execution = executor.getExecution();
+            // handle flow triggers
+            flowTriggerService.computeExecutionsFromFlowTriggers(execution, allFlows, Optional.of(multipleConditionStorage))
+                .forEach(this.executionQueue::emit);
+
+            // 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);
+                });
+
+            // check if there exist a queued execution and submit it to the execution queue
+            if (executor.getFlow().getConcurrency() != null && executor.getFlow().getConcurrency().getBehavior() == Concurrency.Behavior.QUEUE) {
+                executionQueuedStorage.pop(executor.getFlow().getTenantId(),
+                    executor.getFlow().getNamespace(),
+                    executor.getFlow().getId(),
+                    queued -> executionQueue.emit(queued.withState(State.Type.RUNNING))
+                );
+            }
+        }
     }
 
     private Flow transform(Flow flow, Execution execution) {
@@ -939,18 +937,6 @@ public class JdbcExecutor implements ExecutorInterface {
         }
     }
 
-    private boolean deduplicateFlowTrigger(Execution execution, ExecutorState executorState) {
-        Boolean flowTriggerDeduplication = executorState.getFlowTriggerDeduplication();
-
-        if (flowTriggerDeduplication) {
-            log.trace("Duplicate Flow Trigger on execution '{}'", execution.getId());
-            return false;
-        } else {
-            executorState.setFlowTriggerDeduplication(true);
-            return true;
-        }
-    }
-
     private Executor handleFailedExecutionFromExecutor(Executor executor, Exception e) {
         Execution.FailedExecutionWithLog failedExecutionWithLog = executor.getExecution().failedExecutionFromExecutor(e);
 
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
index 02dd6e7a6..4426552fd 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
@@ -1,6 +1,7 @@
 package io.kestra.jdbc.runner;
 
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
@@ -11,6 +12,7 @@ import io.kestra.core.services.ConditionService;
 import io.kestra.core.services.FlowListenersInterface;
 import io.kestra.core.services.FlowService;
 import io.kestra.core.utils.ListUtils;
+import io.kestra.jdbc.JooqDSLContextWrapper;
 import io.kestra.jdbc.repository.AbstractJdbcTriggerRepository;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.annotation.Replaces;
@@ -19,6 +21,10 @@ import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
 
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.function.BiConsumer;
+
 @JdbcRunnerEnabled
 @Singleton
 @Slf4j
@@ -29,6 +35,8 @@ public class JdbcScheduler extends AbstractScheduler {
     private final ConditionService conditionService;
 
     private final FlowRepositoryInterface flowRepository;
+    private final JooqDSLContextWrapper dslContextWrapper;
+
 
     @SuppressWarnings("unchecked")
     @Inject
@@ -43,7 +51,7 @@ public class JdbcScheduler extends AbstractScheduler {
         triggerState = applicationContext.getBean(SchedulerTriggerStateInterface.class);
         conditionService = applicationContext.getBean(ConditionService.class);
         flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
-
+        dslContextWrapper = applicationContext.getBean(JooqDSLContextWrapper.class);
         this.isReady = true;
     }
 
@@ -55,7 +63,7 @@ public class JdbcScheduler extends AbstractScheduler {
             Scheduler.class,
             either -> {
                 if (either.isRight()) {
-                    log.error("Unable to dserialize an execution: {}", either.getRight().getMessage());
+                    log.error("Unable to deserialize an execution: {}", either.getRight().getMessage());
                     return;
                 }
 
@@ -66,13 +74,17 @@ public class JdbcScheduler extends AbstractScheduler {
                         // reset scheduler trigger at end
                         triggerRepository
                             .findByExecution(execution)
-                            .ifPresent(trigger -> triggerRepository.save(trigger.resetExecution()));
+                            .ifPresent(trigger -> {
+                                ((JdbcSchedulerTriggerState) this.triggerState).resetExecution(trigger.resetExecution());
+                            });
                     } else {
                         // update execution state on each state change so the scheduler knows the execution is running
                         triggerRepository
                             .findByExecution(execution)
                             .filter(trigger -> execution.getState().getCurrent() != trigger.getExecutionCurrentState())
-                            .ifPresent(trigger -> triggerRepository.save(Trigger.of(execution, trigger.getDate())));
+                            .ifPresent(trigger -> {
+                                ((JdbcSchedulerTriggerState) this.triggerState).updateExecution(Trigger.of(execution, trigger));
+                            });
                     }
                 }
             }
@@ -90,4 +102,15 @@ public class JdbcScheduler extends AbstractScheduler {
             }
         });
     }
+
+    @Override
+    public void handleNext(List<Flow> flows, ZonedDateTime now, BiConsumer<List<Trigger>, ScheduleContextInterface> consumer) {
+        JdbcSchedulerContext schedulerContext = new JdbcSchedulerContext(this.dslContextWrapper);
+
+        schedulerContext.startTransaction(scheduleContextInterface -> {
+            List<Trigger> triggers = this.triggerState.findByNextExecutionDateReadyForAllTenants(now, scheduleContextInterface);
+
+            consumer.accept(triggers, scheduleContextInterface);
+        });
+    }
 }
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerContext.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerContext.java
new file mode 100644
index 000000000..4c3935e84
--- /dev/null
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerContext.java
@@ -0,0 +1,34 @@
+package io.kestra.jdbc.runner;
+
+import io.kestra.core.schedulers.ScheduleContextInterface;
+import io.kestra.jdbc.JooqDSLContextWrapper;
+import lombok.Getter;
+import org.jooq.DSLContext;
+import org.jooq.impl.DSL;
+
+import java.util.function.Consumer;
+
+@Getter
+public class JdbcSchedulerContext implements ScheduleContextInterface {
+
+    private DSLContext context;
+    private final JooqDSLContextWrapper dslContextWrapper;
+
+    public JdbcSchedulerContext(JooqDSLContextWrapper dslContextWrapper) {
+        this.dslContextWrapper = dslContextWrapper;
+    }
+
+    public void startTransaction(Consumer<ScheduleContextInterface> consumer) {
+        this.dslContextWrapper.transaction(configuration -> {
+            this.context = DSL.using(configuration);
+
+            consumer.accept(this);
+
+            this.commit();
+        });
+    }
+
+    public void commit() {
+        this.context.commit();
+    }
+}
\ No newline at end of file
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
index 2571f84f0..339814d24 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
@@ -1,20 +1,27 @@
 package io.kestra.jdbc.runner;
 
+import io.kestra.core.models.conditions.ConditionContext;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.TriggerContext;
-import io.kestra.core.repositories.TriggerRepositoryInterface;
+import io.kestra.core.schedulers.ScheduleContextInterface;
 import io.kestra.core.schedulers.SchedulerTriggerStateInterface;
+import io.kestra.jdbc.repository.AbstractJdbcTriggerRepository;
 import jakarta.annotation.PostConstruct;
 import jakarta.inject.Singleton;
+import org.apache.commons.lang3.NotImplementedException;
 
+import java.time.ZonedDateTime;
+import java.util.List;
 import java.util.Optional;
 
 @Singleton
 @JdbcRunnerEnabled
 public class JdbcSchedulerTriggerState implements SchedulerTriggerStateInterface {
-    protected TriggerRepositoryInterface triggerRepository;
+    protected AbstractJdbcTriggerRepository triggerRepository;
 
-    public JdbcSchedulerTriggerState(TriggerRepositoryInterface triggerRepository) {
+    public JdbcSchedulerTriggerState(AbstractJdbcTriggerRepository triggerRepository) {
         this.triggerRepository = triggerRepository;
     }
 
@@ -36,9 +43,50 @@ public class JdbcSchedulerTriggerState implements SchedulerTriggerStateInterface
     }
 
     @Override
-    public Trigger save(Trigger trigger) {
-        triggerRepository.save(trigger);
+    public List<Trigger> findAllForAllTenants() {
+        return this.triggerRepository.findAllForAllTenants();
+    }
+
+    @Override
+    public Trigger save(Trigger trigger, ScheduleContextInterface scheduleContextInterface) {
+        this.triggerRepository.save(trigger, scheduleContextInterface);
 
         return trigger;
     }
+
+    @Override
+    public Trigger create(Trigger trigger) {
+
+        return this.triggerRepository.create(trigger);
+    }
+
+    @Override
+    public Trigger update(Trigger trigger) {
+
+        return this.triggerRepository.update(trigger);
+    }
+
+    public Trigger updateExecution(Trigger trigger) {
+
+        return this.triggerRepository.updateExecution(trigger);
+    }
+
+    public Trigger resetExecution(Trigger trigger) {
+
+        return this.triggerRepository.resetExecution(trigger);
+    }
+
+    public Trigger update(Flow flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext) {
+        return this.triggerRepository.update(flow, abstractTrigger, conditionContext);
+    }
+
+    @Override
+    public List<Trigger> findByNextExecutionDateReadyForAllTenants(ZonedDateTime now, ScheduleContextInterface scheduleContext) {
+        return this.triggerRepository.findByNextExecutionDateReadyForAllTenants(now, scheduleContext);
+    }
+
+    @Override
+    public List<Trigger> findByNextExecutionDateReadyForGivenFlows(List<Flow> flows, ZonedDateTime now, ScheduleContextInterface scheduleContext) {
+        throw new NotImplementedException();
+    }
 }
diff --git a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryTriggerRepository.java b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryTriggerRepository.java
index e35183636..e7086af72 100644
--- a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryTriggerRepository.java
+++ b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryTriggerRepository.java
@@ -5,12 +5,13 @@ import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.repositories.ArrayListTotal;
 import io.kestra.core.repositories.TriggerRepositoryInterface;
-
-import java.util.*;
-
 import io.micronaut.data.model.Pageable;
 import jakarta.inject.Singleton;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
 @Singleton
 @MemoryRepositoryEnabled
 public class MemoryTriggerRepository implements TriggerRepositoryInterface {
diff --git a/runner-memory/src/main/java/io/kestra/runner/memory/MemorySchedulerTriggerState.java b/runner-memory/src/main/java/io/kestra/runner/memory/MemorySchedulerTriggerState.java
index a1760e3e8..481c1d5a7 100644
--- a/runner-memory/src/main/java/io/kestra/runner/memory/MemorySchedulerTriggerState.java
+++ b/runner-memory/src/main/java/io/kestra/runner/memory/MemorySchedulerTriggerState.java
@@ -1,20 +1,26 @@
 package io.kestra.runner.memory;
 
+import io.kestra.core.models.conditions.ConditionContext;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.schedulers.ScheduleContextInterface;
 import io.kestra.core.schedulers.SchedulerTriggerStateInterface;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+import org.apache.commons.lang3.NotImplementedException;
 
-import java.util.HashMap;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
-import jakarta.inject.Inject;
-import jakarta.inject.Named;
-import jakarta.inject.Singleton;
-
 @Singleton
 @MemoryQueueEnabled
 public class MemorySchedulerTriggerState implements SchedulerTriggerStateInterface {
@@ -30,10 +36,47 @@ public class MemorySchedulerTriggerState implements SchedulerTriggerStateInterfa
     }
 
     @Override
-    public Trigger save(Trigger trigger) {
+    public List<Trigger> findAllForAllTenants() {
+        return new ArrayList<>(triggers.values());
+    }
+
+    @Override
+    public Trigger save(Trigger trigger, ScheduleContextInterface scheduleContextInterface) {
+        triggers.put(trigger.uid(), trigger);
+        triggerQueue.emit(trigger);
+
+        return trigger;
+    }
+
+    @Override
+    public Trigger create(Trigger trigger) {
         triggers.put(trigger.uid(), trigger);
         triggerQueue.emit(trigger);
 
         return trigger;
     }
+
+    @Override
+    public Trigger update(Trigger trigger) {
+        triggers.put(trigger.uid(), trigger);
+        triggerQueue.emit(trigger);
+
+        return trigger;
+    }
+
+    @Override
+    public Trigger update(Flow flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext) throws Exception {
+        Optional<Trigger> lastTrigger = this.findLast(Trigger.of(flow, abstractTrigger));
+        return this.update(Trigger.of(flow, abstractTrigger, conditionContext, lastTrigger));
+    }
+
+    @Override
+    public List<Trigger> findByNextExecutionDateReadyForAllTenants(ZonedDateTime now, ScheduleContextInterface scheduleContext) {
+        return triggers.values().stream().filter(trigger -> trigger.getNextExecutionDate() == null || trigger.getNextExecutionDate().isBefore(now)).toList();
+    }
+
+    @Override
+    public List<Trigger> findByNextExecutionDateReadyForGivenFlows(List<Flow> flows, ZonedDateTime now, ScheduleContextInterface scheduleContext) {
+        throw new NotImplementedException();
+    }
 }
diff --git a/ui/src/components/admin/Triggers.vue b/ui/src/components/admin/Triggers.vue
index 24e3acacb..25ce58568 100644
--- a/ui/src/components/admin/Triggers.vue
+++ b/ui/src/components/admin/Triggers.vue
@@ -89,6 +89,11 @@
                                 {{ scope.row.updatedDate ? $filters.date(scope.row.updatedDate, "iso") : "" }}
                             </template>
                         </el-table-column>
+                        <el-table-column :label="$t('next execution date')">
+                            <template #default="scope">
+                                {{ scope.row.nextExecutionDate ? $filters.date(scope.row.nextExecutionDate, "iso") : "" }}
+                            </template>
+                        </el-table-column>
                         <el-table-column :label="$t('evaluation lock date')">
                             <template #default="scope">
                                 {{ scope.row.evaluateRunningDate ? $filters.date(scope.row.evaluateRunningDate, "iso") : "" }}
diff --git a/ui/src/routes/routes.js b/ui/src/routes/routes.js
index 65a3165b2..40e21c885 100644
--- a/ui/src/routes/routes.js
+++ b/ui/src/routes/routes.js
@@ -51,7 +51,7 @@ export default [
 
     //Documentation
     {name: "plugins/list", path: "/:tenant?/plugins", component: Plugin},
-    {name: "plugins/view", path: "/:tenant?/plugins/:cls", component: Plugin},
+    {name: "plugins/view", path: "/:tenant?/plugins/:cls",   component: Plugin},
 
     //Templates
     {name: "templates/list", path: "/:tenant?/templates", component: Templates},
diff --git a/ui/src/translations.json b/ui/src/translations.json
index c3e0bda5f..901eb0382 100644
--- a/ui/src/translations.json
+++ b/ui/src/translations.json
@@ -87,6 +87,7 @@
     },
     "created date": "Created date",
     "updated date": "Updated date",
+    "next execution date": "Next execution date",
     "jump to...": "Jump to...",
     "source": "Source",
     "home": "Home",
@@ -658,6 +659,7 @@
     },
     "created date": "Date de création",
     "updated date": "Date de mise à jour",
+    "next execution date": "Prochaine date d'exécution",
     "jump to...": "Aller à...",
     "source": "Source",
     "home": "Accueil",


Test patch:
diff --git a/core/src/main/java/io/kestra/core/utils/TestsUtils.java b/core/src/main/java/io/kestra/core/utils/TestsUtils.java
index 973cf904f..8d33ed3a1 100644
--- a/core/src/main/java/io/kestra/core/utils/TestsUtils.java
+++ b/core/src/main/java/io/kestra/core/utils/TestsUtils.java
@@ -61,6 +61,10 @@ abstract public class TestsUtils {
         return matchingLogs.isEmpty() ? null : matchingLogs.get(0);
     }
 
+    public static List<LogEntry> awaitLogs(List<LogEntry> logs, Integer exactCount) {
+        return awaitLogs(logs, logEntry -> true, exactCount::equals);
+    }
+
     public static List<LogEntry> awaitLogs(List<LogEntry> logs, Predicate<LogEntry> logMatcher, Integer exactCount) {
         return awaitLogs(logs, logMatcher, exactCount::equals);
     }
diff --git a/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java b/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
index e5fbf0d4e..6115f83c9 100644
--- a/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
@@ -4,17 +4,18 @@ import io.kestra.core.models.Label;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.conditions.types.DateTimeBetweenCondition;
 import io.kestra.core.models.conditions.types.DayWeekInMonthCondition;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.input.StringInput;
 import io.kestra.core.models.triggers.AbstractTrigger;
-import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
-import org.junit.jupiter.api.Test;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.tasks.debugs.Return;
 import io.kestra.core.utils.IdUtils;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
 
 import java.time.DayOfWeek;
 import java.time.Duration;
@@ -22,9 +23,10 @@ import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
-import java.util.*;
-
-import jakarta.inject.Inject;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
@@ -358,16 +360,14 @@ class ScheduleTest {
             .withSecond(0)
             .truncatedTo(ChronoUnit.SECONDS);
 
-        Optional<Execution> evaluate = trigger.evaluate(
+        ZonedDateTime evaluate = trigger.nextEvaluationDate(
             conditionContext(trigger),
-            TriggerContext.builder()
+            Optional.of(TriggerContext.builder()
                 .date(date)
-                .build()
+                .build())
         );
 
-        assertThat(evaluate.isPresent(), is(true));
-        var vars = (Map<String, String>) evaluate.get().getVariables().get("schedule");
-        assertThat(dateFromVars(vars.get("date"), date), is(expected));
+        assertThat(evaluate, is(expected));
 
     }
 
diff --git a/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java b/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
index bd8c81e12..cb4aa45dc 100644
--- a/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
+++ b/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
@@ -226,7 +226,7 @@ public abstract class AbstractFlowRepositoryTest {
     @Test
     void findByNamespace() {
         List<Flow> save = flowRepository.findByNamespace(null, "io.kestra.tests");
-        assertThat((long) save.size(), is(Helpers.FLOWS_COUNT - 3));
+        assertThat((long) save.size(), is(Helpers.FLOWS_COUNT - 14));
 
         save = flowRepository.findByNamespace(null, "io.kestra.tests2");
         assertThat((long) save.size(), is(1L));
@@ -395,7 +395,7 @@ public abstract class AbstractFlowRepositoryTest {
     @Test
     void findDistinctNamespace() {
         List<String> distinctNamespace = flowRepository.findDistinctNamespace(null);
-        assertThat((long) distinctNamespace.size(), is(4L));
+        assertThat((long) distinctNamespace.size(), is(5L));
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/core/runners/FlowTriggerCaseTest.java b/core/src/test/java/io/kestra/core/runners/FlowTriggerCaseTest.java
index 79f7d0ad7..a5b72238b 100644
--- a/core/src/test/java/io/kestra/core/runners/FlowTriggerCaseTest.java
+++ b/core/src/test/java/io/kestra/core/runners/FlowTriggerCaseTest.java
@@ -16,6 +16,7 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Singleton
 public class FlowTriggerCaseTest {
@@ -31,9 +32,10 @@ public class FlowTriggerCaseTest {
     protected RunnerUtils runnerUtils;
 
     public void trigger() throws InterruptedException, TimeoutException {
-        CountDownLatch countDownLatch = new CountDownLatch(2);
+        CountDownLatch countDownLatch = new CountDownLatch(3);
         AtomicReference<Execution> flowListener = new AtomicReference<>();
         AtomicReference<Execution> flowListenerNoInput = new AtomicReference<>();
+        AtomicReference<Execution> flowListenerNamespace = new AtomicReference<>();
 
         executionQueue.receive(either -> {
             Execution execution = either.getLeft();
@@ -44,29 +46,36 @@ public class FlowTriggerCaseTest {
                 } else if (flowListener.get() == null && execution.getFlowId().equals("trigger-flow-listener")) {
                     flowListener.set(execution);
                     countDownLatch.countDown();
+                } else if (flowListenerNamespace.get() == null && execution.getFlowId().equals("trigger-flow-listener-namespace-condition")) {
+                    flowListenerNamespace.set(execution);
+                    countDownLatch.countDown();
                 }
             }
         });
 
-        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "trigger-flow");
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests.trigger", "trigger-flow");
 
         assertThat(execution.getTaskRunList().size(), is(1));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
-        countDownLatch.await(10, TimeUnit.SECONDS);
+        assertTrue(countDownLatch.await(15, TimeUnit.SECONDS));
 
         assertThat(flowListener.get().getTaskRunList().size(), is(1));
         assertThat(flowListener.get().getState().getCurrent(), is(State.Type.SUCCESS));
         assertThat(flowListener.get().getTaskRunList().get(0).getOutputs().get("value"), is("childs: from parents: " + execution.getId()));
         assertThat(flowListener.get().getTrigger().getVariables().get("executionId"), is(execution.getId()));
-        assertThat(flowListener.get().getTrigger().getVariables().get("namespace"), is("io.kestra.tests"));
+        assertThat(flowListener.get().getTrigger().getVariables().get("namespace"), is("io.kestra.tests.trigger"));
         assertThat(flowListener.get().getTrigger().getVariables().get("flowId"), is("trigger-flow"));
 
         assertThat(flowListenerNoInput.get().getTaskRunList().size(), is(1));
         assertThat(flowListenerNoInput.get().getTrigger().getVariables().get("executionId"), is(execution.getId()));
-        assertThat(flowListenerNoInput.get().getTrigger().getVariables().get("namespace"), is("io.kestra.tests"));
+        assertThat(flowListenerNoInput.get().getTrigger().getVariables().get("namespace"), is("io.kestra.tests.trigger"));
         assertThat(flowListenerNoInput.get().getTrigger().getVariables().get("flowId"), is("trigger-flow"));
-
         assertThat(flowListenerNoInput.get().getState().getCurrent(), is(State.Type.SUCCESS));
+
+        assertThat(flowListenerNamespace.get().getTaskRunList().size(), is(1));
+        assertThat(flowListenerNamespace.get().getTrigger().getVariables().get("executionId"), is(execution.getId()));
+        assertThat(flowListenerNamespace.get().getTrigger().getVariables().get("namespace"), is("io.kestra.tests.trigger"));
+        assertThat(flowListenerNamespace.get().getTrigger().getVariables().get("flowId"), is("trigger-flow"));
     }
 }
diff --git a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java b/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
index ac40c797c..f714e0465 100644
--- a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
+++ b/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
@@ -39,12 +39,12 @@ public class MultipleConditionTriggerCaseTest {
     public void trigger() throws InterruptedException, TimeoutException {
         CountDownLatch countDownLatch = new CountDownLatch(3);
         ConcurrentHashMap<String, Execution> ended = new ConcurrentHashMap<>();
-        Flow flow = flowRepository.findById(null, "io.kestra.tests", "trigger-multiplecondition-listener").orElseThrow();
+        Flow flow = flowRepository.findById(null, "io.kestra.tests.trigger", "trigger-multiplecondition-listener").orElseThrow();
 
         executionQueue.receive(either -> {
             Execution execution = either.getLeft();
             synchronized (ended) {
-                if (execution.getState().getCurrent() == State.Type.SUCCESS && !execution.getFlowId().equals("trigger-flow-listener-no-condition")) {
+                if (execution.getState().getCurrent() == State.Type.SUCCESS && !execution.getFlowId().equals("trigger-flow-listener-namespace-condition")) {
                     if (!ended.containsKey(execution.getId())) {
                         ended.put(execution.getId(), execution);
                         countDownLatch.countDown();
@@ -54,7 +54,7 @@ public class MultipleConditionTriggerCaseTest {
         });
 
         // first one
-        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "trigger-multiplecondition-flow-a", Duration.ofSeconds(60));
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests.trigger", "trigger-multiplecondition-flow-a", Duration.ofSeconds(60));
         assertThat(execution.getTaskRunList().size(), is(1));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
@@ -63,7 +63,7 @@ public class MultipleConditionTriggerCaseTest {
         assertThat(ended.size(), is(1));
 
         // second one
-        execution = runnerUtils.runOne(null, "io.kestra.tests", "trigger-multiplecondition-flow-b", Duration.ofSeconds(60));
+        execution = runnerUtils.runOne(null, "io.kestra.tests.trigger", "trigger-multiplecondition-flow-b", Duration.ofSeconds(60));
         assertThat(execution.getTaskRunList().size(), is(1));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
@@ -82,7 +82,7 @@ public class MultipleConditionTriggerCaseTest {
         assertThat(triggerExecution.getState().getCurrent(), is(State.Type.SUCCESS));
 
         assertThat(triggerExecution.getTrigger().getVariables().get("executionId"), is(execution.getId()));
-        assertThat(triggerExecution.getTrigger().getVariables().get("namespace"), is("io.kestra.tests"));
+        assertThat(triggerExecution.getTrigger().getVariables().get("namespace"), is("io.kestra.tests.trigger"));
         assertThat(triggerExecution.getTrigger().getVariables().get("flowId"), is("trigger-multiplecondition-flow-b"));
     }
 
@@ -93,7 +93,7 @@ public class MultipleConditionTriggerCaseTest {
         executionQueue.receive(either -> {
             synchronized (ended) {
                 Execution execution = either.getLeft();
-                if (execution.getState().getCurrent().isTerminated() && !execution.getFlowId().equals("trigger-flow-listener-no-condition")) {
+                if (execution.getState().getCurrent().isTerminated() && !execution.getFlowId().equals("trigger-flow-listener-namespace-condition")) {
                     if (!ended.containsKey(execution.getId())) {
                         ended.put(execution.getId(), execution);
                         countDownLatch.countDown();
@@ -103,7 +103,7 @@ public class MultipleConditionTriggerCaseTest {
         });
 
         // first one
-        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "trigger-multiplecondition-flow-c", Duration.ofSeconds(60));
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests.trigger", "trigger-multiplecondition-flow-c", Duration.ofSeconds(60));
         assertThat(execution.getTaskRunList().size(), is(1));
         assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
 
@@ -112,7 +112,7 @@ public class MultipleConditionTriggerCaseTest {
         assertThat(ended.size(), is(1));
 
         // second one
-        execution = runnerUtils.runOne(null, "io.kestra.tests", "trigger-multiplecondition-flow-d", Duration.ofSeconds(60));
+        execution = runnerUtils.runOne(null, "io.kestra.tests.trigger", "trigger-multiplecondition-flow-d", Duration.ofSeconds(60));
         assertThat(execution.getTaskRunList().size(), is(1));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
diff --git a/core/src/test/java/io/kestra/core/runners/RestartCaseTest.java b/core/src/test/java/io/kestra/core/runners/RestartCaseTest.java
index 81b6a288c..be8d64a1e 100644
--- a/core/src/test/java/io/kestra/core/runners/RestartCaseTest.java
+++ b/core/src/test/java/io/kestra/core/runners/RestartCaseTest.java
@@ -48,7 +48,7 @@ public class RestartCaseTest {
 
         // wait
         Execution finishedRestartedExecution = runnerUtils.awaitExecution(
-            execution -> !execution.getFlowId().equals("trigger-flow-listener-no-condition") && execution.getState().getCurrent() == State.Type.SUCCESS,
+            execution -> execution.getState().getCurrent() == State.Type.SUCCESS,
             throwRunnable(() -> {
                 Execution restartedExec = executionService.restart(firstExecution, null);
                 executionQueue.emit(restartedExec);
diff --git a/core/src/test/java/io/kestra/core/runners/RunContextLoggerTest.java b/core/src/test/java/io/kestra/core/runners/RunContextLoggerTest.java
new file mode 100644
index 000000000..b01aa5e38
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/runners/RunContextLoggerTest.java
@@ -0,0 +1,96 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.LogEntry;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.tasks.log.Log;
+import io.kestra.core.utils.IdUtils;
+import io.kestra.core.utils.TestsUtils;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.event.Level;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+
+@MicronautTest
+class RunContextLoggerTest {
+    @Inject
+    @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
+    QueueInterface<LogEntry> logQueue;
+
+    @Inject
+    RunContextFactory runContextFactory;
+
+
+    @Test
+    void logs() {
+        List<LogEntry> logs = new CopyOnWriteArrayList<>();
+        List<LogEntry> matchingLog;
+        logQueue.receive(either -> logs.add(either.getLeft()));
+
+        Flow flow = TestsUtils.mockFlow();
+        Execution execution = TestsUtils.mockExecution(flow, Map.of());
+        Log log = Log.builder().id(IdUtils.create()).type(Log.class.getName()).build();
+
+        RunContextLogger runContextLogger = new RunContextLogger(
+            logQueue,
+            LogEntry.of(execution),
+            Level.TRACE
+        );
+
+        Logger logger = runContextLogger.logger();
+        logger.trace("trace");
+        logger.debug("debug");
+        logger.info("info");
+        logger.warn("warn");
+        logger.error("error");
+
+        matchingLog = TestsUtils.awaitLogs(logs, 5);
+        assertThat(matchingLog.stream().filter(logEntry -> logEntry.getLevel().equals(Level.TRACE)).findFirst().orElse(null).getMessage(), is("trace"));
+        assertThat(matchingLog.stream().filter(logEntry -> logEntry.getLevel().equals(Level.DEBUG)).findFirst().orElse(null).getMessage(), is("debug"));
+        assertThat(matchingLog.stream().filter(logEntry -> logEntry.getLevel().equals(Level.INFO)).findFirst().orElse(null).getMessage(), is("info"));
+        assertThat(matchingLog.stream().filter(logEntry -> logEntry.getLevel().equals(Level.WARN)).findFirst().orElse(null).getMessage(), is("warn"));
+        assertThat(matchingLog.stream().filter(logEntry -> logEntry.getLevel().equals(Level.ERROR)).findFirst().orElse(null).getMessage(), is("error"));
+    }
+
+    @Test
+    void secrets() throws InterruptedException {
+        List<LogEntry> logs = new CopyOnWriteArrayList<>();
+        List<LogEntry> matchingLog;
+        logQueue.receive(either -> logs.add(either.getLeft()));
+
+        Flow flow = TestsUtils.mockFlow();
+        Execution execution = TestsUtils.mockExecution(flow, Map.of());
+        Log log = Log.builder().id(IdUtils.create()).type(Log.class.getName()).build();
+
+        RunContextLogger runContextLogger = new RunContextLogger(
+            logQueue,
+            LogEntry.of(execution),
+            Level.TRACE
+        );
+
+        runContextLogger.usedSecret("doe.com");
+        runContextLogger.usedSecret("myawesomepass");
+
+        Logger logger = runContextLogger.logger();
+        // exception are not handle and secret will not be replaced
+        logger.debug("test {} test", "john@doe.com", new Exception("exception from doe.com"));
+        logger.info("test myawesomepassmyawesomepass myawesomepass myawesomepassmyawesomepass");
+
+        matchingLog = TestsUtils.awaitLogs(logs, 3);
+        assertThat(matchingLog.stream().filter(logEntry -> logEntry.getLevel().equals(Level.DEBUG)).findFirst().orElse(null).getMessage(), is("test john@******* test"));
+        assertThat(matchingLog.stream().filter(logEntry -> logEntry.getLevel().equals(Level.TRACE)).findFirst().orElse(null).getMessage(), containsString("exception from doe.com"));
+        assertThat(matchingLog.stream().filter(logEntry -> logEntry.getLevel().equals(Level.INFO)).findFirst().orElse(null).getMessage(), is("test **masked****************** ************* **************************"));
+    }
+}
\ No newline at end of file
diff --git a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
index dde93a85e..c753da852 100644
--- a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
@@ -29,7 +29,7 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
-@MicronautTest
+@MicronautTest(transactional = false)
 abstract public class AbstractSchedulerTest {
     @Inject
     protected ApplicationContext applicationContext;
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
index fef5505b3..d32cf8a0b 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
@@ -10,7 +10,7 @@ import io.kestra.core.runners.FlowListeners;
 import io.kestra.core.runners.TestMethodScopedWorker;
 import io.kestra.core.runners.Worker;
 import jakarta.inject.Inject;
-import org.junitpioneer.jupiter.RetryingTest;
+import org.junit.jupiter.api.Test;
 
 import java.time.DayOfWeek;
 import java.time.ZonedDateTime;
@@ -22,7 +22,8 @@ import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 
 class SchedulerConditionTest extends AbstractSchedulerTest {
     @Inject
@@ -52,7 +53,7 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
         return createFlow(Collections.singletonList(schedule));
     }
 
-    @RetryingTest(10)
+    @Test
     void schedule() throws Exception {
         // mock flow listeners
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
@@ -60,7 +61,7 @@ class SchedulerConditionTest extends AbstractSchedulerTest {
 
         Flow flow = createScheduleFlow();
 
-        triggerState.save(Trigger.builder()
+        triggerState.create(Trigger.builder()
             .namespace(flow.getNamespace())
             .flowId(flow.getId())
             .flowRevision(flow.getRevision())
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
index 76bdfeea5..5b7449d2f 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
@@ -4,18 +4,18 @@ import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.types.Schedule;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.runners.FlowListeners;
 import io.kestra.core.runners.TestMethodScopedWorker;
 import io.kestra.core.runners.Worker;
+import io.kestra.core.utils.Await;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
-import org.junitpioneer.jupiter.RetryingTest;
-import org.junitpioneer.jupiter.RetryingTest;
+import org.junit.jupiter.api.Test;
 
-import java.lang.reflect.Executable;
 import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
 import java.util.*;
@@ -24,7 +24,8 @@ import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 
 public class SchedulerScheduleTest extends AbstractSchedulerTest {
     @Inject
@@ -39,7 +40,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
 
     private static Flow createScheduleFlow(String zone) {
         Schedule schedule = Schedule.builder()
-            .id("hourly")
+            .id("hourlyfromschedulederschedule")
             .type(Schedule.class.getName())
             .cron("0 * * * *")
             .timezone(zone)
@@ -69,15 +70,17 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
         );
     }
 
-    @RetryingTest(5)
+    @Test
     void schedule() throws Exception {
         // mock flow listeners
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
-        CountDownLatch queueCount = new CountDownLatch(5);
-        CountDownLatch invalidLogCount = new CountDownLatch(5);
+        CountDownLatch queueCount = new CountDownLatch(6);
+        CountDownLatch invalidLogCount = new CountDownLatch(1);
         Set<String> date = new HashSet<>();
         Set<String> executionId = new HashSet<>();
 
+        // Create a flow with a backfill of 5 hours
+        // then flow should be executed 6 times
         Flow invalid = createScheduleFlow("Asia/Delhi");
         Flow flow = createScheduleFlow("Europe/Paris");
 
@@ -123,4 +126,41 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
             assertThat(executionId.size(), is(3));
         }
     }
+
+    // Test to ensure behavior between 0.14 > 0.15
+    @Test
+    void retroSchedule() throws Exception {
+        // mock flow listeners
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+
+        Flow flow = createScheduleFlow("Europe/Paris");
+
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        Trigger trigger = Trigger
+            .builder()
+            .triggerId("hourlyfromschedulederschedule")
+            .flowId(flow.getId())
+            .namespace(flow.getNamespace())
+            .date(ZonedDateTime.now())
+            .build();
+
+        triggerState.create(trigger);
+
+        // scheduler
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
+            worker.run();
+            scheduler.run();
+
+            Await.until(() -> {
+                Optional<Trigger> optionalTrigger = this.triggerState.findLast(trigger);
+                return optionalTrigger.filter(value -> value.getNextExecutionDate() != null).isPresent();
+            });
+
+            assertThat(this.triggerState.findLast(trigger).get().getNextExecutionDate().isAfter(trigger.getDate()), is(true));
+        }
+    }
 }
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
index 22dd7b5cd..97cdeff7e 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
@@ -10,8 +10,7 @@ import io.kestra.core.runners.FlowListeners;
 import io.kestra.core.runners.TestMethodScopedWorker;
 import io.kestra.core.runners.Worker;
 import jakarta.inject.Inject;
-import org.junit.jupiter.api.RepeatedTest;
-import org.junitpioneer.jupiter.RetryingTest;
+import org.junit.jupiter.api.Test;
 
 import java.util.Collections;
 import java.util.List;
@@ -22,7 +21,8 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 
 public class SchedulerThreadTest extends AbstractSchedulerTest {
     @Inject
@@ -50,7 +50,7 @@ public class SchedulerThreadTest extends AbstractSchedulerTest {
         ));
     }
 
-    @RetryingTest(5)
+    @Test
     void thread() throws Exception {
         Flow flow = createThreadFlow();
         CountDownLatch queueCount = new CountDownLatch(2);
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerTriggerStateInterfaceTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerTriggerStateInterfaceTest.java
index 24c7d16a8..6085f0290 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerTriggerStateInterfaceTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerTriggerStateInterfaceTest.java
@@ -1,13 +1,13 @@
 package io.kestra.core.schedulers;
 
-import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
-import org.junit.jupiter.api.Test;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.utils.IdUtils;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
 
 import java.time.ZonedDateTime;
 import java.util.Optional;
-import jakarta.inject.Inject;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
@@ -34,14 +34,14 @@ public abstract class SchedulerTriggerStateInterfaceTest {
         Optional<Trigger> find = triggerState.findLast(builder.build());
         assertThat(find.isPresent(), is(false));
 
-        Trigger save = triggerState.save(builder.build());
+        Trigger save = triggerState.update(builder.build());
 
         find = triggerState.findLast(save);
 
         assertThat(find.isPresent(), is(true));
         assertThat(find.get().getExecutionId(), is(save.getExecutionId()));
 
-        save = triggerState.save(builder.executionId(IdUtils.create()).build());
+        save = triggerState.update(builder.executionId(IdUtils.create()).build());
 
         find = triggerState.findLast(save);
 
diff --git a/core/src/test/java/io/kestra/core/secret/SecretFunctionTest.java b/core/src/test/java/io/kestra/core/secret/SecretFunctionTest.java
index 398bf3d74..bf7fee5cf 100644
--- a/core/src/test/java/io/kestra/core/secret/SecretFunctionTest.java
+++ b/core/src/test/java/io/kestra/core/secret/SecretFunctionTest.java
@@ -2,14 +2,21 @@ package io.kestra.core.secret;
 
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.LogEntry;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.runners.AbstractMemoryRunnerTest;
 import io.kestra.core.runners.RunnerUtils;
+import io.kestra.core.utils.TestsUtils;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
+import jakarta.inject.Named;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
 
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeoutException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -18,6 +25,10 @@ import static org.hamcrest.Matchers.is;
 
 @MicronautTest
 public class SecretFunctionTest extends AbstractMemoryRunnerTest {
+    @Inject
+    @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
+    QueueInterface<LogEntry> logQueue;
+
     @Inject
     private RunnerUtils runnerUtils;
 
@@ -28,9 +39,15 @@ public class SecretFunctionTest extends AbstractMemoryRunnerTest {
     @EnabledIfEnvironmentVariable(named = "SECRET_MY_SECRET", matches = ".*")
     @EnabledIfEnvironmentVariable(named = "SECRET_NEW_LINE", matches = ".*")
     void getSecret() throws TimeoutException {
+        List<LogEntry> logs = new CopyOnWriteArrayList<>();
+        logQueue.receive(either -> logs.add(either.getLeft()));
+
         Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "secrets");
         assertThat(execution.getTaskRunList().get(0).getOutputs().get("value"), is("secretValue"));
-        assertThat(execution.getTaskRunList().get(1).getOutputs().get("value"), is("passwordveryveryveyrlongpasswordveryveryveyrlongpasswordveryveryveyrlongpasswordveryveryveyrlongpasswordveryveryveyrlong"));
+        assertThat(execution.getTaskRunList().get(2).getOutputs().get("value"), is("passwordveryveryveyrlongpasswordveryveryveyrlongpasswordveryveryveyrlongpasswordveryveryveyrlongpasswordveryveryveyrlong"));
+
+        LogEntry matchingLog = TestsUtils.awaitLog(logs, logEntry -> logEntry.getTaskId() != null && logEntry.getTaskId().equals("log-secret"));
+        assertThat(matchingLog.getMessage(), containsString("***"));
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/PauseTest.java b/core/src/test/java/io/kestra/core/tasks/flows/PauseTest.java
index 7e2845330..1fe202d3e 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/PauseTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/PauseTest.java
@@ -72,7 +72,7 @@ public class PauseTest extends AbstractMemoryRunnerTest {
             );
 
             execution = runnerUtils.awaitExecution(
-                e -> !e.getFlowId().equals("trigger-flow-listener-no-condition") && e.getState().getCurrent() == State.Type.SUCCESS,
+                e -> e.getState().getCurrent() == State.Type.SUCCESS,
                 () -> executionQueue.emit(restarted),
                 Duration.ofSeconds(5)
             );
@@ -87,7 +87,7 @@ public class PauseTest extends AbstractMemoryRunnerTest {
             assertThat(execution.getTaskRunList(), hasSize(1));
 
             execution = runnerUtils.awaitExecution(
-                e -> !e.getFlowId().equals("trigger-flow-listener-no-condition") && e.getState().getCurrent() == State.Type.SUCCESS,
+                e -> e.getState().getCurrent() == State.Type.SUCCESS,
                 () -> {},
                 Duration.ofSeconds(5)
             );
diff --git a/core/src/test/java/io/kestra/core/tasks/test/PollingTrigger.java b/core/src/test/java/io/kestra/core/tasks/test/PollingTrigger.java
new file mode 100644
index 000000000..77246e8e7
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/test/PollingTrigger.java
@@ -0,0 +1,42 @@
+package io.kestra.core.tasks.test;
+
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.conditions.ConditionContext;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.models.triggers.PollingTriggerInterface;
+import io.kestra.core.models.triggers.TriggerContext;
+import lombok.Builder;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import jakarta.validation.constraints.NotNull;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Optional;
+
+@SuperBuilder
+@NoArgsConstructor
+public class PollingTrigger extends AbstractTrigger implements PollingTriggerInterface {
+    @PluginProperty
+    @NotNull
+    @Builder.Default
+    private Long duration = 1000L;
+
+    @Override
+    public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) {
+        // Try catch to avoid flakky test
+        try {
+            Thread.sleep(duration);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+
+        return Optional.empty();
+    }
+
+    @Override
+    public Duration getInterval() {
+        return Duration.of(1, ChronoUnit.MINUTES);
+    }
+}
diff --git a/core/src/test/resources/flows/valids/secrets.yaml b/core/src/test/resources/flows/valids/secrets.yaml
index 0bbe91b09..c568f76b4 100644
--- a/core/src/test/resources/flows/valids/secrets.yaml
+++ b/core/src/test/resources/flows/valids/secrets.yaml
@@ -4,7 +4,10 @@ namespace: io.kestra.tests
 tasks:
   - id: get-secret
     type: io.kestra.core.tasks.debugs.Return
-    format: "{{json(secret('my_secret')).secretKey}}"
+    format: "{{ json(secret('my_secret')).secretKey }}"
+  - id: log-secret
+    type: io.kestra.core.tasks.log.Log
+    message: "{{ secret('my_secret') }}"
   - id: get-multiline-secret
     type: io.kestra.core.tasks.debugs.Return
-    format: "{{json(secret('new_line'))}}"
\ No newline at end of file
+    format: "{{ secret('new_line') }}"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/trigger-flow-listener-invalid.yaml b/core/src/test/resources/flows/valids/trigger-flow-listener-invalid.yaml
index 7b0581730..8473653a8 100644
--- a/core/src/test/resources/flows/valids/trigger-flow-listener-invalid.yaml
+++ b/core/src/test/resources/flows/valids/trigger-flow-listener-invalid.yaml
@@ -1,5 +1,5 @@
 id: trigger-flow-listener-invalid
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 
 inputs:
   - id: from-parent
@@ -17,5 +17,5 @@ triggers:
       from-parent: '{{ outputs.invalid.value }}'
     conditions:
       - type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
-        namespace: io.kestra.tests
+        namespace: io.kestra.tests.trigger
         flowId: trigger-flow
diff --git a/core/src/test/resources/flows/valids/trigger-flow-listener-namespace-condition.yaml b/core/src/test/resources/flows/valids/trigger-flow-listener-namespace-condition.yaml
new file mode 100644
index 000000000..d460b949d
--- /dev/null
+++ b/core/src/test/resources/flows/valids/trigger-flow-listener-namespace-condition.yaml
@@ -0,0 +1,18 @@
+id: trigger-flow-listener-namespace-condition
+namespace: io.kestra.tests.trigger
+
+inputs:
+  - id: from-parent
+    type: STRING
+
+tasks:
+  - id: only-listener
+    type: io.kestra.core.tasks.debugs.Return
+    format: "simple return"
+
+triggers:
+  - id: listen-flow
+    type: io.kestra.core.models.triggers.types.Flow
+    conditions:
+      - type: io.kestra.core.models.conditions.types.ExecutionNamespaceCondition
+        namespace: io.kestra.tests.trigger
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/trigger-flow-listener-no-condition.yaml b/core/src/test/resources/flows/valids/trigger-flow-listener-no-condition.yaml
deleted file mode 100644
index 8ae3d2490..000000000
--- a/core/src/test/resources/flows/valids/trigger-flow-listener-no-condition.yaml
+++ /dev/null
@@ -1,15 +0,0 @@
-id: trigger-flow-listener-no-condition
-namespace: io.kestra.tests
-
-inputs:
-  - id: from-parent
-    type: STRING
-
-tasks:
-  - id: only-listener
-    type: io.kestra.core.tasks.debugs.Return
-    format: "simple return"
-
-triggers:
-  - id: listen-flow
-    type: io.kestra.core.models.triggers.types.Flow
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/trigger-flow-listener-no-inputs.yaml b/core/src/test/resources/flows/valids/trigger-flow-listener-no-inputs.yaml
index 2cac21ea5..10c08208e 100644
--- a/core/src/test/resources/flows/valids/trigger-flow-listener-no-inputs.yaml
+++ b/core/src/test/resources/flows/valids/trigger-flow-listener-no-inputs.yaml
@@ -1,5 +1,5 @@
 id: trigger-flow-listener-no-inputs
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 revision: 1
 
 tasks:
@@ -15,7 +15,7 @@ triggers:
         in:
           - SUCCESS
       - type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
-        namespace: io.kestra.tests
+        namespace: io.kestra.tests.trigger
         flowId: trigger-flow
   - id: listen-flow-failed
     type: io.kestra.core.models.triggers.types.Flow
@@ -24,5 +24,5 @@ triggers:
         in:
           - FAILED
       - type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
-        namespace: io.kestra.tests
+        namespace: io.kestra.tests.trigger
         flowId: other-flow
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/trigger-flow-listener.yaml b/core/src/test/resources/flows/valids/trigger-flow-listener.yaml
index 6266f35ae..c22d3a183 100644
--- a/core/src/test/resources/flows/valids/trigger-flow-listener.yaml
+++ b/core/src/test/resources/flows/valids/trigger-flow-listener.yaml
@@ -1,5 +1,5 @@
 id: trigger-flow-listener
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 
 inputs:
   - id: from-parent
@@ -20,5 +20,5 @@ triggers:
         in:
           - SUCCESS
       - type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
-        namespace: io.kestra.tests
+        namespace: io.kestra.tests.trigger
         flowId: trigger-flow
diff --git a/core/src/test/resources/flows/valids/trigger-flow.yaml b/core/src/test/resources/flows/valids/trigger-flow.yaml
index b4266f3d6..f86749491 100644
--- a/core/src/test/resources/flows/valids/trigger-flow.yaml
+++ b/core/src/test/resources/flows/valids/trigger-flow.yaml
@@ -1,5 +1,5 @@
 id: trigger-flow
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 
 tasks:
   - id: only
diff --git a/core/src/test/resources/flows/valids/trigger-multiplecondition-failed.yaml b/core/src/test/resources/flows/valids/trigger-multiplecondition-failed.yaml
index 5b66f0fab..4e438227c 100644
--- a/core/src/test/resources/flows/valids/trigger-multiplecondition-failed.yaml
+++ b/core/src/test/resources/flows/valids/trigger-multiplecondition-failed.yaml
@@ -1,5 +1,5 @@
 id: trigger-multiplecondition-failed
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 
 tasks:
   - id: only-listener
@@ -20,9 +20,9 @@ triggers:
         conditions:
           flow-a:
             type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
-            namespace: io.kestra.tests
+            namespace: io.kestra.tests.trigger
             flowId: trigger-multiplecondition-flow-d
           flow-b:
             type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
-            namespace: io.kestra.tests
+            namespace: io.kestra.tests.trigger
             flowId: trigger-multiplecondition-flow-c
diff --git a/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-a.yaml b/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-a.yaml
index 20dcbfb80..04a28c62d 100644
--- a/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-a.yaml
+++ b/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-a.yaml
@@ -1,5 +1,5 @@
 id: trigger-multiplecondition-flow-a
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 
 tasks:
   - id: only
diff --git a/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-b.yaml b/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-b.yaml
index 38c9ad2b5..d4c1c6621 100644
--- a/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-b.yaml
+++ b/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-b.yaml
@@ -1,5 +1,5 @@
 id: trigger-multiplecondition-flow-b
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 
 tasks:
   - id: only
diff --git a/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-c.yaml b/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-c.yaml
index 2994d1e82..c3c0522aa 100644
--- a/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-c.yaml
+++ b/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-c.yaml
@@ -1,5 +1,5 @@
 id: trigger-multiplecondition-flow-c
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 
 tasks:
   - id: fail
diff --git a/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-d.yaml b/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-d.yaml
index 735809282..1a6918b78 100644
--- a/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-d.yaml
+++ b/core/src/test/resources/flows/valids/trigger-multiplecondition-flow-d.yaml
@@ -1,5 +1,5 @@
 id: trigger-multiplecondition-flow-d
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 
 tasks:
   - id: only
diff --git a/core/src/test/resources/flows/valids/trigger-multiplecondition-listener.yaml b/core/src/test/resources/flows/valids/trigger-multiplecondition-listener.yaml
index eaa5f15b7..cc37c4417 100644
--- a/core/src/test/resources/flows/valids/trigger-multiplecondition-listener.yaml
+++ b/core/src/test/resources/flows/valids/trigger-multiplecondition-listener.yaml
@@ -1,5 +1,5 @@
 id: trigger-multiplecondition-listener
-namespace: io.kestra.tests
+namespace: io.kestra.tests.trigger
 #
 #inputs:
 #  - id: from-parent
@@ -27,9 +27,9 @@ triggers:
               - SUCCESS
           flow-a:
             type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
-            namespace: io.kestra.tests
+            namespace: io.kestra.tests.trigger
             flowId: trigger-multiplecondition-flow-a
           flow-b:
             type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
-            namespace: io.kestra.tests
+            namespace: io.kestra.tests.trigger
             flowId: trigger-multiplecondition-flow-b
diff --git a/jdbc-postgres/src/test/resources/application-test.yml b/jdbc-postgres/src/test/resources/application-test.yml
index 5818f2e3c..2eb503f88 100644
--- a/jdbc-postgres/src/test/resources/application-test.yml
+++ b/jdbc-postgres/src/test/resources/application-test.yml
@@ -84,3 +84,7 @@ kestra:
       min-poll-interval: 10ms
       max-poll-interval: 100ms
       poll-switch-interval: 5s
+
+  tasks:
+    subflow:
+      allow-parameter-outputs: true
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
index e5f979b6e..b1bdedc3e 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
@@ -427,7 +427,7 @@ class FlowControllerTest extends JdbcH2ControllerTest {
         List<String> namespaces = client.toBlocking().retrieve(
             HttpRequest.GET("/api/v1/flows/distinct-namespaces"), Argument.listOf(String.class));
 
-        assertThat(namespaces.size(), is(4));
+        assertThat(namespaces.size(), is(5));
     }
 
     @Test
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/TriggerControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/TriggerControllerTest.java
index 9e010fc46..261a2617b 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/TriggerControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/TriggerControllerTest.java
@@ -1,7 +1,14 @@
 package io.kestra.webserver.controllers;
 
+import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.triggers.Trigger;
+import io.kestra.core.models.triggers.types.Schedule;
+import io.kestra.core.runners.StandAloneRunner;
+import io.kestra.core.tasks.debugs.Return;
+import io.kestra.core.tasks.test.PollingTrigger;
+import io.kestra.core.utils.Await;
 import io.kestra.core.utils.IdUtils;
+import io.kestra.jdbc.JdbcTestUtils;
 import io.kestra.jdbc.repository.AbstractJdbcFlowRepository;
 import io.kestra.jdbc.repository.AbstractJdbcTriggerRepository;
 import io.kestra.webserver.controllers.h2.JdbcH2ControllerTest;
@@ -17,6 +24,13 @@ import org.hamcrest.Matchers;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -32,9 +46,22 @@ class TriggerControllerTest extends JdbcH2ControllerTest {
     @Inject
     AbstractJdbcTriggerRepository jdbcTriggerRepository;
 
+    @Inject
+    private JdbcTestUtils jdbcTestUtils;
+
+    @Inject
+    private StandAloneRunner runner;
+
+
     @BeforeEach
-    protected void init() {
-        super.setup();
+    protected void init() throws IOException, URISyntaxException {
+        jdbcTestUtils.drop();
+        jdbcTestUtils.migrate();
+
+        if (!runner.isRunning()) {
+            runner.setSchedulerEnabled(true);
+            runner.run();
+        }
     }
 
     @SuppressWarnings("unchecked")
@@ -53,8 +80,7 @@ class TriggerControllerTest extends JdbcH2ControllerTest {
         jdbcTriggerRepository.save(trigger.toBuilder().triggerId("schedule-5-min").build());
 
         PagedResults<Trigger> triggers = client.toBlocking().retrieve(HttpRequest.GET("/api/v1/triggers/search?q=schedule-trigger&namespace=io.kestra.tests&sort=triggerId:asc"), Argument.of(PagedResults.class, Trigger.class));
-        assertThat(triggers.getTotal(), is(2L));
-
+        assertThat(triggers.getTotal(), greaterThan(2L));
 
         assertThat(triggers.getResults(), Matchers.hasItems(
                 allOf(
@@ -116,4 +142,41 @@ class TriggerControllerTest extends JdbcH2ControllerTest {
 
         assertThat(e.getStatus(), is(HttpStatus.NOT_FOUND));
     }
+
+    @Test
+    void nextExecutionDate() throws InterruptedException, TimeoutException {
+        Flow flow = generateFlow();
+        jdbcFlowRepository.create(flow, flow.generateSource(), flow);
+        Await.until(
+            () -> client.toBlocking().retrieve(HttpRequest.GET("/api/v1/triggers/search?q=trigger-nextexec"), Argument.of(PagedResults.class, Trigger.class)).getTotal() >= 2,
+            Duration.ofMillis(100),
+            Duration.ofMinutes(2)
+        );
+        PagedResults<Trigger> triggers = client.toBlocking().retrieve(HttpRequest.GET("/api/v1/triggers/search?q=trigger-nextexec"), Argument.of(PagedResults.class, Trigger.class));
+        assertThat(triggers.getResults().get(0).getNextExecutionDate(), notNullValue());
+        assertThat(triggers.getResults().get(1).getNextExecutionDate(), notNullValue());
+    }
+
+    private Flow generateFlow() {
+        return Flow.builder()
+            .id("flow-with-triggers")
+            .namespace("io.kestra.tests.scheduler")
+            .tasks(Collections.singletonList(Return.builder()
+                .id("task")
+                .type(Return.class.getName())
+                .format("return data")
+                .build()))
+            .triggers(List.of(
+                Schedule.builder()
+                    .id("trigger-nextexec-schedule")
+                    .type(Schedule.class.getName())
+                    .cron("*/1 * * * *")
+                    .build(),
+                PollingTrigger.builder()
+                    .id("trigger-nextexec-polling")
+                    .type(PollingTrigger.class.getName())
+                    .build()
+            ))
+            .build();
+    }
 }