added transaction support for tasks

Author: sterlp

core/src/main/java/org/sterl/spring/persistent_tasks/api/RetryStrategy.java

@@ -1,45 +1,114 @@
 package org.sterl.spring.persistent_tasks.api;
 
 import java.time.OffsetDateTime;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.TemporalUnit;
 
 import org.springframework.lang.Nullable;
 
+import lombok.RequiredArgsConstructor;
+
 @FunctionalInterface
 public interface RetryStrategy {
     RetryStrategy NO_RETRY = (c, e) -> false;
     /**
      * One initial execution and after that we will try it 3 more times. Overall 4 executions.
      */
-    RetryStrategy THREE_RETRIES = (c, e) -> c < 4;
+    RetryStrategy THREE_RETRIES = new LinearRetryStrategy(4, ChronoUnit.MINUTES, 1);
     /**
      * One initial execution and after that we will try it 3 more times. Overall 4 executions.
      */
-    RetryStrategy THREE_RETRIES_IMMEDIATELY = new RetryStrategy() {
+    RetryStrategy THREE_RETRIES_IMMEDIATELY = new MultiplicativeRetryStrategy(4, ChronoUnit.MILLIS, 0);
+
+
+    /**
+     * Determines whether a retry should be attempted based on the current
+     * execution count and the provided exception.
+     *
+     * @param executionCount The number of attempts already made.
+     * @param error The exception that triggered the retry.
+     * @return {@code true} if the current execution count is less than
+     *         the maximum execution count; {@code false} otherwise.
+     */
+    boolean shouldRetry(int executionCount, @Nullable Exception error);
+
+    /**
+     * Calculates the time of the next retry attempt based on the current
+     * execution count and the provided exception.
+     *
+     * @param executionCount The number of attempts already made.
+     * @param exception The exception that triggered the retry.
+     * @return The {@link OffsetDateTime} representing the time of the next retry attempt.
+     */
+    default OffsetDateTime retryAt(int executionCount, @Nullable Exception exception) {
+        return OffsetDateTime.now().plusMinutes(executionCount);
+    }
+
+
+    // Default implementations
+    /**
+     * A retry strategy that determines the next retry time by adding a fixed
+     * offset and the execution count to the current time in the specified temporal unit.
+     * 
+     * <p>This strategy can be used to create retry intervals that increase linearly
+     * with the number of attempts, providing a predictable delay pattern.</p>
+     *
+     * <p>Example:
+     * If {@code offset = 5}, {@code unit = ChronoUnit.SECONDS}, and
+     * {@code executionCount = 3}, the next retry will be scheduled after
+     * {@code 5 + 3 = 8 seconds} from the current time.</p>
+     *
+     * <p>Note: The retry attempts will stop once the maximum execution count
+     * ({@code maxExecutionCount}) is reached.</p>
+     *
+     * @author Your Name
+     */
+    @RequiredArgsConstructor
+    public static class LinearRetryStrategy implements RetryStrategy {
+        private final int maxExecutionCount;
+        private final TemporalUnit unit;
+        private final int offset;
+
         @Override
         public boolean shouldRetry(int executionCount, Exception error) {
-            return executionCount < 4;
+            return maxExecutionCount > executionCount;
         }
         @Override
         public OffsetDateTime retryAt(int executionCount, Exception error) {
-            return OffsetDateTime.now();
+            return OffsetDateTime.now().plus(offset + executionCount, unit);
         }
-    };
-
+    }
+    
     /**
-     * Check if a retry should be done.
+     * A retry strategy that determines the next retry time by multiplying
+     * the execution count by a scaling factor and adding the result to the
+     * current time in the specified temporal unit.
+     * 
+     * <p>This strategy can be used to create retry intervals that increase
+     * multiplicatively with the number of attempts, providing a way to progressively
+     * delay retries.</p>
      *
-     * @param executionCount 0 based counter how often the execution was tried
-     * @param error the exception, <code>null</code> on a timeout
-     */
-    boolean shouldRetry(int executionCount, @Nullable Exception error);
-
-    /**
-     * By default a linear retry strategy, adding one minute for each failed try.
+     * <p>Example:
+     * If {@code scalingFactor = 2}, {@code unit = ChronoUnit.SECONDS}, and
+     * {@code executionCount = 3}, the next retry will be scheduled after
+     * {@code 2 * 3 = 6 seconds} from the current time.</p>
      *
-     * @param executionCount 0 based counter how often the execution was tried
-     * @param error the exception, <code>null</code> on a timeout
+     * <p>Note: The retry attempts will stop once the maximum execution count
+     * ({@code maxExecutionCount}) is reached.</p>
      */
-    default OffsetDateTime retryAt(int executionCount, @Nullable Exception error) {
-        return OffsetDateTime.now().plusMinutes(1 + executionCount);
+    @RequiredArgsConstructor
+    public static class MultiplicativeRetryStrategy implements RetryStrategy {
+        private final int maxExecutionCount;
+        private final TemporalUnit unit;
+        private final int scalingFactor;
+
+        @Override
+        public boolean shouldRetry(int executionCount, Exception error) {
+            return maxExecutionCount > executionCount;
+        }
+        @Override
+        public OffsetDateTime retryAt(int executionCount, Exception error) {
+            return OffsetDateTime.now().plus(scalingFactor * executionCount, unit);
+        }
     }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/api/SpringBeanTask.java

@@ -11,4 +11,19 @@ public interface SpringBeanTask<T extends Serializable> extends Consumer<T> {
     default RetryStrategy retryStrategy() {
         return RetryStrategy.THREE_RETRIES;
     }
+    
+    /**
+     * Whether the task is transaction or not. If <code>true</code> the execution
+     * is wrapped into the default transaction template together with the state update
+     * and the following events:
+     * <ol>
+     * <li>org.sterl.spring.persistent_tasks.trigger.event.TriggerRunningEvent</li>
+     * <li>org.sterl.spring.persistent_tasks.trigger.event.TriggerSuccessEvent</li>
+     * <li>org.sterl.spring.persistent_tasks.trigger.event.TriggerFailedEvent</li>
+     * </ol>
+     * @return {@code true} if the task is transactional; {@code false} otherwise.
+     */
+    default boolean isTransactional() {
+        return false;
+    }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/EditTriggerComponent.java

@@ -19,7 +19,7 @@
 import org.sterl.spring.persistent_tasks.shared.model.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.TriggerCompleteEvent;
+import org.sterl.spring.persistent_tasks.trigger.event.TriggerSuccessEvent;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerFailedEvent;
 import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity;
 import org.sterl.spring.persistent_tasks.trigger.repository.TriggerRepository;
@@ -55,8 +55,8 @@ public Optional<TriggerEntity> completeTaskWithStatus(TriggerKey key, Exception
         result.ifPresent(t -> {
             t.complete(e);
 
-            if (t.getData().getStatus() != TriggerStatus.FAILED) {
-                publisher.publishEvent(new TriggerCompleteEvent(t));
+            if (t.getData().getStatus() == TriggerStatus.SUCCESS) {
+                publisher.publishEvent(new TriggerSuccessEvent(t));
                 log.debug("Setting {} to status={} {}", key, t.getData().getStatus(),
                         e == null ? "" : "error=" + e.getClass().getSimpleName());
             } else {

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/RunTriggerComponent.java

@@ -3,10 +3,12 @@
 import java.io.Serializable;
 import java.time.OffsetDateTime;
 import java.util.Optional;
+import java.util.concurrent.Callable;
 
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.lang.Nullable;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.support.TransactionTemplate;
 import org.sterl.spring.persistent_tasks.api.Task;
 import org.sterl.spring.persistent_tasks.task.TaskService;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerRunningEvent;
@@ -23,36 +25,72 @@ public class RunTriggerComponent {
     private final TaskService taskService;
     private final EditTriggerComponent editTrigger;
     private final ApplicationEventPublisher eventPublisher;
+    private final TransactionTemplate trx;
     private final StateSerializer serializer = new StateSerializer();
 
     /**
      * Will execute the given {@link TriggerEntity} and handle any errors etc.
      */
-    public Optional<TriggerEntity> execute(@Nullable TriggerEntity trigger) {
+    public Optional<TriggerEntity> execute(TriggerEntity trigger) {
         if (trigger == null) {
             return Optional.empty();
         }
-        Task<Serializable> task = null;
+        final var taskAndState = getTastAndState(trigger);
+        // something went really wrong this trigger is crap
+        if (taskAndState == null) return Optional.of(trigger);
+
         try {
-            task = taskService.assertIsKnown(trigger.newTaskId());
+            Optional<TriggerEntity> result;
+            if (taskAndState.isTransactional()) {
+                result = trx.execute(t -> taskAndState.call());
+            } else {
+                result = taskAndState.call();
+            }
 
+            return result;
+        } catch (Exception e) {
+            return handleTaskException(taskAndState, e);
+        }
+    }
+    @Nullable
+    private TaskAndState getTastAndState(TriggerEntity trigger) {
+        try {
+            var task = taskService.assertIsKnown(trigger.newTaskId());
+            var state = serializer.deserialize(trigger.getData().getState());
+            return new TaskAndState(task, state, trigger);
+        } catch (Exception e) {
+            // this trigger is somehow crap, no retry and done.
+            handleTaskException(new TaskAndState(null, null, trigger), e);
+            return null;
+        }
+    }
+    @RequiredArgsConstructor
+    private class TaskAndState implements Callable<Optional<TriggerEntity>> {
+        final Task<Serializable> task;
+        final Serializable state;
+        final TriggerEntity trigger;
+
+        boolean isTransactional() {
+            return task.isTransactional();
+        }
+        public Optional<TriggerEntity> call() {
             eventPublisher.publishEvent(new TriggerRunningEvent(trigger));
 
-            task.accept(serializer.deserialize(trigger.getData().getState()));
+            task.accept(state);
 
             var result = editTrigger.completeTaskWithSuccess(trigger.getKey());
             editTrigger.deleteTrigger(trigger);
 
             return result;
-        } catch (Exception e) {
-            return handleTaskException(trigger, task, e);
+            
         }
     }
 
-    private Optional<TriggerEntity> handleTaskException(TriggerEntity trigger,
-            @Nullable Task<Serializable> task,
+    private Optional<TriggerEntity> handleTaskException(TaskAndState taskAndState,
             @Nullable Exception e) {
 
+        var trigger = taskAndState.trigger;
+        var task = taskAndState.task;
         var result = editTrigger.completeTaskWithStatus(trigger.getKey(), e);
 
         if (task != null &&
@@ -73,7 +111,7 @@ private Optional<TriggerEntity> handleTaskException(TriggerEntity trigger,
             }
         } else {
             log.error("{} failed, no more retries! {}", trigger.getKey(), 
-                    e == null ? "No exception given." : e.getMessage());
+                    e == null ? "No exception given." : e.getMessage(), e);
 
             editTrigger.deleteTrigger(trigger);
         }

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/StateSerializer.java

@@ -33,7 +33,7 @@ public Serializable deserialize(byte[] bytes) {
         try (ObjectInput in = new ObjectInputStream(bis)) {
             return (Serializable)in.readObject();
         } catch (Exception ex) {
-            throw new RuntimeException(ex);
+            throw new RuntimeException("Failed to deserialize state of length " + bytes.length, ex);
         }
     }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerCompleteEvent.java renamed to core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerSuccessEvent.java

@@ -2,6 +2,6 @@
 
 import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity;
 
-public record TriggerCompleteEvent(TriggerEntity trigger) implements TriggerLifeCycleEvent {
+public record TriggerSuccessEvent(TriggerEntity trigger) implements TriggerLifeCycleEvent {
 
 }

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/model/TriggerEntity.java

@@ -85,10 +85,15 @@ 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());
@@ -103,4 +108,9 @@ public TriggerEntity runAt(OffsetDateTime runAt) {
         data.setRunAt(runAt);
         return this;
     }
+
+    public TriggerEntity withState(byte[] state) {
+        this.data.setState(state);
+        return this;
+    }
 }

core/src/test/java/org/sterl/spring/persistent_tasks/trigger/TriggerServiceTest.java

@@ -353,4 +353,34 @@ void testRescheduleAbandonedTasks() {
         assertThat(rescheduledTasks).hasSize(1);
         assertThat(rescheduledTasks.get(0).getKey()).isEqualTo(t1.getKey());
     }
+    
+    @Test
+    void testUnknownTriggersNoRetry() {
+        // GIVEN
+        var t = triggerRepository.save(new TriggerEntity(new TriggerKey("fooTask-unknown")));
+        
+        // WHEN
+        runNextTrigger();
+        
+        // WHEN
+        var triggerData = persistentTaskService.getLastTriggerData(t.getKey()).get();
+        assertThat(triggerData.getStatus()).isEqualTo(TriggerStatus.FAILED);
+        assertThat(triggerData.getExceptionName()).isEqualTo(IllegalStateException.class.getName());
+    }
+    
+    @Test
+    void testBadStateNoRetry() {
+        var t = triggerRepository.save(new TriggerEntity(
+                new TriggerKey("slowTask")
+            ).withState(new byte[] {12, 54})
+        );
+        
+        // WHEN
+        runNextTrigger();
+        
+        // WHEN
+        var triggerData = persistentTaskService.getLastTriggerData(t.getKey()).get();
+        assertThat(triggerData.getStatus()).isEqualTo(TriggerStatus.FAILED);
+        assertThat(triggerData.getExceptionName()).isEqualTo(RuntimeException.class.getName());
+    }
 }

ui/src/server-api.d.ts

@@ -37,7 +37,14 @@ export interface HistoryOverview {
 export interface RetryStrategy {
 }
 
+export interface LinearRetryStrategy extends RetryStrategy {
+}
+
+export interface MultiplicativeRetryStrategy extends RetryStrategy {
+}
+
 export interface SpringBeanTask<T> extends Consumer<T> {
+    transactional: boolean;
 }
 
 export interface Task<T> extends SpringBeanTask<T> {