From a5f06290959ada058dd0dbb5b5862914dfbd8301 Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Wed, 5 Aug 2026 18:14:18 -0300 Subject: [PATCH 1/8] Implement retry attempt.duration per-attempt timeout Enforce the retry limit's attempt.duration as a per-attempt timeout on the try block task execution, retrying with a timeout error when an individual attempt exceeds the configured duration. Signed-off-by: Matheus Cruz --- .../impl/executors/TryExecutor.java | 63 ++++++++++++++++--- .../impl/test/RetryTimeoutTest.java | 28 +++++++++ .../try-catch-retry-attempt-duration.yaml | 29 +++++++++ 3 files changed, 110 insertions(+), 10 deletions(-) create mode 100644 impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java index 38ebfa510..05be1c4a3 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java +++ b/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 { @@ -54,6 +58,7 @@ public class TryExecutor extends RegularTaskExecutor { private final TaskExecutor taskExecutor; private final Optional> catchTaskExecutor; private final Optional retryIntervalExecutor; + private final Optional> attemptDuration; private final String errorVariable; public static class TryExecutorBuilder extends RegularTaskExecutorBuilder { @@ -64,6 +69,7 @@ public static class TryExecutorBuilder extends RegularTaskExecutorBuilder taskExecutor; private final Optional> catchTaskExecutor; private final Optional retryIntervalExecutor; + private final Optional> attemptDuration; private String errorVariable; protected TryExecutorBuilder( @@ -83,27 +89,32 @@ protected TryExecutorBuilder( position.copy().addProperty("catch"), catchTaskDo, definition)) : Optional.empty(); Retry retry = catchInfo.getRetry(); - this.retryIntervalExecutor = retry != null ? buildRetryInterval(retry) : Optional.empty(); + RetryPolicy retryPolicy = retry != null ? resolveRetryPolicy(retry) : null; + this.retryIntervalExecutor = + retryPolicy != null ? Optional.of(buildRetryExecutor(retryPolicy)) : Optional.empty(); + this.attemptDuration = + retryPolicy != null ? resolveAttemptDuration(retryPolicy) : Optional.empty(); this.taskExecutor = TaskExecutorHelper.createExecutorList(position, task.getTry(), definition, "try"); } - private Optional buildRetryInterval(Retry retry) { - RetryPolicy retryPolicy = null; + private RetryPolicy resolveRetryPolicy(Retry retry) { if (retry.getRetryPolicyDefinition() != null) { - retryPolicy = retry.getRetryPolicyDefinition(); + return retry.getRetryPolicyDefinition(); } else if (retry.getRetryPolicyReference() != null) { - retryPolicy = + RetryPolicy retryPolicy = workflow .getUse() .getRetries() .getAdditionalProperties() .get(retry.getRetryPolicyReference()); if (retryPolicy == null) { - throw new IllegalStateException("Retry policy " + retryPolicy + " was not found"); + throw new IllegalStateException( + "Retry policy " + retry.getRetryPolicyReference() + " was not found"); } + return retryPolicy; } - return retryPolicy != null ? Optional.of(buildRetryExecutor(retryPolicy)) : Optional.empty(); + return null; } protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { @@ -114,6 +125,16 @@ protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { WorkflowUtils.optionalPredicate(application, retryPolicy.getExceptWhen())); } + private Optional> resolveAttemptDuration( + RetryPolicy retryPolicy) { + RetryLimit limit = retryPolicy.getLimit(); + if (limit != null && limit.getAttempt() != null && limit.getAttempt().getDuration() != null) { + return Optional.of( + WorkflowUtils.fromTimeoutAfter(application, limit.getAttempt().getDuration())); + } + return Optional.empty(); + } + private static int resolveMaxAttempts(RetryLimit limit) { return limit != null && limit.getAttempt() != null ? limit.getAttempt().getCount() @@ -152,6 +173,7 @@ protected TryExecutor(TryExecutorBuilder builder) { this.taskExecutor = builder.taskExecutor; this.catchTaskExecutor = builder.catchTaskExecutor; this.retryIntervalExecutor = builder.retryIntervalExecutor; + this.attemptDuration = builder.attemptDuration; this.errorVariable = builder.errorVariable; } @@ -164,9 +186,30 @@ protected CompletableFuture internalExecute( private CompletableFuture doIt( WorkflowContext workflow, TaskContext taskContext, WorkflowModel model) { retryIntervalExecutor.ifPresent(r -> r.init(workflow, taskContext, model)); - return TaskExecutorHelper.processTaskList( - taskExecutor, workflow, Optional.of(taskContext), model) - .exceptionallyCompose(e -> handleException(e, workflow, taskContext)); + CompletableFuture future = + TaskExecutorHelper.processTaskList(taskExecutor, workflow, Optional.of(taskContext), model); + if (attemptDuration.isPresent()) { + Duration timeout = attemptDuration.get().apply(workflow, taskContext, model); + if (!timeout.isZero()) { + future = + future + .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS) + .exceptionallyCompose( + e -> { + Throwable cause = e instanceof CompletionException ? e.getCause() : e; + if (cause instanceof TimeoutException) { + return CompletableFuture.failedFuture( + new WorkflowException( + WorkflowError.timeout() + .instance(taskContext.position().jsonPointer()) + .build(), + cause)); + } + return CompletableFuture.failedFuture(e); + }); + } + } + return future.exceptionallyCompose(e -> handleException(e, workflow, taskContext)); } private CompletableFuture handleException( diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java index 47898829c..37ad9946a 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java +++ b/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,33 @@ void testRetryEnd() throws IOException { .hasCauseInstanceOf(WorkflowException.class); } + @Test + void testAttemptDuration() throws IOException { + final JsonNode result = JsonUtils.mapper().createObjectNode().put("name", "Matheus"); + apiServer.enqueue( + new MockResponse() + .setHeadersDelay(2, TimeUnit.SECONDS) + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(JsonUtils.mapper().writeValueAsString(result))); + apiServer.enqueue( + new MockResponse() + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(JsonUtils.mapper().writeValueAsString(result))); + CompletableFuture future = + app.workflowDefinition( + readWorkflowFromClasspath( + "workflows-samples/try-catch-retry-attempt-duration.yaml")) + .instance(Map.of()) + .start(); + Awaitility.await() + .atMost(Duration.ofSeconds(5)) + .until(() -> future.join().as(JsonNode.class).orElseThrow().equals(result)); + assertThat(retryListener.taskRetried).hasSize(1); + assertThat(retryListener.taskRetried.get("do/0/tryGetPet/try/0/getPet")).isEqualTo((short) 1); + } + @Test void testTimeout() throws IOException { Map result = diff --git a/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml b/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml new file mode 100644 index 000000000..c8b7bb008 --- /dev/null +++ b/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml @@ -0,0 +1,29 @@ +document: + dsl: '1.0.0' + namespace: test + name: try-catch-retry-attempt-duration + 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: + attempt: + count: 5 + duration: + milliseconds: 50 From ae056db4c2ef0eb96b00c44f3f4e6d01c7bdab3d Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Wed, 5 Aug 2026 18:30:11 -0300 Subject: [PATCH 2/8] I am not a pet Signed-off-by: Matheus Cruz --- .../java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java index 37ad9946a..2580b25a6 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java @@ -189,7 +189,7 @@ void testRetryEnd() throws IOException { @Test void testAttemptDuration() throws IOException { - final JsonNode result = JsonUtils.mapper().createObjectNode().put("name", "Matheus"); + final JsonNode result = JsonUtils.mapper().createObjectNode().put("name", "Luna"); apiServer.enqueue( new MockResponse() .setHeadersDelay(2, TimeUnit.SECONDS) From 42f8808c2947827644a8bb7ef7036fd0aa096901 Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Wed, 5 Aug 2026 18:37:38 -0300 Subject: [PATCH 3/8] Apply copilot suggestions Signed-off-by: Matheus Cruz --- .../io/serverlessworkflow/impl/executors/TryExecutor.java | 6 +++--- .../io/serverlessworkflow/impl/test/RetryTimeoutTest.java | 5 ++--- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java index 05be1c4a3..baac226aa 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java @@ -189,11 +189,11 @@ private CompletableFuture doIt( CompletableFuture future = TaskExecutorHelper.processTaskList(taskExecutor, workflow, Optional.of(taskContext), model); if (attemptDuration.isPresent()) { - Duration timeout = attemptDuration.get().apply(workflow, taskContext, model); - if (!timeout.isZero()) { + long timeoutMillis = attemptDuration.get().apply(workflow, taskContext, model).toMillis(); + if (timeoutMillis > 0) { future = future - .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS) + .orTimeout(timeoutMillis, TimeUnit.MILLISECONDS) .exceptionallyCompose( e -> { Throwable cause = e instanceof CompletionException ? e.getCause() : e; diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java index 2580b25a6..5c5ffddd5 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java @@ -207,9 +207,8 @@ void testAttemptDuration() throws IOException { "workflows-samples/try-catch-retry-attempt-duration.yaml")) .instance(Map.of()) .start(); - Awaitility.await() - .atMost(Duration.ofSeconds(5)) - .until(() -> future.join().as(JsonNode.class).orElseThrow().equals(result)); + Awaitility.await().atMost(Duration.ofSeconds(5)).until(future::isDone); + assertThat(future.join().as(JsonNode.class).orElseThrow()).isEqualTo(result); assertThat(retryListener.taskRetried).hasSize(1); assertThat(retryListener.taskRetried.get("do/0/tryGetPet/try/0/getPet")).isEqualTo((short) 1); } From 56339b2bce40b8120384c77067604c4960db5b4a Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Thu, 6 Aug 2026 19:45:07 -0300 Subject: [PATCH 4/8] Apply pull request suggestions Signed-off-by: Matheus Cruz --- .../impl/executors/TryExecutor.java | 70 ++++++++++--------- .../impl/test/RetryTimeoutTest.java | 29 ++++---- .../try-catch-retry-attempt-duration.yaml | 6 +- 3 files changed, 51 insertions(+), 54 deletions(-) diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java index baac226aa..937225d1e 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java @@ -89,18 +89,17 @@ protected TryExecutorBuilder( position.copy().addProperty("catch"), catchTaskDo, definition)) : Optional.empty(); Retry retry = catchInfo.getRetry(); - RetryPolicy retryPolicy = retry != null ? resolveRetryPolicy(retry) : null; - this.retryIntervalExecutor = - retryPolicy != null ? Optional.of(buildRetryExecutor(retryPolicy)) : Optional.empty(); - this.attemptDuration = - retryPolicy != null ? resolveAttemptDuration(retryPolicy) : Optional.empty(); + Optional retryPolicy = + retry != null ? resolveRetryPolicy(retry) : Optional.empty(); + this.retryIntervalExecutor = retryPolicy.map(this::buildRetryExecutor); + this.attemptDuration = retryPolicy.flatMap(this::resolveAttemptDuration); this.taskExecutor = TaskExecutorHelper.createExecutorList(position, task.getTry(), definition, "try"); } - private RetryPolicy resolveRetryPolicy(Retry retry) { + private Optional resolveRetryPolicy(Retry retry) { if (retry.getRetryPolicyDefinition() != null) { - return retry.getRetryPolicyDefinition(); + return Optional.of(retry.getRetryPolicyDefinition()); } else if (retry.getRetryPolicyReference() != null) { RetryPolicy retryPolicy = workflow @@ -112,9 +111,9 @@ private RetryPolicy resolveRetryPolicy(Retry retry) { throw new IllegalStateException( "Retry policy " + retry.getRetryPolicyReference() + " was not found"); } - return retryPolicy; + return Optional.of(retryPolicy); } - return null; + return Optional.empty(); } protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { @@ -128,11 +127,10 @@ protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { private Optional> resolveAttemptDuration( RetryPolicy retryPolicy) { RetryLimit limit = retryPolicy.getLimit(); - if (limit != null && limit.getAttempt() != null && limit.getAttempt().getDuration() != null) { - return Optional.of( - WorkflowUtils.fromTimeoutAfter(application, limit.getAttempt().getDuration())); - } - return Optional.empty(); + return limit != null && limit.getAttempt() != null && limit.getAttempt().getDuration() != null + ? Optional.of( + WorkflowUtils.fromTimeoutAfter(application, limit.getAttempt().getDuration())) + : Optional.empty(); } private static int resolveMaxAttempts(RetryLimit limit) { @@ -188,30 +186,34 @@ private CompletableFuture doIt( retryIntervalExecutor.ifPresent(r -> r.init(workflow, taskContext, model)); CompletableFuture future = TaskExecutorHelper.processTaskList(taskExecutor, workflow, Optional.of(taskContext), model); - if (attemptDuration.isPresent()) { - long timeoutMillis = attemptDuration.get().apply(workflow, taskContext, model).toMillis(); - if (timeoutMillis > 0) { - future = - future - .orTimeout(timeoutMillis, TimeUnit.MILLISECONDS) - .exceptionallyCompose( - e -> { - Throwable cause = e instanceof CompletionException ? e.getCause() : e; - if (cause instanceof TimeoutException) { - return CompletableFuture.failedFuture( - new WorkflowException( - WorkflowError.timeout() - .instance(taskContext.position().jsonPointer()) - .build(), - cause)); - } - return CompletableFuture.failedFuture(e); - }); - } + long timeoutMillis = + attemptDuration + .map(d -> d.apply(workflow, taskContext, model)) + .orElse(Duration.ZERO) + .toMillis(); + if (timeoutMillis > 0) { + future = + future + .orTimeout(timeoutMillis, TimeUnit.MILLISECONDS) + .exceptionallyCompose(e -> handleTimeoutException(e, taskContext)); } return future.exceptionallyCompose(e -> handleException(e, workflow, taskContext)); } + private CompletableFuture handleTimeoutException( + Throwable e, TaskContext taskContext) { + Throwable cause = e instanceof CompletionException ? e.getCause() : e; + return CompletableFuture.failedFuture( + cause instanceof TimeoutException + ? new WorkflowException( + WorkflowError.timeout() + .instance(taskContext.position().jsonPointer()) + .title(cause.getMessage()) + .build(), + cause) + : e); + } + private CompletableFuture handleException( Throwable e, WorkflowContext workflow, TaskContext taskContext) { if (e instanceof CompletionException) { diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java index 5c5ffddd5..3b4817ece 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java @@ -189,28 +189,23 @@ void testRetryEnd() throws IOException { @Test void testAttemptDuration() throws IOException { - final JsonNode result = JsonUtils.mapper().createObjectNode().put("name", "Luna"); apiServer.enqueue( new MockResponse() .setHeadersDelay(2, TimeUnit.SECONDS) .setResponseCode(200) .setHeader("Content-Type", "application/json") - .setBody(JsonUtils.mapper().writeValueAsString(result))); - apiServer.enqueue( - new MockResponse() - .setResponseCode(200) - .setHeader("Content-Type", "application/json") - .setBody(JsonUtils.mapper().writeValueAsString(result))); - CompletableFuture future = - app.workflowDefinition( - readWorkflowFromClasspath( - "workflows-samples/try-catch-retry-attempt-duration.yaml")) - .instance(Map.of()) - .start(); - Awaitility.await().atMost(Duration.ofSeconds(5)).until(future::isDone); - assertThat(future.join().as(JsonNode.class).orElseThrow()).isEqualTo(result); - assertThat(retryListener.taskRetried).hasSize(1); - assertThat(retryListener.taskRetried.get("do/0/tryGetPet/try/0/getPet")).isEqualTo((short) 1); + .setBody("{}")); + assertThatThrownBy( + () -> + app.workflowDefinition( + readWorkflowFromClasspath( + "workflows-samples/try-catch-retry-attempt-duration.yaml")) + .instance(Map.of()) + .start() + .join()) + .hasCauseInstanceOf(WorkflowException.class) + .cause() + .hasMessageContaining("timeout"); } @Test diff --git a/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml b/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml index c8b7bb008..97d1e1cd5 100644 --- a/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml +++ b/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml @@ -15,8 +15,8 @@ do: catch: errors: with: - type: https://serverlessworkflow.io/spec/1.0.0/errors/timeout - status: 408 + type: https://serverlessworkflow.io/spec/1.0.0/errors/communication + status: 404 retry: delay: milliseconds: 10 @@ -26,4 +26,4 @@ do: attempt: count: 5 duration: - milliseconds: 50 + milliseconds: 50 \ No newline at end of file From 6aa9032cefafba70ecd34690b99a5b9ec95519eb Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Thu, 6 Aug 2026 19:59:12 -0300 Subject: [PATCH 5/8] Apply pull request suggestions Signed-off-by: Matheus Cruz --- .../io/serverlessworkflow/impl/test/RetryTimeoutTest.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java index 3b4817ece..9288a0e8f 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java @@ -203,9 +203,7 @@ void testAttemptDuration() throws IOException { .instance(Map.of()) .start() .join()) - .hasCauseInstanceOf(WorkflowException.class) - .cause() - .hasMessageContaining("timeout"); + .hasCauseInstanceOf(WorkflowException.class); } @Test From 313bda78d397daa4e6a176f44af468b669f5d7bb Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Thu, 6 Aug 2026 20:59:56 -0300 Subject: [PATCH 6/8] Apply pull request suggestions Signed-off-by: Matheus Cruz --- .../impl/executors/TryExecutor.java | 28 +++++++++++++----- .../impl/test/RetryTimeoutTest.java | 26 +++++++++++++++++ ...ry-catch-retry-attempt-duration-retry.yaml | 29 +++++++++++++++++++ 3 files changed, 76 insertions(+), 7 deletions(-) create mode 100644 impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration-retry.yaml diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java index 937225d1e..cd2b06080 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java @@ -184,22 +184,36 @@ protected CompletableFuture internalExecute( private CompletableFuture doIt( WorkflowContext workflow, TaskContext taskContext, WorkflowModel model) { retryIntervalExecutor.ifPresent(r -> r.init(workflow, taskContext, model)); - CompletableFuture future = + CompletableFuture taskFuture = TaskExecutorHelper.processTaskList(taskExecutor, workflow, Optional.of(taskContext), model); long timeoutMillis = attemptDuration .map(d -> d.apply(workflow, taskContext, model)) .orElse(Duration.ZERO) .toMillis(); - if (timeoutMillis > 0) { - future = - future - .orTimeout(timeoutMillis, TimeUnit.MILLISECONDS) - .exceptionallyCompose(e -> handleTimeoutException(e, taskContext)); - } + CompletableFuture future = + timeoutMillis > 0 + ? withAttemptTimeout(taskFuture, timeoutMillis, taskContext) + : taskFuture; return future.exceptionallyCompose(e -> handleException(e, workflow, taskContext)); } + private CompletableFuture withAttemptTimeout( + CompletableFuture taskFuture, long timeoutMillis, TaskContext taskContext) { + CompletableFuture timeoutFuture = new CompletableFuture<>(); + taskFuture.whenComplete( + (result, error) -> { + if (error != null) { + timeoutFuture.completeExceptionally(error); + } else { + timeoutFuture.complete(result); + } + }); + timeoutFuture.orTimeout(timeoutMillis, TimeUnit.MILLISECONDS); + timeoutFuture.whenComplete((r, e) -> taskFuture.cancel(true)); + return timeoutFuture.exceptionallyCompose(e -> handleTimeoutException(e, taskContext)); + } + private CompletableFuture handleTimeoutException( Throwable e, TaskContext taskContext) { Throwable cause = e instanceof CompletionException ? e.getCause() : e; diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java index 9288a0e8f..a67eed794 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java @@ -206,6 +206,32 @@ void testAttemptDuration() throws IOException { .hasCauseInstanceOf(WorkflowException.class); } + @Test + void testAttemptDurationRetry() throws IOException { + final JsonNode result = JsonUtils.mapper().createObjectNode().put("name", "Luna"); + apiServer.enqueue( + new MockResponse() + .setHeadersDelay(2, TimeUnit.SECONDS) + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(JsonUtils.mapper().writeValueAsString(result))); + apiServer.enqueue( + new MockResponse() + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(JsonUtils.mapper().writeValueAsString(result))); + CompletableFuture 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(JsonNode.class).orElseThrow()).isEqualTo(result); + assertThat(retryListener.taskRetried).hasSize(1); + assertThat(retryListener.taskRetried.get("do/0/tryGetPet/try/0/getPet")).isEqualTo((short) 1); + } + @Test void testTimeout() throws IOException { Map result = diff --git a/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration-retry.yaml b/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration-retry.yaml new file mode 100644 index 000000000..3973e9ecb --- /dev/null +++ b/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration-retry.yaml @@ -0,0 +1,29 @@ +document: + dsl: '1.0.0' + namespace: test + name: try-catch-retry-attempt-duration-retry + 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: + attempt: + count: 5 + duration: + milliseconds: 50 \ No newline at end of file From 54d0f45492353296b89ecc1427dbfcebdb37d6f8 Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Thu, 6 Aug 2026 21:07:53 -0300 Subject: [PATCH 7/8] Apply spotless Signed-off-by: Matheus Cruz --- .../io/serverlessworkflow/impl/executors/TryExecutor.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java index cd2b06080..884b28a05 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java @@ -192,9 +192,7 @@ private CompletableFuture doIt( .orElse(Duration.ZERO) .toMillis(); CompletableFuture future = - timeoutMillis > 0 - ? withAttemptTimeout(taskFuture, timeoutMillis, taskContext) - : taskFuture; + timeoutMillis > 0 ? withAttemptTimeout(taskFuture, timeoutMillis, taskContext) : taskFuture; return future.exceptionallyCompose(e -> handleException(e, workflow, taskContext)); } From ae4feeaeff31dc876d9eb936e5601f4f2995a5f6 Mon Sep 17 00:00:00 2001 From: Francisco Javier Tirado Sarti Date: Fri, 7 Aug 2026 10:38:35 +0200 Subject: [PATCH 8/8] [Fix #1526] fjtirado commments Signed-off-by: Francisco Javier Tirado Sarti [Fix #1526] Implementing overall retry timeout Signed-off-by: Francisco Javier Tirado Sarti --- .../impl/executors/TryExecutor.java | 210 ++++++++++-------- .../impl/test/RetryTimeoutTest.java | 39 +++- ...-catch-retry-attempt-duration-overall.yaml | 31 +++ 3 files changed, 187 insertions(+), 93 deletions(-) create mode 100644 impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration-overall.yaml diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java index 884b28a05..7495c951a 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java @@ -59,6 +59,7 @@ public class TryExecutor extends RegularTaskExecutor { private final Optional> catchTaskExecutor; private final Optional retryIntervalExecutor; private final Optional> attemptDuration; + private final Optional> overallDuration; private final String errorVariable; public static class TryExecutorBuilder extends RegularTaskExecutorBuilder { @@ -70,6 +71,7 @@ public static class TryExecutorBuilder extends RegularTaskExecutorBuilder> catchTaskExecutor; private final Optional retryIntervalExecutor; private final Optional> attemptDuration; + private final Optional> overallDuration; private String errorVariable; protected TryExecutorBuilder( @@ -89,31 +91,33 @@ protected TryExecutorBuilder( position.copy().addProperty("catch"), catchTaskDo, definition)) : Optional.empty(); Retry retry = catchInfo.getRetry(); - Optional retryPolicy = - retry != null ? resolveRetryPolicy(retry) : Optional.empty(); + Optional 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 resolveRetryPolicy(Retry retry) { - if (retry.getRetryPolicyDefinition() != null) { - return Optional.of(retry.getRetryPolicyDefinition()); - } else if (retry.getRetryPolicyReference() != null) { - RetryPolicy retryPolicy = - workflow - .getUse() - .getRetries() - .getAdditionalProperties() - .get(retry.getRetryPolicyReference()); - if (retryPolicy == null) { - throw new IllegalStateException( - "Retry policy " + retry.getRetryPolicyReference() + " was not found"); + RetryPolicy retryPolicy = null; + 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 Optional.of(retryPolicy); } - return Optional.empty(); + return Optional.ofNullable(retryPolicy); } protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { @@ -133,6 +137,14 @@ private Optional> resolveAttemptDuration( : Optional.empty(); } + private Optional> 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() @@ -173,12 +185,17 @@ protected TryExecutor(TryExecutorBuilder builder) { this.retryIntervalExecutor = builder.retryIntervalExecutor; this.attemptDuration = builder.attemptDuration; this.errorVariable = builder.errorVariable; + this.overallDuration = builder.overallDuration; } @Override protected CompletableFuture 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 doIt( @@ -186,88 +203,103 @@ private CompletableFuture doIt( retryIntervalExecutor.ifPresent(r -> r.init(workflow, taskContext, model)); CompletableFuture taskFuture = TaskExecutorHelper.processTaskList(taskExecutor, workflow, Optional.of(taskContext), model); - long timeoutMillis = - attemptDuration - .map(d -> d.apply(workflow, taskContext, model)) - .orElse(Duration.ZERO) - .toMillis(); - CompletableFuture future = - timeoutMillis > 0 ? withAttemptTimeout(taskFuture, timeoutMillis, taskContext) : taskFuture; - return future.exceptionallyCompose(e -> handleException(e, workflow, taskContext)); + return cancellingFutureTimeout(taskFuture, attemptDuration, workflow, taskContext, model) + .exceptionallyCompose(e -> handleException(e, workflow, taskContext)); } - private CompletableFuture withAttemptTimeout( - CompletableFuture taskFuture, long timeoutMillis, TaskContext taskContext) { - CompletableFuture timeoutFuture = new CompletableFuture<>(); - taskFuture.whenComplete( - (result, error) -> { - if (error != null) { - timeoutFuture.completeExceptionally(error); - } else { - timeoutFuture.complete(result); - } - }); - timeoutFuture.orTimeout(timeoutMillis, TimeUnit.MILLISECONDS); - timeoutFuture.whenComplete((r, e) -> taskFuture.cancel(true)); - return timeoutFuture.exceptionallyCompose(e -> handleTimeoutException(e, taskContext)); - } - - private CompletableFuture handleTimeoutException( - Throwable e, TaskContext taskContext) { + private CompletableFuture handleException( + Throwable e, WorkflowContext workflow, TaskContext taskContext) { Throwable cause = e instanceof CompletionException ? e.getCause() : e; - return CompletableFuture.failedFuture( - cause instanceof TimeoutException - ? new WorkflowException( - WorkflowError.timeout() - .instance(taskContext.position().jsonPointer()) - .title(cause.getMessage()) - .build(), - cause) - : 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); + } } private CompletableFuture handleException( - Throwable e, WorkflowContext workflow, TaskContext taskContext) { - if (e instanceof CompletionException) { - return handleException(e.getCause(), workflow, taskContext); - } - if (e instanceof WorkflowException) { - WorkflowException exception = (WorkflowException) e; + 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 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 cancellingFutureTimeout( + CompletableFuture originalFuture, + Optional> 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 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> buildErrorFilter(CatchErrors errors) { diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java index a67eed794..81c8866ce 100644 --- a/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java @@ -208,18 +208,18 @@ void testAttemptDuration() throws IOException { @Test void testAttemptDurationRetry() throws IOException { - final JsonNode result = JsonUtils.mapper().createObjectNode().put("name", "Luna"); + String result = "{\"name\":\"Luna\"}"; apiServer.enqueue( new MockResponse() .setHeadersDelay(2, TimeUnit.SECONDS) .setResponseCode(200) .setHeader("Content-Type", "application/json") - .setBody(JsonUtils.mapper().writeValueAsString(result))); + .setBody(result)); apiServer.enqueue( new MockResponse() .setResponseCode(200) .setHeader("Content-Type", "application/json") - .setBody(JsonUtils.mapper().writeValueAsString(result))); + .setBody(result)); CompletableFuture future = app.workflowDefinition( readWorkflowFromClasspath( @@ -227,11 +227,42 @@ void testAttemptDurationRetry() throws IOException { .instance(Map.of()) .start(); Awaitility.await().atMost(Duration.ofSeconds(5)).until(future::isDone); - assertThat(future.join().as(JsonNode.class).orElseThrow()).isEqualTo(result); + 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 result = diff --git a/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration-overall.yaml b/impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration-overall.yaml new file mode 100644 index 000000000..1d7a0cdf3 --- /dev/null +++ b/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 \ No newline at end of file