[Bug]: get() on a SKIPPED parallel branch never returns, and the execution fails #752

Expected Behavior

get() on a branch future returns the branch result or throws an exception. It behaves the same way in the first invocation and on every replay.

The parallel result can record a branch as SKIPPED. For such a branch, get() should throw an exception that names the branch. docs/core/parallel.md should document it. For comparison, docs/core/map.md defines a skipped map item: status SKIPPED, a null result, and a null error.

Actual Behavior

get() on a SKIPPED branch does not return, and it does not throw a branch exception. It waits until the SDK suspends the invocation. The invocation then returns PENDING with no pending operation. In the cloud, the service rejects that output with InvalidParameterValueException: Cannot return PENDING status with no pending operations., the same error as in #736. Every replay reaches the same state, so the execution fails.

This happens in two cases.

Case 1: the branch never started.

1. A completion config such as firstSuccessful() completes the parallel before every branch has started.
2. ParallelOperation.handleCompletion records each branch whose future is not complete as SKIPPED.
3. The SDK does not start that branch later. Nothing completes its future until the SDK suspends the invocation.
4. get() on the branch releases the calling thread and waits. If no other thread is active, the SDK suspends the invocation. The suspension ends the wait with the SDK's internal SuspendExecutionException.
5. On replay, ParallelOperation.branch() reads SKIPPED from the checkpointed result and does not queue the branch. So every replay repeats steps 3 and 4.

Case 1 needs no race, and it fails in the first invocation. The repro uses firstSuccessful(). handleCompletion applies the same rule for every completion config that can complete early: minSuccessful(n), allSuccessful(), toleratedFailureCount(n), and shouldComplete(...).

Case 2: the branch was still running when the parallel completed.

1. With unlimited concurrency, branch-b can still be running when branch-a completes the parallel. The checkpointed result records branch-b as SKIPPED.
2. branch-b finishes later in the same invocation. ChildContextOperation.handleChildContextSuccess does not checkpoint the result, because the parallel is already complete. get() returns the result from memory.
3. A later operation, such as a wait, suspends the invocation. The next invocation replays the handler.
4. On replay, ParallelOperation.branch() reads SKIPPED and does not run branch-b. The get() call that returned a value in the first invocation now waits until the SDK suspends, as in case 1.

In case 2, the same handler code returns a value in the first invocation and fails on replay.

Steps to Reproduce

Both cases run in sdk-integration-tests with LocalDurableTestRunner.

Case 1 is the basic pattern from docs/core/parallel.md with firstSuccessful():

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");
    parallel.get(); // MIN_SUCCESSFUL_REACHED, statuses [SUCCEEDED, SKIPPED]
    a.get(); // "A"
    return b.get(); // never returns a value or throws a branch exception
});

var result = runner.runUntilComplete("input");
assertEquals(ExecutionStatus.PENDING, result.getStatus());

The result has status PENDING and no error. The execution state holds first-successful (Parallel, SUCCEEDED) and branch-a (ParallelBranch, SUCCEEDED). No operation is pending. The invocation logs Invalid suspension. No operation is pending. Each further runner.run("input") gives the same result. The outcome is the same with NestingType.FLAT, and with DurableFuture.allOf(...) over both branches in place of b.get().

Case 2 uses unlimited concurrency. A wait after the parallel forces a replay:

var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
    var config = ParallelConfig.builder()
            .completionConfig(CompletionConfig.firstSuccessful())
            .build();
    var parallel = context.parallel("first-successful", config);
    var a = parallel.branch("branch-a", String.class, ctx -> sleepThen(200, "A"));
    var b = parallel.branch("branch-b", String.class, ctx -> sleepThen(1000, "B"));
    parallel.get(); // statuses [SUCCEEDED, SKIPPED]: branch-b is still running
    var value = a.get() + b.get(); // "AB" in the first invocation
    context.wait("pause", Duration.ofSeconds(1)); // any later suspension causes a replay
    return value;
});

runner.run("input"); // invocation 1
runner.advanceTime(); // completes "pause"
runner.run("input"); // invocation 2, a replay

sleepThen(ms, value) calls Thread.sleep(ms) and returns value.

Invocations 2 to 4 each log Invalid suspension. No operation is pending. The outcome is the same with NestingType.FLAT.

SDK Version

2.2.1. Reproduced on main at ef88276 (2.2.2-SNAPSHOT). No file under sdk/src/main changed between v2.2.1 and ef88276.

Java Version

21

Is this a regression?

Unknown

Additional Context

• [Bug]: Parallel allCompleted can combine a stale branch count with joined state #736 listed this as a follow-up: "Make get() on a persisted skipped branch fail explicitly instead of waiting forever." Fix allCompleted parallel branch registration race #737 fixed the race in [Bug]: Parallel allCompleted can combine a stale branch count with joined state #736 and did not change this behavior. [Bug]: Parallel allCompleted can combine a stale branch count with joined state #736 described the replay of a persisted result. Case 1 also fails in the first invocation.
• A fix for case 2 has to make the first invocation and the replay agree. One option: get() throws for every branch that the checkpointed result records as SKIPPED, including a branch that finished later in the first invocation. Another option: the SDK checkpoints a late branch result so that replay returns it.
• The ParallelResult table in docs/core/parallel.md does not list skipped() or statuses().
• The end state, PENDING with no pending operation, is the state that [Bug]: failed tests due to unexpected PENDING status  #370 and [Bug]: Parallel allCompleted can combine a stale branch count with joined state #736 also reached. How the SDK reports that state is a separate issue.