diff --git a/docs/core/parallel.md b/docs/core/parallel.md index 99bdb26f86..d56723ffc8 100644 --- a/docs/core/parallel.md +++ b/docs/core/parallel.md @@ -39,8 +39,12 @@ try (var parallel = ctx.parallel("work")) { | `size()` | Total number of registered branches | | `succeeded()` | Number of branches that succeeded | | `failed()` | Number of branches that failed | +| `skipped()` | Number of branches recorded as `SKIPPED` | +| `statuses()` | Per-branch status: `SUCCEEDED`, `FAILED`, or `SKIPPED` | | `completionStatus()` | Why the operation completed (`ALL_COMPLETED`, `MIN_SUCCESSFUL_REACHED`, `FAILURE_TOLERANCE_EXCEEDED`) | +When a completion config finishes the parallel operation early, branches that never started — and branches that were still running at completion — are recorded as `SKIPPED`. `get()` on a skipped branch throws `ParallelBranchSkippedException` on the first invocation and on every replay. The SDK does not start or resume that branch later. + ### ParallelConfig Configure concurrency limits and completion criteria: @@ -110,9 +114,29 @@ try { } ``` +A skipped branch is different: the parallel result recorded it as `SKIPPED`, so `get()` throws `ParallelBranchSkippedException` instead of waiting: + +```java +var parallel = ctx.parallel("first-successful", ParallelConfig.builder() + .maxConcurrency(1) + .completionConfig(CompletionConfig.firstSuccessful()) + .build()); +var a = parallel.branch("branch-a", String.class, branchCtx -> "A"); +var b = parallel.branch("branch-b", String.class, branchCtx -> "B"); +parallel.get(); // statuses [SUCCEEDED, SKIPPED] +a.get(); // "A" +try { + b.get(); // throws on the first invocation and on every replay +} catch (ParallelBranchSkippedException e) { + // Branch never started, or was still running when the parallel completed. + // e.getBranchName() is the name passed to branch(). +} +``` + | Exception | When Thrown | |-----------|-------------| | `ParallelBranchFailedException` | Branch failed and the original exception could not be reconstructed | +| `ParallelBranchSkippedException` | Branch was recorded as `SKIPPED` in the parallel result | | User's exception | Branch threw a reconstructable exception — propagated through `get()` | ### Checkpoint-and-Replay @@ -121,4 +145,4 @@ Parallel operations are fully durable. On replay after interruption: - Completed branches return cached results without re-execution - Incomplete branches resume from their last checkpoint -- Branches that never started execute fresh +- Branches recorded as `SKIPPED` are not started; `get()` on those branch futures throws `ParallelBranchSkippedException` diff --git a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ParallelIntegrationTest.java b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ParallelIntegrationTest.java index 8ec7e20578..e83c261a95 100644 --- a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ParallelIntegrationTest.java +++ b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ParallelIntegrationTest.java @@ -7,7 +7,9 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; @@ -19,7 +21,7 @@ import software.amazon.lambda.durable.config.NestingType; import software.amazon.lambda.durable.config.ParallelConfig; import software.amazon.lambda.durable.config.WaitForConditionConfig; -import software.amazon.lambda.durable.exception.IllegalDurableOperationException; +import software.amazon.lambda.durable.exception.ParallelBranchSkippedException; import software.amazon.lambda.durable.model.ConcurrencyCompletionStatus; import software.amazon.lambda.durable.model.ExecutionStatus; import software.amazon.lambda.durable.model.ParallelResult; @@ -1387,12 +1389,104 @@ void testInvalidSuspensionWhenWaitingOnSkippedBranch() { } parallel.get(); - return futures.get(1).get(); + var skipped = assertThrows( + ParallelBranchSkippedException.class, () -> futures.get(1).get()); + assertEquals("branch-2", skipped.getBranchName()); + return "done"; }); - var ex = assertThrows(IllegalDurableOperationException.class, () -> runner.run("test")); - assertTrue(ex.isRetryable()); - assertTrue(ex.getMessage().contains("Invalid suspension. No operation is pending.")); - assertTrue(ex.getMessage().contains("branch-2")); + var result = assertTimeoutPreemptively(Duration.ofSeconds(10), () -> runner.runUntilComplete("test")); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("done", result.getResult(String.class)); + } + + @Test + void getOnSkippedBranchThatNeverStartedThrowsOnFirstRunAndReplay() { + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var config = ParallelConfig.builder() + .maxConcurrency(1) + .completionConfig(CompletionConfig.firstSuccessful()) + .build(); + var parallel = context.parallel("first-successful", config); + var a = parallel.branch("branch-a", String.class, ctx -> "A"); + var b = parallel.branch("branch-b", String.class, ctx -> "B"); + var result = parallel.get(); + assertEquals(ConcurrencyCompletionStatus.MIN_SUCCESSFUL_REACHED, result.completionStatus()); + assertEquals(List.of(ParallelResult.Status.SUCCEEDED, ParallelResult.Status.SKIPPED), result.statuses()); + assertEquals("A", a.get()); + var skipped = assertThrows(ParallelBranchSkippedException.class, b::get); + assertEquals("branch-b", skipped.getBranchName()); + assertEquals("Parallel branch 'branch-b' was skipped", skipped.getMessage()); + return "done"; + }); + + var first = assertTimeoutPreemptively(Duration.ofSeconds(10), () -> runner.runUntilComplete("input")); + assertEquals(ExecutionStatus.SUCCEEDED, first.getStatus()); + assertEquals("done", first.getResult(String.class)); + + var replay = assertTimeoutPreemptively(Duration.ofSeconds(10), () -> runner.run("input")); + assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus()); + assertEquals("done", replay.getResult(String.class)); + } + + @Test + void getOnSkippedInFlightBranchThrowsOnFirstRunAndReplay() throws Exception { + var branchBStarted = new CountDownLatch(1); + var releaseBranchB = new CountDownLatch(1); + var runner = LocalDurableTestRunner.create(String.class, (input, context) -> { + var config = ParallelConfig.builder() + .completionConfig(CompletionConfig.firstSuccessful()) + .build(); + var parallel = context.parallel("first-successful", config); + var b = parallel.branch("branch-b", String.class, ctx -> { + branchBStarted.countDown(); + try { + if (!releaseBranchB.await(10, TimeUnit.SECONDS)) { + throw new IllegalStateException("timed out waiting to release branch-b"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + return "B"; + }); + var a = parallel.branch("branch-a", String.class, ctx -> { + try { + if (!branchBStarted.await(10, TimeUnit.SECONDS)) { + throw new IllegalStateException("timed out waiting for branch-b to start"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + return "A"; + }); + var result = parallel.get(); + assertEquals(ConcurrencyCompletionStatus.MIN_SUCCESSFUL_REACHED, result.completionStatus()); + assertEquals(List.of(ParallelResult.Status.SKIPPED, ParallelResult.Status.SUCCEEDED), result.statuses()); + try { + var skipped = assertThrows(ParallelBranchSkippedException.class, b::get); + assertEquals("branch-b", skipped.getBranchName()); + assertEquals("A", a.get()); + context.wait("pause", Duration.ofSeconds(1)); + var skippedOnReplay = assertThrows(ParallelBranchSkippedException.class, b::get); + assertEquals("branch-b", skippedOnReplay.getBranchName()); + return "done"; + } finally { + releaseBranchB.countDown(); + } + }); + + try { + var first = assertTimeoutPreemptively(Duration.ofSeconds(15), () -> runner.run("input")); + assertEquals(ExecutionStatus.PENDING, first.getStatus()); + + runner.advanceTime(); + var replay = assertTimeoutPreemptively(Duration.ofSeconds(15), () -> runner.run("input")); + assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus()); + assertEquals("done", replay.getResult(String.class)); + } finally { + releaseBranchB.countDown(); + } } } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/ParallelBranchSkippedException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/ParallelBranchSkippedException.java new file mode 100644 index 0000000000..9bee4766d0 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/ParallelBranchSkippedException.java @@ -0,0 +1,32 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.exception; + +/** + * Thrown when {@code get()} is called on a parallel branch that the checkpointed parallel result recorded as + * {@code SKIPPED}. + * + *

This happens when a completion config such as {@code firstSuccessful()} finishes the parallel operation before + * every branch has started, or while a branch is still running. The SDK does not start or resume that branch later, so + * {@code get()} throws this exception on the first invocation and on every replay. + */ +public class ParallelBranchSkippedException extends DurableExecutionException { + private final String branchName; + + public ParallelBranchSkippedException(String branchName) { + super(formatMessage(branchName)); + this.branchName = branchName; + } + + /** Returns the name of the skipped branch. */ + public String getBranchName() { + return branchName; + } + + private static String formatMessage(String branchName) { + if (branchName == null) { + return "Parallel branch was skipped"; + } + return String.format("Parallel branch '%s' was skipped", branchName); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java index cd884796e9..8cefe2de35 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java @@ -242,6 +242,15 @@ private static ExecutionManager getExecutionManager(ListSkipped parallel branches that never started can settle this way because they have no checkpointed operation. + */ + protected void waitForCompletionFuture() { + executionManager.waitForOperationCompletion(this); + } + /** * Waits for the operation to complete. Deregisters the current thread to allow Lambda suspension if the operation * is still in progress, then re-registers when the operation completes. @@ -249,7 +258,7 @@ private static ExecutionManager getExecutionManager(List extends SerializableDurableOperation { private final Function function; private final AtomicBoolean replayChildren = new AtomicBoolean(false); + private final AtomicBoolean skipped = new AtomicBoolean(false); private final AtomicReference> cachedOperationResult = new AtomicReference<>(null); // child context for RunInChildContext @@ -224,8 +226,25 @@ private void handleChildContextFailure(Throwable exception) { OperationUpdate.builder().action(OperationAction.FAIL).error(errorObject)); } + /** + * Completes this branch as skipped so {@link #get()} throws {@link ParallelBranchSkippedException} instead of + * waiting for an operation that will never be started or checkpointed. + */ + void completeAsSkipped() { + skipped.set(true); + markAlreadyCompleted(); + } + + boolean isSkipped() { + return skipped.get(); + } + @Override public T get() { + waitForCompletionFuture(); + if (skipped.get()) { + throw new ParallelBranchSkippedException(getName()); + } var op = waitForOperationCompletion(); if (cachedOperationResult.get() != null) { // we have a result, just return it directly diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java index 41a687e6b6..e2dbd90d35 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java @@ -17,6 +17,7 @@ import software.amazon.lambda.durable.config.ParallelBranchConfig; import software.amazon.lambda.durable.config.ParallelConfig; import software.amazon.lambda.durable.context.DurableContextImpl; +import software.amazon.lambda.durable.exception.ParallelBranchSkippedException; import software.amazon.lambda.durable.execution.ExecutionManager; import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; @@ -85,6 +86,7 @@ protected void handleCompletion(CompletionConfig.CompletionDecision completionDe statuses); var serializedResult = serializeAndDeserializeResult(cachedResult); cachedResult = serializedResult.deserialized(); + completeSkippedBranches(items, statuses); // Branches added after checkpoint will not exist in the checkpointed result, but they'll be in the returned // value from get() method. @@ -95,13 +97,23 @@ protected void handleCompletion(CompletionConfig.CompletionDecision completionDe .contextOptions(ContextOptions.builder().replayChildren(true).build())); } + private void completeSkippedBranches(List> items, List statuses) { + for (int i = 0; i < items.size(); i++) { + if (statuses.get(i) == ParallelResult.Status.SKIPPED) { + items.get(i).completeAsSkipped(); + } + } + } + private ParallelResult.Status getParallelItemStatus(ChildContextOperation childContextOperation) { - if (!childContextOperation.isOperationCompleted()) { + if (!childContextOperation.isOperationCompleted() || childContextOperation.isSkipped()) { return ParallelResult.Status.SKIPPED; } try { childContextOperation.get(); return ParallelResult.Status.SUCCEEDED; + } catch (ParallelBranchSkippedException e) { + return ParallelResult.Status.SKIPPED; } catch (Throwable t) { return ParallelResult.Status.FAILED; } @@ -180,13 +192,22 @@ public DurableFuture branch( var nextBranchIndex = getBranches().size(); - // ConcurrencyOperation will skip this branch if skip=true: - // 1. if the parallel operation is already completed (partialResult is not null) - // 2. if the branch is already skipped in the partialResult or nonexistent in the partialResult - var skip = partialResult != null - && (partialResult.statuses().size() <= nextBranchIndex - || partialResult.statuses().get(nextBranchIndex) == ParallelResult.Status.SKIPPED); + // Skip when the checkpointed or in-memory parallel result already completed without this branch. + var skip = shouldSkipBranch(nextBranchIndex); var serDes = config.serDes() == null ? getContext().getDurableConfig().getSerDes() : config.serDes(); - return enqueueItem(name, func, resultType, serDes, OperationSubType.PARALLEL_BRANCH, skip); + var branch = enqueueItem(name, func, resultType, serDes, OperationSubType.PARALLEL_BRANCH, skip); + if (skip) { + branch.completeAsSkipped(); + } + return branch; + } + + private boolean shouldSkipBranch(int nextBranchIndex) { + var result = partialResult != null ? partialResult : cachedResult; + if (result == null) { + return false; + } + return result.statuses().size() <= nextBranchIndex + || result.statuses().get(nextBranchIndex) == ParallelResult.Status.SKIPPED; } } diff --git a/sdk/src/test/java/software/amazon/lambda/durable/exception/ParallelBranchSkippedExceptionTest.java b/sdk/src/test/java/software/amazon/lambda/durable/exception/ParallelBranchSkippedExceptionTest.java new file mode 100644 index 0000000000..a8b535174b --- /dev/null +++ b/sdk/src/test/java/software/amazon/lambda/durable/exception/ParallelBranchSkippedExceptionTest.java @@ -0,0 +1,25 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.exception; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import org.junit.jupiter.api.Test; + +class ParallelBranchSkippedExceptionTest { + + @Test + void namesTheSkippedBranchInTheMessage() { + var exception = new ParallelBranchSkippedException("branch-b"); + assertEquals("branch-b", exception.getBranchName()); + assertEquals("Parallel branch 'branch-b' was skipped", exception.getMessage()); + } + + @Test + void formatsAFallbackMessageWhenTheBranchNameIsNull() { + var exception = new ParallelBranchSkippedException(null); + assertNull(exception.getBranchName()); + assertEquals("Parallel branch was skipped", exception.getMessage()); + } +} diff --git a/sdk/src/test/java/software/amazon/lambda/durable/operation/ParallelOperationTest.java b/sdk/src/test/java/software/amazon/lambda/durable/operation/ParallelOperationTest.java index 2ec5c76753..130daec4be 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/operation/ParallelOperationTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/operation/ParallelOperationTest.java @@ -9,6 +9,7 @@ import static org.mockito.Mockito.*; import java.lang.reflect.Field; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -33,6 +34,7 @@ import software.amazon.lambda.durable.config.ParallelBranchConfig; import software.amazon.lambda.durable.config.ParallelConfig; import software.amazon.lambda.durable.context.DurableContextImpl; +import software.amazon.lambda.durable.exception.ParallelBranchSkippedException; import software.amazon.lambda.durable.execution.ExecutionManager; import software.amazon.lambda.durable.execution.ThreadContext; import software.amazon.lambda.durable.execution.ThreadType; @@ -365,6 +367,11 @@ void minSuccessful_notExecuteSkippedBranchWhenReplay() { assertEquals(1, result.skipped()); assertEquals(ConcurrencyCompletionStatus.MIN_SUCCESSFUL_REACHED, result.completionStatus()); assertTrue(result.completionStatus().isSucceeded()); + + var skipped = op.getBranches().get(0); + var thrown = assertTimeoutPreemptively( + Duration.ofSeconds(2), () -> assertThrows(ParallelBranchSkippedException.class, skipped::get)); + assertEquals("branch-1", thrown.getBranchName()); } // ===== Context hierarchy =====