[Fix #727] Implementing retries

impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java

@@ -37,7 +37,7 @@ public class TaskContext implements TaskContextData {
   private WorkflowModel rawOutput;
   private Instant completedAt;
   private TransitionInfo transition;
-  private boolean completed;
+  private short retryAttempt;
 
   public TaskContext(
       WorkflowModel input,
@@ -67,6 +67,7 @@ private TaskContext(
     this.input = input;
     this.output = output;
     this.rawOutput = rawOutput;
+    this.retryAttempt = parentContext.map(TaskContext::retryAttempt).orElse((short) 0);
     this.contextVariables =
         parentContext.map(p -> new HashMap<>(p.contextVariables)).orElseGet(HashMap::new);
   }
@@ -110,7 +111,6 @@ public WorkflowModel rawOutput() {
 
   public TaskContext output(WorkflowModel output) {
     this.output = output;
-    this.completed = true;
     return this;
   }
 
@@ -162,7 +162,19 @@ public TaskContext transition(TransitionInfo transition) {
   }
 
   public boolean isCompleted() {
-    return completed;
+    return completedAt != null;
+  }
+
+  public short retryAttempt() {
+    return retryAttempt;
+  }
+
+  public void retryAttempt(short retryAttempt) {
+    this.retryAttempt = retryAttempt;
+  }
+
+  public boolean isRetrying() {
+    return retryAttempt > 0;
   }
 
   @Override
@@ -175,6 +187,8 @@ public String toString() {
         + taskName
         + ", completedAt="
         + completedAt
+        + ", retryAttempt="
+        + retryAttempt
         + "]";
   }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java

@@ -15,17 +15,20 @@
  */
 package io.serverlessworkflow.impl;
 
+import io.serverlessworkflow.api.types.DurationInline;
 import io.serverlessworkflow.api.types.ExportAs;
 import io.serverlessworkflow.api.types.InputFrom;
 import io.serverlessworkflow.api.types.OutputAs;
 import io.serverlessworkflow.api.types.SchemaUnion;
+import io.serverlessworkflow.api.types.TimeoutAfter;
 import io.serverlessworkflow.api.types.UriTemplate;
 import io.serverlessworkflow.impl.expressions.ExpressionDescriptor;
 import io.serverlessworkflow.impl.expressions.ExpressionUtils;
 import io.serverlessworkflow.impl.resources.ResourceLoader;
 import io.serverlessworkflow.impl.schema.SchemaValidator;
 import io.serverlessworkflow.impl.schema.SchemaValidatorFactory;
 import java.net.URI;
+import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import org.slf4j.Logger;
@@ -153,4 +156,42 @@ public static void safeClose(AutoCloseable closeable) {
       }
     }
   }
+
+  public static boolean whenExceptTest(
+      Optional<WorkflowPredicate> whenFilter,
+      Optional<WorkflowPredicate> exceptFilter,
+      WorkflowContext workflow,
+      TaskContext taskContext,
+      WorkflowModel model) {
+    return whenFilter.map(w -> w.test(workflow, taskContext, model)).orElse(true)
+        && exceptFilter.map(w -> !w.test(workflow, taskContext, model)).orElse(true);
+  }
+
+  public static WorkflowValueResolver<Duration> fromTimeoutAfter(
+      WorkflowApplication application, TimeoutAfter timeout) {
+    if (timeout.getDurationExpression() != null) {
+      if (ExpressionUtils.isExpr(timeout.getDurationExpression())) {
+        return (w, f, t) ->
+            Duration.parse(
+                application
+                    .expressionFactory()
+                    .resolveString(ExpressionDescriptor.from(timeout.getDurationExpression()))
+                    .apply(w, f, t));
+      } else {
+        Duration duration = Duration.parse(timeout.getDurationExpression());
+        return (w, f, t) -> duration;
+      }
+    } else if (timeout.getDurationInline() != null) {
+      DurationInline inlineDuration = timeout.getDurationInline();
+      return (w, t, f) ->
+          Duration.ofDays(inlineDuration.getDays())
+              .plus(
+                  Duration.ofHours(inlineDuration.getHours())
+                      .plus(Duration.ofMinutes(inlineDuration.getMinutes()))
+                      .plus(Duration.ofSeconds(inlineDuration.getSeconds()))
+                      .plus(Duration.ofMillis(inlineDuration.getMilliseconds())));
+    } else {
+      return (w, t, f) -> Duration.ZERO;
+    }
+  }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java

@@ -38,6 +38,7 @@
 import io.serverlessworkflow.impl.lifecycle.TaskCancelledEvent;
 import io.serverlessworkflow.impl.lifecycle.TaskCompletedEvent;
 import io.serverlessworkflow.impl.lifecycle.TaskFailedEvent;
+import io.serverlessworkflow.impl.lifecycle.TaskRetriedEvent;
 import io.serverlessworkflow.impl.lifecycle.TaskStartedEvent;
 import io.serverlessworkflow.impl.resources.ResourceLoader;
 import io.serverlessworkflow.impl.schema.SchemaValidator;
@@ -204,9 +205,15 @@ public CompletableFuture<TaskContext> apply(
               .thenCompose(workflowContext.instance()::suspendedCheck)
               .thenApply(
                   t -> {
-                    publishEvent(
-                        workflowContext,
-                        l -> l.onTaskStarted(new TaskStartedEvent(workflowContext, taskContext)));
+                    if (t.isRetrying()) {
+                      publishEvent(
+                          workflowContext,
+                          l -> l.onTaskRetried(new TaskRetriedEvent(workflowContext, taskContext)));
+                    } else {
+                      publishEvent(
+                          workflowContext,
+                          l -> l.onTaskStarted(new TaskStartedEvent(workflowContext, taskContext)));
+                    }
                     inputSchemaValidator.ifPresent(s -> s.validate(t.rawInput()));
                     inputProcessor.ifPresent(
                         p -> taskContext.input(p.apply(workflowContext, t, t.rawInput())));

impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java

@@ -17,6 +17,9 @@
 
 import io.serverlessworkflow.api.types.CatchErrors;
 import io.serverlessworkflow.api.types.ErrorFilter;
+import io.serverlessworkflow.api.types.Retry;
+import io.serverlessworkflow.api.types.RetryBackoff;
+import io.serverlessworkflow.api.types.RetryPolicy;
 import io.serverlessworkflow.api.types.TaskItem;
 import io.serverlessworkflow.api.types.TryTask;
 import io.serverlessworkflow.api.types.TryTaskCatch;
@@ -29,6 +32,12 @@
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.WorkflowUtils;
+import io.serverlessworkflow.impl.executors.retry.ConstantRetryIntervalFunction;
+import io.serverlessworkflow.impl.executors.retry.DefaultRetryExecutor;
+import io.serverlessworkflow.impl.executors.retry.ExponentialRetryIntervalFunction;
+import io.serverlessworkflow.impl.executors.retry.LinearRetryIntervalFunction;
+import io.serverlessworkflow.impl.executors.retry.RetryExecutor;
+import io.serverlessworkflow.impl.executors.retry.RetryIntervalFunction;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -42,6 +51,7 @@ public class TryExecutor extends RegularTaskExecutor<TryTask> {
   private final Optional<Predicate<WorkflowError>> errorFilter;
   private final TaskExecutor<?> taskExecutor;
   private final Optional<TaskExecutor<?>> catchTaskExecutor;
+  private final Optional<RetryExecutor> retryIntervalExecutor;
 
   public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTask> {
 
@@ -50,6 +60,7 @@ public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTas
     private final Optional<Predicate<WorkflowError>> errorFilter;
     private final TaskExecutor<?> taskExecutor;
     private final Optional<TaskExecutor<?>> catchTaskExecutor;
+    private final Optional<RetryExecutor> retryIntervalExecutor;
 
     protected TryExecutorBuilder(
         WorkflowMutablePosition position, TryTask task, WorkflowDefinition definition) {
@@ -60,13 +71,63 @@ protected TryExecutorBuilder(
       this.exceptFilter = WorkflowUtils.optionalPredicate(application, catchInfo.getExceptWhen());
       this.taskExecutor =
           TaskExecutorHelper.createExecutorList(position, task.getTry(), definition);
-      List<TaskItem> catchTask = task.getCatch().getDo();
-      this.catchTaskExecutor =
-          catchTask != null && !catchTask.isEmpty()
-              ? Optional.of(
-                  TaskExecutorHelper.createExecutorList(
-                      position, task.getCatch().getDo(), definition))
-              : Optional.empty();
+      TryTaskCatch catchTask = task.getCatch();
+      if (catchTask != null) {
+        List<TaskItem> catchTaskDo = catchTask.getDo();
+
+        this.catchTaskExecutor =
+            catchTaskDo != null && !catchTaskDo.isEmpty()
+                ? Optional.of(
+                    TaskExecutorHelper.createExecutorList(position, catchTaskDo, definition))
+                : Optional.empty();
+
+        Retry retry = catchTask.getRetry();
+        this.retryIntervalExecutor = retry != null ? buildRetryInterval(retry) : Optional.empty();
+      } else {
+        this.catchTaskExecutor = Optional.empty();
+        this.retryIntervalExecutor = Optional.empty();
+      }
+    }
+
+    private Optional<RetryExecutor> buildRetryInterval(Retry retry) {
+      RetryPolicy retryPolicy = null;
+      if (retry.getRetryPolicyDefinition() != null) {
+        retryPolicy = retry.getRetryPolicyDefinition();
+      } else if (retry.getRetryPolicyReference() != null) {
+        retryPolicy =
+            workflow
+                .getUse()
+                .getRetries()
+                .getAdditionalProperties()
+                .get(retry.getRetryPolicyReference());
+        if (retryPolicy == null) {
+          throw new IllegalStateException("Retry policy " + retryPolicy + " was not found");
+        }
+      }
+      return retryPolicy != null ? Optional.of(buildRetryExecutor(retryPolicy)) : Optional.empty();
+    }
+
+    protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) {
+      return new DefaultRetryExecutor(
+          retryPolicy.getLimit().getAttempt().getCount(),
+          buildIntervalFunction(retryPolicy),
+          WorkflowUtils.optionalPredicate(application, retryPolicy.getWhen()),
+          WorkflowUtils.optionalPredicate(application, retryPolicy.getExceptWhen()));
+    }
+
+    private RetryIntervalFunction buildIntervalFunction(RetryPolicy retryPolicy) {
+      RetryBackoff backoff = retryPolicy.getBackoff();
+      if (backoff.getConstantBackoff() != null) {
+        return new ConstantRetryIntervalFunction(
+            application, retryPolicy.getDelay(), retryPolicy.getJitter());
+      } else if (backoff.getLinearBackoff() != null) {
+        return new LinearRetryIntervalFunction(
+            application, retryPolicy.getDelay(), retryPolicy.getJitter());
+      } else if (backoff.getExponentialBackOff() != null) {
+        return new ExponentialRetryIntervalFunction(
+            application, retryPolicy.getDelay(), retryPolicy.getJitter());
+      }
+      throw new IllegalStateException("A backoff strategy should be set");
     }
 
     @Override
@@ -82,13 +143,19 @@ protected TryExecutor(TryExecutorBuilder builder) {
     this.exceptFilter = builder.exceptFilter;
     this.taskExecutor = builder.taskExecutor;
     this.catchTaskExecutor = builder.catchTaskExecutor;
+    this.retryIntervalExecutor = builder.retryIntervalExecutor;
   }
 
   @Override
   protected CompletableFuture<WorkflowModel> internalExecute(
       WorkflowContext workflow, TaskContext taskContext) {
+    return doIt(workflow, taskContext, taskContext.input());
+  }
+
+  private CompletableFuture<WorkflowModel> doIt(
+      WorkflowContext workflow, TaskContext taskContext, WorkflowModel model) {
     return TaskExecutorHelper.processTaskList(
-            taskExecutor, workflow, Optional.of(taskContext), taskContext.input())
+            taskExecutor, workflow, Optional.of(taskContext), model)
         .exceptionallyCompose(e -> handleException(e, workflow, taskContext));
   }
 
@@ -99,17 +166,27 @@ private CompletableFuture<WorkflowModel> handleException(
     }
     if (e instanceof WorkflowException) {
       WorkflowException exception = (WorkflowException) e;
+      CompletableFuture<WorkflowModel> completable =
+          CompletableFuture.completedFuture(taskContext.rawOutput());
       if (errorFilter.map(f -> f.test(exception.getWorkflowError())).orElse(true)
-          && whenFilter.map(w -> w.test(workflow, taskContext, taskContext.input())).orElse(true)
-          && exceptFilter
-              .map(w -> !w.test(workflow, taskContext, taskContext.input()))
-              .orElse(true)) {
+          && WorkflowUtils.whenExceptTest(
+              whenFilter, exceptFilter, workflow, taskContext, taskContext.rawOutput())) {
         if (catchTaskExecutor.isPresent()) {
-          return TaskExecutorHelper.processTaskList(
-              catchTaskExecutor.get(), workflow, Optional.of(taskContext), taskContext.input());
+          completable =
+              completable.thenCompose(
+                  model ->
+                      TaskExecutorHelper.processTaskList(
+                          catchTaskExecutor.get(), workflow, Optional.of(taskContext), model));
+        }
+        if (retryIntervalExecutor.isPresent()) {
+          completable =
+              completable
+                  .thenCompose(
+                      model -> retryIntervalExecutor.get().retry(workflow, taskContext, model))
+                  .thenCompose(model -> doIt(workflow, taskContext, model));
         }
       }
-      return CompletableFuture.completedFuture(taskContext.rawOutput());
+      return completable;
     } else {
       if (e instanceof RuntimeException) {
         throw (RuntimeException) e;

impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/AbstractRetryIntervalFunction.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors.retry;
+
+import io.serverlessworkflow.api.types.RetryPolicyJitter;
+import io.serverlessworkflow.api.types.TimeoutAfter;
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowUtils;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import java.time.Duration;
+import java.util.Optional;
+
+public abstract class AbstractRetryIntervalFunction implements RetryIntervalFunction {
+
+  private final Optional<WorkflowValueResolver<Duration>> minJitteringResolver;
+  private final Optional<WorkflowValueResolver<Duration>> maxJitteringResolver;
+  private final WorkflowValueResolver<Duration> delayResolver;
+
+  public AbstractRetryIntervalFunction(
+      WorkflowApplication appl, TimeoutAfter delay, RetryPolicyJitter jitter) {
+    if (jitter != null) {
+      minJitteringResolver = Optional.of(WorkflowUtils.fromTimeoutAfter(appl, jitter.getFrom()));
+      maxJitteringResolver = Optional.of(WorkflowUtils.fromTimeoutAfter(appl, jitter.getTo()));
+    } else {
+      minJitteringResolver = Optional.empty();
+      maxJitteringResolver = Optional.empty();
+    }
+    delayResolver = WorkflowUtils.fromTimeoutAfter(appl, delay);
+  }
+
+  @Override
+  public Duration apply(
+      WorkflowContext workflowContext,
+      TaskContext taskContext,
+      WorkflowModel model,
+      short numAttempts) {
+    Duration delay = delayResolver.apply(workflowContext, taskContext, model);
+    Duration minJittering =
+        minJitteringResolver
+            .map(min -> min.apply(workflowContext, taskContext, model))
+            .orElse(Duration.ZERO);
+    Duration maxJittering =
+        maxJitteringResolver
+            .map(max -> max.apply(workflowContext, taskContext, model))
+            .orElse(Duration.ZERO);
+    return calcDelay(delay, numAttempts)
+        .plus(
+            Duration.ofMillis(
+                (long) (minJittering.toMillis() + Math.random() * maxJittering.toMillis())));
+  }
+
+  protected abstract Duration calcDelay(Duration delay, short numAttempts);
+}