[Fix #1601] Implement then on catch

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

@@ -29,7 +29,7 @@ public class CallTaskExecutor<T extends TaskBase> extends RegularTaskExecutor<T>
   private final CallableTask callable;
 
   public static class CallTaskExecutorBuilder<T extends TaskBase>
-      extends RegularTaskExecutorBuilder<T> {
+      extends RegularTaskExecutorBuilder<T, CallTaskExecutor<T>> {
     private CallableTaskFactory callableFactory;
     private List<CallableTaskProxyBuilder> callableProxyBuilders;
     private CallableTask callable;

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

@@ -28,7 +28,7 @@ public class DoExecutor extends RegularTaskExecutor<DoTask> {
 
   private final TaskExecutor<?> taskExecutor;
 
-  public static class DoExecutorBuilder extends RegularTaskExecutorBuilder<DoTask> {
+  public static class DoExecutorBuilder extends RegularTaskExecutorBuilder<DoTask, DoExecutor> {
     private TaskExecutor<?> taskExecutor;
 
     protected DoExecutorBuilder(

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

@@ -53,7 +53,8 @@ public class EmitExecutor extends RegularTaskExecutor<EmitTask> {
           .toList();
   private final EventPropertiesBuilder props;
 
-  public static class EmitExecutorBuilder extends RegularTaskExecutorBuilder<EmitTask> {
+  public static class EmitExecutorBuilder
+      extends RegularTaskExecutorBuilder<EmitTask, EmitExecutor> {
 
     private EventPropertiesBuilder eventBuilder;
 

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

@@ -37,7 +37,7 @@ public class ForExecutor extends RegularTaskExecutor<ForTask> {
   private final Optional<WorkflowPredicate> whileExpr;
   private final TaskExecutor<?> taskExecutor;
 
-  public static class ForExecutorBuilder extends RegularTaskExecutorBuilder<ForTask> {
+  public static class ForExecutorBuilder extends RegularTaskExecutorBuilder<ForTask, ForExecutor> {
     private TaskExecutor<?> taskExecutor;
 
     protected ForExecutorBuilder(

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

@@ -39,7 +39,8 @@ public class ForkExecutor extends RegularTaskExecutor<ForkTask> {
 
   private final boolean compete;
 
-  public static class ForkExecutorBuilder extends RegularTaskExecutorBuilder<ForkTask> {
+  public static class ForkExecutorBuilder
+      extends RegularTaskExecutorBuilder<ForkTask, ForkExecutor> {
 
     private final Map<String, TaskExecutor<?>> taskExecutors;
     private final boolean compete;

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

@@ -47,7 +47,8 @@ public abstract class ListenExecutor extends RegularTaskExecutor<ListenTask> {
   protected final Function<CloudEvent, WorkflowModel> converter;
   protected final EventConsumer eventConsumer;
 
-  public static class ListenExecutorBuilder extends RegularTaskExecutorBuilder<ListenTask> {
+  public static class ListenExecutorBuilder
+      extends RegularTaskExecutorBuilder<ListenTask, ListenExecutor> {
 
     private EventRegistrationBuilderInfo registrationInfo;
     private TaskExecutor<?> loop;

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

@@ -42,7 +42,8 @@ public class RaiseExecutor extends RegularTaskExecutor<RaiseTask> {
 
   private final BiFunction<WorkflowContext, TaskContext, WorkflowError> errorBuilder;
 
-  public static class RaiseExecutorBuilder extends RegularTaskExecutorBuilder<RaiseTask> {
+  public static class RaiseExecutorBuilder
+      extends RegularTaskExecutorBuilder<RaiseTask, RaiseExecutor> {
 
     private final BiFunction<WorkflowContext, TaskContext, WorkflowError> errorBuilder;
     private final WorkflowValueResolver<String> typeFilter;

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

@@ -28,12 +28,14 @@ public abstract class RegularTaskExecutor<T extends TaskBase> extends AbstractTa
 
   protected TransitionInfo transition;
 
-  protected RegularTaskExecutor(RegularTaskExecutorBuilder<T> builder) {
+  protected <V extends RegularTaskExecutor<T>> RegularTaskExecutor(
+      RegularTaskExecutorBuilder<T, V> builder) {
     super(builder);
   }
 
-  public abstract static class RegularTaskExecutorBuilder<T extends TaskBase>
-      extends AbstractTaskExecutorBuilder<T, RegularTaskExecutor<T>> {
+  public abstract static class RegularTaskExecutorBuilder<
+          T extends TaskBase, V extends RegularTaskExecutor<T>>
+      extends AbstractTaskExecutorBuilder<T, V> {
 
     private TransitionInfoBuilder transition;
 
@@ -42,12 +44,13 @@ protected RegularTaskExecutorBuilder(
       super(position, task, definition);
     }
 
+    @Override
     public void connect(Map<String, TaskExecutorBuilder<?>> connections) {
       this.transition = next(task.getThen(), connections);
     }
 
     @Override
-    protected void buildTransition(RegularTaskExecutor<T> instance) {
+    protected void buildTransition(V instance) {
       instance.transition = TransitionInfo.build(transition);
     }
   }
@@ -59,10 +62,8 @@ protected TransitionInfo getSkipTransition() {
 
   protected CompletableFuture<TaskContext> execute(
       WorkflowContext workflow, TaskContext taskContext) {
-    CompletableFuture<TaskContext> future =
-        internalExecute(workflow, taskContext)
-            .thenApply(node -> taskContext.rawOutput(node).transition(transition));
-    return future;
+    return internalExecute(workflow, taskContext)
+        .thenApply(node -> taskContext.rawOutput(node).transition(transition));
   }
 
   protected abstract CompletableFuture<WorkflowModel> internalExecute(

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

@@ -33,7 +33,8 @@ public class RunTaskExecutor extends RegularTaskExecutor<RunTask> {
   private static final ServiceLoader<RunnableTaskBuilder> runnables =
       ServiceLoader.load(RunnableTaskBuilder.class);
 
-  public static class RunTaskExecutorBuilder extends RegularTaskExecutorBuilder<RunTask> {
+  public static class RunTaskExecutorBuilder
+      extends RegularTaskExecutorBuilder<RunTask, RunTaskExecutor> {
     private CallableTask runnable;
 
     protected RunTaskExecutorBuilder(

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

@@ -31,7 +31,7 @@ public class SetExecutor extends RegularTaskExecutor<SetTask> {
 
   private final WorkflowFilter setFilter;
 
-  public static class SetExecutorBuilder extends RegularTaskExecutorBuilder<SetTask> {
+  public static class SetExecutorBuilder extends RegularTaskExecutorBuilder<SetTask, SetExecutor> {
 
     private final WorkflowFilter setFilter;
 

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

@@ -17,6 +17,7 @@
 
 import io.serverlessworkflow.api.types.CatchErrors;
 import io.serverlessworkflow.api.types.ErrorFilter;
+import io.serverlessworkflow.api.types.FlowDirective;
 import io.serverlessworkflow.api.types.Retry;
 import io.serverlessworkflow.api.types.RetryBackoff;
 import io.serverlessworkflow.api.types.RetryLimit;
@@ -42,6 +43,7 @@
 import io.serverlessworkflow.impl.executors.retry.RetryIntervalFunction;
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -60,9 +62,10 @@ public class TryExecutor extends RegularTaskExecutor<TryTask> {
   private final Optional<RetryExecutor> retryIntervalExecutor;
   private final Optional<WorkflowValueResolver<Duration>> attemptDuration;
   private final Optional<WorkflowValueResolver<Duration>> overallDuration;
+  private Optional<TransitionInfo> catchTransition;
   private final String errorVariable;
 
-  public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTask> {
+  public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTask, TryExecutor> {
 
     private final Optional<WorkflowPredicate> whenFilter;
     private final Optional<WorkflowPredicate> exceptFilter;
@@ -72,6 +75,8 @@ public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTas
     private final Optional<RetryExecutor> retryIntervalExecutor;
     private final Optional<WorkflowValueResolver<Duration>> attemptDuration;
     private final Optional<WorkflowValueResolver<Duration>> overallDuration;
+    private final FlowDirective catchDirective;
+    private Optional<TransitionInfoBuilder> catchTransitionBuilder;
     private String errorVariable;
 
     protected TryExecutorBuilder(
@@ -82,6 +87,7 @@ protected TryExecutorBuilder(
       this.errorFilter = buildErrorFilter(catchInfo.getErrors());
       this.whenFilter = WorkflowUtils.optionalPredicate(application, catchInfo.getWhen());
       this.exceptFilter = WorkflowUtils.optionalPredicate(application, catchInfo.getExceptWhen());
+      this.catchDirective = catchInfo.getThen();
       this.errorVariable = catchInfo.getAs();
       List<TaskItem> catchTaskDo = catchInfo.getDo();
       this.catchTaskExecutor =
@@ -99,6 +105,21 @@ protected TryExecutorBuilder(
           TaskExecutorHelper.createExecutorList(position, task.getTry(), definition, "try");
     }
 
+    @Override
+    public void connect(Map<String, TaskExecutorBuilder<?>> connections) {
+      super.connect(connections);
+      this.catchTransitionBuilder =
+          catchDirective != null
+              ? Optional.of(next(catchDirective, connections))
+              : Optional.empty();
+    }
+
+    @Override
+    protected void buildTransition(TryExecutor instance) {
+      super.buildTransition(instance);
+      instance.catchTransition = catchTransitionBuilder.map(TransitionInfo::build);
+    }
+
     private Optional<RetryPolicy> resolveRetryPolicy(Retry retry) {
       RetryPolicy retryPolicy = null;
       if (retry != null) {
@@ -242,16 +263,26 @@ private CompletableFuture<WorkflowModel> handleException(
                     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));
+            completable.thenCompose(
+                model -> {
+                  Optional<CompletableFuture<WorkflowModel>> retryCompletable =
+                      retryIntervalExecutor.orElseThrow().retry(workflow, taskContext, model);
+                  if (retryCompletable.isPresent()) {
+                    return retryCompletable
+                        .orElseThrow()
+                        .thenCompose(innerModel -> doIt(workflow, taskContext, innerModel));
+                  } else if (catchTransition.isPresent()) {
+                    taskContext.transition(catchTransition.orElseThrow());
+                    return CompletableFuture.completedFuture(model);
+                  } else {
+                    return CompletableFuture.failedFuture(exception);
+                  }
+                });
+      } else {
+        catchTransition.ifPresent(t -> taskContext.transition(t));
       }
       return completable;
     } else {
@@ -319,4 +350,11 @@ && compareString(errorFilter.getTitle(), error.title())
   private static boolean compareString(String one, String other) {
     return one == null || one.equals(other);
   }
+
+  @Override
+  protected CompletableFuture<TaskContext> execute(
+      WorkflowContext workflow, TaskContext taskContext) {
+    taskContext.transition(transition);
+    return internalExecute(workflow, taskContext).thenApply(node -> taskContext.rawOutput(node));
+  }
 }

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

@@ -32,7 +32,8 @@ public class WaitExecutor extends RegularTaskExecutor<WaitTask> {
 
   private final WorkflowValueResolver<Duration> durationResolver;
 
-  public static class WaitExecutorBuilder extends RegularTaskExecutorBuilder<WaitTask> {
+  public static class WaitExecutorBuilder
+      extends RegularTaskExecutorBuilder<WaitTask, WaitExecutor> {
     private WorkflowValueResolver<Duration> durationResolver;
 
     protected WaitExecutorBuilder(

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

@@ -232,6 +232,43 @@ void testAttemptDurationRetry() throws IOException {
     assertThat(retryListener.taskRetried.get("do/0/tryGetPet/try/0/getPet")).isEqualTo((short) 1);
   }
 
+  @Test
+  void testAttemptRetryThen() throws IOException {
+    apiServer.enqueue(new MockResponse().setResponseCode(404));
+    apiServer.enqueue(new MockResponse().setResponseCode(404));
+    apiServer.enqueue(
+        new MockResponse()
+            .setResponseCode(200)
+            .setHeader("Content-Type", "application/json")
+            .setBody(JsonUtils.mapper().writeValueAsString("{}")));
+    CompletableFuture<WorkflowModel> future =
+        app.workflowDefinition(
+                readWorkflowFromClasspath("workflows-samples/try-catch-retry-inline-then.yaml"))
+            .instance(Map.of())
+            .start();
+    assertThatThrownBy(() -> future.join()).hasCauseInstanceOf(WorkflowException.class);
+  }
+
+  @Test
+  void testAttemptRetryThenCatch() throws IOException {
+    apiServer.enqueue(new MockResponse().setResponseCode(404));
+    apiServer.enqueue(new MockResponse().setResponseCode(404));
+    apiServer.enqueue(new MockResponse().setResponseCode(404));
+    apiServer.enqueue(new MockResponse().setResponseCode(404));
+    apiServer.enqueue(new MockResponse().setResponseCode(404));
+    apiServer.enqueue(new MockResponse().setResponseCode(404));
+    assertThat(
+            app.workflowDefinition(
+                    readWorkflowFromClasspath("workflows-samples/try-catch-retry-inline-then.yaml"))
+                .instance(Map.of())
+                .start()
+                .join()
+                .asMap()
+                .map(m -> m.get("recovered"))
+                .orElseThrow())
+        .isEqualTo(true);
+  }
+
   @Test
   void testAttemptDurationOverall() throws IOException {
     String result = "{\"name\":\"Luna\"}";
@@ -280,6 +317,7 @@ void testTimeout() throws IOException {
   @ValueSource(
       strings = {
         "workflows-samples/try-catch-match-when.yaml",
+        "workflows-samples/try-catch-match-then.yaml",
         "workflows-samples/try-catch-match-status.yaml",
         "workflows-samples/try-catch-match-details.yaml"
       })

impl/test/src/test/resources/workflows-samples/try-catch-match-then.yaml

@@ -0,0 +1,24 @@
+document:
+  dsl: '1.0.0'
+  namespace: test
+  name: try-catch-match-then
+  version: '0.1.0'
+do:
+  - attemptTask:
+      try:
+          - failingTask:
+                  raise:
+                      error:
+                          type: https://example.com/errors/transient
+                          status: 503
+      catch:
+          when: ${ .status == 503 }
+          then: endGracefully
+  - failWontHappen:
+      raise:
+        error:
+          type: https://serverlessworkflow.io/errors/runtime
+          status: 500
+  - endGracefully:
+      set:
+        recovered: true

impl/test/src/test/resources/workflows-samples/try-catch-retry-inline-then.yaml

@@ -0,0 +1,37 @@
+document:
+  dsl: '1.0.0'
+  namespace: test
+  name: try-catch-retry-inline-then
+  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/communication
+            status: 404
+        then:
+            endGracefully
+        retry:
+          delay:
+            milliseconds: 10
+          backoff:
+            exponential: {}
+          limit:
+            attempt:
+              count: 5
+  - failWontHappen:
+          raise:
+            error:
+              type: https://serverlessworkflow.io/errors/runtime
+              status: 500
+  - endGracefully:
+           set:
+            recovered: true