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..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 @@ -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,8 @@ public class TryExecutor extends RegularTaskExecutor { private final TaskExecutor taskExecutor; 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 { @@ -64,6 +70,8 @@ public static class TryExecutorBuilder extends RegularTaskExecutorBuilder taskExecutor; private final Optional> catchTaskExecutor; private final Optional retryIntervalExecutor; + private final Optional> attemptDuration; + private final Optional> overallDuration; private String errorVariable; protected TryExecutorBuilder( @@ -83,27 +91,33 @@ protected TryExecutorBuilder( position.copy().addProperty("catch"), catchTaskDo, definition)) : Optional.empty(); Retry retry = catchInfo.getRetry(); - this.retryIntervalExecutor = retry != null ? buildRetryInterval(retry) : Optional.empty(); + Optional retryPolicy = 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 buildRetryInterval(Retry retry) { + private Optional resolveRetryPolicy(Retry retry) { RetryPolicy retryPolicy = null; - if (retry.getRetryPolicyDefinition() != null) { - retryPolicy = retry.getRetryPolicyDefinition(); - } else if (retry.getRetryPolicyReference() != null) { - retryPolicy = - workflow - .getUse() - .getRetries() - .getAdditionalProperties() - .get(retry.getRetryPolicyReference()); - if (retryPolicy == null) { - throw new IllegalStateException("Retry policy " + retryPolicy + " was not found"); + if (retry != null) { + if (retry.getRetryPolicyDefinition() != null) { + retryPolicy = retry.getRetryPolicyDefinition(); + } else if (retry.getRetryPolicyReference() != null) { + retryPolicy = + workflow + .getUse() + .getRetries() + .getAdditionalProperties() + .get(retry.getRetryPolicyReference()); + if (retryPolicy == null) { + throw new IllegalStateException( + "Retry policy " + retry.getRetryPolicyReference() + " was not found"); + } } } - return retryPolicy != null ? Optional.of(buildRetryExecutor(retryPolicy)) : Optional.empty(); + return Optional.ofNullable(retryPolicy); } protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { @@ -114,6 +128,23 @@ protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { WorkflowUtils.optionalPredicate(application, retryPolicy.getExceptWhen())); } + private Optional> resolveAttemptDuration( + RetryPolicy retryPolicy) { + RetryLimit limit = retryPolicy.getLimit(); + return limit != null && limit.getAttempt() != null && limit.getAttempt().getDuration() != null + ? Optional.of( + WorkflowUtils.fromTimeoutAfter(application, limit.getAttempt().getDuration())) + : Optional.empty(); + } + + private Optional> resolveOverallDuration( + RetryPolicy retryPolicy) { + RetryLimit limit = retryPolicy.getLimit(); + return limit != null && limit.getDuration() != null + ? Optional.of(WorkflowUtils.fromTimeoutAfter(application, limit.getDuration())) + : Optional.empty(); + } + private static int resolveMaxAttempts(RetryLimit limit) { return limit != null && limit.getAttempt() != null ? limit.getAttempt().getCount() @@ -152,65 +183,123 @@ protected TryExecutor(TryExecutorBuilder builder) { this.taskExecutor = builder.taskExecutor; this.catchTaskExecutor = builder.catchTaskExecutor; this.retryIntervalExecutor = builder.retryIntervalExecutor; + this.attemptDuration = builder.attemptDuration; this.errorVariable = builder.errorVariable; + this.overallDuration = builder.overallDuration; } @Override protected CompletableFuture 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( WorkflowContext workflow, TaskContext taskContext, WorkflowModel model) { retryIntervalExecutor.ifPresent(r -> r.init(workflow, taskContext, model)); - return TaskExecutorHelper.processTaskList( - taskExecutor, workflow, Optional.of(taskContext), model) + CompletableFuture taskFuture = + TaskExecutorHelper.processTaskList(taskExecutor, workflow, Optional.of(taskContext), model); + return cancellingFutureTimeout(taskFuture, attemptDuration, workflow, taskContext, model) .exceptionallyCompose(e -> handleException(e, workflow, taskContext)); } private CompletableFuture handleException( Throwable e, WorkflowContext workflow, TaskContext taskContext) { - if (e instanceof CompletionException) { - return handleException(e.getCause(), workflow, taskContext); + Throwable cause = e instanceof CompletionException ? e.getCause() : e; + if (cause instanceof TimeoutException timeout) { + return handleException(timeoutToWorkflow(timeout, taskContext), workflow, taskContext); + } else if (cause instanceof WorkflowException exception) { + return handleException(exception, workflow, taskContext); + } else { + return CompletableFuture.failedFuture(e); } - if (e instanceof WorkflowException) { - WorkflowException exception = (WorkflowException) e; + } + + private CompletableFuture handleException( + WorkflowException exception, WorkflowContext workflow, TaskContext taskContext) { + WorkflowError error = exception.getWorkflowError(); + if (errorFilter.map(f -> f.test(error)).orElse(true) + && WorkflowUtils.whenExceptTest( + whenFilter, + exceptFilter, + workflow, + taskContext, + workflow.definition().application().modelFactory().fromAny(error))) { CompletableFuture 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 47898829c..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 @@ -34,6 +34,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; import okhttp3.mockwebserver.MockResponse; import okhttp3.mockwebserver.MockWebServer; import org.awaitility.Awaitility; @@ -186,6 +187,82 @@ void testRetryEnd() throws IOException { .hasCauseInstanceOf(WorkflowException.class); } + @Test + void testAttemptDuration() throws IOException { + apiServer.enqueue( + new MockResponse() + .setHeadersDelay(2, TimeUnit.SECONDS) + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody("{}")); + assertThatThrownBy( + () -> + app.workflowDefinition( + readWorkflowFromClasspath( + "workflows-samples/try-catch-retry-attempt-duration.yaml")) + .instance(Map.of()) + .start() + .join()) + .hasCauseInstanceOf(WorkflowException.class); + } + + @Test + void testAttemptDurationRetry() throws IOException { + String result = "{\"name\":\"Luna\"}"; + apiServer.enqueue( + new MockResponse() + .setHeadersDelay(2, TimeUnit.SECONDS) + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(result)); + apiServer.enqueue( + new MockResponse() + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(result)); + CompletableFuture future = + app.workflowDefinition( + readWorkflowFromClasspath( + "workflows-samples/try-catch-retry-attempt-duration-retry.yaml")) + .instance(Map.of()) + .start(); + Awaitility.await().atMost(Duration.ofSeconds(5)).until(future::isDone); + assertThat(future.join().as(String.class).orElseThrow()).isEqualTo(result); + assertThat(retryListener.taskRetried).hasSize(1); + assertThat(retryListener.taskRetried.get("do/0/tryGetPet/try/0/getPet")).isEqualTo((short) 1); + } + + @Test + void testAttemptDurationOverall() throws IOException { + String result = "{\"name\":\"Luna\"}"; + apiServer.enqueue( + new MockResponse() + .setHeadersDelay(1, TimeUnit.SECONDS) + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(result)); + apiServer.enqueue( + new MockResponse() + .setHeadersDelay(1, TimeUnit.SECONDS) + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(result)); + apiServer.enqueue( + new MockResponse() + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(result)); + assertThatThrownBy( + () -> + app.workflowDefinition( + readWorkflowFromClasspath( + "workflows-samples/try-catch-retry-attempt-duration-overall.yaml")) + .instance(Map.of()) + .start() + .join()) + .hasCauseInstanceOf(WorkflowException.class); + } + @Test void testTimeout() throws IOException { Map 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 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 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..97d1e1cd5 --- /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/communication + status: 404 + retry: + delay: + milliseconds: 10 + backoff: + constant: {} + limit: + attempt: + count: 5 + duration: + milliseconds: 50 \ No newline at end of file