Repository: kestra-io/kestra
Issue: Add a generic `stopAfter` property allowing to disable any trigger after a SUCCESS (e.g. HTTP trigger condition) or FAILED (e.g. dependsOnPast behavior) execution
Labels: enhancement

### Feature description

Example syntax:

```yaml
id: hello-world
namespace: dev

tasks:
 - id: hello
   type: io.kestra.core.tasks.log.Log
   message: pausing a trigger after a success

triggers:
 - id: stopAfter
   type: io.kestra.core.models.triggers.types.Schedule
   cron: "0 9 * * *"
   stopAfter: # list of enum-type Execution states - can be set to any Execution states incl. multiple states as it's a list property, similarly to how ExecutionState condition on a Flow trigger accepts multiple possible states
     - SUCCESS # HTTP trigger as exist condition
     - FAILED # dependsOnPast for past Airflow users who want to disable the  executions until a broken pipeline gets fixed
```

**Important notes:**
- Ideally, disabling a trigger should generate a dedicated Audit log in EE e.g. "Trigger state action" with info which process/human disabled the trigger. For now, this doesn't seem feasible; see the comment below.
- This property is meant to be used primarily for a Schedule trigger + triggers with conditions incl. HTTP, JDBC, File Detection triggers; we will add it as a generic property for all triggers to stay flexible but just a note that for message queues, wehook or Flow triggers, this property is less relevant.
- We don't handle any automatic reenabling logic. After a trigger has been disabled, this generates a specific audit log and the user can then take some action based on it e.g. as part of https://github.com/kestra-io/kestra-ee/issues/809 

Issue comments:
@anna-geller there is no audit logs generated when a trigger is automatically disabled due to a stopAfter condition.
Audit logs are for actions made by a user, stopAfter will disable the trigger out of any user context so we cannot create an audit log with our current implementation.

Pull request review comments:

