Implement retry attempt.duration per-attempt timeout - #1600
Open
mcruzdev wants to merge 7 commits into
Open
Conversation
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 <matheuscruz.dev@gmail.com>
Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
This PR adds support for a per-retry-attempt timeout (limit.attempt.duration) in try/catch retry handling, and includes a workflow sample plus a regression test to validate the behavior.
Changes:
- Add
attempt.durationparsing/resolution from retry policy and enforce it viaCompletableFuture.orTimeout(...)duringtryexecution. - Add a new workflow sample exercising
attempt.durationin atry/catch/retryblock. - Add a unit test verifying that a timed-out attempt retries and eventually succeeds.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java | Implements per-attempt timeout handling for retry attempts in TryExecutor. |
| impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java | Adds a test that simulates a slow HTTP response to trigger attempt.duration timeout and verify retry behavior. |
| impl/test/src/test/resources/workflows-samples/try-catch-retry-attempt-duration.yaml | Adds a workflow sample defining limit.attempt.duration under retry policy. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+191
to
+211
| 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); | ||
| }); | ||
| } | ||
| } |
Comment on lines
+210
to
+212
| Awaitility.await() | ||
| .atMost(Duration.ofSeconds(5)) | ||
| .until(() -> future.join().as(JsonNode.class).orElseThrow().equals(result)); |
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.
Suppressed comments (1)
impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java:196
attempt.durationusesorTimeout(timeout.toMillis(), MILLISECONDS)guarded only by!timeout.isZero(). For positive sub-millisecond durations (possible viaDuration.parsein expression/literal),timeout.toMillis()becomes 0 andorTimeout(0, …)throwsIllegalArgumentException, bypassing the intended timeout/retry behavior. Also, negative durations currently fall through toorTimeoutand would throw as well. Guard ontimeout > 0and ensure the millis value is at least 1 when applying the timeout.
if (!timeout.isZero()) {
future =
future
.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS)
Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.
Suppressed comments (1)
impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java:110
- When resolving a retry policy by reference, this code can throw a NullPointerException if the workflow omits the
usesection oruse.retries. Other parts of the codebase (e.g., WorkflowUtils.getTaskTimeout) use explicit null checks with a descriptive message; doing the same here would make misconfigurations much easier to diagnose.
} else if (retry.getRetryPolicyReference() != null) {
RetryPolicy retryPolicy =
workflow
.getUse()
.getRetries()
.getAdditionalProperties()
.get(retry.getRetryPolicyReference());
fjtirado
reviewed
Aug 6, 2026
Comment on lines
101
to
118
| 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; | ||
| } |
Collaborator
There was a problem hiding this comment.
Ok, now I get it, but it not good, in my opinion.
I think this should still return an optional and then use the optional to build the two other optional (map method is your friend) rather than starting a null check party in the caller.
fjtirado
reviewed
Aug 6, 2026
fjtirado
left a comment
fjtirado
left a comment
Collaborator
There was a problem hiding this comment.
Refactor using optional rather than null
Comment on lines
101
to
118
| 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; | ||
| } |
Collaborator
There was a problem hiding this comment.
Ok, now I get it, but it not good, in my opinion.
I think this should still return an optional and then use the optional to build the two other optional (map method is your friend) rather than starting a null check party in the caller.
fjtirado
reviewed
Aug 6, 2026
Comment on lines
+131
to
+135
| if (limit != null && limit.getAttempt() != null && limit.getAttempt().getDuration() != null) { | ||
| return Optional.of( | ||
| WorkflowUtils.fromTimeoutAfter(application, limit.getAttempt().getDuration())); | ||
| } | ||
| return Optional.empty(); |
Collaborator
There was a problem hiding this comment.
Suggested change
| 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 ? | |
| return Optional.of( | |
| WorkflowUtils.fromTimeoutAfter(application, limit.getAttempt().getDuration())) : | |
| return Optional.empty(); |
fjtirado
reviewed
Aug 6, 2026
Comment on lines
+92
to
+96
| 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(); |
Collaborator
There was a problem hiding this comment.
This you should refactor using Optional.map when resolveRetryPolicy return optional
fjtirado
reviewed
Aug 6, 2026
| if (retryPolicy == null) { | ||
| throw new IllegalStateException("Retry policy " + retryPolicy + " was not found"); | ||
| throw new IllegalStateException( | ||
| "Retry policy " + retry.getRetryPolicyReference() + " was not found"); |
fjtirado
reviewed
Aug 6, 2026
Comment on lines
+197
to
+209
| .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); | ||
| }); |
Collaborator
There was a problem hiding this comment.
Lets split this logic in two blocks
Suggested change
| .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); | |
| }); | |
| .exceptionallyCompose(this::handleTimeoutException); |
and add
private CompletableFuture<?> handleTimeoutExcepction (Throwable e) {
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): cause);
}
fjtirado
reviewed
Aug 6, 2026
Comment on lines
+191
to
+193
| if (attemptDuration.isPresent()) { | ||
| long timeoutMillis = attemptDuration.get().apply(workflow, taskContext, model).toMillis(); | ||
| if (timeoutMillis > 0) { |
Collaborator
There was a problem hiding this comment.
The double if can be avoided
Suggested change
| if (attemptDuration.isPresent()) { | |
| long timeoutMillis = attemptDuration.get().apply(workflow, taskContext, model).toMillis(); | |
| if (timeoutMillis > 0) { | |
| long timeoutMillis = attemptDuration.map(d -> d.apply(workflow,taskContext,model).orElse(Duration.ZERO).toMillis() | |
| if ( timeoutMillis>0) { |
fjtirado
reviewed
Aug 6, 2026
Comment on lines
+191
to
+214
| 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<WorkflowModel> 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); | ||
| } |
Collaborator
There was a problem hiding this comment.
I do not think is a good idea to catch retry timeout in the test worflow definition, I think we need a test that fails with retry timeout, proving that the task timed out after the specified delay
fjtirado
reviewed
Aug 6, 2026
Comment on lines
+16
to
+19
| errors: | ||
| with: | ||
| type: https://serverlessworkflow.io/spec/1.0.0/errors/timeout | ||
| status: 408 |
Collaborator
There was a problem hiding this comment.
I think the exception being catch here should be different than timeout to not mistake with the timeout that should be launch is the max retry duration is exceeded.
So use a different one to trigger the retry and to make sure that the overall timeout has worked search for that exception in the unit test
Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>
Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.
Suppressed comments (2)
impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java:214
- In the non-timeout path, this returns
failedFuture(e)even thoughcausehas already been unwrapped fromCompletionException. That can re-wrap non-timeout exceptions and may interfere with later handling (e.g.,handleExceptionseeing aCompletionExceptioninstead of the original cause). Returncause(or rethrowcause) in the: ...branch to preserve the original exception.
private CompletableFuture<WorkflowModel> 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);
impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java:197
- Using a 2-second delayed response in a unit test can unnecessarily slow the suite and can also extend teardown if the server is still serving the delayed response. Since the workflow timeout in the sample is 50ms, consider reducing the delay to a smaller value that is still comfortably above the configured attempt timeout (e.g., a few hundred milliseconds) to keep the test fast and less flaky.
apiServer.enqueue(
new MockResponse()
.setHeadersDelay(2, TimeUnit.SECONDS)
.setResponseCode(200)
.setHeader("Content-Type", "application/json")
.setBody("{}"));
Comment on lines
+187
to
+200
| CompletableFuture<WorkflowModel> future = | ||
| 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)); | ||
| } | ||
| return future.exceptionallyCompose(e -> handleException(e, workflow, taskContext)); |
Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (3)
impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java:228
- In handleTimeoutException(), the non-timeout branch returns the original throwable
einstead of the unwrappedcause. This can leak an extra CompletionException wrapper and interfere with downstream error handling that expects the root cause.
: e);
impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java:109
- resolveRetryPolicy() can throw a NullPointerException when a retry policy reference is used but
workflow.useorworkflow.use.retriesis not defined. This should fail fast with a clear IllegalStateException, consistent with other reference-resolution code (e.g., WorkflowUtils.getTaskTimeout).
RetryPolicy retryPolicy =
workflow
.getUse()
.getRetries()
.getAdditionalProperties()
impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java:206
- testAttemptDuration() only asserts that a WorkflowException is thrown, but doesn’t verify it is specifically the new attempt-duration timeout behavior. Adding an assertion on the WorkflowError status/type would make this test catch regressions where the failure is caused by something else.
.hasCauseInstanceOf(WorkflowException.class);
Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (2)
impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java:224
- Timeout errors created here set the WorkflowError title from the underlying TimeoutException message. JDK TimeoutException messages are typically null, and other timeout paths in this codebase (e.g., AbstractTaskExecutor) don't set a title for timeout errors, so this creates inconsistent error payloads.
WorkflowError.timeout()
.instance(taskContext.position().jsonPointer())
.title(cause.getMessage())
.build(),
impl/test/src/test/java/io/serverlessworkflow/impl/test/RetryTimeoutTest.java:206
- This test only asserts that a WorkflowException occurred, but it doesn't assert that the failure is specifically the expected timeout (status 408). Strengthening the assertion makes the test validate the new attempt.duration behavior more precisely.
.hasCauseInstanceOf(WorkflowException.class);
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Add support for
attempt.durationper attempt timeout.Closes #1526
Many thanks for submitting your Pull Request ❤️!
What this PR does / why we need it:
Special notes for reviewers:
Additional information (if needed):