feat(trigger): add support for concurrent trigger execution (#311)

Fixes: #311

Author: fhussonnois

core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java

@@ -82,6 +82,12 @@ abstract public class AbstractTrigger implements TriggerInterface {
     @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
     private boolean failOnTriggerError = false;
 
+    @PluginProperty(group = PluginProperty.CORE_GROUP)
+    @Schema(
+        title = "Specifies whether a trigger is allowed to start a new execution even if a previous run is still in progress."
+    )
+    private boolean allowConcurrent = false;
+
     /**
      * For backward compatibility: we rename minLogLevel to logLevel.
      * @deprecated use {@link #logLevel} instead

core/src/main/java/io/kestra/core/repositories/TriggerRepositoryInterface.java

@@ -16,8 +16,8 @@
 public interface TriggerRepositoryInterface extends QueryBuilderInterface<Triggers.Fields> {
     Optional<Trigger> findLast(TriggerContext trigger);
 
-    Optional<Trigger> findByExecution(Execution execution);
-
+    Optional<Trigger> findByUid(String uid);
+    
     List<Trigger> findAll(String tenantId);
 
     List<Trigger> findAllForAllTenants();

jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java

@@ -72,12 +72,12 @@ public AbstractJdbcTriggerRepository(io.kestra.jdbc.AbstractJdbcRepository<Trigg
 
     @Override
     public Optional<Trigger> findLast(TriggerContext trigger) {
-        return findOne(DSL.trueCondition(), field("key").eq(trigger.uid()));
+        return findByUid(trigger.uid());
     }
 
     @Override
-    public Optional<Trigger> findByExecution(Execution execution) {
-        return findOne(execution.getTenantId(), field("execution_id").eq(execution.getId()));
+    public Optional<Trigger> findByUid(String uid) {
+        return findOne(DSL.trueCondition(), field("key").eq(uid));
     }
 
     public List<Trigger> findByNextExecutionDateReadyForAllTenants(ZonedDateTime now, ScheduleContextInterface scheduleContextInterface) {

jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java

@@ -13,6 +13,7 @@
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.models.topologies.FlowTopology;
+import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.multipleflows.MultipleCondition;
 import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
 import io.kestra.core.queues.QueueException;
@@ -1143,9 +1144,7 @@ private void toExecution(Executor executor, boolean ignoreFailure) {
                             execution.getTrigger().getId()
                         );
                     } else {
-                        triggerRepository
-                            .findByExecution(execution)
-                            .ifPresent(trigger -> this.triggerState.update(executionService.resetExecution(flow, execution, trigger)));
+                        triggerRepository.findByUid(Trigger.uid(execution)).ifPresent(trigger -> this.triggerState.update(executionService.resetExecution(flow, execution, trigger)));
                     }
                 }
 
@@ -1239,11 +1238,7 @@ private void toExecution(Executor executor, boolean ignoreFailure) {
                 // purge the trigger: reset scheduler trigger at end
                 if (execution.getTrigger() != null) {
                     FlowWithSource flow = executor.getFlow();
-                    triggerRepository
-                        .findByExecution(execution)
-                        .ifPresent(trigger -> {
-                            this.triggerState.update(executionService.resetExecution(flow, execution, trigger));
-                        });
+                    triggerRepository.findByUid(Trigger.uid(execution)).ifPresent(trigger -> this.triggerState.update(executionService.resetExecution(flow, execution, trigger)));
                 }
 
                 // Purge the workerTaskResultQueue and the workerJobQueue

scheduler/src/main/java/io/kestra/scheduler/AbstractScheduler.java

@@ -288,7 +288,7 @@ public void run() {
                         disableInvalidTrigger(workerTriggerResult.getTriggerContext(), e);
                         return;
                     }
-                    this.handleEvaluateWorkerTriggerResult(triggerExecution, nextExecutionDate);
+                    this.handleEvaluateWorkerTriggerResult(triggerExecution, nextExecutionDate, workerTriggerResult.getTrigger());
                 } else {
                     ZonedDateTime nextExecutionDate;
                     try {
@@ -786,7 +786,7 @@ private List<FlowWithSource> getFlowsWithDefaults() {
     }
 
     private void handleEvaluateWorkerTriggerResult(SchedulerExecutionWithTrigger result, ZonedDateTime
-        nextExecutionDate) {
+        nextExecutionDate, AbstractTrigger abstractTrigger) {
         Optional.ofNullable(result)
             .ifPresent(executionWithTrigger -> {
                     log(executionWithTrigger);
@@ -797,6 +797,12 @@ private void handleEvaluateWorkerTriggerResult(SchedulerExecutionWithTrigger res
                         nextExecutionDate
                     );
 
+                    // if the trigger is allowed to run concurrently we do not attached the executio-id to the trigger state
+                    // i.e., the trigger will not be locked
+                    if (abstractTrigger.isAllowConcurrent()) {
+                        trigger = trigger.toBuilder().executionId(null).build();
+                    }
+                
                     // Worker triggers result is evaluated in another thread with the workerTriggerResultQueue.
                     // We can then update the trigger directly.
                     this.saveLastTriggerAndEmitExecution(executionWithTrigger.getExecution(), trigger, triggerToSave -> this.triggerState.update(triggerToSave));
@@ -818,6 +824,12 @@ private void handleEvaluateSchedulingTriggerResult(Schedulable schedule, Schedul
         if (result.getExecution().getState().getCurrent() == State.Type.FAILED) {
             trigger = trigger.resetExecution(State.Type.FAILED);
         }
+        
+        // if the trigger is allowed to run concurrently we do not attached the executio-id to the trigger state
+        // i.e., the trigger will not be locked
+        if (((AbstractTrigger)schedule).isAllowConcurrent()) {
+            trigger = trigger.toBuilder().executionId(null).build();
+        }
 
         // Schedule triggers are being executed directly from the handle method within the context where triggers are locked.
         // So we must save them by passing the scheduleContext.