Fix NPE in TryExecutor when retry policy omits optional fields

experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java

@@ -32,6 +32,7 @@
  */
 
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.function;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.tasks;
 import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.tryCatch;
 import static io.serverlessworkflow.fluent.test.TestSerializationUtils.writeAndReadInMemory;
 import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
@@ -45,6 +46,8 @@
 import io.serverlessworkflow.impl.WorkflowModel;
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -61,6 +64,8 @@ public class FuncTryCatchTest {
   private static final String ORDER_002 = "ORDER#002";
   private static final String ORDER_003 = "ORDER#003";
 
+  private static final String TRANSIENT_ERROR = "ERR_TRANSIENT";
+
   @Test
   void booking_compensation_dsl() throws IOException {
 
@@ -472,6 +477,129 @@ void testCatchAll_WithAnyError() throws IOException {
     }
   }
 
+  @Test
+  void testRetryWithoutBackoff() {
+    AtomicInteger attempts = new AtomicInteger();
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow()
+            .tasks(
+                tryCatch(
+                    "tryTask",
+                    t ->
+                        t.tryCatch(
+                                tasks(
+                                    function(
+                                        "riskyTask",
+                                        (String input) -> {
+                                          if (attempts.incrementAndGet() <= 2) {
+                                            throw new WorkflowException(
+                                                WorkflowError.error(TRANSIENT_ERROR, 503).build());
+                                          }
+                                          return "success";
+                                        },
+                                        String.class)))
+                            .catchHandler(
+                                handler ->
+                                    handler
+                                        .errorsWith(err -> err.type(TRANSIENT_ERROR))
+                                        .retry(
+                                            retry ->
+                                                retry
+                                                    .delay(d -> d.milliseconds(10))
+                                                    .limit(
+                                                        limit -> limit.attempt(a -> a.count(3)))))))
+            .build();
+
+    try (WorkflowApplication application = WorkflowApplication.builder().build()) {
+      WorkflowDefinition definition = application.workflowDefinition(workflow);
+      WorkflowModel result = definition.instance("input").start().join();
+      Assertions.assertThat(result.asText()).hasValue("success");
+      Assertions.assertThat(attempts.get()).isEqualTo(3);
+    }
+  }
+
+  @Test
+  void testRetryWithoutLimit() {
+    AtomicInteger attempts = new AtomicInteger();
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow()
+            .tasks(
+                tryCatch(
+                    "tryTask",
+                    t ->
+                        t.tryCatch(
+                                tasks(
+                                    function(
+                                        "riskyTask",
+                                        (String input) -> {
+                                          if (attempts.incrementAndGet() <= 2) {
+                                            throw new WorkflowException(
+                                                WorkflowError.error(TRANSIENT_ERROR, 503).build());
+                                          }
+                                          return "success";
+                                        },
+                                        String.class)))
+                            .catchHandler(
+                                handler ->
+                                    handler
+                                        .errorsWith(err -> err.type(TRANSIENT_ERROR))
+                                        .retry(
+                                            retry ->
+                                                retry
+                                                    .delay(d -> d.milliseconds(10))
+                                                    .backoff(b -> b.constant("c", "10"))))))
+            .build();
+
+    try (WorkflowApplication application = WorkflowApplication.builder().build()) {
+      WorkflowDefinition definition = application.workflowDefinition(workflow);
+      WorkflowModel result = definition.instance("input").start().join();
+      Assertions.assertThat(result.asText()).hasValue("success");
+      Assertions.assertThat(attempts.get()).isEqualTo(3);
+    }
+  }
+
+  @Test
+  void testRetryWithoutDelay() {
+    AtomicInteger attempts = new AtomicInteger();
+
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow()
+            .tasks(
+                tryCatch(
+                    "tryTask",
+                    t ->
+                        t.tryCatch(
+                                tasks(
+                                    function(
+                                        "riskyTask",
+                                        (String input) -> {
+                                          if (attempts.incrementAndGet() <= 2) {
+                                            throw new WorkflowException(
+                                                WorkflowError.error(TRANSIENT_ERROR, 503).build());
+                                          }
+                                          return "success";
+                                        },
+                                        String.class)))
+                            .catchHandler(
+                                handler ->
+                                    handler
+                                        .errorsWith(err -> err.type(TRANSIENT_ERROR))
+                                        .retry(
+                                            retry ->
+                                                retry
+                                                    .backoff(b -> b.constant("c", "10"))
+                                                    .limit(
+                                                        limit -> limit.attempt(a -> a.count(3)))))))
+            .build();
+
+    try (WorkflowApplication application = WorkflowApplication.builder().build()) {
+      WorkflowDefinition definition = application.workflowDefinition(workflow);
+      WorkflowModel result = definition.instance("input").start().join();
+      Assertions.assertThat(result.asText()).hasValue("success");
+      Assertions.assertThat(attempts.get()).isEqualTo(3);
+    }
+  }
+
   public String reserveStock(String order) {
     log.info("Reserving stock for order: {}", order);
     if (order.equals(ORDER_001)) {

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

@@ -37,10 +37,10 @@ public class TaskContext implements TaskContextData {
   private WorkflowModel rawOutput;
   private Instant completedAt;
   private TransitionInfo transition;
-  private short retryAttempt;
+  private int retryAttempt;
   private int iteration;
   private AuthorizationDescriptor authorization;
-  private Optional<Short> tryRetryCount = Optional.empty();
+  private Optional<Integer> tryRetryCount = Optional.empty();
 
   public TaskContext(
       WorkflowModel input,
@@ -71,7 +71,7 @@ private TaskContext(
     this.output = output;
     this.rawOutput = rawOutput;
     this.retryAttempt =
-        parentContext.map(ctx -> ctx.tryRetryCount.orElse(ctx.retryAttempt())).orElse((short) 0);
+        parentContext.map(ctx -> ctx.tryRetryCount.orElse(ctx.retryAttempt())).orElse(0);
     this.contextVariables =
         parentContext.map(p -> new HashMap<>(p.contextVariables)).orElseGet(HashMap::new);
   }
@@ -173,19 +173,19 @@ public boolean isCompleted() {
   }
 
   @Override
-  public short retryAttempt() {
+  public int retryAttempt() {
     return retryAttempt;
   }
 
-  public void retryAttempt(short retryAttempt) {
+  public void retryAttempt(int retryAttempt) {
     this.retryAttempt = retryAttempt;
   }
 
-  public void tryRetryCount(short tryRetryCount) {
+  public void tryRetryCount(int tryRetryCount) {
     this.tryRetryCount = Optional.of(tryRetryCount);
   }
 
-  public Optional<Short> tryRetryCount() {
+  public Optional<Integer> tryRetryCount() {
     return tryRetryCount;
   }
 

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

@@ -40,5 +40,5 @@ public interface TaskContextData {
 
   int iteration();
 
-  short retryAttempt();
+  int retryAttempt();
 }

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

@@ -187,30 +187,36 @@ public static boolean whenExceptTest(
         && exceptFilter.map(w -> !w.test(workflow, taskContext, model)).orElse(true);
   }
 
+  private static final class ZeroDelayResolverHolder {
+    private static final WorkflowValueResolver<Duration> ZERO_DELAY_RESOLVER =
+        (w, t, f) -> Duration.ZERO;
+  }
+
   public static WorkflowValueResolver<Duration> fromTimeoutAfter(
       WorkflowApplication application, TimeoutAfter timeout) {
-    if (timeout.getDurationExpression() != null) {
-      return (w, f, t) ->
-          Duration.parse(
-              application
-                  .expressionFactory()
-                  .resolveString(ExpressionDescriptor.from(timeout.getDurationExpression()))
-                  .apply(w, f, t));
-    } else if (timeout.getDurationLiteral() != null) {
-      Duration duration = Duration.parse(timeout.getDurationLiteral());
-      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;
+    if (timeout != null) {
+      if (timeout.getDurationExpression() != null) {
+        return (w, f, t) ->
+            Duration.parse(
+                application
+                    .expressionFactory()
+                    .resolveString(ExpressionDescriptor.from(timeout.getDurationExpression()))
+                    .apply(w, f, t));
+      } else if (timeout.getDurationLiteral() != null) {
+        Duration duration = Duration.parse(timeout.getDurationLiteral());
+        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())));
+      }
     }
+    return ZeroDelayResolverHolder.ZERO_DELAY_RESOLVER;
   }
 
   public static Optional<WorkflowValueResolver<Duration>> getTaskTimeout(

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

@@ -19,6 +19,7 @@
 import io.serverlessworkflow.api.types.ErrorFilter;
 import io.serverlessworkflow.api.types.Retry;
 import io.serverlessworkflow.api.types.RetryBackoff;
+import io.serverlessworkflow.api.types.RetryLimit;
 import io.serverlessworkflow.api.types.RetryPolicy;
 import io.serverlessworkflow.api.types.TaskItem;
 import io.serverlessworkflow.api.types.TryTask;
@@ -107,25 +108,34 @@ private Optional<RetryExecutor> buildRetryInterval(Retry retry) {
 
     protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) {
       return new DefaultRetryExecutor(
-          retryPolicy.getLimit().getAttempt().getCount(),
+          resolveMaxAttempts(retryPolicy.getLimit()),
           buildIntervalFunction(retryPolicy),
           WorkflowUtils.optionalPredicate(application, retryPolicy.getWhen()),
           WorkflowUtils.optionalPredicate(application, retryPolicy.getExceptWhen()));
     }
 
+    private static int resolveMaxAttempts(RetryLimit limit) {
+      return limit != null && limit.getAttempt() != null
+          ? limit.getAttempt().getCount()
+          : Integer.MAX_VALUE - 1;
+    }
+
     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());
+      if (backoff != null) {
+        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");
+      return new ConstantRetryIntervalFunction(
+          application, retryPolicy.getDelay(), retryPolicy.getJitter());
     }
 
     @Override

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

@@ -49,21 +49,24 @@ public Duration apply(
       WorkflowContext workflowContext,
       TaskContext taskContext,
       WorkflowModel model,
-      short numAttempts) {
+      int numAttempts) {
     Duration delay = delayResolver.apply(workflowContext, taskContext, model);
     Duration minJittering =
         minJitteringResolver
             .map(min -> min.apply(workflowContext, taskContext, model))
             .orElse(Duration.ZERO);
-    Duration maxJittering =
+    Duration result = calcDelay(delay, numAttempts).plus(minJittering);
+    long 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())));
+            .orElse(Duration.ZERO)
+            .toMillis();
+    long diff = maxJittering - minJittering.toMillis();
+    if (diff > 0) {
+      result = result.plus(Duration.ofMillis(Math.round(Math.random() * diff)));
+    }
+    return result;
   }
 
-  protected abstract Duration calcDelay(Duration delay, short numAttempts);
+  protected abstract Duration calcDelay(Duration delay, int numAttempts);
 }

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

@@ -28,7 +28,7 @@ public ConstantRetryIntervalFunction(
   }
 
   @Override
-  protected Duration calcDelay(Duration delay, short numAttempts) {
+  protected Duration calcDelay(Duration delay, int numAttempts) {
     return delay;
   }
 }

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

@@ -46,7 +46,7 @@ public DefaultRetryExecutor(
   @Override
   public Optional<CompletableFuture<WorkflowModel>> retry(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel model) {
-    short numAttempts = taskContext.tryRetryCount().orElseThrow();
+    int numAttempts = taskContext.tryRetryCount().orElseThrow();
     if (numAttempts++ < maxAttempts
         && WorkflowUtils.whenExceptTest(
             whenFilter, exceptFilter, workflowContext, taskContext, model)) {
@@ -62,7 +62,7 @@ public Optional<CompletableFuture<WorkflowModel>> retry(
   @Override
   public void init(WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel model) {
     if (taskContext.tryRetryCount().isEmpty()) {
-      taskContext.tryRetryCount((short) 0);
+      taskContext.tryRetryCount(0);
     }
   }
 }

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

@@ -22,13 +22,16 @@
 
 public class ExponentialRetryIntervalFunction extends AbstractRetryIntervalFunction {
 
+  // 2^33 will be several years, which I think is too high as delay
+  private static final int MAX_EXPONENTIAL_RETRY = 32;
+
   public ExponentialRetryIntervalFunction(
       WorkflowApplication appl, TimeoutAfter delay, RetryPolicyJitter jitter) {
     super(appl, delay, jitter);
   }
 
   @Override
-  protected Duration calcDelay(Duration delay, short numAttempts) {
-    return delay.multipliedBy(1 << numAttempts);
+  protected Duration calcDelay(Duration delay, int numAttempts) {
+    return delay.multipliedBy(1L << Math.min(numAttempts, MAX_EXPONENTIAL_RETRY));
   }
 }