Implement retry attempt.duration per-attempt timeout

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

@@ -33,17 +33,21 @@
 import io.serverlessworkflow.impl.WorkflowMutablePosition;
 import io.serverlessworkflow.impl.WorkflowPredicate;
 import io.serverlessworkflow.impl.WorkflowUtils;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
 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.time.Duration;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Predicate;
 
 public class TryExecutor extends RegularTaskExecutor<TryTask> {
@@ -54,6 +58,8 @@ public class TryExecutor extends RegularTaskExecutor<TryTask> {
   private final TaskExecutor<?> taskExecutor;
   private final Optional<TaskExecutor<?>> catchTaskExecutor;
   private final Optional<RetryExecutor> retryIntervalExecutor;
+  private final Optional<WorkflowValueResolver<Duration>> attemptDuration;
+  private final Optional<WorkflowValueResolver<Duration>> overallDuration;
   private final String errorVariable;
 
   public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTask> {
@@ -64,6 +70,8 @@ public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTas
     private final TaskExecutor<?> taskExecutor;
     private final Optional<TaskExecutor<?>> catchTaskExecutor;
     private final Optional<RetryExecutor> retryIntervalExecutor;
+    private final Optional<WorkflowValueResolver<Duration>> attemptDuration;
+    private final Optional<WorkflowValueResolver<Duration>> overallDuration;
     private String errorVariable;
 
     protected TryExecutorBuilder(
@@ -83,27 +91,33 @@ protected TryExecutorBuilder(
                       position.copy().addProperty("catch"), catchTaskDo, definition))
               : Optional.empty();
       Retry retry = catchInfo.getRetry();
-      this.retryIntervalExecutor = retry != null ? buildRetryInterval(retry) : Optional.empty();
+      Optional<RetryPolicy> retryPolicy = resolveRetryPolicy(retry);
+      this.retryIntervalExecutor = retryPolicy.map(this::buildRetryExecutor);
+      this.attemptDuration = retryPolicy.flatMap(this::resolveAttemptDuration);
+      this.overallDuration = retryPolicy.flatMap(this::resolveOverallDuration);
       this.taskExecutor =
           TaskExecutorHelper.createExecutorList(position, task.getTry(), definition, "try");
     }
 
-    private Optional<RetryExecutor> buildRetryInterval(Retry retry) {
+    private Optional<RetryPolicy> resolveRetryPolicy(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");
+      if (retry != 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 " + retry.getRetryPolicyReference() + " was not found");
+          }
         }
       }
-      return retryPolicy != null ? Optional.of(buildRetryExecutor(retryPolicy)) : Optional.empty();
+      return Optional.ofNullable(retryPolicy);
     }
 
     protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) {
@@ -114,6 +128,23 @@ protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) {
           WorkflowUtils.optionalPredicate(application, retryPolicy.getExceptWhen()));
     }
 
