- timers can now be disabled

sterlp · sterlp · commit 6671b79d6420 · 2025-06-09T20:52:55.000+02:00
- Resume added
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/history/HistoryTimer.java b/core/src/main/java/org/sterl/spring/persistent_tasks/history/HistoryTimer.java
@@ -7,13 +7,13 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
-import org.sterl.spring.persistent_tasks.scheduler.config.ConditionalSchedulerServiceByProperty;
+import org.sterl.spring.persistent_tasks.shared.TimersEnabled;
 
 import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
-@ConditionalSchedulerServiceByProperty
+@TimersEnabled
 @Service
 @RequiredArgsConstructor
 @Slf4j
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java b/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java
@@ -166,7 +166,7 @@ public SchedulerEntity getStatus() {
     }
 
     @Transactional
-    public List<TriggerEntity> rescheduleAbandonedTasks(OffsetDateTime timeout) {
+    public List<TriggerEntity> rescheduleAbandonedTriggers(OffsetDateTime timeout) {
         var schedulers = editSchedulerStatus.findOnlineSchedulers(timeout);
 
         final List<TriggerKey> runningKeys = this.taskExecutor.getRunningTriggers().stream().map(TriggerEntity::getKey)
@@ -175,7 +175,7 @@ public List<TriggerEntity> rescheduleAbandonedTasks(OffsetDateTime timeout) {
         int running = triggerService.markTriggersAsRunning(runningKeys, name);
         log.atLevel(running > 0 ? Level.INFO : Level.DEBUG).log("({}) - {} trigger(s) are running on {} schedulers", 
                 running, runningKeys, schedulers);
-        return triggerService.rescheduleAbandonedTasks(timeout);
+        return triggerService.rescheduleAbandoned(timeout);
     }
 
     public List<SchedulerEntity> listAll() {
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerTimer.java b/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerTimer.java
@@ -9,21 +9,23 @@
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.sterl.spring.persistent_tasks.scheduler.config.ConditionalSchedulerServiceByProperty;
+import org.sterl.spring.persistent_tasks.shared.TimersEnabled;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
+@TimersEnabled
 @ConditionalSchedulerServiceByProperty
 @Service
 @RequiredArgsConstructor
 @Slf4j
 class SchedulerTimer {
 
-    @Value("${spring.persistent-tasks.task-timeout:PT5M}")
+    @Value("${spring.persistent-tasks.trigger-timeout:PT5M}")
     private Duration taskTimeout = Duration.ofMinutes(5);
     private final Collection<SchedulerService> schedulerServices;
 
-    @Scheduled(fixedDelayString = "${spring.persistent-tasks.poll-rate:30}", timeUnit = TimeUnit.SECONDS)
+    @Scheduled(fixedDelayString = "${spring.persistent-tasks.poll-rate:60}", timeUnit = TimeUnit.SECONDS)
     void triggerNextTasks() {
         for (SchedulerService s : schedulerServices) {
             try {
@@ -35,12 +37,12 @@ void triggerNextTasks() {
         }
     }
 
-    @Scheduled(fixedDelayString = "${spring.persistent-tasks.poll-persistentTask-timeout:300}", timeUnit = TimeUnit.SECONDS)
+    @Scheduled(fixedDelayString = "${spring.persistent-tasks.poll-abandoned-triggers:300}", timeUnit = TimeUnit.SECONDS)
     void rescheduleAbandonedTasks() {
         var timeout = OffsetDateTime.now().minus(taskTimeout);
         for (SchedulerService s : schedulerServices) {
             try {
-                final var count = s.rescheduleAbandonedTasks(timeout);
+                final var count = s.rescheduleAbandonedTriggers(timeout);
                 log.debug("Found {} abandoned tasks for {}. Timeout={}", 
                         count.size(), s.getName(), timeout);
             } catch (Exception e) {
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/TimersEnabled.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/TimersEnabled.java
@@ -0,0 +1,17 @@
+package org.sterl.spring.persistent_tasks.shared;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+@ConditionalOnProperty(name = "spring.persistent-tasks.timers-enabled", havingValue = "true", matchIfMissing = true)
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.TYPE, ElementType.METHOD })
+@Documented
+public @interface TimersEnabled {
+
+}
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerDataRepository.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerDataRepository.java
@@ -1,6 +1,7 @@
 package org.sterl.spring.persistent_tasks.shared.repository;
 
 import java.time.OffsetDateTime;
+import java.util.Optional;
 import java.util.Set;
 
 import org.springframework.data.domain.Page;
@@ -19,6 +20,7 @@
 import org.sterl.spring.persistent_tasks.shared.StringHelper;
 import org.sterl.spring.persistent_tasks.shared.model.HasTriggerData;
 import org.sterl.spring.persistent_tasks.shared.model.QTriggerData;
+import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity;
 
 import com.querydsl.core.BooleanBuilder;
 import com.querydsl.core.types.ExpressionUtils;
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/TriggerService.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/TriggerService.java
@@ -11,9 +11,9 @@
 import org.springframework.lang.Nullable;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
-import org.sterl.spring.persistent_tasks.api.TriggerRequest;
 import org.sterl.spring.persistent_tasks.api.TaskId;
 import org.sterl.spring.persistent_tasks.api.TriggerKey;
+import org.sterl.spring.persistent_tasks.api.TriggerRequest;
 import org.sterl.spring.persistent_tasks.api.TriggerSearch;
 import org.sterl.spring.persistent_tasks.api.TriggerStatus;
 import org.sterl.spring.persistent_tasks.shared.stereotype.TransactionalService;
@@ -185,7 +185,7 @@ public long countTriggers(@Nullable TriggerStatus status) {
      *
      * Retry will be triggered based on the set strategy.
      */
-    public List<TriggerEntity> rescheduleAbandonedTasks(OffsetDateTime timeout) {
+    public List<TriggerEntity> rescheduleAbandoned(OffsetDateTime timeout) {
         final List<TriggerEntity> result = readTrigger.findTriggersLastPingAfter(
                 timeout);
         final var e = new IllegalStateException("Trigger abandoned - timeout: " + timeout);
@@ -197,14 +197,21 @@ public List<TriggerEntity> rescheduleAbandonedTasks(OffsetDateTime timeout) {
         log.debug("rescheduled {} triggers", result.size());
         return result;
     }
+    
+    public List<TriggerEntity> expireTimeoutTriggers() {
+        return readTrigger.findTriggersTimeoutOut(20)
+                          .stream()
+                          .map(editTrigger::expireTrigger)
+                          .toList();
+    }
 
     public long countTriggers() {
         return readTrigger.countByStatus(null);
     }
 
     public Optional<TriggerEntity> updateRunAt(TriggerKey key, OffsetDateTime time) {
         return readTrigger.get(key).map(t -> {
-            if (t.getData().getStatus() != TriggerStatus.WAITING) {
+            if (t.getData().getStatus() == TriggerStatus.RUNNING) {
                 throw new IllegalStateException("Cannot update status of " + key
                         + " as the current status is: " + t.getData().getStatus());
             }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/TriggerTimer.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/TriggerTimer.java
@@ -0,0 +1,26 @@
+package org.sterl.spring.persistent_tasks.trigger;
+
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.sterl.spring.persistent_tasks.shared.TimersEnabled;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@TimersEnabled
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class TriggerTimer {
+    
+    private final TriggerService triggerService;
+
+    @Scheduled(fixedDelayString = "${spring.persistent-tasks.poll-awaiting-trigger-timeout:300}", timeUnit = TimeUnit.SECONDS)
+    void checkAwaitingTriggersForTimeout() {
+        int expired = triggerService.expireTimeoutTriggers().size();
+        if (expired > 0) log.info("{} triggers have not received the signal, wait expired!", expired);
+        else log.debug("No expired triggers found.");
+    }
+}
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/EditTriggerComponent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/EditTriggerComponent.java
@@ -19,7 +19,9 @@
 import org.sterl.spring.persistent_tasks.api.TriggerStatus;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerAddedEvent;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerCanceledEvent;
+import org.sterl.spring.persistent_tasks.trigger.event.TriggerExpiredEvent;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerFailedEvent;
+import org.sterl.spring.persistent_tasks.trigger.event.TriggerResumedEvent;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerRunningEvent;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerSuccessEvent;
 import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity;
@@ -88,14 +90,17 @@ public Optional<TriggerEntity> failTrigger(
     public Optional<TriggerEntity> cancelTask(TriggerKey id, Exception e) {
         return triggerRepository //
                 .findByKey(id) //
-                .map(t -> cancelTask(t, e));
+                .map(t -> cancelTrigger(t, e));
     }
 
-    private TriggerEntity cancelTask(TriggerEntity t, Exception e) {
+    private TriggerEntity cancelTrigger(TriggerEntity t, Exception e) {
         t.cancel(e);
+
         publisher.publishEvent(new TriggerCanceledEvent(
-                t.getId(), t.copyData(),
+                t.getId(), 
+                t.copyData(),
                 stateSerializer.deserializeOrNull(t.getData().getState())));
+
         triggerRepository.delete(t);
         return t;
     }
@@ -121,16 +126,18 @@ public <T extends Serializable> TriggerEntity addTrigger(TriggerRequest<T> tigge
             result = triggerRepository.save(result);
             log.debug("Added trigger={}", result);
         }
+
         publisher.publishEvent(new TriggerAddedEvent(
                 result.getId(), result.copyData(), tigger.state()));
+
         return result;
     }
     
     public Page<TriggerEntity> resume(TriggerRequest<?> trigger) {
         var search = TriggerSearch.forTriggerRequest(trigger);
         search.setStatus(TriggerStatus.AWAITING_SIGNAL);
         
-        var foundTriggers = readTrigger.searchTriggers(TriggerSearch.byCorrelationId(trigger.correlationId()), Pageable.ofSize(100));
+        var foundTriggers = readTrigger.searchTriggers(search, Pageable.ofSize(100));
         
         log.debug("Resuming {} triggers for given data {}", foundTriggers.getSize(), trigger);
         foundTriggers.forEach(t -> {
@@ -140,9 +147,23 @@ public Page<TriggerEntity> resume(TriggerRequest<?> trigger) {
 
             t.setData(newData.getData());
             t.runAt(trigger.runtAt());
+
+            publisher.publishEvent(new TriggerResumedEvent(t.getId(), t.copyData(), trigger.state()));
         });
         return foundTriggers;
     }
+    
+    public TriggerEntity expireTrigger(TriggerEntity t) {
+        t.getData().setStatus(TriggerStatus.EXPIRED_SIGNAL);
+        t.getData().setStart(null);
+        t.getData().setEnd(null);
+        t.getData().updateRunningDuration();
+        
+        publisher.publishEvent(new TriggerExpiredEvent(
+                t.getId(), t.copyData(), 
+                stateSerializer.deserializeOrNull(t.getData().getState())));
+        return t;
+    }
 
     @NonNull
     public <T extends Serializable> List<TriggerEntity> addTriggers(Collection<TriggerRequest<T>> newTriggers) {
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/ReadTriggerComponent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/ReadTriggerComponent.java
@@ -67,4 +67,11 @@ public Page<TriggerEntity> listTriggers(TaskId<? extends Serializable> task, Pag
         if (task == null) return triggerRepository.findAll(page);
         return triggerRepository.findAll(task.name(), page);
     }
+    
+    public List<TriggerEntity> findTriggersTimeoutOut(int max) {
+        return triggerRepository.findByStatusAndRunAtAfter(
+                TriggerStatus.AWAITING_SIGNAL,
+                OffsetDateTime.now(),
+                Pageable.ofSize(max));
+    }
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerExpiredEvent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerExpiredEvent.java
@@ -0,0 +1,19 @@
+package org.sterl.spring.persistent_tasks.trigger.event;
+
+import java.io.Serializable;
+
+import org.sterl.spring.persistent_tasks.shared.model.TriggerData;
+
+/**
+ * Fired if a new trigger is added.
+ * <p>
+ * Inside a transaction, it is save to join or listen for the <code>AFTER_COMMIT</code>
+ * </p>
+ */
+public record TriggerExpiredEvent(long id, TriggerData data, Serializable state) implements TriggerLifeCycleEvent {
+
+    @Override
+    public boolean isDone() {
+        return false;
+    }
+}
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerResumedEvent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerResumedEvent.java
@@ -0,0 +1,19 @@
+package org.sterl.spring.persistent_tasks.trigger.event;
+
+import java.io.Serializable;
+
+import org.sterl.spring.persistent_tasks.shared.model.TriggerData;
+
+/**
+ * Fired if a new trigger is added.
+ * <p>
+ * Inside a transaction, it is save to join or listen for the <code>AFTER_COMMIT</code>
+ * </p>
+ */
+public record TriggerResumedEvent(long id, TriggerData data, Serializable state) implements TriggerLifeCycleEvent {
+
+    @Override
+    public boolean isDone() {
+        return false;
+    }
+}
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/model/TriggerEntity.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/model/TriggerEntity.java
@@ -65,20 +65,31 @@ public TriggerKey getKey() {
         if (data == null) return null;
         return data.getKey();
     }
+    
+    /**
+     * @param e Sets either {@link TriggerStatus#SUCCESS} or {@link TriggerStatus#FAILED}
+     * based if the {@link Exception} is <code>null</code> or not.
+     */
+    public TriggerEntity complete(Exception e) {
+        finishTriggerWithStatus(e == null ? TriggerStatus.SUCCESS : TriggerStatus.FAILED, e);
+        return this;
+    }
 
     public TriggerEntity cancel(Exception e) {
+        finishTriggerWithStatus(TriggerStatus.CANCELED, e);
+        if (e == null) this.data.setExceptionName("PersistentTask canceled");
+        return this;
+    }
+
+    public void finishTriggerWithStatus(TriggerStatus status, Exception e) {
         this.data.setEnd(OffsetDateTime.now());
-        this.data.setStatus(TriggerStatus.CANCELED);
+        this.data.updateRunningDuration();
+        this.data.setStatus(status);
 
-        if (e == null) {
-            this.data.setExceptionName("PersistentTask canceled");
-        } else {
+        if (e != null) {
             this.data.setExceptionName(e.getClass().getName());
             this.data.setLastException(ExceptionUtils.getStackTrace(e));
         }
-
-        this.data.updateRunningDuration();
-        return this;
     }
 
     public TriggerEntity runOn(String runningOn) {
@@ -92,24 +103,6 @@ public TriggerEntity runOn(String runningOn) {
         return this;
     }
 
-    /**
-     * @param e Sets either {@link TriggerStatus#SUCCESS} or {@link TriggerStatus#FAILED}
-     * based if the {@link Exception} is <code>null</code> or not.
-     */
-    public TriggerEntity complete(Exception e) {
-        data.setStatus(TriggerStatus.SUCCESS);
-        data.setEnd(OffsetDateTime.now());
-        data.updateRunningDuration();
-
-        if (e != null) {
-            data.setStatus(TriggerStatus.FAILED);
-            data.setExceptionName(e.getClass().getName());
-            data.setLastException(ExceptionUtils.getStackTrace(e));
-        }
-
-        return this;
-    }
-
     public TriggerEntity runAt(OffsetDateTime runAt) {
         data.setStatus(TriggerStatus.WAITING);
         data.setRunAt(runAt);
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/repository/TriggerRepository.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/repository/TriggerRepository.java
@@ -22,6 +22,7 @@
 import jakarta.persistence.QueryHint;
 
 public interface TriggerRepository extends TriggerDataRepository<TriggerEntity> {
+
     @Query("SELECT e FROM #{#entityName} e WHERE e.data.key = :key")
     Optional<TriggerEntity> findByKey(@Param("key") TriggerKey key);
 
@@ -72,4 +73,14 @@ int markTriggersAsRunning(
             """)
     List<TriggerEntity> findTriggersLastPingAfter(
             @Param("lastPing") OffsetDateTime lastPing);
+    
+    @Query("""
+            SELECT e FROM #{#entityName} e
+            WHERE  e.data.status = :status
+            AND    e.data.runAt <= :runAt
+            """)
+    List<TriggerEntity> findByStatusAndRunAtAfter(
+            @Param("status") TriggerStatus status,
+            @Param("runAt") OffsetDateTime runAt,
+            Pageable page);
 }
diff --git a/core/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/core/src/main/resources/META-INF/additional-spring-configuration-metadata.json
diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/TaskFailoverTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/TaskFailoverTest.java
diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/TriggerServiceTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/TriggerServiceTest.java
diff --git a/core/src/test/resources/application.yml b/core/src/test/resources/application.yml
diff --git a/ui/pom.xml b/ui/pom.xml

Original file line number	Diff line number	Diff line change
`@@ -166,7 +166,7 @@ public SchedulerEntity getStatus() {`
`166`	`166`	`}`
`167`	`167`
`168`	`168`	`@Transactional`
`169`		`- public List<TriggerEntity> rescheduleAbandonedTasks(OffsetDateTime timeout) {`
	`169`	`+ public List<TriggerEntity> rescheduleAbandonedTriggers(OffsetDateTime timeout) {`
`170`	`170`	`var schedulers = editSchedulerStatus.findOnlineSchedulers(timeout);`
`171`	`171`
`172`	`172`	`final List<TriggerKey> runningKeys = this.taskExecutor.getRunningTriggers().stream().map(TriggerEntity::getKey)`
`@@ -175,7 +175,7 @@ public List<TriggerEntity> rescheduleAbandonedTasks(OffsetDateTime timeout) {`
`175`	`175`	`int running = triggerService.markTriggersAsRunning(runningKeys, name);`
`176`	`176`	`log.atLevel(running > 0 ? Level.INFO : Level.DEBUG).log("({}) - {} trigger(s) are running on {} schedulers",`
`177`	`177`	`running, runningKeys, schedulers);`
`178`		`- return triggerService.rescheduleAbandonedTasks(timeout);`
	`178`	`+ return triggerService.rescheduleAbandoned(timeout);`
`179`	`179`	`}`
`180`	`180`
`181`	`181`	`public List<SchedulerEntity> listAll() {`
Original file line number	Diff line number	Diff line change
`@@ -67,4 +67,11 @@ public Page<TriggerEntity> listTriggers(TaskId<? extends Serializable> task, Pag`
`67`	`67`	`if (task == null) return triggerRepository.findAll(page);`
`68`	`68`	`return triggerRepository.findAll(task.name(), page);`
`69`	`69`	`}`
	`70`	`+`
	`71`	`+ public List<TriggerEntity> findTriggersTimeoutOut(int max) {`
	`72`	`+ return triggerRepository.findByStatusAndRunAtAfter(`
	`73`	`+ TriggerStatus.AWAITING_SIGNAL,`
	`74`	`+ OffsetDateTime.now(),`
	`75`	`+ Pageable.ofSize(max));`
	`76`	`+ }`
`70`	`77`	`}`