feat(DurableExecution): incremental, heterogeneous Parallel API (#2519)

[GarrettBeatty]

Description

Implements #2519: an additive, branch-oriented parallel API for Amazon.Lambda.DurableExecution supporting heterogeneous per-branch result types and incremental branch registration, alongside the existing homogeneous ParallelAsync<T> overloads (which are unchanged).

await using var parallel = ctx.CreateParallel(name: "process-order");

IParallelBranch<InventoryReservation> inventory = parallel.BranchAsync(
    "inventory", async (branch, ct) => await ReserveInventoryAsync(branch, ct));
IParallelBranch<PaymentAuthorization> payment = parallel.BranchAsync(
    "payment", async (branch, ct) => await AuthorizePaymentAsync(branch, ct));

IBatchResult summary = await parallel.CompleteAsync();

InventoryReservation reserved = await inventory;   // own concrete type — no shared base, cast, or envelope
PaymentAuthorization  authed   = await payment;

What & why

Today every branch of a Parallel must share one generic result type T, forcing unrelated branch contracts into object, a common base type, or a wrapper. This adds a branch-scoped generic API (matching the Java SDK's ParallelDurableFuture) that gives each branch its own compile-time type, replay-safe per-branch deserialization, and incremental composition (register/start branches as work is discovered, then seal).

New public API

• IDurableContext.CreateParallel(name?, config?) → IDurableParallel
• IDurableParallel : IAsyncDisposable — BranchAsync<T>(name, func), CompleteAsync(ct)
• IParallelBranch<T> — awaitable typed handle exposing Name / Index / Status

Design

• Each branch runs as an existing ChildContextOperation<T> with the same deterministic child op id (hash("{parentId}-{index}")) and the same parent CONTEXT/Parallel BatchSummary checkpoint shape as batch Parallel — so replay, checkpoints, and reconstruction are identical and interoperable.
• Branches start on registration, gated by a shared MaxConcurrency semaphore and a cooperative short-circuit token; CompleteAsync seals, awaits per CompletionConfig, and checkpoints the aggregate.
• Deterministic replay: branch identity is positional (register the same branches in the same order); a name change at an index throws NonDeterministicExecutionException. Terminal-parent replay reconstructs from the frozen inline summary without re-running (re-running only overflow-stripped branches).
• DisposeAsync auto-completes if CompleteAsync wasn't called, so await using always writes the terminal checkpoint.
• MaxConcurrency, CompletionConfig, NestingType, cancellation, and the registered ILambdaSerializer are honored unchanged.
• Refactors BatchSummary (de)serialization + overflow handling out of ConcurrentOperation<T> into a shared BatchSummaryCodec so the batch and incremental paths can't diverge on the wire format.

Testing

• 16 unit tests (IncrementalParallelOperationTests): fresh happy path, heterogeneous types, deterministic ids, MaxConcurrency, completion short-circuit/skip, failure surfacing, empty, replay reconstruct (inline + failed branch), name-drift, and STARTED-parent replay. All 428 unit tests pass.
• 2 integration tests, both verified green against the durable-execution service:
  • IncrementalParallelHeterogeneousTest — string/int/POCO branches round-trip end-to-end.
  • IncrementalParallelReplayTest — deterministic replay across both the STARTED-parent Run path and the terminal-reconstruct resume (each branch step executes exactly once).
• Docs: new "Incremental, heterogeneous branches" section in docs/core/parallel.md.

Note for reviewers

Because IParallelBranch<T> is awaitable, a bare parallel.BranchAsync(...) statement whose result is ignored trips CS4014 under the repo's warnings-as-errors — callers must capture or discard (_ =) the handle. Flagging in case the team prefers a non-awaitable handle + explicit GetResultAsync().

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[Copilot]

🟡 Changes recommended

Replay validation, cancellation, completion concurrency, and branch result consistency contain unresolved correctness issues.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds an incremental parallel API supporting heterogeneous typed branches while preserving existing batch APIs and checkpoint formats.

Changes:

• Adds CreateParallel, typed branch handles, and orchestration logic.
• Extracts shared batch-summary serialization.
• Adds unit, integration, and documentation coverage.

File summaries

File	Description
IncrementalParallelOperationTests.cs	Tests incremental execution and replay.
IncrementalParallelReplayFunction.csproj	Configures replay test function.
IncrementalParallelReplayFunction/Function.cs	Exercises replay paths.
IncrementalParallelHeterogeneousFunction.csproj	Configures heterogeneous test function.
IncrementalParallelHeterogeneousFunction/Function.cs	Exercises typed branches.
IncrementalParallelReplayTest.cs	Validates replay integration.
IncrementalParallelHeterogeneousTest.cs	Validates heterogeneous integration.
IParallelBranch.cs	Defines typed awaitable handles.
IncrementalParallelOperation.cs	Implements incremental orchestration.
ConcurrentOperation.cs	Uses shared summary codec.
BatchSummaryCodec.cs	Centralizes summary serialization.
IDurableParallel.cs	Defines the public parallel API.
IDurableContext.cs	Exposes CreateParallel.
DurableContext.cs	Constructs incremental operations.
docs/core/parallel.md	Documents the new API.

Review details

Suppressed comments (2)

Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs:160

• Serialize the value before completing the public result task. With NestingType.Flat, Serialize(value) can throw after TrySetResult; the catch path then records a failed outcome but cannot replace the already-successful handle result, so summary.HasFailure/Status report failure while await branch returns a value.

            var value = await run().ConfigureAwait(false);
            if (_frozenStatus is null) _status = (int)BatchItemStatus.Succeeded;
            _result.TrySetResult(value);
            return BranchOutcome.Success(Index, Name, Serialize(value));

Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs:331

• Handle missing/corrupt terminal summaries explicitly. ParseSummary returns null for these payloads, but the operation remains in Terminal mode; every branch is then resolved as skipped and CompleteAsync synthesizes AllCompleted, masking a previously terminal checkpoint. Recover from child checkpoints where possible or fail replay rather than returning a false success.

        if (terminal)
        {
            _mode = ParallelExecutionMode.Terminal;
            _frozenSummary = BatchSummaryCodec.ParseSummary(existing!.ContextDetails?.Result);
            _startTask = Task.CompletedTask;

• Files reviewed: 16/16 changed files
• Comments generated: 7
• Review effort level: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Pull request overview

Copilot reviewed 24 out of 24 changed files in this pull request and generated 2 comments.