Patch that resolved the issue:
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 dd156d0d03..d20c689929 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
@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.Condition;
+import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.micronaut.core.annotation.Introspected;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -56,6 +57,12 @@ abstract public class AbstractTrigger {
 
     private Level logLevel;
 
+    @PluginProperty
+    @Schema(
+        title = "List of execution states after which a trigger should be stopped (a.k.a. disabled)."
+    )
+    private List<State.Type> stopAfter;
+
     /**
      * For backward compatibility: we rename minLogLevel to logLevel.
      * @deprecated use {@link #logLevel} instead
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 f23640730c..4a121f367e 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
@@ -73,6 +73,7 @@ public static Trigger of(Flow flow, AbstractTrigger abstractTrigger) {
             .flowId(flow.getId())
             .flowRevision(flow.getRevision())
             .triggerId(abstractTrigger.getId())
+            .stopAfter(abstractTrigger.getStopAfter())
             .build();
     }
 
@@ -88,6 +89,7 @@ public static Trigger of(TriggerContext triggerContext) {
             .triggerId(triggerContext.getTriggerId())
             .date(triggerContext.getDate())
             .backfill(triggerContext.getBackfill())
+            .stopAfter(triggerContext.getStopAfter())
             .disabled(triggerContext.getDisabled())
             .build();
     }
@@ -105,6 +107,7 @@ public static Trigger of(TriggerContext triggerContext, ZonedDateTime nextExecut
             .date(triggerContext.getDate())
             .nextExecutionDate(nextExecutionDate)
             .backfill(triggerContext.getBackfill())
+            .stopAfter(triggerContext.getStopAfter())
             .disabled(triggerContext.getDisabled())
             .build();
     }
@@ -126,6 +129,7 @@ public static Trigger of(TriggerContext triggerContext, Execution execution) {
             .updatedDate(Instant.now())
             .nextExecutionDate(triggerContext.getNextExecutionDate())
             .backfill(triggerContext.getBackfill())
+            .stopAfter(triggerContext.getStopAfter())
             .disabled(triggerContext.getDisabled())
             .build();
     }
@@ -148,6 +152,7 @@ public static Trigger of(TriggerContext triggerContext, Execution execution, Zon
             .updatedDate(Instant.now())
             .nextExecutionDate(nextExecutionDate)
             .backfill(triggerContext.getBackfill())
+            .stopAfter(triggerContext.getStopAfter())
             .disabled(triggerContext.getDisabled())
             .build();
     }
@@ -170,6 +175,7 @@ public static Trigger of(Execution execution, Trigger trigger) {
             .executionCurrentState(execution.getState().getCurrent())
             .updatedDate(Instant.now())
             .backfill(trigger.getBackfill())
+            .stopAfter(trigger.getStopAfter())
             .disabled(trigger.getDisabled())
             .build();
     }
@@ -191,6 +197,7 @@ public static Trigger of(Trigger trigger, ZonedDateTime evaluateRunningDate) {
             .evaluateRunningDate(evaluateRunningDate)
             .updatedDate(Instant.now())
             .backfill(trigger.getBackfill())
+            .stopAfter(trigger.getStopAfter())
             .disabled(trigger.getDisabled())
             .build();
     }
@@ -204,6 +211,7 @@ public static Trigger of(Flow flow, AbstractTrigger abstractTrigger, ConditionCo
             .triggerId(abstractTrigger.getId())
             .date(ZonedDateTime.now().truncatedTo(ChronoUnit.SECONDS))
             .nextExecutionDate(((PollingTriggerInterface) abstractTrigger).nextEvaluationDate(conditionContext, lastTrigger))
+            .stopAfter(abstractTrigger.getStopAfter())
             .disabled(lastTrigger.map(TriggerContext::getDisabled).orElse(Boolean.FALSE))
             .build();
     }
@@ -234,7 +242,24 @@ public static Trigger update(Trigger currentTrigger, Trigger newTrigger) {
             .build();
     }
 
-    public Trigger resetExecution() {
+    public Trigger resetExecution(State.Type executionEndState) {
+        // switch disabled automatically if the executionEndState is one of the stopAfter states
+        Boolean disabled = this.getStopAfter() != null ? this.getStopAfter().contains(executionEndState) : this.getDisabled();
+
+        return Trigger.builder()
+            .tenantId(this.getTenantId())
+            .namespace(this.getNamespace())
+            .flowId(this.getFlowId())
+            .flowRevision(this.getFlowRevision())
+            .triggerId(this.getTriggerId())
+            .date(this.getDate())
+            .nextExecutionDate(this.getNextExecutionDate())
+            .stopAfter(this.getStopAfter())
+            .disabled(disabled)
+            .build();
+    }
+
+    public Trigger unlock() {
         return Trigger.builder()
             .tenantId(this.getTenantId())
             .namespace(this.getNamespace())
@@ -244,8 +269,11 @@ public Trigger resetExecution() {
             .date(this.getDate())
             .nextExecutionDate(this.getNextExecutionDate())
             .backfill(this.getBackfill())
+            .stopAfter(this.getStopAfter())
+            .disabled(this.getDisabled())
             .build();
     }
+
     // if the next date is after the backfill end, we remove the backfill
     // if not, we update the backfill with the next Date
     // which will be the base date to calculate the next one
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 07ad964b90..066186c2e3 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
@@ -1,5 +1,6 @@
 package io.kestra.core.models.triggers;
 
+import io.kestra.core.models.flows.State;
 import io.kestra.core.utils.IdUtils;
 import io.micronaut.core.annotation.Introspected;
 import io.micronaut.core.annotation.Nullable;
@@ -12,6 +13,7 @@
 import lombok.experimental.SuperBuilder;
 
 import java.time.ZonedDateTime;
+import java.util.List;
 
 @SuperBuilder(toBuilder = true)
 @ToString
@@ -43,6 +45,9 @@ public class TriggerContext {
     @Nullable
     private Backfill backfill;
 
+    @Nullable
+    private List<State.Type> stopAfter;
+
     @Builder.Default
     private Boolean disabled = Boolean.FALSE;
 
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 f12746d80e..2be39f0c6b 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -199,6 +199,7 @@ private void initializedTriggers(List<Flow> flows) {
                             .triggerId(abstractTrigger.getId())
                             .date(now())
                             .nextExecutionDate(nextExecutionDate)
+                            .stopAfter(abstractTrigger.getStopAfter())
                             .build();
                         this.triggerState.create(newTrigger);
                     } catch (Exception e) {
@@ -274,8 +275,11 @@ private void handle() {
             }
 
             triggers.forEach(trigger -> schedulableNextDate.remove(trigger.uid()));
+            List<Trigger> triggerContextsToEvaluate = triggers.stream()
+                .filter(trigger -> Boolean.FALSE.equals(trigger.getDisabled()))
+                .toList();
 
-            List<FlowWithTriggers> schedulable = this.computeSchedulable(flowListeners.flows(), triggers, scheduleContext);
+            List<FlowWithTriggers> schedulable = this.computeSchedulable(flowListeners.flows(), triggerContextsToEvaluate, scheduleContext);
 
             metricRegistry
                 .counter(MetricRegistry.SCHEDULER_LOOP_COUNT)
@@ -299,7 +303,7 @@ private void handle() {
                     .abstractTrigger(flowWithTriggers.getAbstractTrigger())
                     .pollingTrigger((PollingTriggerInterface) flowWithTriggers.getAbstractTrigger())
                     .conditionContext(flowWithTriggers.getConditionContext())
-                    .triggerContext(flowWithTriggers.TriggerContext.toBuilder().date(now()).build())
+                    .triggerContext(flowWithTriggers.TriggerContext.toBuilder().date(now()).stopAfter(flowWithTriggers.getAbstractTrigger().getStopAfter()).build())
                     .build())
                 .filter(f -> f.getTriggerContext().getEvaluateRunningDate() == null && !f.getTriggerContext().getDisabled())
                 .filter(this::isExecutionNotRunning)
@@ -393,6 +397,7 @@ private void handleEvaluatePollingTriggerResult(SchedulerExecutionWithTrigger re
                         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);
@@ -670,6 +675,7 @@ public static FlowWithPollingTriggerNextDate of(FlowWithPollingTrigger f) {
                     .date(f.getTriggerContext().getNextExecutionDate())
                     .nextExecutionDate(f.getTriggerContext().getNextExecutionDate())
                     .backfill(f.getTriggerContext().getBackfill())
+                    .stopAfter(f.getTriggerContext().getStopAfter())
                     .build()
                 )
                 .next(f.getTriggerContext().getNextExecutionDate())
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 11257313e6..fe738c97b4 100644
--- a/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/DefaultScheduler.java
@@ -63,7 +63,7 @@ public void run() {
                 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.update(trigger.resetExecution());
+                    triggerState.update(trigger.resetExecution(execution.getState().getCurrent()));
                     watchingTrigger.remove(execution.getId());
                 } else {
                     triggerState.update(Trigger.of(execution, trigger));
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 94bf153801..ed5d4e6b90 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcTriggerRepository.java
@@ -181,30 +181,6 @@ public Trigger updateExecution(Trigger trigger) {
             });
     }
 
-    // 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
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 4426552fd6..10ca77e1fe 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcScheduler.java
@@ -75,7 +75,7 @@ public void run() {
                         triggerRepository
                             .findByExecution(execution)
                             .ifPresent(trigger -> {
-                                ((JdbcSchedulerTriggerState) this.triggerState).resetExecution(trigger.resetExecution());
+                                this.triggerState.update(trigger.resetExecution(execution.getState().getCurrent()));
                             });
                     } else {
                         // update execution state on each state change so the scheduler knows the execution is running
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 339814d24d..8f47587c6d 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
@@ -71,11 +71,6 @@ 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);
     }
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/TriggerController.java b/webserver/src/main/java/io/kestra/webserver/controllers/TriggerController.java
index 6cab96b95c..406a58df3e 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/TriggerController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/TriggerController.java
@@ -75,7 +75,7 @@ public HttpResponse<Trigger> unlock(
             throw new IllegalStateException("Trigger is not locked");
         }
 
-        trigger = trigger.resetExecution();
+        trigger = trigger.unlock();
         triggerQueue.emit(trigger);
 
         return HttpResponse.ok(trigger);


Test patch:
diff --git a/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java b/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
index 4995d13efd..fed1a8ca2d 100644
--- a/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
+++ b/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
@@ -123,9 +123,11 @@ void trigger() throws URISyntaxException {
 
             Map<String, Object> jsonSchema = jsonSchemaGenerator.generate(AbstractTrigger.class, AbstractTrigger.class);
 
+            System.out.println(jsonSchema.get("properties"));
             assertThat((Map<String, Object>) jsonSchema.get("properties"), allOf(
-                Matchers.aMapWithSize(1),
-                hasKey("conditions")
+                Matchers.aMapWithSize(2),
+                hasKey("conditions"),
+                hasKey("stopAfter")
             ));
         });
     }
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 c753da8522..fdf84b1b2b 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.List;
 import java.util.Optional;
 
-@MicronautTest(transactional = false)
+@MicronautTest(transactional = false, rebuildContext = true) // rebuild context to lower possible flaky tests
 abstract public class AbstractSchedulerTest {
     @Inject
     protected ApplicationContext applicationContext;
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
index 6db7791848..17e518d409 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerPollingTriggerTest.java
@@ -1,30 +1,34 @@
 package io.kestra.core.schedulers;
 
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.repositories.LocalFlowRepositoryLoader;
+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.runners.FlowListeners;
+import io.kestra.core.runners.TestMethodScopedWorker;
 import io.kestra.core.runners.Worker;
+import io.kestra.core.tasks.executions.Fail;
+import io.kestra.core.tasks.test.PollingTrigger;
+import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
-import java.util.Objects;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 
-@Disabled("Causes all sort of issues on CI")
-@MicronautTest(transactional = false)
-public class SchedulerPollingTriggerTest {
+@MicronautTest(transactional = false, rebuildContext = true) // without 'rebuildContext = true' the second test fail
+public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
     @Inject
     private ApplicationContext applicationContext;
 
@@ -34,39 +38,78 @@ public class SchedulerPollingTriggerTest {
     @Inject
     private FlowListeners flowListenersService;
 
-    @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    private QueueInterface<Execution> executionQueue;
-
-    @Inject
-    private LocalFlowRepositoryLoader repositoryLoader;
-
 
     @Test
     void pollingTrigger() throws Exception {
+        // mock flow listener
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        PollingTrigger pollingTrigger = createPollingTrigger(null);
+        Flow flow = createPollingTriggerFlow(pollingTrigger);
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
         CountDownLatch queueCount = new CountDownLatch(1);
 
         try (
-            AbstractScheduler scheduler = new DefaultScheduler(
-                this.applicationContext,
-                this.flowListenersService,
-                this.triggerState
-            );
-            Worker worker = new Worker(applicationContext, 8, null)
+            AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+            Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)
         ) {
             AtomicReference<Execution> last = new AtomicReference<>();
 
             Runnable executionQueueStop = executionQueue.receive(execution -> {
-                last.set(execution.getLeft());
-
-                queueCount.countDown();
-                assertThat(execution.getLeft().getFlowId(), is("polling-trigger"));
+                if (execution.getLeft().getFlowId().equals(flow.getId())) {
+                    last.set(execution.getLeft());
+                    queueCount.countDown();
+                }
             });
 
             worker.run();
             scheduler.run();
 
-            repositoryLoader.load(Objects.requireNonNull(SchedulerPollingTriggerTest.class.getClassLoader().getResource("flows/trigger")));
+            queueCount.await(10, TimeUnit.SECONDS);
+            // close the execution queue consumer
+            executionQueueStop.run();
+
+            assertThat(queueCount.getCount(), is(0L));
+            assertThat(last.get(), notNullValue());
+        }
+    }
+
+    @Test
+    void pollingTriggerStopAfter() throws Exception {
+        // mock flow listener
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        PollingTrigger pollingTrigger = createPollingTrigger(List.of(State.Type.FAILED));
+        Flow flow = createPollingTriggerFlow(pollingTrigger)
+            .toBuilder()
+            .tasks(List.of(Fail.builder().id("fail").type(Fail.class.getName()).build()))
+            .build();
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        CountDownLatch queueCount = new CountDownLatch(2);
+
+        try (
+            AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+            Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)
+        ) {
+            AtomicReference<Execution> last = new AtomicReference<>();
+
+            Runnable executionQueueStop = executionQueue.receive(execution -> {
+                if (execution.getLeft().getFlowId().equals(flow.getId())) {
+                    last.set(execution.getLeft());
+                    queueCount.countDown();
+
+                    if (execution.getLeft().getState().getCurrent() == State.Type.CREATED) {
+                        executionQueue.emit(execution.getLeft().withState(State.Type.FAILED));
+                    }
+                }
+            });
+
+            worker.run();
+            scheduler.run();
 
             queueCount.await(10, TimeUnit.SECONDS);
             // close the execution queue consumer
@@ -74,6 +117,32 @@ void pollingTrigger() throws Exception {
 
             assertThat(queueCount.getCount(), is(0L));
             assertThat(last.get(), notNullValue());
+
+            // Assert that the trigger is now disabled.
+            // It needs to await on assertion as it will be disabled AFTER we receive a success execution.
+            Trigger trigger = Trigger.of(flow, pollingTrigger);
+            Await.until(() -> this.triggerState.findLast(trigger).map(t -> t.getDisabled()).orElse(false), Duration.ofMillis(100), Duration.ofSeconds(10));
         }
     }
+
+    private Flow createPollingTriggerFlow(PollingTrigger pollingTrigger) {
+        return createFlow(Collections.singletonList(pollingTrigger));
+    }
+
+    private PollingTrigger createPollingTrigger(List<State.Type> stopAfter) {
+        return PollingTrigger.builder()
+            .id("polling-trigger")
+            .type(PollingTrigger.class.getName())
+            .duration(500L)
+            .stopAfter(stopAfter)
+            .build();
+    }
+
+    private AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy) {
+        return new DefaultScheduler(
+            applicationContext,
+            flowListenersServiceSpy,
+            triggerState
+        );
+    }
 }
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 3f8eb81e3a..7555f02590 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
@@ -40,21 +40,24 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
     @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
     protected QueueInterface<LogEntry> logQueue;
 
-    private static Flow createScheduleFlow(String zone, String triggerId, boolean invalid) {
-        Schedule schedule = Schedule.builder()
+    private Schedule.ScheduleBuilder<?, ?> createScheduleTrigger(String zone, String cron, String triggerId, boolean invalid) {
+        return Schedule.builder()
             .id(triggerId + (invalid ? "-invalid" : ""))
             .type(Schedule.class.getName())
-            .cron("0 * * * *")
+            .cron(cron)
             .timezone(zone)
             .inputs(Map.of(
                 "testInputs", "test-inputs"
-            ))
-            .build();
+            ));
+    }
+
+    private Flow createScheduleFlow(String zone, String triggerId, boolean invalid) {
+        Schedule schedule = createScheduleTrigger(zone, "0 * * * *", triggerId, invalid).build();
 
         return createFlow(Collections.singletonList(schedule));
     }
 
-    private static ZonedDateTime date(int minus) {
+    private ZonedDateTime date(int minus) {
         return ZonedDateTime.now()
             .minusHours(minus)
             .truncatedTo(ChronoUnit.HOURS);
@@ -285,4 +288,58 @@ void disabled() throws Exception {
             assertThat(lastTrigger.getNextExecutionDate().truncatedTo(ChronoUnit.HOURS).isEqual(now), is(true));
         }
     }
+
+    @Test
+    void stopAfterSchedule() throws Exception {
+        // mock flow listeners
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+        Schedule schedule = createScheduleTrigger("Europe/Paris", "* * * * *", "stopAfter", false).build();
+        Flow flow = createFlow(Collections.singletonList(schedule));
+        doReturn(List.of(flow))
+            .when(flowListenersServiceSpy)
+            .flows();
+
+        // to avoid waiting too much before a trigger execution, we add a last trigger with a date now - 1m.
+        Trigger lastTrigger = Trigger
+            .builder()
+            .triggerId("stopAfter")
+            .flowId(flow.getId())
+            .namespace(flow.getNamespace())
+            .date(ZonedDateTime.now().minusMinutes(1L))
+            .build();
+        triggerState.create(lastTrigger);
+
+        CountDownLatch queueCount = new CountDownLatch(2);
+
+        // scheduler
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
+            // wait for execution
+            Runnable assertionStop = executionQueue.receive(either -> {
+                Execution execution = either.getLeft();
+                assertThat(execution.getInputs().get("testInputs"), is("test-inputs"));
+                assertThat(execution.getInputs().get("def"), is("awesome"));
+                assertThat(execution.getFlowId(), is(flow.getId()));
+
+                queueCount.countDown();
+                if (execution.getState().getCurrent() == State.Type.CREATED) {
+                    executionQueue.emit(execution.withState(State.Type.SUCCESS));
+                }
+            });
+
+            worker.run();
+            scheduler.run();
+
+            queueCount.await(1, TimeUnit.MINUTES);
+            // needed for RetryingTest to work since there is no context cleaning between method => we have to clear assertion receiver manually
+            assertionStop.run();
+
+            assertThat(queueCount.getCount(), is(0L));
+
+            // Assert that the trigger is now disabled.
+            // It needs to await on assertion as it will be disabled AFTER we receive a success execution.
+            Trigger trigger = Trigger.of(flow, schedule);
+            Await.until(() -> this.triggerState.findLast(trigger).map(t -> t.getDisabled()).orElse(false), Duration.ofMillis(100), Duration.ofSeconds(10));
+        }
+    }
 }
diff --git a/core/src/test/resources/flows/trigger/polling-trigger.yaml b/core/src/test/resources/flows/trigger/polling-trigger.yaml
deleted file mode 100644
index 4f5e4b98b0..0000000000
--- a/core/src/test/resources/flows/trigger/polling-trigger.yaml
+++ /dev/null
@@ -1,11 +0,0 @@
-id: polling-trigger
-namespace: io.kestra.tests.trigger
-
-triggers:
-  - id: watch
-    type: io.kestra.core.tasks.test.PollingTrigger
-
-tasks:
-  - id: end
-    type: io.kestra.core.tasks.debugs.Return
-    format: "{{task.id}} > {{taskrun.startDate}}"