Skip to content

Commit f92c103

Browse files
mcruzdevfjtirado
mcruzdev
and
fjtirado
authored
Implement retry attempt.duration per-attempt timeout (#1600)
* 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 <matheuscruz.dev@gmail.com> * I am not a pet Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com> * Apply copilot suggestions Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com> * Apply pull request suggestions Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com> * Apply pull request suggestions Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com> * Apply pull request suggestions Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com> * Apply spotless Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com> * [Fix #1526] fjtirado commments Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> [Fix #1526] Implementing overall retry timeout Signed-off-by: Francisco Javier Tirado Sarti <ftrados@ibm.com> --------- Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com> Signed-off-by: Francisco Javier Tirado Sarti <ftrados@ibm.com> Co-authored-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>
1 parent 3066bc7 commit f92c103

5 files changed

Lines changed: 307 additions & 52 deletions

File tree

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

Lines changed: 141 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -33,17 +33,21 @@
3333
import io.serverlessworkflow.impl.WorkflowMutablePosition;
3434
import io.serverlessworkflow.impl.WorkflowPredicate;
3535
import io.serverlessworkflow.impl.WorkflowUtils;
36+
import io.serverlessworkflow.impl.WorkflowValueResolver;
3637
import io.serverlessworkflow.impl.executors.retry.ConstantRetryIntervalFunction;
3738
import io.serverlessworkflow.impl.executors.retry.DefaultRetryExecutor;
3839
import io.serverlessworkflow.impl.executors.retry.ExponentialRetryIntervalFunction;
3940
import io.serverlessworkflow.impl.executors.retry.LinearRetryIntervalFunction;
4041
import io.serverlessworkflow.impl.executors.retry.RetryExecutor;
4142
import io.serverlessworkflow.impl.executors.retry.RetryIntervalFunction;
43+
import java.time.Duration;
4244
import java.util.List;
4345
import java.util.Objects;
4446
import java.util.Optional;
4547
import java.util.concurrent.CompletableFuture;
4648
import java.util.concurrent.CompletionException;
49+
import java.util.concurrent.TimeUnit;
50+
import java.util.concurrent.TimeoutException;
4751
import java.util.function.Predicate;
4852

4953
public class TryExecutor extends RegularTaskExecutor<TryTask> {
@@ -54,6 +58,8 @@ public class TryExecutor extends RegularTaskExecutor<TryTask> {
5458
private final TaskExecutor<?> taskExecutor;
5559
private final Optional<TaskExecutor<?>> catchTaskExecutor;
5660
private final Optional<RetryExecutor> retryIntervalExecutor;
61+
private final Optional<WorkflowValueResolver<Duration>> attemptDuration;
62+
private final Optional<WorkflowValueResolver<Duration>> overallDuration;
5763
private final String errorVariable;
5864

5965
public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTask> {
@@ -64,6 +70,8 @@ public static class TryExecutorBuilder extends RegularTaskExecutorBuilder<TryTas
6470
private final TaskExecutor<?> taskExecutor;
6571
private final Optional<TaskExecutor<?>> catchTaskExecutor;
6672
private final Optional<RetryExecutor> retryIntervalExecutor;
73+
private final Optional<WorkflowValueResolver<Duration>> attemptDuration;
74+
private final Optional<WorkflowValueResolver<Duration>> overallDuration;
6775
private String errorVariable;
6876

6977
protected TryExecutorBuilder(
@@ -83,27 +91,33 @@ protected TryExecutorBuilder(
8391
position.copy().addProperty("catch"), catchTaskDo, definition))
8492
: Optional.empty();
8593
Retry retry = catchInfo.getRetry();
86-
this.retryIntervalExecutor = retry != null ? buildRetryInterval(retry) : Optional.empty();
94+
Optional<RetryPolicy> retryPolicy = resolveRetryPolicy(retry);
95+
this.retryIntervalExecutor = retryPolicy.map(this::buildRetryExecutor);
96+
this.attemptDuration = retryPolicy.flatMap(this::resolveAttemptDuration);
97+
this.overallDuration = retryPolicy.flatMap(this::resolveOverallDuration);
8798
this.taskExecutor =
8899
TaskExecutorHelper.createExecutorList(position, task.getTry(), definition, "try");
89100
}
90101

91-
private Optional<RetryExecutor> buildRetryInterval(Retry retry) {
102+
private Optional<RetryPolicy> resolveRetryPolicy(Retry retry) {
92103
RetryPolicy retryPolicy = null;
93-
if (retry.getRetryPolicyDefinition() != null) {
94-
retryPolicy = retry.getRetryPolicyDefinition();
95-
} else if (retry.getRetryPolicyReference() != null) {
96-
retryPolicy =
97-
workflow
98-
.getUse()
99-
.getRetries()
100-
.getAdditionalProperties()
101-
.get(retry.getRetryPolicyReference());
102-
if (retryPolicy == null) {
103-
throw new IllegalStateException("Retry policy " + retryPolicy + " was not found");
104+
if (retry != null) {
105+
if (retry.getRetryPolicyDefinition() != null) {
106+
retryPolicy = retry.getRetryPolicyDefinition();
107+
} else if (retry.getRetryPolicyReference() != null) {
108+
retryPolicy =
109+
workflow
110+
.getUse()
111+
.getRetries()
112+
.getAdditionalProperties()
113+
.get(retry.getRetryPolicyReference());
114+
if (retryPolicy == null) {
115+
throw new IllegalStateException(
116+
"Retry policy " + retry.getRetryPolicyReference() + " was not found");
117+
}
104118
}
105119
}
106-
return retryPolicy != null ? Optional.of(buildRetryExecutor(retryPolicy)) : Optional.empty();
120+
return Optional.ofNullable(retryPolicy);
107121
}
108122

109123
protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) {
@@ -114,6 +128,23 @@ protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) {
114128
WorkflowUtils.optionalPredicate(application, retryPolicy.getExceptWhen()));
115129
}
116130

131+
private Optional<WorkflowValueResolver<Duration>> resolveAttemptDuration(
132+
RetryPolicy retryPolicy) {
133+
RetryLimit limit = retryPolicy.getLimit();
134+
return limit != null && limit.getAttempt() != null && limit.getAttempt().getDuration() != null
135+
? Optional.of(
136+
WorkflowUtils.fromTimeoutAfter(application, limit.getAttempt().getDuration()))
137+
: Optional.empty();
138+
}
139+
140+
private Optional<WorkflowValueResolver<Duration>> resolveOverallDuration(
141+
RetryPolicy retryPolicy) {
142+
RetryLimit limit = retryPolicy.getLimit();
143+
return limit != null && limit.getDuration() != null
144+
? Optional.of(WorkflowUtils.fromTimeoutAfter(application, limit.getDuration()))
145+
: Optional.empty();
146+
}
147+
117148
private static int resolveMaxAttempts(RetryLimit limit) {
118149
return limit != null && limit.getAttempt() != null
119150
? limit.getAttempt().getCount()
@@ -152,65 +183,123 @@ protected TryExecutor(TryExecutorBuilder builder) {
152183
this.taskExecutor = builder.taskExecutor;
153184
this.catchTaskExecutor = builder.catchTaskExecutor;
154185
this.retryIntervalExecutor = builder.retryIntervalExecutor;
186+
this.attemptDuration = builder.attemptDuration;
155187
this.errorVariable = builder.errorVariable;
188+
this.overallDuration = builder.overallDuration;
156189
}
157190

158191
@Override
159192
protected CompletableFuture<WorkflowModel> internalExecute(
160193
WorkflowContext workflow, TaskContext taskContext) {
161-
return doIt(workflow, taskContext, taskContext.input());
194+
WorkflowModel model = taskContext.input();
195+
return cancellingFutureTimeout(
196+
doIt(workflow, taskContext, model), overallDuration, workflow, taskContext, model)
197+
.exceptionallyCompose(
198+
e -> CompletableFuture.failedFuture(timeoutToWorkflow(e, taskContext)));
162199
}
163200

164201
private CompletableFuture<WorkflowModel> doIt(
165202
WorkflowContext workflow, TaskContext taskContext, WorkflowModel model) {
166203
retryIntervalExecutor.ifPresent(r -> r.init(workflow, taskContext, model));
167-
return TaskExecutorHelper.processTaskList(
168-
taskExecutor, workflow, Optional.of(taskContext), model)
204+
CompletableFuture<WorkflowModel> taskFuture =
205+
TaskExecutorHelper.processTaskList(taskExecutor, workflow, Optional.of(taskContext), model);
206+
return cancellingFutureTimeout(taskFuture, attemptDuration, workflow, taskContext, model)
169207
.exceptionallyCompose(e -> handleException(e, workflow, taskContext));
170208
}
171209

172210
private CompletableFuture<WorkflowModel> handleException(
173211
Throwable e, WorkflowContext workflow, TaskContext taskContext) {
174-
if (e instanceof CompletionException) {
175-
return handleException(e.getCause(), workflow, taskContext);
212+
Throwable cause = e instanceof CompletionException ? e.getCause() : e;
213+
if (cause instanceof TimeoutException timeout) {
214+
return handleException(timeoutToWorkflow(timeout, taskContext), workflow, taskContext);
215+
} else if (cause instanceof WorkflowException exception) {
216+
return handleException(exception, workflow, taskContext);
217+
} else {
218+
return CompletableFuture.failedFuture(e);
176219
}
177-
if (e instanceof WorkflowException) {
178-
WorkflowException exception = (WorkflowException) e;
220+
}
221+
222+
private CompletableFuture<WorkflowModel> handleException(
223+
WorkflowException exception, WorkflowContext workflow, TaskContext taskContext) {
224+
WorkflowError error = exception.getWorkflowError();
225+
if (errorFilter.map(f -> f.test(error)).orElse(true)
226+
&& WorkflowUtils.whenExceptTest(
227+
whenFilter,
228+
exceptFilter,
229+
workflow,
230+
taskContext,
231+
workflow.definition().application().modelFactory().fromAny(error))) {
179232
CompletableFuture<WorkflowModel> completable =
180233
CompletableFuture.completedFuture(taskContext.rawOutput());
181-
WorkflowError error = exception.getWorkflowError();
182-
if (errorFilter.map(f -> f.test(error)).orElse(true)
183-
&& WorkflowUtils.whenExceptTest(
184-
whenFilter,
185-
exceptFilter,
186-
workflow,
187-
taskContext,
188-
workflow.definition().application().modelFactory().fromAny(error))) {
189-
if (errorVariable != null) {
190-
taskContext.variables().put(errorVariable, error);
191-
}
192-
if (catchTaskExecutor.isPresent()) {
193-
completable =
194-
completable.thenCompose(
195-
model ->
196-
TaskExecutorHelper.processTaskList(
197-
catchTaskExecutor.get(), workflow, Optional.of(taskContext), model));
198-
}
199-
if (retryIntervalExecutor.isPresent()) {
200-
completable =
201-
completable
202-
.thenCompose(
203-
model ->
204-
retryIntervalExecutor
205-
.get()
206-
.retry(workflow, taskContext, model)
207-
.orElse(CompletableFuture.failedFuture(e)))
208-
.thenCompose(model -> doIt(workflow, taskContext, model));
209-
}
210-
return completable;
234+
235+
if (errorVariable != null) {
236+
taskContext.variables().put(errorVariable, error);
237+
}
238+
if (catchTaskExecutor.isPresent()) {
239+
completable =
240+
completable.thenCompose(
241+
model ->
242+
TaskExecutorHelper.processTaskList(
243+
catchTaskExecutor.get(), workflow, Optional.of(taskContext), model));
244+
}
245+
if (retryIntervalExecutor.isPresent()) {
246+
completable =
247+
completable
248+
.thenCompose(
249+
model ->
250+
retryIntervalExecutor
251+
.get()
252+
.retry(workflow, taskContext, model)
253+
.orElse(CompletableFuture.failedFuture(exception)))
254+
.thenCompose(model -> doIt(workflow, taskContext, model));
255+
}
256+
return completable;
257+
} else {
258+
return CompletableFuture.failedFuture(exception);
259+
}
260+
}
261+
262+
private static WorkflowException timeoutToWorkflow(
263+
TimeoutException timeout, TaskContext taskContext) {
264+
return new WorkflowException(
265+
WorkflowError.timeout()
266+
.instance(taskContext.position().jsonPointer())
267+
.title(timeout.getMessage())
268+
.build(),
269+
timeout);
270+
}
271+
272+
private static Throwable timeoutToWorkflow(Throwable ex, TaskContext taskContext) {
273+
Throwable cause = ex instanceof CompletionException ? ex.getCause() : ex;
274+
return cause instanceof TimeoutException timeout ? timeoutToWorkflow(timeout, taskContext) : ex;
275+
}
276+
277+
private static CompletableFuture<WorkflowModel> cancellingFutureTimeout(
278+
CompletableFuture<WorkflowModel> originalFuture,
279+
Optional<WorkflowValueResolver<Duration>> duration,
280+
WorkflowContext workflowContext,
281+
TaskContext taskContext,
282+
WorkflowModel model) {
283+
long timeout =
284+
duration
285+
.map(d -> d.apply(workflowContext, taskContext, model))
286+
.orElse(Duration.ZERO)
287+
.toMillis();
288+
return timeout > 0
289+
? originalFuture
290+
.copy()
291+
.orTimeout(timeout, TimeUnit.MILLISECONDS)
292+
.whenComplete((v, e) -> cancelIfTimeout(e, originalFuture))
293+
: originalFuture;
294+
}
295+
296+
private static void cancelIfTimeout(Throwable e, CompletableFuture<WorkflowModel> taskFuture) {
297+
if (!taskFuture.isDone()) {
298+
Throwable realException = e instanceof CompletionException ? e.getCause() : e;
299+
if (realException instanceof TimeoutException) {
300+
taskFuture.cancel(true);
211301
}
212302
}
213-
return CompletableFuture.failedFuture(e);
214303
}
215304

216305
private static Optional<Predicate<WorkflowError>> buildErrorFilter(CatchErrors errors) {

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

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import java.util.Map;
3535
import java.util.concurrent.CompletableFuture;
3636
import java.util.concurrent.ConcurrentHashMap;
37+
import java.util.concurrent.TimeUnit;
3738
import okhttp3.mockwebserver.MockResponse;
3839
import okhttp3.mockwebserver.MockWebServer;
3940
import org.awaitility.Awaitility;
@@ -186,6 +187,82 @@ void testRetryEnd() throws IOException {
186187
.hasCauseInstanceOf(WorkflowException.class);
187188
}
188189

190+
@Test
191+
void testAttemptDuration() throws IOException {
192+
apiServer.enqueue(
193+
new MockResponse()
194+
.setHeadersDelay(2, TimeUnit.SECONDS)
195+
.setResponseCode(200)
196+
.setHeader("Content-Type", "application/json")
197+
.setBody("{}"));
198+
assertThatThrownBy(
199+
() ->
200+
app.workflowDefinition(
201+
readWorkflowFromClasspath(
202+
"workflows-samples/try-catch-retry-attempt-duration.yaml"))
203+
.instance(Map.of())
204+
.start()
205+
.join())
206+
.hasCauseInstanceOf(WorkflowException.class);
207+
}
208+
209+
@Test
210+
void testAttemptDurationRetry() throws IOException {
211+
String result = "{\"name\":\"Luna\"}";
212+
apiServer.enqueue(
213+
new MockResponse()
214+
.setHeadersDelay(2, TimeUnit.SECONDS)
215+
.setResponseCode(200)
216+
.setHeader("Content-Type", "application/json")
217+
.setBody(result));
218+
apiServer.enqueue(
219+
new MockResponse()
220+
.setResponseCode(200)
221+
.setHeader("Content-Type", "application/json")
222+
.setBody(result));
223+
CompletableFuture<WorkflowModel> future =
224+
app.workflowDefinition(
225+
readWorkflowFromClasspath(
226+
"workflows-samples/try-catch-retry-attempt-duration-retry.yaml"))
227+
.instance(Map.of())
228+
.start();
229+
Awaitility.await().atMost(Duration.ofSeconds(5)).until(future::isDone);
230+
assertThat(future.join().as(String.class).orElseThrow()).isEqualTo(result);
231+
assertThat(retryListener.taskRetried).hasSize(1);
232+
assertThat(retryListener.taskRetried.get("do/0/tryGetPet/try/0/getPet")).isEqualTo((short) 1);
233+
}
234+
235+
@Test
236+
void testAttemptDurationOverall() throws IOException {
237+
String result = "{\"name\":\"Luna\"}";
238+
apiServer.enqueue(
239+
new MockResponse()
240+
.setHeadersDelay(1, TimeUnit.SECONDS)
241+
.setResponseCode(200)
242+
.setHeader("Content-Type", "application/json")
243+
.setBody(result));
244+
apiServer.enqueue(
245+
new MockResponse()
246+
.setHeadersDelay(1, TimeUnit.SECONDS)
247+
.setResponseCode(200)
248+
.setHeader("Content-Type", "application/json")
249+
.setBody(result));
250+
apiServer.enqueue(
251+
new MockResponse()
252+
.setResponseCode(200)
253+
.setHeader("Content-Type", "application/json")
254+
.setBody(result));
255+
assertThatThrownBy(
256+
() ->
257+
app.workflowDefinition(
258+
readWorkflowFromClasspath(
259+
"workflows-samples/try-catch-retry-attempt-duration-overall.yaml"))
260+
.instance(Map.of())
261+
.start()
262+
.join())
263+
.hasCauseInstanceOf(WorkflowException.class);
264+
}
265+
189266
@Test
190267
void testTimeout() throws IOException {
191268
Map<String, Object> result =

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

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
document:
2+
dsl: '1.0.0'
3+
namespace: test
4+
name: try-catch-retry-attempt-duration-overall
5+
version: '0.1.0'
6+
do:
7+
- tryGetPet:
8+
try:
9+
- getPet:
10+
call: http
11+
with:
12+
method: get
13+
endpoint: http://localhost:9797
14+
redirect: true
15+
catch:
16+
errors:
17+
with:
18+
type: https://serverlessworkflow.io/spec/1.0.0/errors/timeout
19+
status: 408
20+
retry:
21+
delay:
22+
milliseconds: 10
23+
backoff:
24+
constant: {}
25+
limit:
26+
duration:
27+
milliseconds: 100
28+
attempt:
29+
count: 5
30+
duration:
31+
milliseconds: 50

0 commit comments

Comments
 (0)