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

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`

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/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}.
+ *
+ * <p>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);
+    }
+}

sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java

@@ -242,14 +242,23 @@ private static ExecutionManager getExecutionManager(List<? extends BaseDurableOp
         return executionManager;
     }
 
+    /**
+     * Waits for this operation's completion future without looking up a checkpointed {@code Operation}.
+     *
+     * <p>Skipped 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.
      *
      * @return the completed operation
      */
     protected Operation waitForOperationCompletion() {
-        executionManager.waitForOperationCompletion(this);
+        waitForCompletionFuture();
 
         if (isVirtual) {
             // We don't  store virtual operations so they don't corresponding Operation in storage

sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java

@@ -27,6 +27,7 @@
 import software.amazon.lambda.durable.exception.DurableOperationException;
 import software.amazon.lambda.durable.exception.MapIterationFailedException;
 import software.amazon.lambda.durable.exception.ParallelBranchFailedException;
+import software.amazon.lambda.durable.exception.ParallelBranchSkippedException;
 import software.amazon.lambda.durable.exception.StepFailedException;
 import software.amazon.lambda.durable.exception.StepInterruptedException;
 import software.amazon.lambda.durable.exception.UnrecoverableDurableExecutionException;
@@ -53,6 +54,7 @@ public class ChildContextOperation<T> extends SerializableDurableOperation<T> {
 
     private final Function<DurableContext, T> function;
     private final AtomicBoolean replayChildren = new AtomicBoolean(false);
+    private final AtomicBoolean skipped = new AtomicBoolean(false);
     private final AtomicReference<DeserializedOperationResult<T>> 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

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<ChildContextOperation<?>> items, List<ParallelResult.Status> 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 <T> DurableFuture<T> 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;
     }
 }

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