Repository navigation
Conversation
get() on a parallel branch recorded as SKIPPED now throws ParallelBranchSkippedException instead of suspending with no pending operations. This covers branches that never started and in-flight branches skipped at completion, on the first invocation and on replay.
Move the existing waitForOperationCompletion javadoc back onto that method and document waitForCompletionFuture separately.
zhongkechen
left a comment
zhongkechen
left a comment
There was a problem hiding this comment.
Found three issues in this revision. Validation: 97 existing focused tests passed. Targeted integration repros confirmed all three findings; the anyOf replay and changing-result repros also passed against the parent commit.
| 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(); |
There was a problem hiding this comment.
[P1] Preserve anyOf() results across replay
Immediately completing skipped futures changes which branch wins DurableFuture.anyOf(). With firstSuccessful() and maxConcurrency(1), register the winning branch first, then call anyOf(skipped, winner) while it is running. This returns the winner initially, but after a later suspension it throws ParallelBranchSkippedException because replay immediately settles the skipped future. Previously successful handler code now fails on replay.
A deterministic integration repro observes [winner, winner] on the parent commit and [winner, SKIPPED:skipped] on this PR. Preserve the observable winner when skipped futures participate in anyOf().
| statuses); | ||
| var serializedResult = serializeAndDeserializeResult(cachedResult); | ||
| cachedResult = serializedResult.deserialized(); | ||
| completeSkippedBranches(items, statuses); |
There was a problem hiding this comment.
[P2] Make the completion snapshot and skip transition atomic
A branch can finish after the status snapshot records it as SKIPPED but before completeSkippedBranches() runs, particularly during result serialization. Its get() can then return a successful value before completeAsSkipped() changes that same future to throw.
A controlled integration repro gates ParallelResult serialization while branch B finishes: the first b.get() returns "B", but after parallel.get(), the second throws ParallelBranchSkippedException. The parent commit returns "B" both times. Coordinate the snapshot and skip transition with branch completion so an already-observed outcome cannot change.
| childContextOperation.get(); | ||
| return ParallelResult.Status.SUCCEEDED; | ||
| } catch (ParallelBranchSkippedException e) { | ||
| return ParallelResult.Status.SKIPPED; |
There was a problem hiding this comment.
[P2] Keep propagated nested exceptions classified as failures
An outer branch can fail by calling get() on a skipped inner branch. Its checkpoint correctly records FAILED, but this catch classifies the outer branch as SKIPPED, reports failed() == 0, and replaces the inner exception with one naming the outer branch.
A nested-parallel integration repro confirms that the outer checkpoint is FAILED while the returned statuses are [SKIPPED] and outerBranch.get() names outer-branch instead of inner-skipped. Use the branch's explicit isSkipped() state to identify skips; an exception propagated from its body remains a failure.
anyOf ignores a branch that settled only because it was skipped, so replay returns the same winner. The parallel snapshot marks an incomplete branch skipped under the same lock that publishes its result, and a nested skip exception stays a failure of the branch that propagated it.
|
The Oct 21 merge only brings in main, so the three comments were still open. They're fixed in aed55d3. anyOf no longer lets a skipped branch win while another future can still settle, so replay returns the same winner. The parallel snapshot and the skip happen under the same lock as branch completion, so get() can't return a value and then throw. A nested skip exception stays a failure of the outer branch; only an explicit skip is SKIPPED, and get() still names the inner branch. |
firstSuccessful() can checkpoint a parallel operation before a later branch is registered. get() on that branch then suspends with no pending operation. Decide the skip and publish the completion snapshot under one lock, and settle any branch still missing from that result.
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| private void markSkipped() { | ||
| skipped.set(true); | ||
| markAlreadyCompleted(); | ||
| } |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_kc5klhh4uqegosyk6mmj4px7do
[P1] Publish skipped results only after the parallel checkpoint succeeds
settleForParallelSnapshot() calls this before ParallelOperation.handleCompletion() serializes or submits its SUCCEED checkpoint. A caller waiting in branch.get() can therefore catch the skip and checkpoint subsequent work before persisted state records that skip. If the invocation is interrupted in this window, replay sees an incomplete parent and can execute and complete the previously skipped branch, changing control flow against those subsequent checkpoints. Reserve the skip under the lock, but delay both future completion and the exception exposed by get() until the parent checkpoint succeeds. Add an interruption/replay test covering this window.
To suppress repeats, reply /ai verdict deferred, /ai verdict false-positive, or /ai verdict out-of-scope on the first nonblank line. Add rationale on later lines. A new /ai verdict accepted allows future repeats again.
| if (eligible.isEmpty() || !isSkipSettlement(settled)) { | ||
| return settled; | ||
| } | ||
| pending.remove(settled); |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_dp6iycqxibkyp44yj6sjqf6w7i
[P1] Handle exhaustion when the last waiting candidate becomes skipped
Consider DurableFuture.anyOf(b) called before another branch wins a firstSuccessful() parallel. When b becomes skipped, this iteration removes it because eligible was nonempty. The next iteration passes an empty list to awaitFirstSettledOperation(), whose CompletableFuture.anyOf() never completes. The caller can hang or trigger invalid suspension instead of receiving ParallelBranchSkippedException; replay with an already-skipped branch behaves differently. Retain a skipped fallback and return it when candidates are exhausted. Test a branch that becomes skipped after the wait begins.
To suppress repeats, reply /ai verdict deferred, /ai verdict false-positive, or /ai verdict out-of-scope on the first nonblank line. Add rationale on later lines. A new /ai verdict accepted allows future repeats again.
| if (skipped.get()) { | ||
| return; | ||
| } |
There was a problem hiding this comment.
Codex AI review · Finding arf_v1_x2dftr772dldvhbwsumgln4mwz
[P2] Preserve the terminal plugin hook for started skipped branches
A running NestingType.FLAT branch has already emitted onOperationStart. Once it is marked skipped, this return—and the equivalent failure path—bypasses its existing fireOnOperationEnd() call. markSkipped() emits no replacement, so the hook never arrives even when the branch's user code finishes. This drops the branch span entirely in ExecutionOtelPlugin, which creates operation spans only in onOperationEnd, and bypasses custom plugin cleanup. Emit the terminal hook exactly once for started skipped branches while continuing to suppress result publication, and cover both success and failure exits.
To suppress repeats, reply /ai verdict deferred, /ai verdict false-positive, or /ai verdict out-of-scope on the first nonblank line. Add rationale on later lines. A new /ai verdict accepted allows future repeats again.
Codex AI reviewThree actionable findings:
Tests were not executed under the read-only review constraints. Reviewed commit |
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
Issue Link, if available
Fixes #752
Description
get()on a parallel branch recorded asSKIPPEDno longer waits until the SDK suspends with no pending operations. It throwsParallelBranchSkippedException, which names the branch, on the first invocation and on replay.This covers both cases from the issue:
firstSuccessful()(or another early-completion config) finished the parallel first.SKIPPED.get()now throws on that first invocation as well, so the first run and replay agree.docs/core/parallel.mddocumentsskipped(),statuses(), and the new exception.ExecutionManageris unchanged. The skip settlement lives inParallelOperationplus a smallChildContextOperation/BaseDurableOperationwait path soget()can complete without a checkpointedOperation.DurableContextis unchanged.Demo/Screenshots
N/A
Checklist
Testing
Unit Tests
Have unit tests been written for these changes?
Yes.
ParallelBranchSkippedExceptionTestcovers the named-branch message.ParallelOperationTest.minSuccessful_notExecuteSkippedBranchWhenReplaynow asserts thatget()on the skipped branch throwsParallelBranchSkippedException.Integration Tests
Have integration tests been written for these changes?
Yes. Local runner tests in
ParallelIntegrationTestusefirstSuccessful()with two branches and callget()on the skipped one, with a bounded timeout:maxConcurrency(1)), first run and replaytestInvalidSuspensionWhenWaitingOnSkippedBranchnow expects the named skipped exception instead of invalid suspension.Examples
Has a new example been added for the change? (if applicable)
No. Existing parallel examples still pass; the behavior is documented in
docs/core/parallel.md.This change was prepared with AI assistance and reviewed by the author.