diff --git a/.autover/changes/12a4a1f7-d59f-4544-aa1f-6db30289485e.json b/.autover/changes/12a4a1f7-d59f-4544-aa1f-6db30289485e.json index c2b4327cc..1dca7e54c 100644 --- a/.autover/changes/12a4a1f7-d59f-4544-aa1f-6db30289485e.json +++ b/.autover/changes/12a4a1f7-d59f-4544-aa1f-6db30289485e.json @@ -2,7 +2,7 @@ "Projects": [ { "Name": "Amazon.Lambda.DurableExecution", - "Type": "Minor", + "Type": "Major", "ChangelogMessages": [ "Behavior change: steps and child contexts now round-trip their result through the configured serializer on a fresh (non-replay) success, deserializing the just-written checkpoint before returning \u2014 matching replay semantics. A custom per-operation serializer\u0027s transform is now reflected in the operation result on the first execution, not only on replay. The returned value is a fresh deserialized instance rather than the exact object the step/child body produced. Overflow (replay-children) results are unaffected." ] diff --git a/.autover/changes/add-incremental-heterogeneous-parallel.json b/.autover/changes/add-incremental-heterogeneous-parallel.json new file mode 100644 index 000000000..e59b54b2d --- /dev/null +++ b/.autover/changes/add-incremental-heterogeneous-parallel.json @@ -0,0 +1,12 @@ +{ + "Projects": [ + { + "Name": "Amazon.Lambda.DurableExecution", + "Type": "Minor", + "ChangelogMessages": [ + "Added an incremental, branch-oriented parallel API (IDurableContext.CreateParallel, IDurableParallel, IParallelBranch) supporting heterogeneous per-branch result types and incremental branch registration, alongside the existing homogeneous ParallelAsync overloads. Each branch declares its own result type and returns an awaitable typed handle; branches start on registration (respecting MaxConcurrency) and the operation is sealed with CompleteAsync. Honors the existing MaxConcurrency, CompletionConfig, NestingType, cancellation, deterministic replay, and ILambdaSerializer behavior.", + "CreateParallel supports per-operation and per-branch result serialization: ParallelConfig.ItemSerializer sets the operation-level serializer for all branch results, and IDurableParallel.Branch accepts an optional per-branch ILambdaSerializer override, falling back to ItemSerializer and then the globally-registered serializer." + ] + } + ] +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs b/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs index 1144c6bf1..5cd0d7e94 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs @@ -192,6 +192,29 @@ private Task> RunCallback( return op.ExecuteAsync(cancellationToken); } + public IDurableParallel CreateParallel( + string? name = null, + ParallelConfig? config = null) + { + var effectiveConfig = config ?? new ParallelConfig(); + // Operation-level default for per-branch result serialization: the config's + // ItemSerializer if set, else the globally-registered serializer. Individual + // branches may still override this via Branch's serializer parameter. + // + // Resolved LAZILY: a workflow that overrides the serializer on every Branch + // call (the AOT/per-branch scenario) must not be forced to register a global + // serializer. GetRequired is deferred to the factory below and invoked only + // when a branch actually falls back to this operation-level default. + var lambdaContext = LambdaContext; + Func defaultSerializerFactory = + () => effectiveConfig.ItemSerializer ?? LambdaSerializerHelper.GetRequired(lambdaContext); + + var operationId = _idGenerator.NextId(); + return new Internal.IncrementalParallelOperation( + operationId, name, _idGenerator.ParentId, effectiveConfig, defaultSerializerFactory, MakeChildFactory(), + _state, _terminationManager, _workflowCancellation, _durableExecutionArn, _batcher); + } + public Task> ParallelAsync( IReadOnlyList>> branches, string? name = null, diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs b/Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs index 9d536f5d3..cf531a4f3 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs @@ -367,6 +367,50 @@ Task WaitForConditionAsync( string? name = null, CancellationToken cancellationToken = default); + /// + /// Create an incremental, branch-oriented parallel operation. Unlike the + /// + /// overloads — which take a complete branch list up front and share one result + /// type — the returned lets you register branches + /// one at a time via + /// , + /// each with its own result type (heterogeneous), starting each branch as it is + /// registered. Call + /// to seal registration and obtain the aggregate . + /// + /// + /// Use await using so the operation is sealed and its terminal checkpoint + /// written even if + /// is not called. Branch identity is positional and deterministic across + /// replays, so register the same branches in the same order every invocation — + /// derive any dynamic branch set from a checkpointed step. Per-branch results + /// are serialized via the registered on + /// . Honors the same + /// , + /// , and + /// as the homogeneous API. + /// + /// + /// Optional human-readable name for the parallel operation. It surfaces on the + /// wire OperationUpdate.Name field and in execution traces. The + /// deterministic operation ID is positional (derived from the call order, not + /// from this name); however, when provided, the name becomes part of the + /// operation's deterministic definition and is validated on replay — changing it + /// across deployments for an in-flight execution throws + /// , so keep it stable (or leave + /// it null) for the life of an execution. Defaults to null. + /// + /// + /// Optional parallel configuration. Defaults are used when null. + /// + /// + /// An for registering branches and awaiting the + /// aggregate result. + /// + IDurableParallel CreateParallel( + string? name = null, + ParallelConfig? config = null); + /// /// Execute multiple branches concurrently. Each branch runs inside its own /// child context; per-branch results are aggregated into an diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/IDurableParallel.cs b/Libraries/src/Amazon.Lambda.DurableExecution/IDurableParallel.cs new file mode 100644 index 000000000..6de69c093 --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution/IDurableParallel.cs @@ -0,0 +1,130 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +namespace Amazon.Lambda.DurableExecution; + +/// +/// An incremental, branch-oriented parallel operation created by +/// . Branches +/// are registered one at a time via +/// , +/// each with its own result type (heterogeneous), and each begins executing +/// immediately (subject to ). Call +/// to seal +/// registration, await the branches according to the +/// , and obtain the aggregate result. +/// +/// +/// This is an additive alternative to the homogeneous +/// +/// overloads, which accept a complete branch list up front and share one result +/// type. Use CreateParallel when branches return unrelated types, or when +/// branches are discovered incrementally (for example, tool calls derived from a +/// checkpointed plan) and earlier branches should start before later ones are known. +/// +/// Deterministic replay. Branch identity is positional: the n-th +/// +/// call reuses the n-th deterministic operation ID. Workflow code must therefore +/// register the same branches in the same order across invocations — produce any +/// dynamic branch list inside a checkpointed +/// so replay sees the same set. +/// +/// +/// Disposal. seals and +/// completes the operation if +/// was not called, +/// so an await using block always writes the parallel's terminal checkpoint. +/// Calling +/// explicitly is recommended so you can capture the aggregate result. +/// +/// +public interface IDurableParallel : IAsyncDisposable +{ + /// + /// Registers a branch and immediately begins executing it (respecting + /// ). Returns a typed + /// handle for retrieving the branch's result. + /// + /// + /// The branch runs inside its own child context with a deterministic + /// operation-ID space; its result is serialized to a checkpoint via the + /// registered on + /// . Per-branch + /// failures are captured on the handle and aggregated into the + /// result — a + /// branch failure never throws out of this method. + /// + /// The branch's result type. + /// + /// Human-readable branch name. Required; surfaces on + /// OperationUpdate.Name and must remain stable at a given branch index + /// across deployments (a drift is a non-deterministic-execution error). + /// + /// + /// The branch body. Receives its own and a + /// linking the SDK's + /// workflow-shutdown signal with the operation's completion-policy + /// short-circuit, and returns the branch's result. + /// + /// + /// Optional serializer for this branch's result payload. When + /// null (default), the branch uses the operation-level serializer — + /// if set on + /// , otherwise + /// the globally-registered on + /// . Because each branch + /// declares its own result type, a per-branch serializer lets one branch use a + /// bespoke serializer (for example a source-generated context for AOT) without + /// affecting sibling branches. It is part of the deterministic definition: the same + /// branch must be able to deserialize a result it previously serialized, so pass the + /// same serializer at a given branch index across replays. + /// + /// A typed handle for awaiting the branch's result. + /// + /// or is null. + /// + /// + /// The operation has already been sealed by + /// . + /// + /// + /// The operation has already been disposed. + /// + /// + /// On replay, the at this branch index differs from the + /// name recorded in the checkpoint for a previous invocation (branch name drift). + /// + IParallelBranch Branch( + string name, + Func> func, + Amazon.Lambda.Core.ILambdaSerializer? serializer = null); + + /// + /// Seals registration (no further branches may be added), awaits the + /// registered branches according to the + /// , checkpoints the aggregate + /// outcome, and returns it. Idempotent — repeated calls return the same result. + /// + /// + /// Like the homogeneous parallel API, this never throws on per-branch failure: + /// inspect / + /// , or await individual branch + /// handles, to observe failures. It does propagate workflow-level errors (for + /// example ) and cancellation. + /// + /// The does not interrupt in-flight branch + /// settlement. Because branches begin executing when they are registered (before + /// CompleteAsync is called), this token is neither retroactively linked + /// into already-running branch bodies nor into the wait for them to settle — + /// those observe the SDK's workflow-shutdown signal (and the completion-policy + /// short-circuit) instead. This call blocks until every dispatched branch reaches + /// its terminal checkpoint; the token is observed only after that, so cancellation + /// is surfaced just before the aggregate result would be returned rather than + /// cutting the wait short. Dispatched branches always run to a terminal checkpoint + /// so replay stays deterministic, matching . + /// + /// + /// A token to observe for cancellation. + /// The aggregate summarizing branch outcomes. + Task CompleteAsync(CancellationToken cancellationToken = default); +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/IParallelBranch.cs b/Libraries/src/Amazon.Lambda.DurableExecution/IParallelBranch.cs new file mode 100644 index 000000000..2300b3b76 --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution/IParallelBranch.cs @@ -0,0 +1,65 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using System.Runtime.CompilerServices; + +namespace Amazon.Lambda.DurableExecution; + +/// +/// A typed handle to a single branch registered on an +/// via . +/// Unlike the homogeneous +/// API — where every branch shares one result type T — each branch on an +/// declares its own result type, so a single +/// parallel operation can mix, for example, an InventoryReservation branch +/// with a PaymentAuthorization branch. +/// +/// +/// The handle is await-able: await branch yields the branch's typed +/// result once it succeeds, or rethrows the branch's failure (a +/// ) if it failed. Awaiting a branch that was +/// skipped by the operation's short-circuit (its +/// is ) throws a +/// — inspect before +/// awaiting when a completion policy may skip branches. +/// +/// Typically you await the handle after +/// +/// has sealed and resolved the operation, mirroring the Java SDK's +/// future.get() after the try-with-resources block. A branch may +/// still be awaited earlier; the await simply completes when the branch does. +/// +/// +/// The branch's result type. +public interface IParallelBranch +{ + /// + /// The branch name supplied at registration. Surfaces on the wire + /// OperationUpdate.Name field and in execution traces. + /// + string Name { get; } + + /// + /// Zero-based registration order of this branch within its parallel + /// operation. Stable across replays. The branch's deterministic operation ID + /// is derived from the one-based position (hash("{parentId}-{Index+1}")), + /// so the first branch (Index 0) uses suffix 1. + /// + int Index { get; } + + /// + /// The branch's outcome. until the + /// branch settles (and permanently for a branch skipped by a completion-policy + /// short-circuit), then or + /// . + /// + BatchItemStatus Status { get; } + + /// + /// Enables await branch. Yields the branch's typed result on success, + /// rethrows its on failure, or throws a + /// if the branch was skipped. + /// + /// An awaiter over the branch's result. + TaskAwaiter GetAwaiter(); +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummaryCodec.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummaryCodec.cs new file mode 100644 index 000000000..032c421f5 --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummaryCodec.cs @@ -0,0 +1,86 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using System.Text; +using System.Text.Json; + +namespace Amazon.Lambda.DurableExecution.Internal; + +/// +/// Shared (de)serialization primitives for the payload +/// stored on a concurrent operation's parent CONTEXT checkpoint. Centralising the +/// wire-string mappings and the payload serialize / overflow check here keeps the +/// batch () and incremental +/// () parallel implementations in exact +/// agreement on the on-the-wire format, so a checkpoint written by one is +/// reconstructable by the other. +/// +internal static class BatchSummaryCodec +{ + /// + /// Serializes the summary to its JSON payload using the source-generated + /// (trim/AOT safe). + /// + public static string ToPayload(BatchSummary summary) + => JsonSerializer.Serialize(summary, BatchJsonContext.Default.BatchSummary); + + /// + /// True when exceeds the per-operation checkpoint + /// byte limit and must be re-emitted stripped (statuses only) with + /// ReplayChildren=true. + /// + public static bool IsOverflow(string payload) + => Encoding.UTF8.GetByteCount(payload) > DurableConstants.MaxOperationCheckpointBytes; + + /// + /// Deserializes a from a checkpoint payload, + /// tolerating null/empty/corrupt payloads by returning null (callers + /// fall back to inferring per-unit status from child checkpoints). + /// + public static BatchSummary? ParseSummary(string? payload) + { + if (string.IsNullOrEmpty(payload)) return null; + try + { + return JsonSerializer.Deserialize(payload, BatchJsonContext.Default.BatchSummary); + } + catch (JsonException) + { + // Tolerate older / corrupted payloads — fall back to inferring status + // from per-unit checkpoints. + return null; + } + } + + public static string SerializeStatus(BatchItemStatus status) => status switch + { + BatchItemStatus.Succeeded => "SUCCEEDED", + BatchItemStatus.Failed => "FAILED", + BatchItemStatus.Started => "STARTED", + _ => throw new ArgumentOutOfRangeException(nameof(status)) + }; + + public static BatchItemStatus DeserializeStatus(string? wire) => wire switch + { + "SUCCEEDED" => BatchItemStatus.Succeeded, + "FAILED" => BatchItemStatus.Failed, + "STARTED" => BatchItemStatus.Started, + _ => BatchItemStatus.Started + }; + + public static string SerializeCompletionReason(CompletionReason reason) => reason switch + { + CompletionReason.AllCompleted => "ALL_COMPLETED", + CompletionReason.MinSuccessfulReached => "MIN_SUCCESSFUL_REACHED", + CompletionReason.FailureToleranceExceeded => "FAILURE_TOLERANCE_EXCEEDED", + _ => throw new ArgumentOutOfRangeException(nameof(reason)) + }; + + public static CompletionReason DeserializeCompletionReason(string? wire) => wire switch + { + "ALL_COMPLETED" => CompletionReason.AllCompleted, + "MIN_SUCCESSFUL_REACHED" => CompletionReason.MinSuccessfulReached, + "FAILURE_TOLERANCE_EXCEEDED" => CompletionReason.FailureToleranceExceeded, + _ => CompletionReason.AllCompleted + }; +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CompletionPolicy.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CompletionPolicy.cs index b1573a593..a02cb153b 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CompletionPolicy.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CompletionPolicy.cs @@ -44,8 +44,8 @@ public CompletionPolicy(CompletionConfig config) /// failed. Reads slightly-stale counters by design (see the dispatch loop); /// is the authoritative verdict. /// - public bool ShouldStopDispatching(int succeeded, int failed, int totalBranches) - => MinSuccessfulReached(succeeded) || FailureToleranceExceeded(failed, totalBranches); + public bool ShouldStopDispatching(int succeeded, int failed, int totalBranches, bool evaluatePercentage = true) + => MinSuccessfulReached(succeeded) || FailureToleranceExceeded(failed, totalBranches, evaluatePercentage); /// /// Final verdict once all dispatched branches have settled. Failure tolerance @@ -77,7 +77,7 @@ private bool MinSuccessfulReached(int succeeded) // ToleratedFailureCount = 0) and the empty config are therefore equivalent; // CompletionConfig.AllCompleted() sets ToleratedFailureCount = int.MaxValue to // stay lenient. - private bool FailureToleranceExceeded(int failed, int totalBranches) + private bool FailureToleranceExceeded(int failed, int totalBranches, bool evaluatePercentage = true) { if (_failFastOnAnyFailure) return failed > 0; @@ -85,7 +85,7 @@ private bool FailureToleranceExceeded(int failed, int totalBranches) if (_toleratedFailureCount is { } tfc && failed > tfc) return true; - if (_toleratedFailurePercentage is { } tfp && totalBranches > 0 && + if (evaluatePercentage && _toleratedFailurePercentage is { } tfp && totalBranches > 0 && (double)failed / totalBranches > tfp) { return true; diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs index 44658262f..b8000b077 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs @@ -3,7 +3,6 @@ using System.IO; using System.Text; -using System.Text.Json; using Amazon.Lambda; using Amazon.Lambda.Core; using SdkContextOptions = Amazon.Lambda.Model.ContextOptions; @@ -390,7 +389,7 @@ private async Task> ReplayChildrenAsync(Operation frozen, Cancel { cancellationToken.ThrowIfCancellationRequested(); - var summary = ParseSummary(frozen.ContextDetails?.Result); + var summary = BatchSummaryCodec.ParseSummary(frozen.ContextDetails?.Result); var unitCount = UnitCount; var items = new List>(unitCount); @@ -401,7 +400,7 @@ private async Task> ReplayChildrenAsync(Operation frozen, Cancel // Frozen per-unit status is authoritative. var status = summaryEntry != null - ? DeserializeStatus(summaryEntry.Status) + ? BatchSummaryCodec.DeserializeStatus(summaryEntry.Status) : BatchItemStatus.Started; // Same unit-name drift check as ReconstructFromCheckpoints: code must @@ -459,7 +458,7 @@ private async Task> ReplayChildrenAsync(Operation frozen, Cancel // Completion reason is pinned from the frozen summary; fall back to // recomputing only if the summary is absent/corrupt. var completionReason = summary != null - ? DeserializeCompletionReason(summary.CompletionReason) + ? BatchSummaryCodec.DeserializeCompletionReason(summary.CompletionReason) : ComputeCompletionReason(items, unitCount); // No re-checkpoint: the parent is already terminal in state. Return the @@ -707,7 +706,7 @@ BatchSummary BuildSummary(bool includeInline) var s = new BatchSummary { - CompletionReason = SerializeCompletionReason(completionReason), + CompletionReason = BatchSummaryCodec.SerializeCompletionReason(completionReason), Units = new List(UnitCount) }; for (var i = 0; i < UnitCount; i++) @@ -718,7 +717,7 @@ BatchSummary BuildSummary(bool includeInline) { Index = i, Name = item?.Name ?? unitName, - Status = SerializeStatus(item?.Status ?? BatchItemStatus.Started) + Status = BatchSummaryCodec.SerializeStatus(item?.Status ?? BatchItemStatus.Started) }; // Persist each unit's result/error inline on the parent summary — // for BOTH Nested and Flat units. The service collapses completed @@ -741,18 +740,18 @@ BatchSummary BuildSummary(bool includeInline) } var summary = BuildSummary(includeInline: true); - var payload = JsonSerializer.Serialize(summary, BatchJsonContext.Default.BatchSummary); + var payload = BatchSummaryCodec.ToPayload(summary); // Overflow: the inline per-unit results pushed the summary over the // checkpoint limit. Re-emit a stripped summary (statuses only) and flag // ReplayChildren so replay reconstructs the values by re-executing units. // Applies to both Nested and Flat now that both inline their results. var overflow = - Encoding.UTF8.GetByteCount(payload) > DurableConstants.MaxOperationCheckpointBytes; + BatchSummaryCodec.IsOverflow(payload); if (overflow) { summary = BuildSummary(includeInline: false); - payload = JsonSerializer.Serialize(summary, BatchJsonContext.Default.BatchSummary); + payload = BatchSummaryCodec.ToPayload(summary); } // Always checkpoint as SUCCEED — even when FailureToleranceExceeded. @@ -776,7 +775,7 @@ await EnqueueAsync(new SdkOperationUpdate private IBatchResult ReconstructFromCheckpoints(Operation parent) { - var summary = ParseSummary(parent.ContextDetails?.Result); + var summary = BatchSummaryCodec.ParseSummary(parent.ContextDetails?.Result); var items = new List>(UnitCount); for (var i = 0; i < UnitCount; i++) @@ -787,7 +786,7 @@ private IBatchResult ReconstructFromCheckpoints(Operation parent) var summaryEntry = summary?.Units.FirstOrDefault(b => b.Index == i); BatchItemStatus status = summaryEntry != null - ? DeserializeStatus(summaryEntry.Status) + ? BatchSummaryCodec.DeserializeStatus(summaryEntry.Status) : InferStatusFromChildOp(childOp); // Prefer the name that was checkpointed at the moment the batch @@ -866,7 +865,7 @@ private IBatchResult ReconstructFromCheckpoints(Operation parent) } var completionReason = summary != null - ? DeserializeCompletionReason(summary.CompletionReason) + ? BatchSummaryCodec.DeserializeCompletionReason(summary.CompletionReason) : ComputeCompletionReason(items, UnitCount); return new BatchResult(items, completionReason); @@ -883,53 +882,6 @@ private static BatchItemStatus InferStatusFromChildOp(Operation? childOp) }; } - private static BatchSummary? ParseSummary(string? payload) - { - if (string.IsNullOrEmpty(payload)) return null; - try - { - return JsonSerializer.Deserialize(payload, BatchJsonContext.Default.BatchSummary); - } - catch (JsonException) - { - // Tolerate older / corrupted payloads — fall back to inferring status - // from per-unit checkpoints. - return null; - } - } - - private static string SerializeStatus(BatchItemStatus status) => status switch - { - BatchItemStatus.Succeeded => "SUCCEEDED", - BatchItemStatus.Failed => "FAILED", - BatchItemStatus.Started => "STARTED", - _ => throw new ArgumentOutOfRangeException(nameof(status)) - }; - - private static BatchItemStatus DeserializeStatus(string? wire) => wire switch - { - "SUCCEEDED" => BatchItemStatus.Succeeded, - "FAILED" => BatchItemStatus.Failed, - "STARTED" => BatchItemStatus.Started, - _ => BatchItemStatus.Started - }; - - private static string SerializeCompletionReason(CompletionReason reason) => reason switch - { - CompletionReason.AllCompleted => "ALL_COMPLETED", - CompletionReason.MinSuccessfulReached => "MIN_SUCCESSFUL_REACHED", - CompletionReason.FailureToleranceExceeded => "FAILURE_TOLERANCE_EXCEEDED", - _ => throw new ArgumentOutOfRangeException(nameof(reason)) - }; - - private static CompletionReason DeserializeCompletionReason(string? wire) => wire switch - { - "ALL_COMPLETED" => CompletionReason.AllCompleted, - "MIN_SUCCESSFUL_REACHED" => CompletionReason.MinSuccessfulReached, - "FAILURE_TOLERANCE_EXCEEDED" => CompletionReason.FailureToleranceExceeded, - _ => CompletionReason.AllCompleted - }; - private T DeserializeResult(string serialized) { var bytes = Encoding.UTF8.GetBytes(serialized); diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs new file mode 100644 index 000000000..e0ce3a541 --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/IncrementalParallelOperation.cs @@ -0,0 +1,898 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using System.IO; +using System.Runtime.CompilerServices; +using System.Text; +using Amazon.Lambda; +using Amazon.Lambda.Core; +using SdkContextOptions = Amazon.Lambda.Model.ContextOptions; +using SdkOperationUpdate = Amazon.Lambda.Model.OperationUpdate; + +namespace Amazon.Lambda.DurableExecution.Internal; + +/// +/// Which replay branch the operation is on, decided once from the parent CONTEXT +/// checkpoint at construction. +/// +internal enum ParallelExecutionMode +{ + /// No terminal parent checkpoint: run branches (fresh, or STARTED/PENDING + /// where each branch replays from its own checkpoint). The parent SUCCEED is + /// written by . + Run, + + /// Parent already terminal: reconstruct branch outcomes from the frozen + /// (re-running a branch only to recover a value that + /// was stripped on overflow). The parent is NOT re-checkpointed. + Terminal +} + +/// +/// Type-erased outcome of a single branch, gathered by the orchestrator to build +/// the parent without knowing each branch's T. +/// +internal readonly struct BranchOutcome +{ + public int Index { get; init; } + public string? Name { get; init; } + public BatchItemStatus Status { get; init; } + + /// Serialized branch result (succeeded branches only). + public string? SerializedResult { get; init; } + + /// Branch error (failed branches only). + public ErrorObject? Error { get; init; } + + public static BranchOutcome Success(int index, string? name, string? serialized) => + new() { Index = index, Name = name, Status = BatchItemStatus.Succeeded, SerializedResult = serialized }; + + public static BranchOutcome Failure(int index, string? name, ErrorObject error) => + new() { Index = index, Name = name, Status = BatchItemStatus.Failed, Error = error }; + + public static BranchOutcome Skipped(int index, string? name) => + new() { Index = index, Name = name, Status = BatchItemStatus.Started }; +} + +/// +/// Type-erased view the orchestrator holds over each branch handle, so it can +/// await settlement and read per-branch identity/status without the branch's +/// generic parameter. +/// +internal interface IParallelBranchController +{ + int Index { get; } + string Name { get; } + BatchItemStatus Status { get; } + + /// + /// Completes (never faults for a graceful per-branch failure) with the branch's + /// . Faults only for workflow-level errors + /// (e.g. ) or control-token + /// cancellation, which the orchestrator surfaces. + /// + Task Settlement { get; } +} + +/// +/// Typed, awaitable handle for a single branch of an +/// . Backs the public +/// ; also exposes the type-erased +/// the orchestrator uses to aggregate. +/// +internal sealed class IncrementalParallelBranch : IParallelBranch, IParallelBranchController +{ + private readonly TaskCompletionSource _result = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly ILambdaSerializer _serializer; + private readonly string _childSubType; + + // Frozen status wins over the live re-run outcome on the overflow-recovery path + // (Terminal mode): the checkpointed verdict is authoritative even if a + // non-deterministic body re-executes to a different result. + private BatchItemStatus? _frozenStatus; + private volatile int _status = (int)BatchItemStatus.Started; + + public IncrementalParallelBranch(int index, string name, ILambdaSerializer serializer, string childSubType) + { + Index = index; + Name = name; + _serializer = serializer; + _childSubType = childSubType; + + // Per-branch failures are intentionally consumed via CompleteAsync, so a + // caller may never await this handle. Observe the fault here so a discarded + // failed handle can never surface as an UnobservedTaskException. + _ = _result.Task.ContinueWith( + static t => { _ = t.Exception; }, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + + public string Name { get; } + public int Index { get; } + public BatchItemStatus Status => _frozenStatus ?? (BatchItemStatus)_status; + public Task Settlement { get; private set; } = null!; + + public TaskAwaiter GetAwaiter() => _result.Task.GetAwaiter(); + + /// + /// Run mode: wrap a live child-context execution. + /// is set only on the overflow-recovery path, where the checkpointed status is + /// authoritative and the run merely recovers the stripped value. + /// + public void Launch( + Func> run, + CancellationToken shortCircuitToken, + CancellationToken controlToken, + BatchItemStatus? frozenStatus = null) + { + _frozenStatus = frozenStatus; + Settlement = ExecuteAsync(run, shortCircuitToken, controlToken); + } + + /// + /// Terminal mode: resolve the branch directly from the frozen summary without + /// running it (the common, non-overflow reconstruct path). + /// + public void ResolveFromInline(BatchItemStatus status, string? serializedResult, ErrorObject? error) + { + _frozenStatus = status; + switch (status) + { + case BatchItemStatus.Succeeded: + _result.TrySetResult(Deserialize(serializedResult)); + Settlement = Task.FromResult(BranchOutcome.Success(Index, Name, serializedResult)); + break; + case BatchItemStatus.Failed: + _result.TrySetException(BuildError(error)); + Settlement = Task.FromResult(BranchOutcome.Failure(Index, Name, error ?? new ErrorObject { ErrorMessage = "Branch failed" })); + break; + default: + _result.TrySetException(SkippedError()); + Settlement = Task.FromResult(BranchOutcome.Skipped(Index, Name)); + break; + } + } + + private async Task ExecuteAsync( + Func> run, + CancellationToken shortCircuitToken, + CancellationToken controlToken) + { + try + { + var value = await run().ConfigureAwait(false); + if (_frozenStatus is null) _status = (int)BatchItemStatus.Succeeded; + _result.TrySetResult(value); + return BranchOutcome.Success(Index, Name, Serialize(value)); + } + catch (ChildContextException ex) + { + if (_frozenStatus is null) _status = (int)BatchItemStatus.Failed; + _result.TrySetException(ex); + return BranchOutcome.Failure(Index, Name, ErrorObject.FromException(ex)); + } + catch (DurableExecutionException ex) + { + // Workflow-level error (e.g. NonDeterministicExecutionException): not a + // graceful per-branch failure. Fault the settlement so the orchestrator + // surfaces it out of CompleteAsync. Also fault _result before rethrowing + // so a caller that catches the workflow-level fault out of CompleteAsync + // and then awaits this branch handle observes the same fault instead of + // hanging forever on a never-completed result (every other arm below + // completes _result). + _result.TrySetException(ex); + throw; + } + catch (OperationCanceledException) + when (shortCircuitToken.IsCancellationRequested && !controlToken.IsCancellationRequested + && _frozenStatus is null) + { + // Cooperative bail: a sibling satisfied the CompletionConfig before this + // branch acquired its concurrency slot (or its body honored the bail + // token). Record it as skipped — never a failure. + // + // Excluded when _frozenStatus is set: an overflow-recovery re-run is + // launched only to recover a value the frozen summary already recorded + // as Succeeded/Failed. It is isolated from _shortCircuitCts (see + // LaunchRunBranch), so this arm should not fire for it — but guard + // regardless so a stray short-circuit can never resolve _result to a + // SkippedError while Status reports the frozen terminal verdict, which + // would make `await branch` throw for a branch whose Status==Succeeded. + if (_frozenStatus is null) _status = (int)BatchItemStatus.Started; + _result.TrySetException(SkippedError()); + return BranchOutcome.Skipped(Index, Name); + } + catch (OperationCanceledException) when (controlToken.IsCancellationRequested) + { + // Caller-cancel or workflow shutdown: propagate. + _result.TrySetCanceled(); + throw; + } + catch (OperationCanceledException ex) + { + var wrapped = Wrap(ex); + if (_frozenStatus is null) _status = (int)BatchItemStatus.Failed; + _result.TrySetException(wrapped); + return BranchOutcome.Failure(Index, Name, ErrorObject.FromException(wrapped)); + } + catch (Exception ex) + { + var wrapped = Wrap(ex); + if (_frozenStatus is null) _status = (int)BatchItemStatus.Failed; + _result.TrySetException(wrapped); + return BranchOutcome.Failure(Index, Name, ErrorObject.FromException(wrapped)); + } + } + + private ChildContextException Wrap(Exception ex) => new(ex.Message, ex) + { + SubType = _childSubType, + ErrorType = ex.GetType().FullName + }; + + private ChildContextException BuildError(ErrorObject? error) => + new(error?.ErrorMessage ?? "Branch failed") + { + SubType = _childSubType, + ErrorType = error?.ErrorType, + ErrorData = error?.ErrorData, + OriginalStackTrace = error?.StackTrace + }; + + private DurableExecutionException SkippedError() => new( + $"Parallel branch '{Name}' (index {Index}) did not execute: the parallel " + + $"operation completed before it started (completion-policy short-circuit). " + + $"Inspect the branch's Status before awaiting it."); + + private string Serialize(T value) + { + using var ms = new MemoryStream(); + _serializer.Serialize(value, ms); + return Encoding.UTF8.GetString(ms.ToArray()); + } + + private T Deserialize(string? serialized) + { + if (serialized == null) return default!; + var bytes = Encoding.UTF8.GetBytes(serialized); + using var ms = new MemoryStream(bytes); + return _serializer.Deserialize(ms); + } +} + +/// +/// Incremental, heterogeneous parallel orchestrator implementing +/// . Each branch runs as a +/// under the SAME deterministic child +/// operation-ID scheme (hash("{parentId}-{index+1}"), where index is +/// the zero-based branch registration order, so the ID suffix is one-based — +/// positions 1..n) and the SAME parent +/// checkpoint shape as the batch +/// , so a checkpoint written by one is +/// reconstructable by the other. Branch identity is positional: register the same +/// branches in the same order across replays. +/// +internal sealed class IncrementalParallelOperation : IDurableParallel +{ + private readonly string _operationId; + private readonly string? _name; + private readonly string? _parentId; + private readonly CompletionPolicy _policy; + private readonly int? _maxConcurrency; + private readonly bool _isVirtual; + // Operation-level default serializer, resolved LAZILY. A workflow that overrides + // the serializer on every Branch must not be forced to register a global + // serializer just to construct the operation (the AOT/per-branch scenario), so + // the factory — which may call LambdaSerializerHelper.GetRequired and throw when + // no global serializer exists — is invoked only when a branch actually falls back + // to this default. Memoized in _defaultSerializer under _lock. + private readonly Func _defaultSerializerFactory; + private ILambdaSerializer? _defaultSerializer; + private readonly Func _childContextFactory; + private readonly ExecutionState _state; + private readonly TerminationManager _termination; + private readonly WorkflowCancellation _workflowCancellation; + private readonly string _durableExecutionArn; + private readonly CheckpointBatcher? _batcher; + + private readonly object _lock = new(); + private readonly List _branches = new(); + private readonly SemaphoreSlim? _semaphore; + private readonly CancellationTokenSource _shortCircuitCts = new(); + private readonly CancellationTokenSource _dispatchCts; + + private readonly ParallelExecutionMode _mode; + private readonly BatchSummary? _frozenSummary; + private readonly Task _startTask; + + private int _succeeded; + private int _failed; + private int _registeredCount; + private bool _sealed; + private volatile bool _sealedVolatile; + private bool _disposed; + private Task? _completion; + + public IncrementalParallelOperation( + string operationId, + string? name, + string? parentId, + ParallelConfig config, + Func defaultSerializerFactory, + Func childContextFactory, + ExecutionState state, + TerminationManager termination, + WorkflowCancellation workflowCancellation, + string durableExecutionArn, + CheckpointBatcher? batcher = null) + { + _operationId = operationId; + _name = name; + _parentId = parentId; + _policy = new CompletionPolicy(config.CompletionConfig); + _maxConcurrency = config.MaxConcurrency; + _isVirtual = config.NestingType == NestingType.Flat; + _defaultSerializerFactory = defaultSerializerFactory; + _childContextFactory = childContextFactory; + _state = state; + _termination = termination; + _workflowCancellation = workflowCancellation; + _durableExecutionArn = durableExecutionArn; + _batcher = batcher; + + _semaphore = _maxConcurrency is { } mc ? new SemaphoreSlim(mc, mc) : null; + _dispatchCts = CancellationTokenSource.CreateLinkedTokenSource( + _shortCircuitCts.Token, workflowCancellation.Token); + + // The parent operation position has been reached — mirror the base + // DurableOperation.ExecuteAsync bookkeeping for the parent CONTEXT op. + _state.ValidateReplayConsistency(_operationId, OperationTypes.Context, _name); + _state.TrackReplay(_operationId); + + var existing = _state.GetOperation(_operationId); + if (existing == null) + { + // Fresh: emit the parent CONTEXT START so the service has a parent + // record if a branch suspends. Enqueued once here so it is ordered + // before any branch's child START. + _mode = ParallelExecutionMode.Run; + _startTask = EnqueueAsync(new SdkOperationUpdate + { + Id = _operationId, + ParentId = _parentId, + Type = OperationTypes.Context, + Action = OperationAction.START, + SubType = OperationSubTypes.Parallel, + Name = _name + }); + } + else + { + // Mirror ConcurrentOperation.ReplayAsync: only SUCCEEDED reconstructs, + // only STARTED/PENDING re-run, and any other status is a replay + // mismatch. The parent parallel only ever checkpoints SUCCEED, so a + // FAILED/CANCELLED/STOPPED/TIMED_OUT parent must never be silently + // re-run (which would overwrite the prior terminal outcome). + switch (existing.Status) + { + case OperationStatuses.Succeeded: + _mode = ParallelExecutionMode.Terminal; + _frozenSummary = BatchSummaryCodec.ParseSummary(existing.ContextDetails?.Result); + _startTask = Task.CompletedTask; + break; + case OperationStatuses.Started: + case OperationStatuses.Pending: + // Children replay from their own checkpoints; the parent START + // is not re-emitted (the original is authoritative). + _mode = ParallelExecutionMode.Run; + _startTask = Task.CompletedTask; + break; + default: + throw new NonDeterministicExecutionException( + $"Parallel operation '{_name ?? _operationId}' has unexpected status " + + $"'{existing.Status}' on replay."); + } + } + } + + public IParallelBranch Branch( + string name, + Func> func, + ILambdaSerializer? serializer = null) + { + if (name == null) throw new ArgumentNullException(nameof(name)); + if (func == null) throw new ArgumentNullException(nameof(func)); + + lock (_lock) + { + if (_disposed) throw new ObjectDisposedException(nameof(IDurableParallel)); + if (_sealed) + throw new InvalidOperationException( + "Cannot register a branch after the parallel operation has been sealed by CompleteAsync() or disposal."); + + var index = _branches.Count; // zero-based branch index + var childOpId = OperationIdGenerator.HashOperationId($"{_operationId}-{index + 1}"); + // Per-branch serializer override, else the operation-level default + // (ParallelConfig.ItemSerializer ?? the globally-registered serializer), + // resolved lazily here so a workflow overriding the serializer on every + // branch never triggers the global-serializer lookup. Memoized under _lock. + var branchSerializer = serializer ?? (_defaultSerializer ??= _defaultSerializerFactory()); + var handle = new IncrementalParallelBranch(index, name, branchSerializer, OperationSubTypes.ParallelBranch); + + var summaryEntry = FindSummaryUnit(index); + + // Strict name-drift check: a branch's name must be stable at its index + // across deployments (matches the batch Parallel reconstruct check). + if (summaryEntry?.Name != null && summaryEntry.Name != name) + { + throw new NonDeterministicExecutionException( + $"Non-deterministic execution detected for parallel branch {index} of operation " + + $"'{_name ?? _operationId}': expected checkpointed name '{summaryEntry.Name}' but " + + $"the current registration used '{name}'. Code must not change the order or name of " + + $"branches between deployments."); + } + + if (_mode == ParallelExecutionMode.Terminal) + { + ResolveTerminalBranch(handle, name, childOpId, func, summaryEntry, branchSerializer); + } + else + { + LaunchRunBranch(handle, name, childOpId, func, branchSerializer); + } + + _branches.Add(handle); + _registeredCount = _branches.Count; + return handle; + } + } + + public Task CompleteAsync(CancellationToken cancellationToken = default) + { + lock (_lock) + { + // Cache the in-progress task (not just the finished result) so two + // concurrent CompleteAsync calls — or DisposeAsync racing one — share a + // single completion and enqueue exactly one parent SUCCEED. + if (_completion != null) return _completion; + _sealed = true; + _sealedVolatile = true; + _completion = CompleteCoreAsync(cancellationToken); + return _completion; + } + } + + private async Task CompleteCoreAsync(CancellationToken cancellationToken) + { + // Registration is sealed: the denominator is now known, so re-evaluate the + // completion policy (including percentage-based tolerance, which is + // suppressed pre-seal) and signal any in-flight branches to bail. + if (ShouldStopDispatchingNow()) + { + try { _shortCircuitCts.Cancel(); } + catch (ObjectDisposedException) { } + } + + // Ensure the parent START is durably enqueued even for an empty operation. + await _startTask.ConfigureAwait(false); + + var controllers = SnapshotControllers(); + + // Await every branch settlement. Task.WhenAll surfaces only the first + // exception; swallow here and inspect each below so a workflow-level fault + // is surfaced deterministically and graceful failures aggregate. + if (controllers.Count > 0) + { + try { await Task.WhenAll(controllers.Select(c => c.Settlement)).ConfigureAwait(false); } + catch { /* inspected below */ } + } + + foreach (var c in controllers) + { + var s = c.Settlement; + if (s.IsFaulted && s.Exception is { } agg) + { + foreach (var inner in agg.InnerExceptions) + { + if (inner is DurableExecutionException dex && inner is not ChildContextException) + throw dex; + } + } + } + + // A torn-down operation propagates cancellation rather than a synthesized verdict. + _workflowCancellation.Token.ThrowIfCancellationRequested(); + cancellationToken.ThrowIfCancellationRequested(); + + IBatchResult result = _mode == ParallelExecutionMode.Terminal + ? BuildTerminalResult(controllers) + : await BuildAndCheckpointRunResultAsync(controllers, cancellationToken).ConfigureAwait(false); + + return result; + } + + public async ValueTask DisposeAsync() + { + Task? completion; + lock (_lock) + { + if (_disposed) return; + _disposed = true; + // If CompleteAsync was never called, complete now so the parent's + // terminal checkpoint is written — otherwise replay would see a STARTED + // parent forever and re-run the whole operation. + completion = _completion; + } + + try + { + completion ??= CompleteAsync(CancellationToken.None); + await completion.ConfigureAwait(false); + } + catch + { + // DisposeAsync must never throw. A completion fault (e.g. a + // NonDeterministicExecutionException, or the secondary effect of a + // Branch that already threw during registration) is either + // already surfaced to a caller that awaited CompleteAsync, or will + // resurface on the next invocation's replay. Swallow it here so + // `await using` teardown stays clean. + } + finally + { + _shortCircuitCts.Dispose(); + _dispatchCts.Dispose(); + _semaphore?.Dispose(); + } + } + + // ── Run mode ──────────────────────────────────────────────────────── + + private void LaunchRunBranch( + IncrementalParallelBranch handle, + string name, + string childOpId, + Func> func, + ILambdaSerializer branchSerializer, + BatchItemStatus? frozenStatus = null) + { + // An overflow-recovery re-run (frozenStatus set) exists only to recover a + // value the frozen summary already recorded as terminal; it MUST run to + // completion. A completion-policy short-circuit (OnBranchSettled → + // _shortCircuitCts.Cancel()) must never cancel it — a cancelled recovery + // branch would hit the cooperative-bail arm and lose the recovered value + // even though its Status is the frozen Succeeded/Failed. So isolate it from + // _shortCircuitCts/_dispatchCts: it observes only workflow shutdown, and is + // never handed a bail token. Regular Run-mode branches honor the + // short-circuit exactly as before. + var isRecovery = frozenStatus.HasValue; + var dispatchToken = isRecovery ? _workflowCancellation.Token : _dispatchCts.Token; + var bailToken = isRecovery ? CancellationToken.None : _shortCircuitCts.Token; + + async Task Run() + { + // Parent START must be enqueued before this branch's child START. + await _startTask.ConfigureAwait(false); + + if (_semaphore != null) + { + await _semaphore.WaitAsync(dispatchToken).ConfigureAwait(false); + } + + try + { + // A short-circuit may have fired while waiting on the semaphore. + dispatchToken.ThrowIfCancellationRequested(); + + var childOp = new ChildContextOperation( + childOpId, + name, + _operationId, + func, + new ChildContextConfig { SubType = OperationSubTypes.ParallelBranch }, + branchSerializer, + _childContextFactory, + _state, + _termination, + _workflowCancellation, + _durableExecutionArn, + _batcher, + bailToken, + isVirtual: _isVirtual); + + // Branch child ops receive CancellationToken.None here — they re-link + // workflow-shutdown and the cooperative-bail token internally, and + // their checkpoint writes must not observe shutdown mid-flush. + return await childOp.ExecuteAsync(CancellationToken.None).ConfigureAwait(false); + } + finally + { + _semaphore?.Release(); + } + } + + handle.Launch(Run, bailToken, _workflowCancellation.Token, frozenStatus); + ObserveSettlement(handle.Settlement); + } + + private void ObserveSettlement(Task settlement) + { + _ = settlement.ContinueWith( + OnBranchSettled, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + + private void OnBranchSettled(Task settlement) + { + if (settlement.Status != TaskStatus.RanToCompletion) return; + + switch (settlement.Result.Status) + { + case BatchItemStatus.Succeeded: Interlocked.Increment(ref _succeeded); break; + case BatchItemStatus.Failed: Interlocked.Increment(ref _failed); break; + } + + // The deciding completion usually lands after all currently-registered + // branches were dispatched, so re-check here and signal stragglers to bail. + if (ShouldStopDispatchingNow()) + { + try { _shortCircuitCts.Cancel(); } + catch (ObjectDisposedException) { } + } + } + + // During incremental registration the "total" is the number registered so far; + // percentage-based tolerance is evaluated against that running total. MinSuccessful + // and count-based tolerance don't depend on the total. + // During incremental registration the denominator is unknown, so percentage-based + // failure tolerance must NOT drive short-circuiting (a premature ratio like 1/1 + // could skip branches that would have lowered the final ratio). MinSuccessful and + // absolute-count tolerance are denominator-independent and always apply; the + // percentage component is enabled only once registration is sealed, and the final + // verdict (ComputeCompletionReason) always uses the true total. + private bool ShouldStopDispatchingNow() => _policy.ShouldStopDispatching( + Volatile.Read(ref _succeeded), Volatile.Read(ref _failed), Volatile.Read(ref _registeredCount), + evaluatePercentage: _sealedVolatile); + + private async Task BuildAndCheckpointRunResultAsync( + IReadOnlyList controllers, + CancellationToken cancellationToken) + { + var outcomes = new List(controllers.Count); + foreach (var c in controllers) + { + var s = c.Settlement; + outcomes.Add(s.Status == TaskStatus.RanToCompletion + ? s.Result + : BranchOutcome.Skipped(c.Index, c.Name)); // defensive; faults handled above + } + + var reason = ComputeCompletionReason(outcomes); + await CheckpointParentSucceedAsync(outcomes, reason, cancellationToken).ConfigureAwait(false); + return BuildResult(outcomes, reason); + } + + private CompletionReason ComputeCompletionReason(IReadOnlyList outcomes) + { + var succeeded = 0; + var failed = 0; + foreach (var o in outcomes) + { + if (o.Status == BatchItemStatus.Succeeded) succeeded++; + else if (o.Status == BatchItemStatus.Failed) failed++; + } + + var total = outcomes.Count; + var started = total - succeeded - failed; + return _policy.Evaluate(succeeded, failed, started, total); + } + + private async Task CheckpointParentSucceedAsync( + IReadOnlyList outcomes, + CompletionReason reason, + CancellationToken cancellationToken) + { + BatchSummary Build(bool includeInline) + { + var s = new BatchSummary + { + CompletionReason = BatchSummaryCodec.SerializeCompletionReason(reason), + Units = new List(outcomes.Count) + }; + foreach (var o in outcomes) + { + var unit = new BatchUnitSummary + { + Index = o.Index, + Name = o.Name, + Status = BatchSummaryCodec.SerializeStatus(o.Status) + }; + if (includeInline) + { + if (o.Status == BatchItemStatus.Succeeded) unit.Result = o.SerializedResult; + else if (o.Status == BatchItemStatus.Failed) unit.Error = o.Error; + } + s.Units.Add(unit); + } + return s; + } + + var summary = Build(includeInline: true); + var payload = BatchSummaryCodec.ToPayload(summary); + + var overflow = BatchSummaryCodec.IsOverflow(payload); + if (overflow) + { + summary = Build(includeInline: false); + payload = BatchSummaryCodec.ToPayload(summary); + } + + await EnqueueAsync(new SdkOperationUpdate + { + Id = _operationId, + ParentId = _parentId, + Type = OperationTypes.Context, + Action = OperationAction.SUCCEED, + SubType = OperationSubTypes.Parallel, + Name = _name, + Payload = payload, + ContextOptions = overflow ? new SdkContextOptions { ReplayChildren = true } : null + }, cancellationToken).ConfigureAwait(false); + } + + // ── Terminal (reconstruct) mode ───────────────────────────────────── + + private void ResolveTerminalBranch( + IncrementalParallelBranch handle, + string name, + string childOpId, + Func> func, + BatchUnitSummary? summaryEntry, + ILambdaSerializer branchSerializer) + { + // A branch registered now but absent from the frozen summary (registered + // after the original seal) never ran — surface it as skipped. + if (summaryEntry == null) + { + handle.ResolveFromInline(BatchItemStatus.Started, null, null); + return; + } + + var status = BatchSummaryCodec.DeserializeStatus(summaryEntry.Status); + + switch (status) + { + case BatchItemStatus.Succeeded when summaryEntry.Result != null: + handle.ResolveFromInline(BatchItemStatus.Succeeded, summaryEntry.Result, null); + break; + case BatchItemStatus.Failed when summaryEntry.Error != null: + handle.ResolveFromInline(BatchItemStatus.Failed, null, summaryEntry.Error); + break; + case BatchItemStatus.Succeeded: + case BatchItemStatus.Failed: + // Overflow: the inline value/error was stripped. Re-run the branch to + // recover it from the branch's own checkpoint; the frozen status stays + // authoritative. + LaunchRunBranch(handle, name, childOpId, func, branchSerializer, frozenStatus: status); + break; + default: + handle.ResolveFromInline(BatchItemStatus.Started, null, null); + break; + } + } + + private IBatchResult BuildTerminalResult(IReadOnlyList controllers) + { + // Prefer the frozen summary (authoritative for status + completion reason). + // Fall back to the registered controllers when the payload is missing/corrupt. + if (_frozenSummary != null) + { + // Positional replay contract: the same branches must be registered in the + // same order every invocation. A different count (a branch added or removed + // vs. the sealed run) would silently skip or double-count units, so reject it. + if (_registeredCount != _frozenSummary.Units.Count) + { + throw new NonDeterministicExecutionException( + $"Non-deterministic execution detected for parallel operation " + + $"'{_name ?? _operationId}': registered {_registeredCount} branch(es) on replay " + + $"but the checkpoint recorded {_frozenSummary.Units.Count}. Code must register the " + + $"same branches in the same order between deployments."); + } + + var succeeded = 0; + var failed = 0; + var started = 0; + foreach (var u in _frozenSummary.Units) + { + switch (BatchSummaryCodec.DeserializeStatus(u.Status)) + { + case BatchItemStatus.Succeeded: succeeded++; break; + case BatchItemStatus.Failed: failed++; break; + default: started++; break; + } + } + var reason = BatchSummaryCodec.DeserializeCompletionReason(_frozenSummary.CompletionReason); + return new IncrementalBatchResult(reason, succeeded, failed, started, _frozenSummary.Units.Count); + } + + var outcomes = new List(controllers.Count); + foreach (var c in controllers) + { + var s = c.Settlement; + outcomes.Add(s.Status == TaskStatus.RanToCompletion ? s.Result : BranchOutcome.Skipped(c.Index, c.Name)); + } + return BuildResult(outcomes, ComputeCompletionReason(outcomes)); + } + + // ── Shared helpers ────────────────────────────────────────────────── + + private static IBatchResult BuildResult(IReadOnlyList outcomes, CompletionReason reason) + { + var succeeded = 0; + var failed = 0; + var started = 0; + foreach (var o in outcomes) + { + switch (o.Status) + { + case BatchItemStatus.Succeeded: succeeded++; break; + case BatchItemStatus.Failed: failed++; break; + default: started++; break; + } + } + return new IncrementalBatchResult(reason, succeeded, failed, started, outcomes.Count); + } + + private BatchUnitSummary? FindSummaryUnit(int index) + { + if (_frozenSummary == null) return null; + foreach (var u in _frozenSummary.Units) + { + if (u.Index == index) return u; + } + return null; + } + + private IReadOnlyList SnapshotControllers() + { + lock (_lock) return _branches.ToArray(); + } + + private Task EnqueueAsync(SdkOperationUpdate update, CancellationToken cancellationToken = default) + => _batcher?.EnqueueAsync(update, cancellationToken) ?? Task.CompletedTask; +} + +/// +/// Non-generic returned by +/// . Per-branch typed +/// values are retrieved from the individual +/// handles; this type carries only the aggregate bookkeeping. +/// +internal sealed class IncrementalBatchResult : IBatchResult +{ + public IncrementalBatchResult( + CompletionReason completionReason, + int successCount, + int failureCount, + int startedCount, + int totalCount) + { + CompletionReason = completionReason; + SuccessCount = successCount; + FailureCount = failureCount; + StartedCount = startedCount; + TotalCount = totalCount; + } + + public CompletionReason CompletionReason { get; } + public bool HasFailure => FailureCount > 0; + public int SuccessCount { get; } + public int FailureCount { get; } + public int StartedCount { get; } + public int TotalCount { get; } +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs index bb580fbfd..d9ed786a0 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/StepOperation.cs @@ -215,6 +215,12 @@ private async Task ExecuteFunc(int attemptNumber, CancellationToken cancellat using var linked = CancellationTokenSource.CreateLinkedTokenSource( cancellationToken, _workflowCancellation.Token); + // Only the user-func call is guarded by the retry/fail funnel. The + // SUCCEED checkpoint and the result round-trip below sit OUTSIDE this + // try (mirroring ChildContextOperation): once SUCCEED is enqueued, no + // later failure may enqueue a conflicting RETRY/FAIL for an operation + // that has already committed a terminal SUCCEED. + T result; try { var stepContext = new StepContext(OperationId, attemptNumber, _logger); @@ -223,7 +229,6 @@ private async Task ExecuteFunc(int attemptNumber, CancellationToken cancellat // lines with the operation id, name, and current attempt. Wrap // only the user-func call — checkpoint emission shouldn't carry // step metadata into any side-channel logging. - T result; using (_logger.BeginScope(new Dictionary { ["operationId"] = OperationId, @@ -233,27 +238,6 @@ private async Task ExecuteFunc(int attemptNumber, CancellationToken cancellat { result = await _func(stepContext, linked.Token); } - - var serialized = SerializeResult(result); - await EnqueueAsync(new SdkOperationUpdate - { - Id = OperationId, - ParentId = ParentId, - Type = OperationTypes.Step, - Action = OperationAction.SUCCEED, - SubType = OperationSubTypes.Step, - Name = Name, - Payload = serialized - }, cancellationToken); - - // Round-trip the just-written checkpoint so the value the workflow - // observes on this fresh execution is the deserialized-from-checkpoint - // value, exactly as it would be on replay. This makes a custom - // (possibly non-round-tripping) StepConfig.Serializer's transform - // visible in the step result on the first run, not just on replay. - // Behavior change: the returned object is no longer the same instance - // the step body produced. See the AutoVer change note. - return DeserializeResult(serialized); } catch (OperationCanceledException) when (linked.IsCancellationRequested) { @@ -272,6 +256,34 @@ await EnqueueAsync(new SdkOperationUpdate // falls through here and is treated as a normal step failure. return await HandleStepFailureAsync(ex, attemptNumber, cancellationToken); } + + var serialized = SerializeResult(result); + await EnqueueAsync(new SdkOperationUpdate + { + Id = OperationId, + ParentId = ParentId, + Type = OperationTypes.Step, + Action = OperationAction.SUCCEED, + SubType = OperationSubTypes.Step, + Name = Name, + Payload = serialized + }, cancellationToken); + + // Round-trip the just-written checkpoint so the value the workflow + // observes on this fresh execution is the deserialized-from-checkpoint + // value, exactly as it would be on replay. This makes a custom + // (possibly non-round-tripping) StepConfig.Serializer's transform + // visible in the step result on the first run, not just on replay. + // Behavior change: the returned object is no longer the same instance + // the step body produced. See the AutoVer change note. + // + // This deserialize sits OUTSIDE the try above by design: it runs AFTER + // the SUCCEED checkpoint is enqueued, so a serializer that cannot + // deserialize its own just-written payload surfaces the fault directly + // to the caller instead of funneling into HandleStepFailureAsync — which + // would enqueue a RETRY/FAIL conflicting with the SUCCEED and risk a + // duplicate side effect on retry. Mirrors ChildContextOperation. + return DeserializeResult(serialized); } /// diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md b/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md index 7a258717a..f6fdbe24b 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md +++ b/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md @@ -24,6 +24,59 @@ Task> ParallelAsync( Each branch receives its own `IDurableContext` and a `CancellationToken` (linking the caller-supplied token with the SDK's workflow-shutdown signal — see [Cancellation](cancellation.md)), so a branch can itself use steps, waits, and nested durable operations. Branch results are serialized to per-branch checkpoints via the `ILambdaSerializer` registered on `ILambdaContext.Serializer`. The operation `name` is used for observability and to derive the deterministic operation ID, so keep it stable across deployments. +## Incremental, heterogeneous branches (`CreateParallel`) + +`ParallelAsync` takes a complete branch list up front and every branch shares one result type `T`. When branches return **unrelated types**, or are **discovered incrementally**, use `CreateParallel` instead. It returns an `IDurableParallel` you register branches on one at a time — each with its own result type — and each branch **begins executing as soon as it is registered** (subject to `MaxConcurrency`), so earlier independent work makes progress while later branches are still being assembled. + +```csharp +await using var parallel = ctx.CreateParallel(name: "process-order"); + +IParallelBranch inventory = parallel.Branch( + "inventory", async (branch, ct) => await ReserveInventoryAsync(branch, ct)); + +IParallelBranch payment = parallel.Branch( + "payment", async (branch, ct) => await AuthorizePaymentAsync(branch, ct)); + +if (plan.RequiresComplianceReview) +{ + // Branches can be added conditionally / incrementally. + _ = parallel.Branch("compliance", + async (branch, ct) => await ReviewComplianceAsync(branch, ct)); +} + +// Seal registration, await the branches per CompletionConfig, checkpoint the aggregate. +IBatchResult summary = await parallel.CompleteAsync(); + +// Each handle yields its own concrete type — no shared base type, casts, or envelopes. +InventoryReservation reservedInventory = await inventory; +PaymentAuthorization authorizedPayment = await payment; +``` + +`Branch` returns an awaitable `IParallelBranch` handle exposing `Name`, `Index`, and `Status`. `await handle` yields the branch's typed result, rethrows its `ChildContextException` on failure, or throws a `DurableExecutionException` if the branch was skipped by a completion-policy short-circuit (inspect `Status` first when that's possible). `CompleteAsync()` returns the non-generic aggregate `IBatchResult` (counts + `CompletionReason`); it is idempotent and never throws on per-branch failure. `DisposeAsync` (via `await using`) seals and completes the operation if you did not call `CompleteAsync`, so the parallel's terminal checkpoint is always written. + +> **Deterministic replay applies unchanged.** Branch identity is positional: the n-th `Branch` call reuses the n-th deterministic operation ID, so workflow code must register the same branches in the same order across invocations (a name change at a given index throws `NonDeterministicExecutionException`). Produce any dynamic branch set inside a checkpointed `StepAsync` so replay sees the same branches. `MaxConcurrency`, `CompletionConfig`, `NestingType`, cancellation, and the checkpoint format are identical to `ParallelAsync` — `CreateParallel` writes the same `Parallel` / `ParallelBranch` checkpoints, so it is purely an additive, front-end alternative. + +The homogeneous `ParallelAsync` overloads remain the simplest choice for a fixed set of same-typed branches and convenient `GetResults()` usage. + +### Per-branch serialization + +Because each branch declares its own result type, each may also use its own serializer. `Branch` takes an optional `ILambdaSerializer? serializer`; when omitted a branch uses the operation-level default — `ParallelConfig.ItemSerializer` if set on `CreateParallel`, otherwise the globally-registered `ILambdaContext.Serializer`. This lets one branch opt into a bespoke serializer (for example a source-generated `JsonSerializerContext` for Native AOT) without affecting sibling branches. + +```csharp +await using var parallel = ctx.CreateParallel(name: "process-order"); + +// Uses the operation-level / global serializer. +var inventory = parallel.Branch("inventory", async (b, ct) => await ReserveAsync(b, ct)); + +// Overrides serialization for just this branch. +var payment = parallel.Branch( + "payment", + async (b, ct) => await AuthorizeAsync(b, ct), + serializer: PaymentSerializerContext.Default.CreateLambdaSerializer()); +``` + +Like every other part of the workflow definition, a branch's serializer is re-resolved on replay, so a branch must be able to deserialize a result it previously serialized — keep the serializer stable at a given branch index across deployments. + ## Example Fan out three independent lookups and collect the results: diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/README.md b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/README.md index d4ce87efd..7b83c82da 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/README.md +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/README.md @@ -42,7 +42,7 @@ Conformance/ ## Coverage -All nine suites are implemented (one handler project per requirement id): +All ten suites are implemented (one handler project per requirement id): | Suite | Ids | Handlers | |-------|-----|----------| @@ -55,6 +55,7 @@ All nine suites are implemented (one handler project per requirement id): | `wait_for_callback` | 7-1 .. 7-15 | 15 | | `parallel` | 8-1 .. 8-22 (8-15 n/a) | 21 | | `map` | 9-1 .. 9-18 (9-14 n/a) | 17 | +| `static_typing` | 12-1 .. 12-2 | 2 | A few requirement ids have no .NET handler because the SDK intentionally lacks the feature they exercise (e.g. per-item / whole-result serdes slots in `map`); diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/scripts/build_examples.sh b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/scripts/build_examples.sh index 4ce59e368..6707491ab 100755 --- a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/scripts/build_examples.sh +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/scripts/build_examples.sh @@ -10,7 +10,7 @@ # ./build_examples.sh [operation...] # # Operations (default: every suite directory found next to the templates): -# step wait callback child invoke parallel map wait_for_callback wait_for_condition +# step wait callback child invoke parallel map static_typing wait_for_callback wait_for_condition # # Examples: # ./build_examples.sh step diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelEarlyStart/Function.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelEarlyStart/Function.cs new file mode 100644 index 000000000..a37e9ac33 --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelEarlyStart/Function.cs @@ -0,0 +1,44 @@ +// 12-2: Parallel branch starts before registration is sealed +using Amazon.Lambda.Core; +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.RuntimeSupport; +using Amazon.Lambda.Serialization.SystemTextJson; + +namespace ParallelEarlyStart; + +public class Function +{ + public static async Task Main(string[] args) + { + var handler = new Function(); + var serializer = new DefaultLambdaJsonSerializer(); + using var handlerWrapper = HandlerWrapper.GetHandlerWrapper(handler.Handler, serializer); + using var bootstrap = new LambdaBootstrap(handlerWrapper); + await bootstrap.RunAsync(); + } + + public Task Handler( + DurableExecutionInvocationInput input, ILambdaContext context) + => DurableFunction.WrapAsync>(Workflow, input, context); + + private async Task> Workflow(object? input, IDurableContext context) + { + string firstResult; + IParallelBranch second; + + await using (var parallel = context.CreateParallel( + name: "early-start", + config: new ParallelConfig { MaxConcurrency = 1 })) + { + IParallelBranch first = parallel.Branch( + "first", (_, _) => Task.FromResult("ready")); + firstResult = await first; + + second = parallel.Branch( + "second", (_, _) => Task.FromResult(firstResult + "-second")); + await parallel.CompleteAsync(); + } + + return new List { firstResult, await second }; + } +} diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelEarlyStart/ParallelEarlyStart.csproj b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelEarlyStart/ParallelEarlyStart.csproj new file mode 100644 index 000000000..4b354e178 --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelEarlyStart/ParallelEarlyStart.csproj @@ -0,0 +1,19 @@ + + + + net8.0 + Exe + true + bootstrap + enable + enable + + + + + + + + + + diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelTypedBranches/Function.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelTypedBranches/Function.cs new file mode 100644 index 000000000..1f9bc3efd --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelTypedBranches/Function.cs @@ -0,0 +1,52 @@ +// 12-1: Parallel with independently typed heterogeneous branch handles +using Amazon.Lambda.Core; +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.RuntimeSupport; +using Amazon.Lambda.Serialization.SystemTextJson; + +namespace ParallelTypedBranches; + +public class Function +{ + public static async Task Main(string[] args) + { + var handler = new Function(); + var serializer = new DefaultLambdaJsonSerializer(); + using var handlerWrapper = HandlerWrapper.GetHandlerWrapper(handler.Handler, serializer); + using var bootstrap = new LambdaBootstrap(handlerWrapper); + await bootstrap.RunAsync(); + } + + public Task Handler( + DurableExecutionInvocationInput input, ILambdaContext context) + => DurableFunction.WrapAsync>(Workflow, input, context); + + private async Task> Workflow(object? input, IDurableContext context) + { + IParallelBranch inventory; + IParallelBranch payment; + IParallelBranch> quote; + + await using (var parallel = context.CreateParallel( + name: "typed-branches", + config: new ParallelConfig { MaxConcurrency = 1 })) + { + inventory = parallel.Branch( + "inventory", (_, _) => Task.FromResult("reserved")); + payment = parallel.Branch( + "payment", (_, _) => Task.FromResult(200)); + quote = parallel.Branch( + "quote", (_, _) => Task.FromResult(new Dictionary + { + ["currency"] = "USD" + })); + + await parallel.CompleteAsync(); + } + + string inventoryResult = await inventory; + int paymentResult = await payment; + Dictionary quoteResult = await quote; + return new List { inventoryResult, paymentResult, quoteResult }; + } +} diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelTypedBranches/ParallelTypedBranches.csproj b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelTypedBranches/ParallelTypedBranches.csproj new file mode 100644 index 000000000..4b354e178 --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/static_typing/ParallelTypedBranches/ParallelTypedBranches.csproj @@ -0,0 +1,19 @@ + + + + net8.0 + Exe + true + bootstrap + enable + enable + + + + + + + + + + diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/template_static_typing.yaml b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/template_static_typing.yaml new file mode 100644 index 000000000..599898e05 --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/Conformance/template_static_typing.yaml @@ -0,0 +1,68 @@ +AWSTemplateFormatVersion: '2010-09-09' +Transform: AWS::Serverless-2016-10-31 +Description: Durable Execution Conformance Test Examples - .NET (Static Typing) +Globals: + Function: + Runtime: dotnet8 + Timeout: 60 + MemorySize: 512 + +Resources: + DurableFunctionRole: + Type: AWS::IAM::Role + Properties: + AssumeRolePolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Principal: + Service: lambda.amazonaws.com + Action: sts:AssumeRole + ManagedPolicyArns: + - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole + Policies: + - PolicyName: DurableExecutionPolicy + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Action: + - lambda:CheckpointDurableExecution + - lambda:GetDurableExecutionState + Resource: '*' + + ParallelTypedBranches: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["12-1"] + Metadata: + BuildMethod: makefile + Properties: + CodeUri: publish/ParallelTypedBranches/ + Handler: bootstrap + Description: Parallel with independently typed heterogeneous branch handles + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 + + ParallelEarlyStart: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: ["12-2"] + Metadata: + BuildMethod: makefile + Properties: + CodeUri: publish/ParallelEarlyStart/ + Handler: bootstrap + Description: Parallel branch completes before later branches are registered + Role: + Fn::GetAtt: + - DurableFunctionRole + - Arn + DurableConfig: + RetentionPeriodInDays: 7 + ExecutionTimeout: 300 diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/IncrementalParallelHeterogeneousTest.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/IncrementalParallelHeterogeneousTest.cs new file mode 100644 index 000000000..033a39df7 --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/IncrementalParallelHeterogeneousTest.cs @@ -0,0 +1,79 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using System.Linq; +using System.Text; +using Amazon.Lambda.Model; +using Xunit; +using Xunit.Abstractions; + +namespace Amazon.Lambda.DurableExecution.IntegrationTests; + +public class IncrementalParallelHeterogeneousTest +{ + private readonly ITestOutputHelper _output; + public IncrementalParallelHeterogeneousTest(ITestOutputHelper output) => _output = output; + + /// + /// End-to-end incremental, heterogeneous parallel: three branches registered + /// one at a time via CreateParallel/Branch return three + /// unrelated types (string, int, POCO), each retrieved through its own typed + /// handle. Validates the parent CONTEXT and per-branch CONTEXT checkpoints all + /// land in the service-side history with the correct names, and that the + /// heterogeneous per-branch values round-trip into the user-visible result. + /// + [Fact] + public async Task IncrementalParallel_HeterogeneousBranches_Succeed() + { + await using var deployment = await DurableFunctionDeployment.CreateAsync( + DurableFunctionDeployment.FindTestFunctionDir("IncrementalParallelHeterogeneousFunction"), + "iphetero", _output); + + var (invokeResponse, executionName) = await deployment.InvokeAsync("""{"orderId": "p1"}"""); + Assert.Equal(200, invokeResponse.StatusCode); + + var responsePayload = Encoding.UTF8.GetString(invokeResponse.Payload.ToArray()); + _output.WriteLine($"Response: {responsePayload}"); + + var arn = await deployment.FindDurableExecutionArnByNameAsync(executionName, TimeSpan.FromSeconds(60)); + Assert.NotNull(arn); + + var status = await deployment.PollForCompletionAsync(arn!, TimeSpan.FromSeconds(60)); + Assert.Equal("SUCCEEDED", status, ignoreCase: true); + + // Each heterogeneous branch's typed result surfaces in the user payload. + Assert.Contains("reserved-p1", responsePayload); // string branch + // Distinguishing token for the int branch: a bare "200" is tautologically + // satisfied by the POCO branch's "USD:4200" substring, so assert the + // property-qualified value instead. OrderResult is serialized with + // DefaultLambdaJsonSerializer (PascalCase property names), so Payment=200 + // renders as "Payment":200. + Assert.Contains("\"Payment\":200", responsePayload); // int branch + Assert.Contains("USD:4200", responsePayload); // POCO branch + + // History is eventually consistent — wait until the parent CONTEXT and all + // three child CONTEXT checkpoints are visible. + var history = await deployment.WaitForHistoryAsync( + arn!, + h => (h.Events?.Count(e => e.EventType == EventType.ContextStarted) ?? 0) >= 4 + && (h.Events?.Count(e => e.EventType == EventType.ContextSucceeded) ?? 0) >= 4, + TimeSpan.FromSeconds(60)); + var events = history.Events ?? new List(); + + // Parent + 3 branches = 4 ContextStarted, 4 ContextSucceeded. + Assert.Equal(4, events.Count(e => e.EventType == EventType.ContextStarted)); + Assert.Equal(4, events.Count(e => e.EventType == EventType.ContextSucceeded)); + + var startedNames = events + .Where(e => e.EventType == EventType.ContextStarted) + .Select(e => e.Name) + .ToList(); + Assert.Contains("process-order", startedNames); + Assert.Contains("inventory", startedNames); + Assert.Contains("payment", startedNames); + Assert.Contains("shipping", startedNames); + + // No branch failed. + Assert.Empty(events.Where(e => e.EventType == EventType.ContextFailed)); + } +} diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/IncrementalParallelReplayTest.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/IncrementalParallelReplayTest.cs new file mode 100644 index 000000000..7d20a338d --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/IncrementalParallelReplayTest.cs @@ -0,0 +1,111 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using System.Linq; +using System.Security.Cryptography; +using System.Text; +using Amazon.Lambda.Model; +using Xunit; +using Xunit.Abstractions; + +namespace Amazon.Lambda.DurableExecution.IntegrationTests; + +public class IncrementalParallelReplayTest +{ + private readonly ITestOutputHelper _output; + public IncrementalParallelReplayTest(ITestOutputHelper output) => _output = output; + + private static string HashOpId(string raw) + { + var bytes = Encoding.UTF8.GetBytes(raw); + var hash = SHA256.HashData(bytes); + var sb = new StringBuilder(hash.Length * 2); + foreach (var b in hash) sb.Append(b.ToString("x2")); + return sb.ToString(); + } + + /// + /// Deterministic replay of the incremental CreateParallel API across + /// both replay paths. Three branches each do a step (generating a GUID) then a + /// durable wait; a further wait runs after the parallel completes. This forces + /// (1) a Run-mode replay while the parent CONTEXT is still STARTED, and (2) a + /// terminal-reconstruct resume once the parent CONTEXT is SUCCEEDED. Verifies: + /// 1. Branch operation IDs match SHA-256("<parentId>-<n>"). + /// 2. Each branch's "generate" step succeeds EXACTLY once — proving neither + /// the STARTED-parent replay nor the terminal reconstruct re-executes a + /// branch body. + /// 3. The run spans multiple invocations (suspend/resume actually happened). + /// + [Fact] + public async Task IncrementalParallel_ReplayDeterminism_AcrossRunAndTerminalPaths() + { + await using var deployment = await DurableFunctionDeployment.CreateAsync( + DurableFunctionDeployment.FindTestFunctionDir("IncrementalParallelReplayFunction"), + "ipreplay", _output); + + var (invokeResponse, executionName) = await deployment.InvokeAsync("""{"orderId": "p6"}"""); + var responsePayload = Encoding.UTF8.GetString(invokeResponse.Payload.ToArray()); + _output.WriteLine($"Response: {responsePayload}"); + + var arn = await deployment.FindDurableExecutionArnByNameAsync(executionName, TimeSpan.FromSeconds(60)); + Assert.NotNull(arn); + + var status = await deployment.PollForCompletionAsync(arn!, TimeSpan.FromSeconds(180)); + Assert.Equal("SUCCEEDED", status, ignoreCase: true); + + // The parallel parent is the first root-level operation -> SHA256("1"). + var parentOpId = HashOpId("1"); + var expectedBranchIds = new[] + { + HashOpId($"{parentOpId}-1"), + HashOpId($"{parentOpId}-2"), + HashOpId($"{parentOpId}-3"), + }; + + var history = await deployment.WaitForHistoryAsync( + arn!, + h => + { + var events = h.Events ?? new List(); + // Parent + 3 branch CONTEXTs all succeeded. + if (events.Count(e => e.EventType == EventType.ContextSucceeded) < 4) return false; + // Each branch ran one step and one wait, plus the post-parallel wait. + if (events.Count(e => e.EventType == EventType.StepSucceeded) < 3) return false; + if (events.Count(e => e.EventType == EventType.WaitSucceeded) < 4) return false; + return true; + }, + TimeSpan.FromSeconds(120)); + var allEvents = history.Events ?? new List(); + + // 1. Branch operation IDs match the deterministic hash. + var observedBranchIds = allEvents + .Where(e => e.EventType == EventType.ContextStarted && e.Id != null && e.Id != parentOpId) + .Select(e => e.Id) + .Distinct() + .ToList(); + Assert.Equal(3, observedBranchIds.Count); + foreach (var expected in expectedBranchIds) + { + Assert.Contains(expected, observedBranchIds); + } + + // 2. Each branch's "generate" step succeeded exactly once — no branch body + // re-executed on either the STARTED-parent replay or the terminal resume. + var generateSucceeded = allEvents + .Where(e => e.EventType == EventType.StepSucceeded && e.Name == "generate") + .ToList(); + Assert.Equal(3, generateSucceeded.Count); + + // 3. Parent + 3 branches succeeded once each. + Assert.Equal(4, allEvents.Count(e => e.EventType == EventType.ContextSucceeded)); + + // 4. The run spans multiple invocations (branch waits + post-parallel wait). + var invocations = allEvents.Where(e => e.InvocationCompletedDetails != null).ToList(); + Assert.True( + invocations.Count >= 2, + $"Expected >= 2 InvocationCompleted events (suspend + resume), got {invocations.Count}"); + + // 5. The user-visible response carries the joined per-branch results. + Assert.Contains("completed", responsePayload, StringComparison.OrdinalIgnoreCase); + } +} diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelHeterogeneousFunction/Function.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelHeterogeneousFunction/Function.cs new file mode 100644 index 000000000..ba5c09a6b --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelHeterogeneousFunction/Function.cs @@ -0,0 +1,91 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using Amazon.Lambda.Core; +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.RuntimeSupport; +using Amazon.Lambda.Serialization.SystemTextJson; + +namespace DurableExecutionTestFunction; + +/// +/// Deployed entry point exercising the incremental, heterogeneous branch API +/// (). Three branches return three +/// unrelated types (a string, an int, and a POCO); each is retrieved through its +/// own typed handle with no shared base type, +/// cast, or envelope. Validates that heterogeneous per-branch results round-trip +/// through the service checkpoint history end-to-end. +/// +public class Function +{ + public static async Task Main(string[] args) + { + var handler = new Function(); + var serializer = new DefaultLambdaJsonSerializer(); + using var handlerWrapper = HandlerWrapper.GetHandlerWrapper(handler.Handler, serializer); + using var bootstrap = new LambdaBootstrap(handlerWrapper); + await bootstrap.RunAsync(); + } + + public Task Handler( + DurableExecutionInvocationInput input, ILambdaContext context) + => DurableFunction.WrapAsync(Workflow, input, context); + + private static async Task Workflow(OrderRequest input, IDurableContext context) + { + var orderId = input?.OrderId ?? "unknown"; + + await using var parallel = context.CreateParallel(name: "process-order"); + + // Each branch declares its own result type — string, int, and Money. + IParallelBranch inventory = parallel.Branch( + "inventory", + async (branch, ct) => await branch.StepAsync( + (_, _) => Task.FromResult($"reserved-{orderId}"), name: "reserve")); + + IParallelBranch payment = parallel.Branch( + "payment", + async (branch, ct) => await branch.StepAsync( + (_, _) => Task.FromResult(200), name: "charge")); + + IParallelBranch shipping = parallel.Branch( + "shipping", + async (branch, ct) => await branch.StepAsync( + (_, _) => Task.FromResult(new Money { Currency = "USD", Amount = 4200 }), name: "quote")); + + IBatchResult summary = await parallel.CompleteAsync(); + + var reservedInventory = await inventory; + var authorizedPayment = await payment; + var shippingQuote = await shipping; + + return new OrderResult + { + Inventory = reservedInventory, + Payment = authorizedPayment, + Shipping = $"{shippingQuote.Currency}:{shippingQuote.Amount}", + SuccessCount = summary.SuccessCount, + TotalCount = summary.TotalCount, + }; + } +} + +public class OrderRequest +{ + public string? OrderId { get; set; } +} + +public class OrderResult +{ + public string Inventory { get; set; } = ""; + public int Payment { get; set; } + public string Shipping { get; set; } = ""; + public int SuccessCount { get; set; } + public int TotalCount { get; set; } +} + +public class Money +{ + public string Currency { get; set; } = ""; + public int Amount { get; set; } +} diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelHeterogeneousFunction/IncrementalParallelHeterogeneousFunction.csproj b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelHeterogeneousFunction/IncrementalParallelHeterogeneousFunction.csproj new file mode 100644 index 000000000..f8bf7fd0c --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelHeterogeneousFunction/IncrementalParallelHeterogeneousFunction.csproj @@ -0,0 +1,18 @@ + + + + net10.0 + Exe + true + bootstrap + enable + enable + + + + + + + + + diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelReplayFunction/Function.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelReplayFunction/Function.cs new file mode 100644 index 000000000..fdeaa9534 --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelReplayFunction/Function.cs @@ -0,0 +1,90 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using Amazon.Lambda.Core; +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.RuntimeSupport; +using Amazon.Lambda.Serialization.SystemTextJson; + +namespace DurableExecutionTestFunction; + +/// +/// Deployed entry point exercising deterministic replay of the incremental +/// () API across two distinct replay +/// paths: +/// +/// Each branch does a step (generating a GUID) then a durable wait. The +/// wait suspends the whole invocation, so the parallel re-runs with its +/// parent CONTEXT still STARTED — branches replay from their own checkpoints +/// and the cached GUID must survive. +/// After the parallel completes, a second durable wait suspends again. On +/// that resume the parent CONTEXT is already SUCCEEDED, so the incremental +/// operation takes the terminal-reconstruct path: branch handles resolve +/// from the frozen inline summary WITHOUT re-running, and the aggregate is +/// rebuilt from the checkpoint. +/// +/// If replay determinism were broken, the per-branch GUIDs would change between +/// invocations, or a branch step would re-execute (surfacing as duplicate +/// StepSucceeded events). +/// +public class Function +{ + public static async Task Main(string[] args) + { + var handler = new Function(); + var serializer = new DefaultLambdaJsonSerializer(); + using var handlerWrapper = HandlerWrapper.GetHandlerWrapper(handler.Handler, serializer); + using var bootstrap = new LambdaBootstrap(handlerWrapper); + await bootstrap.RunAsync(); + } + + public Task Handler( + DurableExecutionInvocationInput input, ILambdaContext context) + => DurableFunction.WrapAsync(Workflow, input, context); + + private static async Task Workflow(TestEvent input, IDurableContext context) + { + await using var parallel = context.CreateParallel(name: "fanout"); + + IParallelBranch a = parallel.Branch("a", BranchAsync); + IParallelBranch b = parallel.Branch("b", BranchAsync); + IParallelBranch c = parallel.Branch("c", BranchAsync); + + var summary = await parallel.CompleteAsync(); + + // Retrieve each branch's typed result through its own handle. + var joined = string.Join(",", await a, await b, await c); + + // Force a resume where the parallel is ALREADY terminal, so CreateParallel + // takes the terminal-reconstruct path on the next invocation. + await context.WaitAsync(TimeSpan.FromSeconds(2), name: "post-boundary"); + + return new TestResult + { + Status = "completed", + Data = joined, + SuccessCount = summary.SuccessCount + }; + } + + private static async Task BranchAsync(IDurableContext ctx, CancellationToken cancellationToken) + { + var generatedId = await ctx.StepAsync( + async (_, _) => { await Task.CompletedTask; return Guid.NewGuid().ToString(); }, + name: "generate"); + + // Suspend/resume cycle so the parallel replays with its parent still STARTED. + await ctx.WaitAsync(TimeSpan.FromSeconds(2), name: "boundary"); + + return generatedId; + } +} + +public class TestEvent { public string? OrderId { get; set; } } + +public class TestResult +{ + public string? Status { get; set; } + public string? Data { get; set; } + public int SuccessCount { get; set; } +} diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelReplayFunction/IncrementalParallelReplayFunction.csproj b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelReplayFunction/IncrementalParallelReplayFunction.csproj new file mode 100644 index 000000000..f8bf7fd0c --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/IncrementalParallelReplayFunction/IncrementalParallelReplayFunction.csproj @@ -0,0 +1,18 @@ + + + + net10.0 + Exe + true + bootstrap + enable + enable + + + + + + + + + diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/IncrementalParallelOperationTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/IncrementalParallelOperationTests.cs new file mode 100644 index 000000000..64d4f6514 --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/IncrementalParallelOperationTests.cs @@ -0,0 +1,904 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.DurableExecution.Internal; +using Amazon.Lambda.Serialization.SystemTextJson; +using Amazon.Lambda.TestUtilities; +using Xunit; + +namespace Amazon.Lambda.DurableExecution.Tests; + +/// +/// Tests for the incremental, heterogeneous parallel API +/// ( / ). +/// The checkpoint shape mirrors the batch ParallelOperation<T>, so these +/// reuse the same IdAt/ChildIdAt/CreateContext harness as +/// . +/// +public class IncrementalParallelOperationTests +{ + /// Reproduces the Id that emits for the n-th root-level operation. + private static string IdAt(int position) => OperationIdGenerator.HashOperationId(position.ToString()); + + /// The hashed ID of the n-th child operation under . + private static string ChildIdAt(string parentOpId, int position) => + OperationIdGenerator.HashOperationId($"{parentOpId}-{position}"); + + private static (DurableContext context, RecordingBatcher recorder, TerminationManager tm, ExecutionState state) + CreateContext(InitialExecutionState? initialState = null) + { + var state = new ExecutionState(); + state.LoadFromCheckpoint(initialState); + var tm = new TerminationManager(); + var idGen = new OperationIdGenerator(); + var lambdaContext = new TestLambdaContext { Serializer = new DefaultLambdaJsonSerializer() }; + var recorder = new RecordingBatcher(); + var context = new DurableContext(state, tm, new WorkflowCancellation(tm), idGen, "arn:test", lambdaContext, recorder.Batcher); + return (context, recorder, tm, state); + } + + public sealed class Money + { + public string Currency { get; set; } = ""; + public int Amount { get; set; } + } + + // ────────────────────────────────────────────────────────────────────── + // Fresh execution — happy paths + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task CreateParallel_FreshExecution_HeterogeneousBranches_ResolveTypedResults() + { + var (context, recorder, tm, _) = CreateContext(); + + IParallelBranch inventory; + IParallelBranch payment; + IParallelBranch shipping; + + await using (var parallel = context.CreateParallel(name: "process-order")) + { + inventory = parallel.Branch("inventory", async (_, _) => { await Task.Yield(); return "reserved"; }); + payment = parallel.Branch("payment", async (_, _) => { await Task.Yield(); return 200; }); + shipping = parallel.Branch("shipping", async (_, _) => + { + await Task.Yield(); + return new Money { Currency = "USD", Amount = 4200 }; + }); + + var summary = await parallel.CompleteAsync(); + + Assert.False(tm.IsTerminated); + Assert.Equal(3, summary.TotalCount); + Assert.Equal(3, summary.SuccessCount); + Assert.Equal(0, summary.FailureCount); + Assert.False(summary.HasFailure); + Assert.Equal(CompletionReason.AllCompleted, summary.CompletionReason); + } + + // Each branch handle yields its own concrete type — no shared T, no casts. + Assert.Equal("reserved", await inventory); + Assert.Equal(200, await payment); + var ship = await shipping; + Assert.Equal("USD", ship.Currency); + Assert.Equal(4200, ship.Amount); + + Assert.Equal(BatchItemStatus.Succeeded, inventory.Status); + Assert.Equal(0, inventory.Index); + Assert.Equal(2, shipping.Index); + + await recorder.Batcher.DrainAsync(); + var contextActions = recorder.Flushed.Where(o => o.Type == "CONTEXT") + .Select(o => $"{o.SubType}:{o.Action}").ToArray(); + // Parent START + 3 child STARTs + 3 child SUCCEEDs + parent SUCCEED + Assert.Equal(8, contextActions.Length); + Assert.Equal("Parallel:START", contextActions[0]); + Assert.Equal("Parallel:SUCCEED", contextActions[^1]); + } + + [Fact] + public async Task CreateParallel_BranchOperationIds_AreDeterministic() + { + var (context, recorder, _, _) = CreateContext(); + + await using (var parallel = context.CreateParallel()) + { + _ = parallel.Branch("a", async (_, _) => { await Task.Yield(); return "a"; }); + _ = parallel.Branch("b", async (_, _) => { await Task.Yield(); return "b"; }); + await parallel.CompleteAsync(); + } + + await recorder.Batcher.DrainAsync(); + + var parentOpId = IdAt(1); + var branchStarts = recorder.Flushed + .Where(o => o.Type == "CONTEXT" && o.SubType == "ParallelBranch" && o.Action == "START") + .ToArray(); + Assert.Equal(2, branchStarts.Length); + Assert.Contains(branchStarts, o => o.Id == ChildIdAt(parentOpId, 1)); + Assert.Contains(branchStarts, o => o.Id == ChildIdAt(parentOpId, 2)); + } + + [Fact] + public async Task CreateParallel_EmptyOperation_FlushesStartAndSucceed() + { + var (context, recorder, _, _) = CreateContext(); + + IBatchResult summary; + await using (var parallel = context.CreateParallel()) + { + summary = await parallel.CompleteAsync(); + } + + Assert.Equal(0, summary.TotalCount); + Assert.Equal(CompletionReason.AllCompleted, summary.CompletionReason); + + await recorder.Batcher.DrainAsync(); + var contextActions = recorder.Flushed.Where(o => o.Type == "CONTEXT") + .Select(o => $"{o.SubType}:{o.Action}").ToArray(); + Assert.Equal(new[] { "Parallel:START", "Parallel:SUCCEED" }, contextActions); + } + + [Fact] + public async Task CreateParallel_NamesPropagateToCheckpointAndHandle() + { + var (context, recorder, _, _) = CreateContext(); + + await using (var parallel = context.CreateParallel(name: "fanout")) + { + var a = parallel.Branch("alpha", async (_, _) => { await Task.Yield(); return 1; }); + var b = parallel.Branch("beta", async (_, _) => { await Task.Yield(); return 2; }); + await parallel.CompleteAsync(); + Assert.Equal("alpha", a.Name); + Assert.Equal("beta", b.Name); + } + + await recorder.Batcher.DrainAsync(); + var branchSucceeds = recorder.Flushed + .Where(o => o.Type == "CONTEXT" && o.SubType == "ParallelBranch" && o.Action == "SUCCEED") + .ToArray(); + Assert.Contains(branchSucceeds, o => o.Name == "alpha"); + Assert.Contains(branchSucceeds, o => o.Name == "beta"); + } + + [Fact] + public async Task CreateParallel_NestedSucceeded_InlinesPerBranchResultsOnParentPayload() + { + var (context, recorder, _, _) = CreateContext(); + + await using (var parallel = context.CreateParallel(name: "fanout")) + { + _ = parallel.Branch("i", async (_, _) => { await Task.Yield(); return 100; }); + _ = parallel.Branch("p", async (_, _) => { await Task.Yield(); return 200; }); + await parallel.CompleteAsync(); + } + + await recorder.Batcher.DrainAsync(); + var parentSucceed = Assert.Single(recorder.Flushed.Where(o => + o.Type == "CONTEXT" && o.SubType == "Parallel" && $"{o.Action}" == "SUCCEED")); + var summary = System.Text.Json.JsonSerializer.Deserialize(parentSucceed.Payload!); + Assert.NotNull(summary); + Assert.Equal("100", summary!.Units[0].Result); + Assert.Equal("200", summary.Units[1].Result); + } + + // ────────────────────────────────────────────────────────────────────── + // Failure handling + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task CreateParallel_DefaultFailFast_BranchFailure_SurfacesOnResultAndHandle() + { + var (context, _, _, _) = CreateContext(); + + IParallelBranch ok; + IParallelBranch bad; + IBatchResult summary; + + await using (var parallel = context.CreateParallel()) + { + ok = parallel.Branch("ok", async (_, _) => { await Task.Yield(); return 1; }); + bad = parallel.Branch("bad", async (_, _) => + { + await Task.Yield(); + throw new InvalidOperationException("branch boom"); + }); + // Never throws on per-branch failure. + summary = await parallel.CompleteAsync(); + } + + Assert.True(summary.HasFailure); + Assert.Equal(CompletionReason.FailureToleranceExceeded, summary.CompletionReason); + Assert.Equal(1, summary.FailureCount); + + Assert.Equal(1, await ok); + Assert.Equal(BatchItemStatus.Failed, bad.Status); + var ex = await Assert.ThrowsAsync(async () => await bad); + Assert.Contains("branch boom", ex.Message); + } + + [Fact] + public async Task CreateParallel_AllCompleted_PartialFailureDoesNotExceedTolerance() + { + var (context, _, _, _) = CreateContext(); + + IBatchResult summary; + await using (var parallel = context.CreateParallel( + config: new ParallelConfig { CompletionConfig = CompletionConfig.AllCompleted() })) + { + _ = parallel.Branch("ok", async (_, _) => { await Task.Yield(); return 1; }); + _ = parallel.Branch("bad", async (_, _) => { await Task.Yield(); throw new InvalidOperationException("x"); }); + summary = await parallel.CompleteAsync(); + } + + Assert.Equal(CompletionReason.AllCompleted, summary.CompletionReason); + Assert.Equal(1, summary.SuccessCount); + Assert.Equal(1, summary.FailureCount); + Assert.True(summary.HasFailure); + } + + // ────────────────────────────────────────────────────────────────────── + // MaxConcurrency + completion short-circuit + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task CreateParallel_MaxConcurrency_LimitsInFlight() + { + var (context, _, _, _) = CreateContext(); + + var inFlight = 0; + var maxObserved = 0; + var gate = new object(); + + await using (var parallel = context.CreateParallel(config: new ParallelConfig { MaxConcurrency = 2 })) + { + for (var i = 0; i < 6; i++) + { + _ = parallel.Branch($"b{i}", async (_, ct) => + { + lock (gate) { inFlight++; maxObserved = Math.Max(maxObserved, inFlight); } + await Task.Delay(20, ct); + lock (gate) { inFlight--; } + return 1; + }); + } + await parallel.CompleteAsync(); + } + + Assert.True(maxObserved <= 2, $"Observed concurrency {maxObserved} exceeded MaxConcurrency = 2"); + } + + [Fact] + public async Task CreateParallel_FirstSuccessful_WithMaxConcurrency1_SkipsTrailingBranches() + { + var (context, _, _, _) = CreateContext(); + + IParallelBranch last; + IBatchResult summary; + + await using (var parallel = context.CreateParallel(config: new ParallelConfig + { + MaxConcurrency = 1, + CompletionConfig = CompletionConfig.FirstSuccessful() + })) + { + _ = parallel.Branch("b0", async (_, _) => { await Task.Yield(); return 1; }); + _ = parallel.Branch("b1", async (_, _) => { await Task.Yield(); return 2; }); + last = parallel.Branch("b2", async (_, _) => { await Task.Yield(); return 3; }); + summary = await parallel.CompleteAsync(); + } + + Assert.True(summary.SuccessCount >= 1); + Assert.Equal(0, summary.FailureCount); + Assert.Equal(3, summary.TotalCount); + // With MaxConcurrency=1 the completion policy stops dispatching once the first + // success lands, but because branches start on registration, how many of the + // already-in-flight branches run to completion before the short-circuit is + // observed is timing-dependent. The deterministic invariants: no failures, the + // reason reflects an early success (MinSuccessfulReached when a branch was + // skipped, AllCompleted when every branch happened to finish), and a skipped + // branch's handle throws on await. + Assert.True( + summary.CompletionReason == CompletionReason.MinSuccessfulReached + || summary.CompletionReason == CompletionReason.AllCompleted); + Assert.Equal(3, summary.SuccessCount + summary.StartedCount); + + if (last.Status == BatchItemStatus.Started) + { + await Assert.ThrowsAsync(async () => await last); + } + } + + // ────────────────────────────────────────────────────────────────────── + // Registration guardrails + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task CreateParallel_RegisterAfterComplete_Throws() + { + var (context, _, _, _) = CreateContext(); + + var parallel = context.CreateParallel(); + _ = parallel.Branch("a", async (_, _) => { await Task.Yield(); return 1; }); + await parallel.CompleteAsync(); + + Assert.Throws(() => + _ = parallel.Branch("late", async (_, _) => { await Task.Yield(); return 2; })); + + await parallel.DisposeAsync(); + } + + [Fact] + public async Task CreateParallel_CompleteAsync_IsIdempotent() + { + var (context, recorder, _, _) = CreateContext(); + + var parallel = context.CreateParallel(); + _ = parallel.Branch("a", async (_, _) => { await Task.Yield(); return 1; }); + var first = await parallel.CompleteAsync(); + var second = await parallel.CompleteAsync(); + Assert.Same(first, second); + await parallel.DisposeAsync(); + + await recorder.Batcher.DrainAsync(); + // Exactly one parent SUCCEED despite two CompleteAsync calls + dispose. + Assert.Single(recorder.Flushed.Where(o => + o.Type == "CONTEXT" && o.SubType == "Parallel" && o.Action == "SUCCEED")); + } + + [Fact] + public async Task CreateParallel_DisposeWithoutComplete_StillCheckpointsParent() + { + var (context, recorder, _, _) = CreateContext(); + + await using (var parallel = context.CreateParallel(name: "auto")) + { + _ = parallel.Branch("a", async (_, _) => { await Task.Yield(); return 1; }); + // No explicit CompleteAsync — DisposeAsync must seal + checkpoint. + } + + await recorder.Batcher.DrainAsync(); + Assert.Single(recorder.Flushed.Where(o => + o.Type == "CONTEXT" && o.SubType == "Parallel" && o.Action == "SUCCEED")); + } + + // ────────────────────────────────────────────────────────────────────── + // Replay — terminal parent reconstructs without re-running branches + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task CreateParallel_ReplaySucceeded_RebuildsFromInlineSummary_WithoutRerunning() + { + var parentOpId = IdAt(1); + var summaryJson = """ + {"CompletionReason":"ALL_COMPLETED","Units":[ + {"Index":0,"Name":"inventory","Status":"SUCCEEDED","Result":"\"reserved\""}, + {"Index":1,"Name":"payment","Status":"SUCCEEDED","Result":"200"} + ]} + """; + + var (context, recorder, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = OperationStatuses.Succeeded, + SubType = OperationSubTypes.Parallel, + Name = "process-order", + ContextDetails = new ContextDetails { Result = summaryJson } + } + } + }); + + var executed = false; + IParallelBranch inventory; + IParallelBranch payment; + IBatchResult summary; + + await using (var parallel = context.CreateParallel(name: "process-order")) + { + inventory = parallel.Branch("inventory", async (_, _) => { executed = true; await Task.Yield(); return "LIVE"; }); + payment = parallel.Branch("payment", async (_, _) => { executed = true; await Task.Yield(); return -1; }); + summary = await parallel.CompleteAsync(); + } + + Assert.False(executed); + Assert.Equal("reserved", await inventory); + Assert.Equal(200, await payment); + Assert.Equal(2, summary.SuccessCount); + Assert.Equal(CompletionReason.AllCompleted, summary.CompletionReason); + + await recorder.Batcher.DrainAsync(); + Assert.Empty(recorder.Flushed); // terminal parent → no re-checkpoint + } + + [Fact] + public async Task CreateParallel_ReplaySucceeded_FailedBranch_AwaitRethrows() + { + var parentOpId = IdAt(1); + var summaryJson = """ + {"CompletionReason":"FAILURE_TOLERANCE_EXCEEDED","Units":[ + {"Index":0,"Name":"ok","Status":"SUCCEEDED","Result":"1"}, + {"Index":1,"Name":"bad","Status":"FAILED","Error":{"ErrorType":"System.InvalidOperationException","ErrorMessage":"boom"}} + ]} + """; + + var (context, _, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = OperationStatuses.Succeeded, + SubType = OperationSubTypes.Parallel, + Name = "fanout", + ContextDetails = new ContextDetails { Result = summaryJson } + } + } + }); + + IParallelBranch ok; + IParallelBranch bad; + IBatchResult summary; + + await using (var parallel = context.CreateParallel(name: "fanout")) + { + ok = parallel.Branch("ok", async (_, _) => { await Task.Yield(); return -1; }); + bad = parallel.Branch("bad", async (_, _) => { await Task.Yield(); return -1; }); + summary = await parallel.CompleteAsync(); + } + + Assert.Equal(CompletionReason.FailureToleranceExceeded, summary.CompletionReason); + Assert.True(summary.HasFailure); + Assert.Equal(1, await ok); + var ex = await Assert.ThrowsAsync(async () => await bad); + Assert.Contains("boom", ex.Message); + } + + [Fact] + public async Task CreateParallel_ReplaySucceeded_OverflowStrippedResult_ReRunsBranchToRecoverValue() + { + // Overflow-recovery arm of ResolveTerminalBranch: the parent checkpointed + // SUCCEEDED, but the summary exceeded the payload cap so it was written with + // the inline per-branch Result stripped (unit Status=SUCCEEDED, Result=null). + // On replay such a unit is routed through LaunchRunBranch(frozenStatus), which + // RE-RUNS the branch body to recover the stripped value while keeping the + // frozen SUCCEEDED verdict authoritative and NOT re-checkpointing the parent. + var parentOpId = IdAt(1); + var summaryJson = """ + {"CompletionReason":"ALL_COMPLETED","Units":[ + {"Index":0,"Name":"inventory","Status":"SUCCEEDED"} + ]} + """; + + var (context, recorder, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = OperationStatuses.Succeeded, + SubType = OperationSubTypes.Parallel, + Name = "process-order", + ContextDetails = new ContextDetails { Result = summaryJson } + } + } + }); + + var executed = false; + IParallelBranch inventory; + IBatchResult summary; + + await using (var parallel = context.CreateParallel(name: "process-order")) + { + inventory = parallel.Branch("inventory", async (_, _) => + { + executed = true; // the recovered value can only come from a re-run + await Task.Yield(); + return "recovered-reserved"; + }); + summary = await parallel.CompleteAsync(); + } + + // (a) The stripped value is not inline, so the branch body had to re-run. + Assert.True(executed); + // (b) The handle resolves to the value the re-run recovered — not the default + // a broken (inline-null) resolution would have produced. + Assert.Equal("recovered-reserved", await inventory); + // (c) The frozen verdict wins even though the body re-executed. + Assert.Equal(BatchItemStatus.Succeeded, inventory.Status); + Assert.Equal(1, summary.SuccessCount); + Assert.Equal(0, summary.FailureCount); + Assert.Equal(CompletionReason.AllCompleted, summary.CompletionReason); + + await recorder.Batcher.DrainAsync(); + // (d) A terminal parent is never re-checkpointed: no parent Parallel SUCCEED. + Assert.DoesNotContain(recorder.Flushed, o => + o.Type == "CONTEXT" && o.SubType == "Parallel" && o.Action == "SUCCEED"); + } + + [Fact] + public async Task CreateParallel_ReplayFailed_OverflowStrippedError_ReRunsBranchToRecoverFailure() + { + // Same overflow-recovery arm for a FAILED unit whose inline Error was stripped + // (unit Status=FAILED, Error=null). The body re-runs (and fails again), the + // recovered failure surfaces on the handle, the frozen FAILED verdict stays + // authoritative, and the parent is not re-checkpointed. + var parentOpId = IdAt(1); + var summaryJson = """ + {"CompletionReason":"FAILURE_TOLERANCE_EXCEEDED","Units":[ + {"Index":0,"Name":"bad","Status":"FAILED"} + ]} + """; + + var (context, recorder, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = OperationStatuses.Succeeded, + SubType = OperationSubTypes.Parallel, + Name = "fanout", + ContextDetails = new ContextDetails { Result = summaryJson } + } + } + }); + + var executed = false; + IParallelBranch bad; + IBatchResult summary; + + await using (var parallel = context.CreateParallel(name: "fanout")) + { + bad = parallel.Branch("bad", async (_, _) => + { + executed = true; + await Task.Yield(); + throw new InvalidOperationException("recovered-boom"); + }); + summary = await parallel.CompleteAsync(); + } + + // (a) The stripped error is not inline, so the branch body had to re-run. + Assert.True(executed); + // (b) The recovered failure surfaces on the handle with the re-run's message. + var ex = await Assert.ThrowsAsync(async () => await bad); + Assert.Contains("recovered-boom", ex.Message); + // (c) The frozen FAILED verdict wins. + Assert.Equal(BatchItemStatus.Failed, bad.Status); + Assert.Equal(1, summary.FailureCount); + Assert.True(summary.HasFailure); + Assert.Equal(CompletionReason.FailureToleranceExceeded, summary.CompletionReason); + + await recorder.Batcher.DrainAsync(); + // (d) A terminal parent is never re-checkpointed: no parent Parallel SUCCEED. + Assert.DoesNotContain(recorder.Flushed, o => + o.Type == "CONTEXT" && o.SubType == "Parallel" && o.Action == "SUCCEED"); + } + + [Fact] + public async Task CreateParallel_ReplayNameDrift_Throws() + { + var parentOpId = IdAt(1); + var summaryJson = """ + {"CompletionReason":"ALL_COMPLETED","Units":[ + {"Index":0,"Name":"inventory","Status":"SUCCEEDED","Result":"1"} + ]} + """; + + var (context, _, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = OperationStatuses.Succeeded, + SubType = OperationSubTypes.Parallel, + Name = "fanout", + ContextDetails = new ContextDetails { Result = summaryJson } + } + } + }); + + var parallel = context.CreateParallel(name: "fanout"); + // Registered a branch whose name drifted from the checkpointed "inventory". + Assert.Throws(() => + _ = parallel.Branch("renamed", async (_, _) => { await Task.Yield(); return 1; })); + await parallel.DisposeAsync(); + } + + // ────────────────────────────────────────────────────────────────────── + // Replay — STARTED parent re-runs branches (children replay from own checkpoints) + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task CreateParallel_ReplayStartedParent_ReRunsBranches_AndCheckpointsSucceed() + { + var parentOpId = IdAt(1); + var (context, recorder, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = OperationStatuses.Started, + SubType = OperationSubTypes.Parallel, + Name = "fanout" + } + } + }); + + var runCount = 0; + IBatchResult summary; + await using (var parallel = context.CreateParallel(name: "fanout")) + { + _ = parallel.Branch("a", async (_, _) => { Interlocked.Increment(ref runCount); await Task.Yield(); return 1; }); + _ = parallel.Branch("b", async (_, _) => { Interlocked.Increment(ref runCount); await Task.Yield(); return 2; }); + summary = await parallel.CompleteAsync(); + } + + Assert.Equal(2, runCount); // children re-run (no terminal checkpoints for them) + Assert.Equal(2, summary.SuccessCount); + + await recorder.Batcher.DrainAsync(); + // STARTED parent is not re-emitted, but the terminal SUCCEED is written now. + var parentActions = recorder.Flushed + .Where(o => o.Type == "CONTEXT" && o.SubType == "Parallel") + .Select(o => $"{o.Action}").ToArray(); + Assert.DoesNotContain("START", parentActions); + Assert.Contains("SUCCEED", parentActions); + } + + // ────────────────────────────────────────────────────────────────────── + // Replay guardrails from review (issue #2519 Copilot feedback) + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public void CreateParallel_ReplayUnexpectedParentStatus_Throws() + { + // The parent parallel only ever checkpoints SUCCEED. A terminal status the + // SDK never writes (CANCELLED/STOPPED/TIMED_OUT/FAILED) is a replay mismatch + // and must not silently re-run and overwrite the prior outcome. + var parentOpId = IdAt(1); + var (context, _, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = "CANCELLED", + SubType = OperationSubTypes.Parallel, + Name = "fanout" + } + } + }); + + Assert.Throws(() => context.CreateParallel(name: "fanout")); + } + + [Fact] + public async Task CreateParallel_ReplayBranchCountMismatch_Throws() + { + // Registering a different number of branches than the frozen summary recorded + // violates the positional replay contract. + var parentOpId = IdAt(1); + var summaryJson = """ + {"CompletionReason":"ALL_COMPLETED","Units":[ + {"Index":0,"Name":"inventory","Status":"SUCCEEDED","Result":"1"}, + {"Index":1,"Name":"payment","Status":"SUCCEEDED","Result":"2"} + ]} + """; + + var (context, _, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = OperationStatuses.Succeeded, + SubType = OperationSubTypes.Parallel, + Name = "fanout", + ContextDetails = new ContextDetails { Result = summaryJson } + } + } + }); + + var parallel = context.CreateParallel(name: "fanout"); + _ = parallel.Branch("inventory", async (_, _) => { await Task.Yield(); return 1; }); + // Only one branch registered, but the checkpoint recorded two. + await Assert.ThrowsAsync(async () => await parallel.CompleteAsync()); + await parallel.DisposeAsync(); // must not throw a secondary exception + } + + [Fact] + public void CompletionPolicy_PercentageTolerance_NotEvaluatedBeforeSeal() + { + // A percentage-based tolerance must not short-circuit against an incomplete + // denominator: 1 failure out of 1 registered-so-far is 100%, but with two + // more registrations pending the true ratio may be under threshold. + var policy = new CompletionPolicy(new CompletionConfig { ToleratedFailurePercentage = 0.5 }); + + // Pre-seal: percentage suppressed → do NOT stop dispatching. + Assert.False(policy.ShouldStopDispatching(succeeded: 0, failed: 1, totalBranches: 1, evaluatePercentage: false)); + + // Post-seal with the true denominator: 1/3 <= 0.5 → still do not stop. + Assert.False(policy.ShouldStopDispatching(succeeded: 0, failed: 1, totalBranches: 3, evaluatePercentage: true)); + + // Post-seal, genuinely over threshold: 2/3 > 0.5 → stop. + Assert.True(policy.ShouldStopDispatching(succeeded: 0, failed: 2, totalBranches: 3, evaluatePercentage: true)); + } + + // ────────────────────────────────────────────────────────────────────── + // Per-branch serialization (stacked on feature/per-step-serializer) + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task CreateParallel_PerBranchSerializer_UsedForThatBranchOnly() + { + var (context, _, _, _) = CreateContext(); + var custom = new CountingSerializer(); + + IParallelBranch a; + IParallelBranch b; + await using (var parallel = context.CreateParallel()) + { + // Branch "a" overrides its serializer; branch "b" uses the global default. + a = parallel.Branch("a", async (_, _) => { await Task.Yield(); return 7; }, serializer: custom); + b = parallel.Branch("b", async (_, _) => { await Task.Yield(); return 8; }); + await parallel.CompleteAsync(); + } + + // Results round-trip correctly regardless of which serializer produced them. + Assert.Equal(7, await a); + Assert.Equal(8, await b); + + // The per-branch serializer was exercised for branch "a". + Assert.True(custom.SerializeCount > 0, "custom per-branch serializer should have serialized branch a's result"); + } + + [Fact] + public async Task CreateParallel_ItemSerializer_AppliesToAllBranchesByDefault() + { + var (context, _, _, _) = CreateContext(); + var shared = new CountingSerializer(); + + await using (var parallel = context.CreateParallel(config: new ParallelConfig { ItemSerializer = shared })) + { + _ = parallel.Branch("a", async (_, _) => { await Task.Yield(); return 1; }); + _ = parallel.Branch("b", async (_, _) => { await Task.Yield(); return 2; }); + var summary = await parallel.CompleteAsync(); + Assert.Equal(2, summary.SuccessCount); + } + + // The operation-level ItemSerializer served both branches. + Assert.True(shared.SerializeCount >= 2, "ItemSerializer should serialize every branch result by default"); + } + + // ────────────────────────────────────────────────────────────────────── + // Deferred operation-level serializer resolution (comment 4) + // ────────────────────────────────────────────────────────────────────── + + private static DurableContext CreateContextNoGlobalSerializer(out RecordingBatcher recorder) + { + var state = new ExecutionState(); + state.LoadFromCheckpoint(null); + var tm = new TerminationManager(); + var idGen = new OperationIdGenerator(); + var lambdaContext = new TestLambdaContext(); // NO Serializer registered + recorder = new RecordingBatcher(); + return new DurableContext(state, tm, new WorkflowCancellation(tm), idGen, "arn:test", lambdaContext, recorder.Batcher); + } + + [Fact] + public async Task CreateParallel_NoGlobalSerializer_AllBranchesOverride_DoesNotThrow() + { + // AOT / per-branch scenario: with no global serializer registered, + // CreateParallel and Branch must not eagerly demand one. Resolution of the + // operation-level default (LambdaSerializerHelper.GetRequired) is deferred and + // never reached when every branch supplies its own serializer. + var context = CreateContextNoGlobalSerializer(out _); + var custom = new CountingSerializer(); + + IParallelBranch a; + IParallelBranch b; + await using (var parallel = context.CreateParallel()) // must NOT throw + { + a = parallel.Branch("a", async (_, _) => { await Task.Yield(); return 1; }, serializer: custom); + b = parallel.Branch("b", async (_, _) => { await Task.Yield(); return 2; }, serializer: custom); + var summary = await parallel.CompleteAsync(); + Assert.Equal(2, summary.SuccessCount); + } + + Assert.Equal(1, await a); + Assert.Equal(2, await b); + } + + [Fact] + public async Task CreateParallel_NoGlobalSerializer_BranchWithoutOverride_ThrowsOnThatBranch() + { + // The deferral does not swallow the requirement: a branch that omits its + // serializer falls back to the operation-level default, which resolves + // GetRequired and throws when no global serializer exists — but only then, + // not at CreateParallel time. + var context = CreateContextNoGlobalSerializer(out _); + + await using var parallel = context.CreateParallel(); // deferred: does NOT throw here + Assert.Throws(() => + parallel.Branch("a", async (_, _) => { await Task.Yield(); return 1; })); + } + + // ────────────────────────────────────────────────────────────────────── + // Workflow-level fault surfaces on the handle rather than hanging (comment 7) + // ────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task CreateParallel_BranchThrowsWorkflowError_AwaitingHandleFaults_NotHangs() + { + // A workflow-level fault (NonDeterministicExecutionException) is rethrown out + // of the branch's ExecuteAsync. _result must be faulted before the rethrow so + // a caller that catches the fault out of CompleteAsync and then awaits the + // branch handle observes the same fault instead of hanging on a handle whose + // result was never completed. + var (context, _, _, _) = CreateContext(); + + await using var parallel = context.CreateParallel(); + var bad = parallel.Branch("bad", async (_, _) => + { + await Task.Yield(); + throw new NonDeterministicExecutionException("boom"); + }); + + // The workflow-level fault propagates out of CompleteAsync. + await Assert.ThrowsAsync(async () => await parallel.CompleteAsync()); + + // Awaiting the handle must COMPLETE (faulted), not hang. Guard with a timeout + // so a regression fails the test deterministically instead of blocking it. + async Task AwaitHandle() => await bad; + var handleTask = AwaitHandle(); + var finished = await Task.WhenAny(handleTask, Task.Delay(TimeSpan.FromSeconds(10))); + Assert.Same(handleTask, finished); + await Assert.ThrowsAsync(async () => await handleTask); + } + + /// + /// Delegating that counts calls, so a + /// test can assert which serializer a branch used. + /// + private sealed class CountingSerializer : Amazon.Lambda.Core.ILambdaSerializer + { + private readonly DefaultLambdaJsonSerializer _inner = new(); + public int SerializeCount; + public int DeserializeCount; + + public T Deserialize(System.IO.Stream requestStream) + { + Interlocked.Increment(ref DeserializeCount); + return _inner.Deserialize(requestStream); + } + + public void Serialize(T response, System.IO.Stream responseStream) + { + Interlocked.Increment(ref SerializeCount); + _inner.Serialize(response, responseStream); + } + } +} diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/PerOperationSerializerTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/PerOperationSerializerTests.cs index 48b342377..3261b3599 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/PerOperationSerializerTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/PerOperationSerializerTests.cs @@ -120,6 +120,62 @@ public async Task Step_Replay_UsesPerOpSerializerToDeserialize() Assert.Equal(0, global.DeserializeCount); } + // ---------------------------------------------------------------- Step round-trip failure (fresh success) + + /// + /// Serializer that serializes normally but throws on deserialize — models a + /// custom serializer that cannot round-trip its own just-written payload. + /// + private sealed class DeserializeThrowingSerializer : ILambdaSerializer + { + private readonly ILambdaSerializer _inner = new DefaultLambdaJsonSerializer(); + public sealed class CannotDeserialize : Exception { } + + public T Deserialize(Stream requestStream) => throw new CannotDeserialize(); + public void Serialize(T response, Stream responseStream) => _inner.Serialize(response, responseStream); + } + + [Fact] + public async Task Step_FreshSuccess_RoundTripDeserializeFailure_SurfacesFaultWithoutRetryOrFail() + { + // Comment 1 (BLOCKER): the fresh-success round-trip deserialize sits OUTSIDE + // the try that funnels step failures into HandleStepFailureAsync. A serializer + // that cannot deserialize its own just-written payload must surface the fault + // directly — it must NOT be caught and turned into a RETRY/FAIL checkpoint, + // because the SUCCEED has already been committed and a conflicting terminal + // record would risk a duplicate side effect on retry. + var state = new ExecutionState(); + state.LoadFromCheckpoint(null); + var tm = new TerminationManager(); + var idGen = new OperationIdGenerator(); + var lambdaContext = new TestLambdaContext { Serializer = new DefaultLambdaJsonSerializer() }; + var recorder = new RecordingBatcher(); + var ctx = new DurableContext(state, tm, new WorkflowCancellation(tm), idGen, TestArn, lambdaContext, recorder.Batcher); + + var perOp = new DeserializeThrowingSerializer(); + + // The raw deserialize fault propagates — NOT wrapped in StepException, which + // is what HandleStepFailureAsync would have produced had the deserialize been + // caught by the retry/fail funnel. + await Assert.ThrowsAsync(async () => + await ctx.StepAsync( + async (_, _) => { await Task.CompletedTask; return 42; }, + name: "s", + config: new StepConfig { Serializer = perOp })); + + await recorder.Batcher.DrainAsync(); + var stepActions = recorder.Flushed + .Where(o => o.Type == OperationTypes.Step) + .Select(o => o.Action) + .ToList(); + + // The terminal SUCCEED was committed, and crucially NO RETRY/FAIL was enqueued + // for the same operation afterward. + Assert.Contains(OperationAction.SUCCEED, stepActions); + Assert.DoesNotContain(OperationAction.FAIL, stepActions); + Assert.DoesNotContain(OperationAction.RETRY, stepActions); + } + // ---------------------------------------------------------------- Callback (deserialize side) [Fact]