+    private Optional<WorkflowValueResolver<Duration>> resolveAttemptDuration(
+        RetryPolicy retryPolicy) {
+      RetryLimit limit = retryPolicy.getLimit();
+      return limit != null && limit.getAttempt() != null && limit.getAttempt().getDuration() != null
+          ? Optional.of(
+              WorkflowUtils.fromTimeoutAfter(application, limit.getAttempt().getDuration()))
+          : Optional.empty();
+    }
+
+    private Optional<WorkflowValueResolver<Duration>> resolveOverallDuration(
+        RetryPolicy retryPolicy) {
+      RetryLimit limit = retryPolicy.getLimit();
+      return limit != null && limit.getDuration() != null
+          ? Optional.of(WorkflowUtils.fromTimeoutAfter(application, limit.getDuration()))
+          : Optional.empty();
+    }
+
     private static int resolveMaxAttempts(RetryLimit limit) {
       return limit != null && limit.getAttempt() != null
           ? limit.getAttempt().getCount()
@@ -152,65 +183,123 @@ protected TryExecutor(TryExecutorBuilder builder) {
     this.taskExecutor = builder.taskExecutor;
     this.catchTaskExecutor = builder.catchTaskExecutor;
     this.retryIntervalExecutor = builder.retryIntervalExecutor;
+    this.attemptDuration = builder.attemptDuration;
     this.errorVariable = builder.errorVariable;
+    this.overallDuration = builder.overallDuration;
   }
 
   @Override
   protected CompletableFuture<WorkflowModel> internalExecute(
       WorkflowContext workflow, TaskContext taskContext) {
-    return doIt(workflow, taskContext, taskContext.input());
+    WorkflowModel model = taskContext.input();
+    return cancellingFutureTimeout(
+            doIt(workflow, taskContext, model), overallDuration, workflow, taskContext, model)
+        .exceptionallyCompose(
+            e -> CompletableFuture.failedFuture(timeoutToWorkflow(e, taskContext)));
   }
 
   private CompletableFuture<WorkflowModel> doIt(
       WorkflowContext workflow, TaskContext taskContext, WorkflowModel model) {
     retryIntervalExecutor.ifPresent(r -> r.init(workflow, taskContext, model));
-    return TaskExecutorHelper.processTaskList(
-            taskExecutor, workflow, Optional.of(taskContext), model)
+    CompletableFuture<WorkflowModel> taskFuture =
+        TaskExecutorHelper.processTaskList(taskExecutor, workflow, Optional.of(taskContext), model);
+    return cancellingFutureTimeout(taskFuture, attemptDuration, workflow, taskContext, model)
         .exceptionallyCompose(e -> handleException(e, workflow, taskContext));
   }
 
   private CompletableFuture<WorkflowModel> handleException(
       Throwable e, WorkflowContext workflow, TaskContext taskContext) {
-    if (e instanceof CompletionException) {
-      return handleException(e.getCause(), workflow, taskContext);
+    Throwable cause = e instanceof CompletionException ? e.getCause() : e;
+    if (cause instanceof TimeoutException timeout) {
+      return handleException(timeoutToWorkflow(timeout, taskContext), workflow, taskContext);
+    } else if (cause instanceof WorkflowException exception) {
+      return handleException(exception, workflow, taskContext);
+    } else {
+      return CompletableFuture.failedFuture(e);
     }
-    if (e instanceof WorkflowException) {
-      WorkflowException exception = (WorkflowException) e;
+  }
+
+  private CompletableFuture<WorkflowModel> handleException(
+      WorkflowException exception, WorkflowContext workflow, TaskContext taskContext) {
+    WorkflowError error = exception.getWorkflowError();
+    if (errorFilter.map(f -> f.test(error)).orElse(true)
+        && WorkflowUtils.whenExceptTest(
+            whenFilter,
+            exceptFilter,
+            workflow,
+            taskContext,
+            workflow.definition().application().modelFactory().fromAny(error))) {
       CompletableFuture<WorkflowModel> completable =
           CompletableFuture.completedFuture(taskContext.rawOutput());
-      WorkflowError error = exception.getWorkflowError();
-      if (errorFilter.map(f -> f.test(error)).orElse(true)
-          && WorkflowUtils.whenExceptTest(
-              whenFilter,
-              exceptFilter,
-              workflow,
-              taskContext,
-              workflow.definition().application().modelFactory().fromAny(error))) {
-        if (errorVariable != null) {
-          taskContext.variables().put(errorVariable, error);
-        }
-        if (catchTaskExecutor.isPresent()) {
-          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)
-                              .orElse(CompletableFuture.failedFuture(e)))
-                  .thenCompose(model -> doIt(workflow, taskContext, model));
-        }
-        return completable;
+
+      if (errorVariable != null) {
+        taskContext.variables().put(errorVariable, error);
+      }
+      if (catchTaskExecutor.isPresent()) {
+        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)
+                            .orElse(CompletableFuture.failedFuture(exception)))
+                .thenCompose(model -> doIt(workflow, taskContext, model));
+      }
+      return completable;
+    } else {
+      return CompletableFuture.failedFuture(exception);
+    }
+  }
+
+  private static WorkflowException timeoutToWorkflow(
+      TimeoutException timeout, TaskContext taskContext) {
+    return new WorkflowException(
+        WorkflowError.timeout()
+            .instance(taskContext.position().jsonPointer())
+            .title(timeout.getMessage())
+            .build(),
+        timeout);
+  }
+
+  private static Throwable timeoutToWorkflow(Throwable ex, TaskContext taskContext) {
+    Throwable cause = ex instanceof CompletionException ? ex.getCause() : ex;
+    return cause instanceof TimeoutException timeout ? timeoutToWorkflow(timeout, taskContext) : ex;
+  }
+
+  private static CompletableFuture<WorkflowModel> cancellingFutureTimeout(
+      CompletableFuture<WorkflowModel> originalFuture,
+      Optional<WorkflowValueResolver<Duration>> duration,
+      WorkflowContext workflowContext,
+      TaskContext taskContext,
+      WorkflowModel model) {
+    long timeout =
+        duration
+            .map(d -> d.apply(workflowContext, taskContext, model))
+            .orElse(Duration.ZERO)
+            .toMillis();
+    return timeout > 0
+        ? originalFuture
+            .copy()
+            .orTimeout(timeout, TimeUnit.MILLISECONDS)
+            .whenComplete((v, e) -> cancelIfTimeout(e, originalFuture))
+        : originalFuture;
+  }
+
+  private static void cancelIfTimeout(Throwable e, CompletableFuture<WorkflowModel> taskFuture) {
+    if (!taskFuture.isDone()) {
+      Throwable realException = e instanceof CompletionException ? e.getCause() : e;
+      if (realException instanceof TimeoutException) {
+        taskFuture.cancel(true);
       }
     }
-    return CompletableFuture.failedFuture(e);
   }
 
   private static Optional<Predicate<WorkflowError>> buildErrorFilter(CatchErrors errors) {

impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java

@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.MockWebServer;
 import org.awaitility.Awaitility;
@@ -186,6 +187,82 @@ void testRetryEnd() throws IOException {
         .hasCauseInstanceOf(WorkflowException.class);
   }
 
+  @Test
+  void testAttemptDuration() throws IOException {
+    apiServer.enqueue(
+        new MockResponse()
+            .setHeadersDelay(2, TimeUnit.SECONDS)
+            .setResponseCode(200)
+            .setHeader("Content-Type", "application/json")
+            .setBody("{}"));
+    assertThatThrownBy(
+            () ->
+                app.workflowDefinition(
+                        readWorkflowFromClasspath(
+                            "workflows-samples/try-catch-retry-attempt-duration.yaml"))
+                    .instance(Map.of())
+                    .start()
+                    .join())
+        .hasCauseInstanceOf(WorkflowException.class);
+  }
+
+  @Test
+  void testAttemptDurationRetry() throws IOException {
+    String result = "{\"name\":\"Luna\"}";
+    apiServer.enqueue(
+        new MockResponse()
+            .setHeadersDelay(2, TimeUnit.SECONDS)
+            .setResponseCode(200)
+            .setHeader("Content-Type", "application/json")
+            .setBody(result));
+    apiServer.enqueue(
+        new MockResponse()
+            .setResponseCode(200)
+            .setHeader("Content-Type", "application/json")
+            .setBody(result));
+    CompletableFuture<WorkflowModel> future =
+        app.workflowDefinition(
+                readWorkflowFromClasspath(
+                    "workflows-samples/try-catch-retry-attempt-duration-retry.yaml"))
+            .instance(Map.of())
+            .start();
+    Awaitility.await().atMost(Duration.ofSeconds(5)).until(future::isDone);
+    assertThat(future.join().as(String.class).orElseThrow()).isEqualTo(result);
+    assertThat(retryListener.taskRetried).hasSize(1);
+    assertThat(retryListener.taskRetried.get("do/0/tryGetPet/try/0/getPet")).isEqualTo((short) 1);
+  }
+
+  @Test
+  void testAttemptDurationOverall() throws IOException {
+    String result = "{\"name\":\"Luna\"}";
+    apiServer.enqueue(
+        new MockResponse()
+            .setHeadersDelay(1, TimeUnit.SECONDS)
+            .setResponseCode(200)
+            .setHeader("Content-Type", "application/json")
+            .setBody(result));
+    apiServer.enqueue(
+        new MockResponse()
+            .setHeadersDelay(1, TimeUnit.SECONDS)
+            .setResponseCode(200)
+            .setHeader("Content-Type", "application/json")
+            .setBody(result));
+    apiServer.enqueue(
+        new MockResponse()
+            .setResponseCode(200)
+            .setHeader("Content-Type", "application/json")
+            .setBody(result));
+    assertThatThrownBy(
+            () ->
+                app.workflowDefinition(
+                        readWorkflowFromClasspath(
+                            "workflows-samples/try-catch-retry-attempt-duration-overall.yaml"))
+                    .instance(Map.of())
+                    .start()
+                    .join())
+        .hasCauseInstanceOf(WorkflowException.class);
+  }
+
   @Test
   void testTimeout() throws IOException {
     Map<String, Object> result =

impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration-overall.yaml

@@ -0,0 +1,31 @@
+document:
+  dsl: '1.0.0'
+  namespace: test
+  name: try-catch-retry-attempt-duration-overall
+  version: '0.1.0'
+do:
+  - tryGetPet:
+      try:
+        - getPet:
+            call: http
+            with:
+              method: get
+              endpoint: http://localhost:9797
+              redirect: true
+      catch:
+        errors:
+          with:
+            type: https://serverlessworkflow.io/spec/1.0.0/errors/timeout
+            status: 408
+        retry:
+          delay:
+            milliseconds: 10
+          backoff:
+            constant: {}
+          limit:
+            duration:
+                milliseconds: 100
+            attempt:
+              count: 5
+              duration:
+                milliseconds: 50