fix(sdk): throw when getting a skipped parallel branch

docs/core/parallel.md

@@ -39,8 +39,14 @@ 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`. The snapshot and the skip happen together with that branch's own completion, so `get()` does not return a value and later throw for the same call sequence. `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.
+
+`DurableFuture.anyOf` does not select a skipped branch while another future can still settle, so the winner from the first invocation is the winner on replay. If every future passed to `anyOf` is skipped, it throws `ParallelBranchSkippedException`.
+
 ### ParallelConfig
 
 Configure concurrency limits and completion criteria:
@@ -110,9 +116,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 +147,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`

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();
+        }
     }
 }

sdk-integration-tests/src/test/java/software/amazon/lambda/durable/ParallelSkippedBranchReviewTest.java

@@ -0,0 +1,216 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.lambda.model.OperationStatus;
+import software.amazon.lambda.durable.config.CompletionConfig;
+import software.amazon.lambda.durable.config.ParallelConfig;
+import software.amazon.lambda.durable.exception.ParallelBranchSkippedException;
+import software.amazon.lambda.durable.model.ExecutionStatus;
+import software.amazon.lambda.durable.model.ParallelResult;
+import software.amazon.lambda.durable.serde.JacksonSerDes;
+import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
+
+/**
+ * Repros for the three review findings on skipped parallel branches: {@code anyOf} replay, the completion-snapshot
+ * race, and nested skip exceptions classified as failures.
+ */
+class ParallelSkippedBranchReviewTest {
+
+    private static final Duration DEADLINE = Duration.ofSeconds(20);
+
+    @Test
+    void anyOfKeepsTheRunningWinnerWhenTheOtherBranchIsSkippedOnReplay() {
+        var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
+            var handlerThread = Thread.currentThread();
+            var parallel = context.parallel(
+                    "first-successful",
+                    ParallelConfig.builder()
+                            .maxConcurrency(1)
+                            .completionConfig(CompletionConfig.firstSuccessful())
+                            .build());
+            var winner = parallel.branch("winner", String.class, ctx -> {
+                waitUntilParked(handlerThread);
+                return "winner";
+            });
+            var skipped = parallel.branch("skipped", String.class, ctx -> "skipped");
+            var first = String.valueOf(DurableFuture.anyOf(skipped, winner));
+            context.wait("pause", Duration.ofSeconds(1));
+            var second = String.valueOf(DurableFuture.anyOf(skipped, winner));
+            return first + "," + second;
+        });
+
+        var pending = assertTimeoutPreemptively(DEADLINE, () -> runner.run("input"));
+        assertEquals(ExecutionStatus.PENDING, pending.getStatus());
+
+        runner.advanceTime();
+        var replay = assertTimeoutPreemptively(DEADLINE, () -> runner.run("input"));
+        assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus());
+        assertEquals("winner,winner", replay.getResult(String.class));
+    }
+
+    @Test
+    void branchGetKeepsTheSameOutcomeWhenItFinishesDuringResultSerialization() {
+        var branchRunning = new CountDownLatch(1);
+        var allowBranch = new CountDownLatch(1);
+        var serDes = new ParallelResultSerializationGate(allowBranch);
+        var config = DurableConfig.builder().withSerDes(serDes).build();
+        var runner = LocalDurableTestRunner.create(
+                String.class,
+                (input, context) -> {
+                    var parallel = context.parallel(
+                            "first-successful",
+                            ParallelConfig.builder()
+                                    .completionConfig(CompletionConfig.firstSuccessful())
+                                    .build());
+                    var slow = parallel.branch("branch-b", String.class, ctx -> {
+                        branchRunning.countDown();
+                        await(allowBranch, "release branch-b");
+                        return "B";
+                    });
+                    parallel.branch("branch-a", String.class, ctx -> {
+                        await(branchRunning, "branch-b to start");
+                        return "A";
+                    });
+                    var result = parallel.get();
+                    return "after-parallel|" + observe(slow) + "|" + result.statuses();
+                },
+                config);
+
+        try {
+            var execution = assertTimeoutPreemptively(DEADLINE, () -> runner.runUntilComplete("input"));
+            assertEquals(ExecutionStatus.SUCCEEDED, execution.getStatus());
+            assertEquals("after-parallel|SKIPPED:branch-b|[SKIPPED, SUCCEEDED]", execution.getResult(String.class));
+        } finally {
+            allowBranch.countDown();
+        }
+    }
+
+    @Test
+    void skippedBranchStaysSkippedOnReplayWhenParallelResultIsAlreadyCheckpointed() {
+        var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
+            var parallel = context.parallel(
+                    "first-successful",
+                    ParallelConfig.builder()
+                            .completionConfig(CompletionConfig.firstSuccessful())
+                            .build());
+            parallel.branch("branch-a", String.class, ctx -> "A");
+            var skipped = parallel.branch("branch-b", String.class, ctx -> "B");
+            var result = parallel.get();
+            context.step("record", String.class, stepCtx -> observe(skipped) + "|" + result.statuses());
+            return "done";
+        });
+
+        var first = assertTimeoutPreemptively(DEADLINE, () -> runner.runUntilComplete("input"));
+        assertEquals(ExecutionStatus.SUCCEEDED, first.getStatus());
+
+        var replay = assertTimeoutPreemptively(DEADLINE, () -> runner.run("input"));
+        assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus());
+        assertEquals("done", replay.getResult(String.class));
+    }
+
+    @Test
+    void nestedSkipExceptionStaysAFailedOuterBranch() {
+        var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
+            var outer = context.parallel("outer", ParallelConfig.builder().build());
+            var outerBranch = outer.branch("outer-branch", String.class, ctx -> {
+                var inner = ctx.parallel(
+                        "inner",
+                        ParallelConfig.builder()
+                                .maxConcurrency(1)
+                                .completionConfig(CompletionConfig.firstSuccessful())
+                                .build());
+                inner.branch("inner-winner", String.class, innerCtx -> "ok");
+                var skipped = inner.branch("inner-skipped", String.class, innerCtx -> "no");
+                inner.get();
+                skipped.get();
+                return "unreachable";
+            });
+
+            var result = outer.get();
+            assertEquals(List.of(ParallelResult.Status.FAILED), result.statuses());
+            assertEquals(1, result.failed());
+            assertEquals(0, result.skipped());
+            var thrown = assertThrows(ParallelBranchSkippedException.class, outerBranch::get);
+            assertEquals("inner-skipped", thrown.getBranchName());
+            return "done";
+        });
+
+        var first = assertTimeoutPreemptively(DEADLINE, () -> runner.runUntilComplete("input"));
+        assertEquals(ExecutionStatus.SUCCEEDED, first.getStatus());
+        assertEquals("done", first.getResult(String.class));
+        var checkpoint = runner.getOperation("outer-branch");
+        assertEquals(OperationStatus.FAILED, checkpoint.getStatus());
+        assertTrue(checkpoint.getContextDetails().error().errorMessage().contains("inner-skipped"));
+
+        var replay = assertTimeoutPreemptively(DEADLINE, () -> runner.run("input"));
+        assertEquals(ExecutionStatus.SUCCEEDED, replay.getStatus());
+        assertEquals("done", replay.getResult(String.class));
+    }
+
+    private static void waitUntilParked(Thread handlerThread) {
+        var deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+        while (handlerThread.getState() != Thread.State.WAITING
+                && handlerThread.getState() != Thread.State.TIMED_WAITING) {
+            if (System.nanoTime() >= deadline) {
+                throw new IllegalStateException("timed out waiting for anyOf to park");
+            }
+            Thread.onSpinWait();
+        }
+    }
+
+    private static void await(CountDownLatch latch, String description) {
+        try {
+            if (!latch.await(10, TimeUnit.SECONDS)) {
+                throw new IllegalStateException("timed out waiting for " + description);
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("interrupted while waiting for " + description, e);
+        }
+    }
+
+    private static String observe(DurableFuture<?> future) {
+        try {
+            return String.valueOf(future.get());
+        } catch (ParallelBranchSkippedException e) {
+            return "SKIPPED:" + e.getBranchName();
+        }
+    }
+
+    /** Blocks inside {@link ParallelResult} serialization so a branch can finish in that window. */
+    private static final class ParallelResultSerializationGate implements SerDes {
+        private final JacksonSerDes delegate = new JacksonSerDes();
+        private final CountDownLatch allowBranch;
+        private final AtomicBoolean gated = new AtomicBoolean();
+
+        private ParallelResultSerializationGate(CountDownLatch allowBranch) {
+            this.allowBranch = allowBranch;
+        }
+
+        @Override
+        public String serialize(Object value) {
+            if (value instanceof ParallelResult && gated.compareAndSet(false, true)) {
+                allowBranch.countDown();
+            }
+            return delegate.serialize(value);
+        }
+
+        @Override
+        public <T> T deserialize(String data, TypeToken<T> typeToken) {
+            return delegate.deserialize(data, typeToken);
+        }
+    }
+}