diff --git a/.autover/changes/durable-concurrent-js-parity.json b/.autover/changes/durable-concurrent-js-parity.json new file mode 100644 index 000000000..b46781f1b --- /dev/null +++ b/.autover/changes/durable-concurrent-js-parity.json @@ -0,0 +1,11 @@ +{ + "Projects": [ + { + "Name": "Amazon.Lambda.DurableExecution", + "Type": "Minor", + "ChangelogMessages": [ + "Align ParallelAsync/MapAsync failure semantics with the JS/Python SDKs (breaking, preview). ParallelAsync and MapAsync no longer throw on failure — they always return an IBatchResult; inspect CompletionReason/HasFailure or call ThrowIfError(). The ParallelException and MapException types are removed. MapConfig.CompletionConfig now defaults to AllSuccessful() (fail-fast), matching ParallelConfig. An empty CompletionConfig() is now fail-fast (any failure resolves FailureToleranceExceeded); use CompletionConfig.AllCompleted() to run every unit regardless of failures. MapConfig is now generic (MapConfig) so ItemNamer is typed Func instead of Func. Preview." + ] + } + ] +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/CompletionConfig.cs b/Libraries/src/Amazon.Lambda.DurableExecution/CompletionConfig.cs index 8ae0fb59b..a5bbd9c99 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/CompletionConfig.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/CompletionConfig.cs @@ -90,20 +90,29 @@ public double? ToleratedFailurePercentage } /// - /// All items must succeed. Equivalent to - /// = 0. The default for - /// . + /// All items must succeed — any single failure resolves the batch with + /// . Equivalent to + /// = 0, and to a default (empty) + /// . The default for both + /// and + /// , matching the JS/Python SDKs. /// public static CompletionConfig AllSuccessful() => new() { ToleratedFailureCount = 0 }; /// /// Run every branch regardless of failures; surface failures per-item via - /// . Resolution does not auto-throw — - /// the caller can inspect the result and call - /// if they want strict-success - /// behavior. + /// . Never resolves with + /// . The caller + /// inspects the result and can call + /// if it wants strict-success behavior. /// - public static CompletionConfig AllCompleted() => new(); + /// + /// Sets to so it + /// stays lenient. An empty is fail-fast + /// (equivalent to ), so leniency must be opted into + /// explicitly. + /// + public static CompletionConfig AllCompleted() => new() { ToleratedFailureCount = int.MaxValue }; /// /// Resolve once at least one branch has succeeded. Branches that were not diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/CompletionReason.cs b/Libraries/src/Amazon.Lambda.DurableExecution/CompletionReason.cs index 5d7a97805..db3221945 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/CompletionReason.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/CompletionReason.cs @@ -24,9 +24,11 @@ public enum CompletionReason /// /// or - /// was exceeded. - /// The batch is considered failed and surfaces a - /// when awaited. + /// was exceeded (a + /// default/empty is fail-fast, so any failure + /// trips this). The batch is considered failed: + /// is true and surfaces the + /// first item error. The operation itself does not throw. /// FailureToleranceExceeded } diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs b/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs index 5ff6c1bab..b062d4faa 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs @@ -258,7 +258,7 @@ public Task> MapAsync( IReadOnlyList items, Func, CancellationToken, Task> func, string? name = null, - MapConfig? config = null, + MapConfig? config = null, CancellationToken cancellationToken = default) => RunMap(items, func, name, config, cancellationToken); @@ -266,13 +266,13 @@ private Task> RunMap( IReadOnlyList items, Func, CancellationToken, Task> func, string? name, - MapConfig? config, + MapConfig? config, CancellationToken cancellationToken) { if (items == null) throw new ArgumentNullException(nameof(items)); if (func == null) throw new ArgumentNullException(nameof(func)); - var effectiveConfig = config ?? new MapConfig(); + var effectiveConfig = config ?? new MapConfig(); var serializer = LambdaSerializerHelper.GetRequired(LambdaContext); diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/DurableExecutionException.cs b/Libraries/src/Amazon.Lambda.DurableExecution/DurableExecutionException.cs index e4748b381..7f8707966 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/DurableExecutionException.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/DurableExecutionException.cs @@ -98,69 +98,3 @@ public ChildContextException(string message) : base(message) { } /// Creates a wrapping an inner exception. public ChildContextException(string message, Exception innerException) : base(message, innerException) { } } - -/// -/// Thrown when a parallel operation resolves with -/// . The aggregate -/// is preserved on so callers -/// can inspect per-branch outcomes. -/// -/// -/// This is the base type for parallel failures. Subclasses may be added in -/// future releases (for example, a dedicated -/// ParallelFailureToleranceExceededException); catching -/// remains forward-compatible. -/// -public class ParallelException : DurableExecutionException -{ - /// - /// The aggregate result of the parallel operation. Type-erased — cast to - /// IBatchResult<T> if the per-branch result type is known. - /// - public IBatchResult? Result { get; init; } - - /// - /// Why the parallel operation resolved. - /// - public CompletionReason CompletionReason { get; init; } - - /// Creates an empty . - public ParallelException() { } - /// Creates a with the given message. - public ParallelException(string message) : base(message) { } - /// Creates a wrapping an inner exception. - public ParallelException(string message, Exception innerException) : base(message, innerException) { } -} - -/// -/// Thrown when a map operation resolves with -/// . The aggregate -/// is preserved on so callers -/// can inspect per-item outcomes. -/// -/// -/// This is the base type for map failures. Subclasses may be added in future -/// releases; catching remains forward-compatible. -/// A dedicated type (rather than reusing ) lets -/// callers pattern-match which concurrent operation failed. -/// -public class MapException : DurableExecutionException -{ - /// - /// The aggregate result of the map operation. Type-erased — cast to - /// IBatchResult<T> if the per-item result type is known. - /// - public IBatchResult? Result { get; init; } - - /// - /// Why the map operation resolved. - /// - public CompletionReason CompletionReason { get; init; } - - /// Creates an empty . - public MapException() { } - /// Creates a with the given message. - public MapException(string message) : base(message) { } - /// Creates a wrapping an inner exception. - public MapException(string message, Exception innerException) : base(message, innerException) { } -} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/IBatchResult.cs b/Libraries/src/Amazon.Lambda.DurableExecution/IBatchResult.cs index a93e46190..f449110f4 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/IBatchResult.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/IBatchResult.cs @@ -4,9 +4,9 @@ namespace Amazon.Lambda.DurableExecution; /// -/// Non-generic marker for . Used by -/// so callers can hold a reference to -/// the aggregate result without knowing the per-branch type at compile time. +/// Non-generic marker for . Lets callers hold a +/// reference to the aggregate result, or read the completion bookkeeping, without +/// knowing the per-branch type at compile time. /// public interface IBatchResult { diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs b/Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs index 2760ab0ee..9d536f5d3 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs @@ -377,9 +377,11 @@ Task WaitForConditionAsync( /// /// On per-branch failure (a branch's user function throws), the failure is /// captured on the corresponding instead of - /// aborting the parallel. The parallel only throws - /// when - /// criteria are violated. Use + /// aborting the parallel. The parallel NEVER throws on failure — it always + /// returns an . By default + /// (, fail-fast) any branch failure + /// resolves it with ; + /// failures surface via . Use /// for explicit strict-success /// semantics. Per-branch results are serialized to checkpoints using the /// registered on @@ -449,8 +451,8 @@ Task> ParallelAsync( /// Process a collection of items concurrently, running /// once per item. Each item runs inside its own child context; per-item /// results are aggregated into an . Items - /// are dispatched up to ; the aggregate - /// resolves according to . + /// are dispatched up to ; the aggregate + /// resolves according to . /// /// /// The per-item function receives the durable context, the item, its @@ -458,13 +460,17 @@ Task> ParallelAsync( /// linking the /// caller-supplied token with the SDK's workflow-shutdown signal (it is also /// tripped cooperatively when a sibling item satisfies the - /// and the map short-circuits). On + /// and the map short-circuits). On /// per-item failure (the user function throws), the failure is captured on /// the corresponding instead of aborting - /// the map. By default () every - /// item runs and failures surface via ; - /// the map throws only when - /// criteria are violated. Use + /// the map. The map NEVER throws on failure — it always returns an + /// . By default + /// (, fail-fast) any item failure + /// resolves the map with + /// ; use + /// to run every item regardless. + /// Failures surface via / + /// ; call /// for explicit /// strict-success semantics. Per-item results are serialized to checkpoints /// using the registered on @@ -474,7 +480,7 @@ Task> MapAsync( IReadOnlyList items, Func, CancellationToken, Task> func, string? name = null, - MapConfig? config = null, + MapConfig? config = null, CancellationToken cancellationToken = default); } diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CompletionPolicy.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CompletionPolicy.cs index b12c8d0b3..b1573a593 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CompletionPolicy.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/CompletionPolicy.cs @@ -22,11 +22,20 @@ internal readonly struct CompletionPolicy private readonly int? _toleratedFailureCount; private readonly double? _toleratedFailurePercentage; + // True when NO completion criteria are set at all (every field null). This is + // the fail-fast default, matching the JS/Python SDKs: an empty CompletionConfig + // fails the batch on the first failed unit. Setting ANY field (even + // MinSuccessful alone) opts out of this and uses the explicit thresholds below. + private readonly bool _failFastOnAnyFailure; + public CompletionPolicy(CompletionConfig config) { _minSuccessful = config.MinSuccessful; _toleratedFailureCount = config.ToleratedFailureCount; _toleratedFailurePercentage = config.ToleratedFailurePercentage; + _failFastOnAnyFailure = _minSuccessful is null + && _toleratedFailureCount is null + && _toleratedFailurePercentage is null; } /// @@ -61,12 +70,18 @@ public CompletionReason Evaluate(int succeeded, int failed, int started, int tot private bool MinSuccessfulReached(int succeeded) => _minSuccessful is { } min && succeeded >= min; - // Failure count or ratio STRICTLY exceeds a configured threshold. Only a - // threshold that was explicitly set can trip this — an "empty" CompletionConfig - // (all properties null) is permissive. CompletionConfig.AllSuccessful() opts - // into fail-fast by setting ToleratedFailureCount = 0. + // Failure count or ratio STRICTLY exceeds a configured threshold. An "empty" + // CompletionConfig (all properties null) is FAIL-FAST — any failure trips it — + // matching the JS/Python SDKs. Setting any explicit threshold opts out of that + // and uses the thresholds below. CompletionConfig.AllSuccessful() (which sets + // 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) { + if (_failFastOnAnyFailure) + return failed > 0; + if (_toleratedFailureCount is { } tfc && failed > tfc) return true; diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs index 33ba97965..9e247f329 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs @@ -31,17 +31,20 @@ namespace Amazon.Lambda.DurableExecution.Internal; /// SUCCEED with summary payload (). /// SUCCEEDED: parent payload supplies the snapshot of per-unit /// statuses + completion reason; per-unit results are deserialised from the -/// children's own CONTEXT checkpoints. If the completion reason is -/// , throws the -/// subclass exception carrying the rebuilt . +/// children's own CONTEXT checkpoints. The rebuilt +/// is returned regardless of completion +/// reason. /// STARTED / PENDING: re-execute (children replay from their /// own checkpoints). /// -/// Per-unit errors do NOT abort the operation directly — the orchestrator catches -/// each unit's , records it as a failed +/// Per-unit errors do NOT abort the operation — the orchestrator catches each +/// unit's , records it as a failed /// , and consults the -/// after every completion. Only when the completion config marks the run as -/// does it throw. +/// after every completion only to decide whether to stop dispatching. The +/// operation ALWAYS returns an — it never throws on +/// failure, matching the JS/Python/Java SDKs. Callers inspect +/// / +/// or call to surface a failure. /// internal abstract class ConcurrentOperation : DurableOperation> { @@ -104,21 +107,12 @@ protected ConcurrentOperation( /// Singular operation noun used in messages (e.g. "Parallel" / "Map"). protected abstract string OperationNoun { get; } - /// Plural unit noun used in messages (e.g. "branches" / "items"). - protected abstract string UnitNounPlural { get; } - /// /// Resolves the unit at into its display name and the /// function to run inside the unit's child context. /// protected abstract (string? Name, Func> Func) GetUnit(int index); - /// - /// Builds the subclass-specific exception thrown when the operation resolves - /// with . - /// - protected abstract DurableExecutionException CreateException(string message, IBatchResult result); - // ── Orchestration ─────────────────────────────────────────────────── protected override async Task> StartAsync(CancellationToken cancellationToken) @@ -159,12 +153,10 @@ protected override Task> ReplayAsync(Operation existing, Cancell case OperationStatuses.Succeeded: // The parent always checkpoints as SUCCEED — even when - // CompletionReason is FailureToleranceExceeded. Reconstruct - // the BatchResult and throw if it was a tolerance failure. - var result = ReconstructFromCheckpoints(existing); - if (result.CompletionReason == CompletionReason.FailureToleranceExceeded) - throw BuildException(result); - return Task.FromResult(result); + // CompletionReason is FailureToleranceExceeded. Reconstruct and + // return the BatchResult; the operation never throws on failure + // (the caller inspects CompletionReason / calls ThrowIfError). + return Task.FromResult(ReconstructFromCheckpoints(existing)); case OperationStatuses.Started: case OperationStatuses.Pending: @@ -382,17 +374,11 @@ void SignalShortCircuit() var completionReason = ComputeCompletionReason(items, unitCount); var result = new BatchResult(items, completionReason); - var failureException = completionReason == CompletionReason.FailureToleranceExceeded - ? BuildException(result) - : null; - await CheckpointParentResultAsync(result, completionReason, cancellationToken); - if (failureException != null) - { - throw failureException; - } - + // Never throw on failure — always return the aggregate result. The caller + // inspects CompletionReason / HasFailure or calls ThrowIfError. Matches + // the JS/Python/Java SDKs. return result; } @@ -476,15 +462,9 @@ private async Task> ReplayChildrenAsync(Operation frozen, Cancel ? DeserializeCompletionReason(summary.CompletionReason) : ComputeCompletionReason(items, unitCount); - var result = new BatchResult(items, completionReason); - - // No re-checkpoint: the parent is already terminal in state. - if (completionReason == CompletionReason.FailureToleranceExceeded) - { - throw BuildException(result); - } - - return result; + // No re-checkpoint: the parent is already terminal in state. Return the + // reconstructed result regardless of completion reason — never throw. + return new BatchResult(items, completionReason); } private async Task RunUnitAsync( @@ -698,14 +678,6 @@ private CompletionReason ComputeCompletionReason(IReadOnlyList> it return _policy.Evaluate(succeeded, failed, started, totalCount); } - private DurableExecutionException BuildException(IBatchResult result) - { - var message = - $"{OperationNoun} operation failed: failure tolerance exceeded " + - $"({result.FailureCount} of {result.TotalCount} {UnitNounPlural} failed)."; - return CreateException(message, result); - } - private async Task CheckpointParentResultAsync( BatchResult result, CompletionReason completionReason, diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs index a93a07fb3..21299a3cf 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs @@ -14,14 +14,13 @@ namespace Amazon.Lambda.DurableExecution.Internal; /// checkpoint, and replay logic lives in ; /// this subclass supplies only the map-specific bits: how to turn an item index /// into a (name, func) pair (the per-item callback receives the item, its -/// index, and the full source list), the Map sub-type labels, and the -/// factory. +/// index, and the full source list) and the Map sub-type labels. /// internal sealed class MapOperation : ConcurrentOperation { private readonly IReadOnlyList _items; private readonly Func, CancellationToken, Task> _func; - private readonly Func? _itemNamer; + private readonly Func? _itemNamer; public MapOperation( string operationId, @@ -29,7 +28,7 @@ public MapOperation( string? parentId, IReadOnlyList items, Func, CancellationToken, Task> func, - MapConfig config, + MapConfig config, ILambdaSerializer serializer, Func childContextFactory, ExecutionState state, @@ -51,7 +50,6 @@ public MapOperation( protected override string ParentSubType => OperationSubTypes.Map; protected override string ChildSubType => OperationSubTypes.MapItem; protected override string OperationNoun => "Map"; - protected override string UnitNounPlural => "items"; protected override (string? Name, Func> Func) GetUnit(int index) { @@ -61,7 +59,7 @@ protected override (string? Name, Func _func(ctx, item, index, _items, ct)); } - - protected override DurableExecutionException CreateException(string message, IBatchResult result) - { - return new MapException(message) - { - Result = result, - CompletionReason = result.CompletionReason - }; - } } diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ParallelOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ParallelOperation.cs index 4d1b2bf41..ab7327db0 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ParallelOperation.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ParallelOperation.cs @@ -43,20 +43,10 @@ public ParallelOperation( protected override string ParentSubType => OperationSubTypes.Parallel; protected override string ChildSubType => OperationSubTypes.ParallelBranch; protected override string OperationNoun => "Parallel"; - protected override string UnitNounPlural => "branches"; protected override (string? Name, Func> Func) GetUnit(int index) { var branch = _branches[index]; return (branch.Name, branch.Func); } - - protected override DurableExecutionException CreateException(string message, IBatchResult result) - { - return new ParallelException(message) - { - Result = result, - CompletionReason = result.CompletionReason - }; - } } diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/MapConfig.cs b/Libraries/src/Amazon.Lambda.DurableExecution/MapConfig.cs index 0f33b9b14..9a58ea489 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/MapConfig.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/MapConfig.cs @@ -2,8 +2,12 @@ namespace Amazon.Lambda.DurableExecution; /// /// Configuration for -/// . +/// . /// +/// +/// The type of each item processed by the map. Generic so +/// receives the strongly-typed item rather than object. +/// /// /// Per-item checkpoint payloads are serialized via the /// registered on @@ -11,7 +15,7 @@ namespace Amazon.Lambda.DurableExecution; /// configured via LambdaBootstrapBuilder.Create(handler, serializer)); /// this config does not expose a serializer slot. /// -public sealed class MapConfig +public sealed class MapConfig { private int? _maxConcurrency; @@ -38,19 +42,21 @@ public int? MaxConcurrency /// /// When the map operation is considered complete. Defaults to - /// — every item runs regardless - /// of per-item failures, which are surfaced via - /// rather than thrown. + /// (fail-fast) — any item failure + /// resolves the map with + /// , matching the + /// JS/Python SDKs and . /// /// - /// This differs from , which - /// defaults to (fail-fast). For - /// fail-fast map behavior — any item failure surfaces a - /// when the result is awaited — set this to - /// , or call - /// on the result. + /// The map never throws on failure — it always returns an + /// . Inspect + /// / + /// or call + /// to surface failures. For + /// run-everything semantics, set this to + /// . /// - public CompletionConfig CompletionConfig { get; set; } = CompletionConfig.AllCompleted(); + public CompletionConfig CompletionConfig { get; set; } = CompletionConfig.AllSuccessful(); /// /// How item branches are represented in the checkpoint graph. Defaults to @@ -65,10 +71,10 @@ public int? MaxConcurrency /// /// Optional function to generate a custom name for each item's branch. - /// Receives the item and its zero-based index, and returns the branch name - /// surfaced in execution traces and on . - /// When null (default), branches are named by index ("0", - /// "1", ...). + /// Receives the strongly-typed item and its zero-based index, and returns the + /// branch name surfaced in execution traces and on + /// . When null (default), branches are + /// named by index ("0", "1", ...). /// - public Func? ItemNamer { get; set; } + public Func? ItemNamer { get; set; } } diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/ParallelConfig.cs b/Libraries/src/Amazon.Lambda.DurableExecution/ParallelConfig.cs index 3d08d1636..69b82a3bc 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/ParallelConfig.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/ParallelConfig.cs @@ -41,10 +41,17 @@ public int? MaxConcurrency /// /// When the parallel operation is considered complete. Defaults to - /// — any single branch failure - /// surfaces as a when the parallel result - /// is awaited. + /// (fail-fast) — any single + /// branch failure resolves the operation with + /// . /// + /// + /// The parallel operation never throws on failure — it always returns an + /// . Inspect + /// / + /// or call + /// to surface failures. + /// public CompletionConfig CompletionConfig { get; set; } = CompletionConfig.AllSuccessful(); /// diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md b/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md index 1b77c333c..317a907fd 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md +++ b/Libraries/src/Amazon.Lambda.DurableExecution/docs/core/parallel.md @@ -44,7 +44,7 @@ var batch = await ctx.ParallelAsync( var quotes = batch.GetResults(); // all three, in original branch order ``` -With the default completion policy (`AllSuccessful`), any single branch failure surfaces as a `ParallelException` when the result is awaited. +With the default completion policy (`AllSuccessful`, fail-fast), any single branch failure resolves the batch with `CompletionReason.FailureToleranceExceeded` and `HasFailure == true`. The operation never throws on failure — inspect the result or call `ThrowIfError()`. ## Configuration @@ -53,11 +53,11 @@ public sealed class ParallelConfig { public int? MaxConcurrency { get; set; } // null = unlimited; must be >= 1 when set public CompletionConfig CompletionConfig { get; set; } = CompletionConfig.AllSuccessful(); - public NestingType NestingType { get; set; } = NestingType.Nested; // Flat is reserved — throws NotSupportedException + public NestingType NestingType { get; set; } = NestingType.Nested; } ``` -`MaxConcurrency` bounds how many branches run at once via a semaphore — useful to avoid overwhelming a downstream service. `NestingType.Nested` (default) gives each branch a full child context visible in traces; `NestingType.Flat` is reserved for a future checkpoint optimization and currently throws `NotSupportedException`. +`MaxConcurrency` bounds how many branches run at once via a semaphore — useful to avoid overwhelming a downstream service. `NestingType.Nested` (default) gives each branch a full child context visible in traces; `NestingType.Flat` runs branches in virtual contexts that emit no per-branch `CONTEXT` checkpoint, recording per-branch results inline on the parallel operation's payload instead — fewer checkpoints, at the cost of trace granularity. ## Completion policies @@ -65,8 +65,8 @@ public sealed class ParallelConfig | Factory | Behavior | | --- | --- | -| `CompletionConfig.AllSuccessful()` | Every branch must succeed (equivalent to `ToleratedFailureCount = 0`). The first failure resolves the batch as failed. **Default.** | -| `CompletionConfig.AllCompleted()` | Run every branch to a terminal state regardless of failures; never auto-throws. Inspect `Succeeded` / `Failed` (or call `ThrowIfError`) afterward. | +| `CompletionConfig.AllSuccessful()` | Every branch must succeed (equivalent to `ToleratedFailureCount = 0`, and to a default/empty `CompletionConfig`). Any failure resolves the batch as `FailureToleranceExceeded`. **Default.** | +| `CompletionConfig.AllCompleted()` | Run every branch to a terminal state regardless of failures (`ToleratedFailureCount = int.MaxValue`). Inspect `Succeeded` / `Failed` (or call `ThrowIfError`) afterward. | | `CompletionConfig.FirstSuccessful()` | Resolve as soon as one branch succeeds (`MinSuccessful = 1`). Branches not yet dispatched are reported as `Started`. | For finer control, set the properties yourself: @@ -121,18 +121,17 @@ foreach (var item in batch.Failed) var succeeded = batch.GetResults(); ``` -With the default `AllSuccessful` policy, awaiting a batch in which a branch failed throws `ParallelException`. The exception carries the type-erased `Result` (cast to `IBatchResult` to inspect per-branch detail) and the `CompletionReason`: +The operation never throws on failure — even under the default `AllSuccessful` (fail-fast) policy, a batch with a failed branch resolves with `CompletionReason.FailureToleranceExceeded` and `HasFailure == true` (matching the JS/Python/Java SDKs). Inspect the result, or call `ThrowIfError()` to opt into surfacing the first branch failure as an exception: ```csharp -try -{ - var batch = await ctx.ParallelAsync(branches, name: "fan-out"); -} -catch (ParallelException ex) +var batch = await ctx.ParallelAsync(branches, name: "fan-out"); + +if (batch.HasFailure) { - var result = (IBatchResult?)ex.Result; ctx.Logger.LogWarning( "Parallel operation failed ({Reason}); {Failed} of {Total} branches failed.", - ex.CompletionReason, result?.FailureCount, result?.TotalCount); + batch.CompletionReason, batch.FailureCount, batch.TotalCount); + + batch.ThrowIfError(); // rethrow the first branch's DurableExecutionException, if desired } ``` diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Analyzers.Tests/DurableStubs.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Analyzers.Tests/DurableStubs.cs index 2e7c36ebd..e6004de87 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Analyzers.Tests/DurableStubs.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Analyzers.Tests/DurableStubs.cs @@ -32,7 +32,7 @@ public sealed class CallbackConfig { } public sealed class WaitForCallbackConfig { } public sealed class InvokeConfig { } public sealed class ParallelConfig { } - public sealed class MapConfig { } + public sealed class MapConfig { } public sealed class WaitForConditionConfig { } public readonly struct DurableBranch { public DurableBranch(string name, Func> func) { } } public interface IBatchResult { } @@ -59,7 +59,7 @@ public interface IDurableContext Task> ParallelAsync(IReadOnlyList>> branches, string name = null, ParallelConfig config = null, CancellationToken cancellationToken = default); Task> ParallelAsync(IReadOnlyList> branches, string name = null, ParallelConfig config = null, CancellationToken cancellationToken = default); - Task> MapAsync(IReadOnlyList items, Func, CancellationToken, Task> func, string name = null, MapConfig config = null, CancellationToken cancellationToken = default); + Task> MapAsync(IReadOnlyList items, Func, CancellationToken, Task> func, string name = null, MapConfig config = null, CancellationToken cancellationToken = default); } } "; diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/MapFailureToleranceTest.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/MapFailureToleranceTest.cs index 25d773f03..cec7d1b3c 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/MapFailureToleranceTest.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/MapFailureToleranceTest.cs @@ -12,15 +12,15 @@ public class MapFailureToleranceTest public MapFailureToleranceTest(ITestOutputHelper output) => _output = output; /// - /// Five items, two fail, ToleratedFailureCount=1. The map must surface a - /// with reason - /// ; the workflow must - /// terminate FAILED. Validates the failure-tolerance short-circuit and that - /// MapException (not ParallelException) propagates as the - /// workflow's terminal error. + /// Five items, two fail, ToleratedFailureCount=1. The map resolves with + /// but does NOT throw + /// (JS parity): the workflow completes SUCCEEDED and the returned + /// BatchResult reports the completion reason. Validates the + /// failure-tolerance short-circuit and that the parent CONTEXT is checkpointed + /// ContextSucceeded (the completion reason lives inside the payload). /// [Fact] - public async Task Map_FailureToleranceExceeded_FailsWorkflow() + public async Task Map_FailureToleranceExceeded_CompletesWithReason() { await using var deployment = await DurableFunctionDeployment.CreateAsync( DurableFunctionDeployment.FindTestFunctionDir("MapFailureToleranceFunction"), @@ -30,24 +30,18 @@ public async Task Map_FailureToleranceExceeded_FailsWorkflow() var responsePayload = Encoding.UTF8.GetString(invokeResponse.Payload.ToArray()); _output.WriteLine($"Response: {responsePayload}"); - // Failed workflows return null payload to the Invoke caller — locate the - // execution by name to inspect its terminal status. var arn = await deployment.FindDurableExecutionArnByNameAsync(executionName, TimeSpan.FromSeconds(60)); Assert.NotNull(arn); + // The operation no longer throws on failure tolerance, so the workflow + // itself succeeds — the failure surfaces in the result payload, not as a + // terminal workflow error. var status = await deployment.PollForCompletionAsync(arn!, TimeSpan.FromSeconds(60)); - Assert.Equal("FAILED", status, ignoreCase: true); + Assert.Equal("SUCCEEDED", status, ignoreCase: true); var execution = await deployment.GetExecutionAsync(arn!); - Assert.NotNull(execution.Error); - // MapException is the terminal error type the SDK throws when the - // failure-tolerance short-circuit fires. - var errorType = execution.Error.ErrorType ?? string.Empty; - var errorMessage = execution.Error.ErrorMessage ?? string.Empty; - Assert.True( - errorType.Contains("MapException", StringComparison.Ordinal) - || errorMessage.Contains("Map", StringComparison.OrdinalIgnoreCase), - $"Expected error to indicate MapException; got type='{errorType}' message='{errorMessage}'"); + Assert.Null(execution.Error); + Assert.Contains("FailureToleranceExceeded", responsePayload, StringComparison.Ordinal); // History: parent CONTEXT and at least 2 failed item contexts visible. var history = await deployment.WaitForHistoryAsync( @@ -64,11 +58,8 @@ public async Task Map_FailureToleranceExceeded_FailsWorkflow() // The parent context (named "tolerance") is checkpointed ContextSucceeded // even when the failure tolerance is exceeded: ConcurrentOperation always // writes the parent batch summary with action SUCCEED (the completion - // reason lives inside the payload), then the SDK throws MapException - // AFTER the checkpoint. This matches the Python/JS/Java wire format. The - // workflow-level failure is asserted above via PollForCompletionAsync == - // FAILED and the MapException error type — the parent CONTEXT itself is - // NOT recorded as ContextFailed. + // reason lives inside the payload). This matches the Python/JS/Java wire + // format. The parent CONTEXT itself is NOT recorded as ContextFailed. var parentSucceeded = events.FirstOrDefault(e => e.EventType == EventType.ContextSucceeded && e.Name == "tolerance"); Assert.NotNull(parentSucceeded); diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/ParallelFailureToleranceTest.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/ParallelFailureToleranceTest.cs index 46d6de202..c826838c5 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/ParallelFailureToleranceTest.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/ParallelFailureToleranceTest.cs @@ -15,14 +15,15 @@ public class ParallelFailureToleranceTest public ParallelFailureToleranceTest(ITestOutputHelper output) => _output = output; /// - /// Five branches, two fail, ToleratedFailureCount=1. The parallel must surface a - /// with reason - /// ; the workflow must - /// terminate FAILED. Validates the failure-tolerance short-circuit and that - /// ParallelException propagates as the workflow's terminal error. + /// Five branches, two fail, ToleratedFailureCount=1. The parallel resolves with + /// but does NOT throw + /// (JS parity): the workflow completes SUCCEEDED and the returned + /// BatchResult reports the completion reason. Validates the + /// failure-tolerance short-circuit and that the parent CONTEXT is checkpointed + /// ContextSucceeded (the completion reason lives inside the payload). /// [Fact] - public async Task Parallel_FailureToleranceExceeded_FailsWorkflow() + public async Task Parallel_FailureToleranceExceeded_CompletesWithReason() { await using var deployment = await DurableFunctionDeployment.CreateAsync( DurableFunctionDeployment.FindTestFunctionDir("ParallelFailureToleranceFunction"), @@ -32,24 +33,18 @@ public async Task Parallel_FailureToleranceExceeded_FailsWorkflow() var responsePayload = Encoding.UTF8.GetString(invokeResponse.Payload.ToArray()); _output.WriteLine($"Response: {responsePayload}"); - // Failed workflows return null payload to the Invoke caller — locate the - // execution by name to inspect its terminal status. var arn = await deployment.FindDurableExecutionArnByNameAsync(executionName, TimeSpan.FromSeconds(60)); Assert.NotNull(arn); + // The operation no longer throws on failure tolerance, so the workflow + // itself succeeds — the failure surfaces in the result payload, not as a + // terminal workflow error. var status = await deployment.PollForCompletionAsync(arn!, TimeSpan.FromSeconds(60)); - Assert.Equal("FAILED", status, ignoreCase: true); + Assert.Equal("SUCCEEDED", status, ignoreCase: true); var execution = await deployment.GetExecutionAsync(arn!); - Assert.NotNull(execution.Error); - // ParallelException is the terminal error type the SDK throws when the - // failure-tolerance short-circuit fires. - var errorType = execution.Error.ErrorType ?? string.Empty; - var errorMessage = execution.Error.ErrorMessage ?? string.Empty; - Assert.True( - errorType.Contains("ParallelException", StringComparison.Ordinal) - || errorMessage.Contains("Parallel", StringComparison.OrdinalIgnoreCase), - $"Expected error to indicate ParallelException; got type='{errorType}' message='{errorMessage}'"); + Assert.Null(execution.Error); + Assert.Contains("FailureToleranceExceeded", responsePayload, StringComparison.Ordinal); // History: parent CONTEXT and at least 2 failed branch contexts visible. var history = await deployment.WaitForHistoryAsync( @@ -68,11 +63,8 @@ public async Task Parallel_FailureToleranceExceeded_FailsWorkflow() // The parent context (named "tolerance") is checkpointed ContextSucceeded // even when the failure tolerance is exceeded: ConcurrentOperation always // writes the parent batch summary with action SUCCEED (the completion - // reason lives inside the payload), then the SDK throws ParallelException - // AFTER the checkpoint. This matches the Python/JS/Java wire format. The - // workflow-level failure is asserted above via PollForCompletionAsync == - // FAILED and the ParallelException error type — the parent CONTEXT itself - // is NOT recorded as ContextFailed. + // reason lives inside the payload). This matches the Python/JS/Java wire + // format. The parent CONTEXT itself is NOT recorded as ContextFailed. var parentSucceeded = events.FirstOrDefault(e => e.EventType == EventType.ContextSucceeded && e.Name == "tolerance"); Assert.NotNull(parentSucceeded); diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/PortableScenariosCloudTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/PortableScenariosCloudTests.cs index 94b1fd08a..55ed0d23b 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/PortableScenariosCloudTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/PortableScenariosCloudTests.cs @@ -284,7 +284,7 @@ public async Task Parallel_FirstSuccessful_ShortCircuits() } [Fact] - public async Task Parallel_FailureTolerance_Exceeded_Fails() + public async Task Parallel_FailureTolerance_Exceeded_ResolvesWithReason() { await using var deployment = await DurableFunctionDeployment.CreateAsync( DurableFunctionDeployment.FindTestFunctionDir("ParallelFailureToleranceFunction"), @@ -339,7 +339,7 @@ public async Task Map_FirstSuccessful_ShortCircuits() } [Fact] - public async Task Map_FailureTolerance_Exceeded_Fails() + public async Task Map_FailureTolerance_Exceeded_ResolvesWithReason() { await using var deployment = await DurableFunctionDeployment.CreateAsync( DurableFunctionDeployment.FindTestFunctionDir("MapFailureToleranceFunction"), diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/MapFlatNestingFunction/Function.cs b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/MapFlatNestingFunction/Function.cs index 80f9181e5..2ae86dc0c 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/MapFlatNestingFunction/Function.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests/TestFunctions/MapFlatNestingFunction/Function.cs @@ -46,7 +46,7 @@ private async Task Workflow(TestEvent input, IDurableContext context return generatedId; }, name: "fanout", - config: new MapConfig { NestingType = NestingType.Flat }); + config: new MapConfig { NestingType = NestingType.Flat }); var joined = string.Join(",", batch.GetResults()); return new TestResult { Status = "completed", Data = joined }; diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Shared/Scenarios/PortableScenarios.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Shared/Scenarios/PortableScenarios.cs index a8c9b551c..3b03eb15e 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Shared/Scenarios/PortableScenarios.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Shared/Scenarios/PortableScenarios.cs @@ -424,14 +424,18 @@ public static async Task ParallelFirstSuccessfulAsync(IDurableTestRunnerBehavioral half of parallel failure tolerance: exceeding tolerance fails the workflow. + /// + /// Behavioral half of parallel failure tolerance: exceeding tolerance resolves + /// the batch with FailureToleranceExceeded. The operation does NOT throw + /// (JS parity) — the workflow completes and reports the completion reason. + /// public static async Task ParallelFailureToleranceAsync(IDurableTestRunner runner) { var result = await runner.RunAsync(new BatchRequest { OrderId = "tol" }); - Assert.True(result.IsFailed); - Assert.NotNull(result.Error); - Assert.Contains("ParallelException", result.Error!.ErrorType ?? string.Empty); + result.EnsureSucceeded(); + Assert.Equal("FailureToleranceExceeded", result.Result!.CompletionReason); + Assert.Equal(2, result.Result.FailureCount); } /// @@ -485,14 +489,18 @@ public static async Task MapFirstSuccessfulAsync(IDurableTestRunnerBehavioral half of map failure tolerance: exceeding tolerance fails the workflow with MapException. + /// + /// Behavioral half of map failure tolerance: exceeding tolerance resolves the + /// batch with FailureToleranceExceeded. The operation does NOT throw + /// (JS parity) — the workflow completes and reports the completion reason. + /// public static async Task MapFailureToleranceAsync(IDurableTestRunner runner) { var result = await runner.RunAsync(new BatchRequest { OrderId = "tol" }); - Assert.True(result.IsFailed); - Assert.NotNull(result.Error); - Assert.Contains("MapException", result.Error!.ErrorType ?? string.Empty); + result.EnsureSucceeded(); + Assert.Equal("FailureToleranceExceeded", result.Result!.CompletionReason); + Assert.Equal(2, result.Result.FailureCount); } /// Behavioral half of map max-concurrency: all six items complete (wave timing is cloud-only). diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Shared/Workflows/DurableWorkflows.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Shared/Workflows/DurableWorkflows.cs index 74211298c..7783bf6dc 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Shared/Workflows/DurableWorkflows.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Shared/Workflows/DurableWorkflows.cs @@ -73,7 +73,7 @@ await ctx.StepAsync( async (_, _) => { await Task.CompletedTask; return (index + 1) * 10m; }, name: "price"), name: "price_items", - config: new MapConfig { ItemNamer = (sku, index) => $"sku-{sku}" }); + config: new MapConfig { ItemNamer = (sku, index) => $"sku-{sku}" }); var pricedResults = priced.GetResults(); var orderTotal = await context.StepAsync( async (_, _) => { await Task.CompletedTask; return pricedResults.Sum(); }, @@ -669,7 +669,11 @@ public static async Task FirstSuccessfulAsync(BatchRequest input, I }; } - /// Two branches throw with tolerance=1, so the parallel exceeds tolerance and throws. + /// + /// Two branches throw with tolerance=1, so the parallel resolves with + /// . The operation does + /// NOT throw (JS parity) — the workflow inspects the result and reports it. + /// public static async Task FailureToleranceAsync(BatchRequest input, IDurableContext context) { var batch = await context.ParallelAsync( @@ -684,7 +688,13 @@ public static async Task FailureToleranceAsync(BatchRequest input, name: "tolerance", config: new ParallelConfig { CompletionConfig = new CompletionConfig { ToleratedFailureCount = 1 } }); - return new BatchResult { Status = "should_not_reach", SuccessCount = batch.SuccessCount }; + return new BatchResult + { + Status = "completed", + SuccessCount = batch.SuccessCount, + FailureCount = batch.FailureCount, + CompletionReason = batch.CompletionReason.ToString(), + }; } /// Six branches, MaxConcurrency=2; each waits then timestamps. All must succeed. @@ -735,12 +745,17 @@ await ctx.StepAsync( async (_, _) => { await Task.CompletedTask; return $"{orderId}-{input.OrderId}"; }, name: "process"), name: "process_all", - config: new MapConfig { ItemNamer = (item, index) => $"item-{item}" }); + config: new MapConfig { ItemNamer = (item, index) => $"item-{item}" }); return new BatchResult { Status = "completed", Data = string.Join(",", batch.GetResults()) }; } - /// Middle item throws; Map's default AllCompleted tolerates it. + /// + /// Middle item throws. Under the fail-fast default the map resolves with + /// , but with unlimited + /// concurrency all three items are dispatched before any completes, so the + /// result still reports two successes and one failure. The map does not throw. + /// public static async Task PartialFailureAsync(BatchRequest input, IDurableContext context) { var items = new[] { "ok1", "boom", "ok2" }; @@ -753,7 +768,8 @@ public static async Task PartialFailureAsync(BatchRequest input, ID throw new InvalidOperationException("intentional partial failure"); return item; }, - name: "partial"); + name: "partial", + config: new MapConfig { CompletionConfig = CompletionConfig.AllCompleted() }); var errorSummary = string.Join("|", batch.GetErrors().Select(e => $"{e.GetType().Name}:{e.Message}")); return new BatchResult @@ -777,7 +793,7 @@ public static async Task FirstSuccessfulAsync(BatchRequest input, I return index; }, name: "race", - config: new MapConfig { CompletionConfig = CompletionConfig.FirstSuccessful() }); + config: new MapConfig { CompletionConfig = CompletionConfig.FirstSuccessful() }); var winner = batch.Succeeded.FirstOrDefault(); return new BatchResult @@ -791,7 +807,11 @@ public static async Task FirstSuccessfulAsync(BatchRequest input, I }; } - /// Two items throw with tolerance=1, so the map exceeds tolerance and throws MapException. + /// + /// Two items throw with tolerance=1, so the map resolves with + /// . The operation does + /// NOT throw (JS parity) — the workflow inspects the result and reports it. + /// public static async Task FailureToleranceAsync(BatchRequest input, IDurableContext context) { var items = new[] { "ok1", "bad1", "ok2", "bad2", "ok3" }; @@ -805,9 +825,15 @@ public static async Task FailureToleranceAsync(BatchRequest input, return item; }, name: "tolerance", - config: new MapConfig { CompletionConfig = new CompletionConfig { ToleratedFailureCount = 1 } }); + config: new MapConfig { CompletionConfig = new CompletionConfig { ToleratedFailureCount = 1 } }); - return new BatchResult { Status = "should_not_reach", SuccessCount = batch.SuccessCount }; + return new BatchResult + { + Status = "completed", + SuccessCount = batch.SuccessCount, + FailureCount = batch.FailureCount, + CompletionReason = batch.CompletionReason.ToString(), + }; } /// Six items, MaxConcurrency=2; each waits then timestamps. All must succeed. @@ -822,7 +848,7 @@ public static async Task MaxConcurrencyAsync(BatchRequest input, ID return DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); }, name: "throttled", - config: new MapConfig { MaxConcurrency = 2, CompletionConfig = CompletionConfig.AllCompleted() }); + config: new MapConfig { MaxConcurrency = 2, CompletionConfig = CompletionConfig.AllCompleted() }); return new BatchResult { diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/PortableScenariosLocalTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/PortableScenariosLocalTests.cs index 8c9d7ff9b..9a0947f1c 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/PortableScenariosLocalTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/PortableScenariosLocalTests.cs @@ -180,7 +180,7 @@ public async Task Parallel_FirstSuccessful_ShortCircuits() } [Fact] - public async Task Parallel_FailureTolerance_Exceeded_Fails() + public async Task Parallel_FailureTolerance_Exceeded_ResolvesWithReason() { await using var runner = new DurableTestRunner(ParallelWorkflows.FailureToleranceAsync); await PortableScenarios.ParallelFailureToleranceAsync(runner); @@ -215,7 +215,7 @@ public async Task Map_FirstSuccessful_ShortCircuits() } [Fact] - public async Task Map_FailureTolerance_Exceeded_Fails() + public async Task Map_FailureTolerance_Exceeded_ResolvesWithReason() { await using var runner = new DurableTestRunner(MapWorkflows.FailureToleranceAsync); await PortableScenarios.MapFailureToleranceAsync(runner); diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs index 5eca6bb45..0c3649aa9 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs @@ -147,7 +147,7 @@ public async Task MapAsync_ItemNamer_PropagatesNameToCheckpointAndItem() new[] { "order-1", "order-2" }, async (ctx, item, index, all, _) => { await Task.Yield(); return item.Length; }, name: "process_orders", - config: new MapConfig { ItemNamer = (item, index) => $"Order-{item}" }); + config: new MapConfig { ItemNamer = (item, index) => $"Order-{item}" }); Assert.Equal("Order-order-1", result.All[0].Name); Assert.Equal("Order-order-2", result.All[1].Name); @@ -182,15 +182,17 @@ public async Task MapAsync_EmptyCollection_ReturnsEmptyResultWithAllCompleted() } // ────────────────────────────────────────────────────────────────────── - // CompletionConfig — Map's permissive default vs fail-fast opt-in + // CompletionConfig — fail-fast default (JS parity); operations never throw // ────────────────────────────────────────────────────────────────────── [Fact] - public async Task MapAsync_AllCompletedDefault_PartialFailureDoesNotThrow() + public async Task MapAsync_DefaultFailFast_PartialFailureResolvesFailureTolerance() { - // Map's default CompletionConfig is AllCompleted() (permissive), unlike - // Parallel's AllSuccessful(). A single item failure is captured rather - // than thrown. + // Map's default CompletionConfig is AllSuccessful() (fail-fast), matching + // Parallel and the JS/Python SDKs. A single item failure resolves the map + // with FailureToleranceExceeded, but the map NEVER throws — the failure is + // captured on the result. With unlimited concurrency all three items are + // dispatched before any completes, so two still succeed. var (context, _, _, _) = CreateContext(); var result = await context.MapAsync( @@ -205,7 +207,7 @@ public async Task MapAsync_AllCompletedDefault_PartialFailureDoesNotThrow() Assert.True(result.HasFailure); Assert.Equal(2, result.SuccessCount); Assert.Equal(1, result.FailureCount); - Assert.Equal(CompletionReason.AllCompleted, result.CompletionReason); + Assert.Equal(CompletionReason.FailureToleranceExceeded, result.CompletionReason); Assert.Equal(new[] { 1, 3 }, result.GetResults()); var errors = result.GetErrors(); @@ -214,33 +216,34 @@ public async Task MapAsync_AllCompletedDefault_PartialFailureDoesNotThrow() } [Fact] - public async Task MapAsync_AllSuccessfulOptIn_OneFailureThrowsMapException() + public async Task MapAsync_AllCompletedOptIn_PartialFailureIsTolerated() { + // AllCompleted() opts out of fail-fast: every item runs and the batch + // resolves AllCompleted despite the failure. Still no throw. var (context, _, _, _) = CreateContext(); - var ex = await Assert.ThrowsAsync(() => - context.MapAsync( - new[] { 1, 2, 3 }, - async (ctx, item, index, all, _) => - { - await Task.Yield(); - if (item == 2) throw new InvalidOperationException("item boom"); - return item; - }, - config: new MapConfig { CompletionConfig = CompletionConfig.AllSuccessful() })); + var result = await context.MapAsync( + new[] { 1, 2, 3 }, + async (ctx, item, index, all, _) => + { + await Task.Yield(); + if (item == 2) throw new InvalidOperationException("oops"); + return item; + }, + config: new MapConfig { CompletionConfig = CompletionConfig.AllCompleted() }); - Assert.Equal(CompletionReason.FailureToleranceExceeded, ex.CompletionReason); - Assert.NotNull(ex.Result); - var typed = Assert.IsAssignableFrom>(ex.Result); - Assert.Equal(1, typed.FailureCount); - Assert.Equal(2, typed.SuccessCount); + Assert.True(result.HasFailure); + Assert.Equal(2, result.SuccessCount); + Assert.Equal(1, result.FailureCount); + Assert.Equal(CompletionReason.AllCompleted, result.CompletionReason); + Assert.Equal(new[] { 1, 3 }, result.GetResults()); } [Fact] - public async Task MapAsync_ThrowIfError_ThrowsUnderPermissiveDefault() + public async Task MapAsync_ThrowIfError_ThrowsAfterFailure() { - // The permissive default does not auto-throw; ThrowIfError is the - // explicit strict-success check. + // The operation never auto-throws; ThrowIfError is the explicit + // strict-success check the caller opts into. var (context, _, _, _) = CreateContext(); var result = await context.MapAsync( @@ -258,25 +261,26 @@ public async Task MapAsync_ThrowIfError_ThrowsUnderPermissiveDefault() } [Fact] - public async Task MapAsync_ToleratedFailureCount_ExceededThrows() + public async Task MapAsync_ToleratedFailureCount_ExceededResolvesFailureTolerance() { var (context, _, _, _) = CreateContext(); - var ex = await Assert.ThrowsAsync(() => - context.MapAsync( - new[] { 1, 2, 3 }, - async (ctx, item, index, all, _) => - { - await Task.Yield(); - if (item != 3) throw new InvalidOperationException($"fail-{item}"); - return item; - }, - config: new MapConfig - { - CompletionConfig = new CompletionConfig { ToleratedFailureCount = 1 } - })); + var result = await context.MapAsync( + new[] { 1, 2, 3 }, + async (ctx, item, index, all, _) => + { + await Task.Yield(); + if (item != 3) throw new InvalidOperationException($"fail-{item}"); + return item; + }, + config: new MapConfig + { + CompletionConfig = new CompletionConfig { ToleratedFailureCount = 1 } + }); - Assert.Equal(CompletionReason.FailureToleranceExceeded, ex.CompletionReason); + Assert.Equal(CompletionReason.FailureToleranceExceeded, result.CompletionReason); + Assert.True(result.HasFailure); + Assert.Equal(2, result.FailureCount); } // ────────────────────────────────────────────────────────────────────── @@ -294,7 +298,7 @@ public async Task MapAsync_FirstSuccessful_ResolvesAfterFirstSuccess() var result = await context.MapAsync( new[] { 1, 2, 3 }, async (ctx, item, index, all, _) => { await Task.Yield(); return item; }, - config: new MapConfig + config: new MapConfig { MaxConcurrency = 1, CompletionConfig = CompletionConfig.FirstSuccessful() @@ -337,7 +341,7 @@ public async Task MapAsync_MaxConcurrency_LimitsInFlight() lock (lockObj) inFlight--; return item; }, - config: new MapConfig { MaxConcurrency = 2 }); + config: new MapConfig { MaxConcurrency = 2 }); Assert.Equal(5, result.SuccessCount); Assert.True(maxObserved <= 2, $"Observed concurrency {maxObserved} exceeded MaxConcurrency = 2"); @@ -353,7 +357,7 @@ public async Task MapAsync_MaxConcurrencyAtLeastItemCount_RunsWithoutSemaphore() var result = await context.MapAsync( new[] { 1, 2, 3 }, async (ctx, item, index, all, _) => { await Task.Yield(); return item; }, - config: new MapConfig { MaxConcurrency = 10 }); + config: new MapConfig { MaxConcurrency = 10 }); Assert.Equal(3, result.SuccessCount); Assert.Equal(new[] { 1, 2, 3 }, result.GetResults()); @@ -362,7 +366,7 @@ public async Task MapAsync_MaxConcurrencyAtLeastItemCount_RunsWithoutSemaphore() [Fact] public void MapConfig_MaxConcurrency_OutOfRange_Throws() { - var config = new MapConfig(); + var config = new MapConfig(); Assert.Throws(() => config.MaxConcurrency = 0); Assert.Throws(() => config.MaxConcurrency = -1); config.MaxConcurrency = 1; @@ -370,12 +374,12 @@ public void MapConfig_MaxConcurrency_OutOfRange_Throws() } [Fact] - public void MapConfig_DefaultCompletionConfig_IsAllCompleted() + public void MapConfig_DefaultCompletionConfig_IsAllSuccessful() { - // Guards the intentional divergence from ParallelConfig (AllSuccessful). - var config = new MapConfig(); - // AllCompleted() == empty CompletionConfig (no failure thresholds). - Assert.Null(config.CompletionConfig.ToleratedFailureCount); + // Map now defaults to fail-fast (AllSuccessful), matching ParallelConfig + // and the JS/Python SDKs. AllSuccessful() sets ToleratedFailureCount = 0. + var config = new MapConfig(); + Assert.Equal(0, config.CompletionConfig.ToleratedFailureCount); Assert.Null(config.CompletionConfig.MinSuccessful); Assert.Null(config.CompletionConfig.ToleratedFailurePercentage); } @@ -393,7 +397,7 @@ public async Task MapAsync_NestingTypeFlat_SuppressesPerItemContextOps() new[] { 1, 2, 3 }, async (ctx, item, index, all, _) => { await Task.Yield(); return item * 10; }, name: "doubler", - config: new MapConfig { NestingType = NestingType.Flat }); + config: new MapConfig { NestingType = NestingType.Flat }); Assert.Equal(new[] { 10, 20, 30 }, result.GetResults()); Assert.Equal(CompletionReason.AllCompleted, result.CompletionReason); @@ -420,7 +424,7 @@ await context.MapAsync( async (ctx, item, index, all, _) => await ctx.StepAsync(async (_, _) => { await Task.Yield(); return item * 10; }), name: "doubler", - config: new MapConfig { NestingType = NestingType.Flat }); + config: new MapConfig { NestingType = NestingType.Flat }); await recorder.Batcher.DrainAsync(); @@ -475,7 +479,7 @@ public async Task MapAsync_NestingTypeFlat_ReplaySucceeded_RebuildsFromInlinePay new[] { 1, 2 }, async (ctx, item, index, all, _) => { executed = true; await Task.Yield(); return item * 999; }, name: "doubler", - config: new MapConfig { NestingType = NestingType.Flat }); + config: new MapConfig { NestingType = NestingType.Flat }); Assert.False(executed); Assert.Equal(new[] { 10, 20 }, result.GetResults()); @@ -646,7 +650,7 @@ public async Task MapAsync_ReplayMixedStatus_PreservesStartedShortCircuited() } [Fact] - public async Task MapAsync_ReplayFailed_RebuildsResultAndThrows() + public async Task MapAsync_ReplayFailed_RebuildsResultWithoutThrowing() { var parentOpId = IdAt(1); var i0 = ChildIdAt(parentOpId, 1); @@ -685,15 +689,16 @@ public async Task MapAsync_ReplayFailed_RebuildsResultAndThrows() } }); - var ex = await Assert.ThrowsAsync(() => - context.MapAsync( - new[] { 1 }, - async (ctx, item, index, all, _) => { await Task.Yield(); return 999; }, - name: "m")); + var result = await context.MapAsync( + new[] { 1 }, + async (ctx, item, index, all, _) => { await Task.Yield(); return 999; }, + name: "m"); - Assert.Equal(CompletionReason.FailureToleranceExceeded, ex.CompletionReason); - var typed = Assert.IsAssignableFrom>(ex.Result); - Assert.Equal(1, typed.FailureCount); + // Replay reconstructs the frozen FailureToleranceExceeded result and + // returns it — the operation never throws (JS parity). + Assert.Equal(CompletionReason.FailureToleranceExceeded, result.CompletionReason); + Assert.True(result.HasFailure); + Assert.Equal(1, result.FailureCount); } [Fact] @@ -741,7 +746,7 @@ await Assert.ThrowsAsync(() => async (ctx, item, index, all, _) => { await Task.Yield(); return 999; }, name: "m", // Namer now yields "renamed" instead of the checkpointed "alpha". - config: new MapConfig { ItemNamer = (item, index) => "renamed" })); + config: new MapConfig { ItemNamer = (item, index) => "renamed" })); } // ────────────────────────────────────────────────────────────────────── diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/ParallelOperationTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/ParallelOperationTests.cs index 750750f0d..c6eda2b4a 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/ParallelOperationTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/ParallelOperationTests.cs @@ -180,7 +180,7 @@ public async Task ParallelAsync_EmptyBranches_ReturnsEmptyResultWithAllCompleted // ────────────────────────────────────────────────────────────────────── [Fact] - public async Task ParallelAsync_AllSuccessfulDefault_OneFailureThrowsParallelException() + public async Task ParallelAsync_AllSuccessfulDefault_OneFailureResolvesFailureTolerance() { var (context, _, _, _) = CreateContext(); @@ -194,23 +194,23 @@ public async Task ParallelAsync_AllSuccessfulDefault_OneFailureThrowsParallelExc var branch0Done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var branch2Done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - var ex = await Assert.ThrowsAsync(() => - context.ParallelAsync(new Func>[] + // The default is fail-fast, but the operation NEVER throws (JS parity) — + // the failure surfaces on the returned result. + var result = await context.ParallelAsync(new Func>[] + { + async (_, _) => { await Task.Yield(); branch0Done.SetResult(); return 1; }, + async (_, _) => { - async (_, _) => { await Task.Yield(); branch0Done.SetResult(); return 1; }, - async (_, _) => - { - await Task.WhenAll(branch0Done.Task, branch2Done.Task); - throw new InvalidOperationException("branch boom"); - }, - async (_, _) => { await Task.Yield(); branch2Done.SetResult(); return 3; }, - })); - - Assert.Equal(CompletionReason.FailureToleranceExceeded, ex.CompletionReason); - Assert.NotNull(ex.Result); - var typed = Assert.IsAssignableFrom>(ex.Result); - Assert.Equal(1, typed.FailureCount); - Assert.Equal(2, typed.SuccessCount); + await Task.WhenAll(branch0Done.Task, branch2Done.Task); + throw new InvalidOperationException("branch boom"); + }, + async (_, _) => { await Task.Yield(); branch2Done.SetResult(); return 3; }, + }); + + Assert.Equal(CompletionReason.FailureToleranceExceeded, result.CompletionReason); + Assert.True(result.HasFailure); + Assert.Equal(1, result.FailureCount); + Assert.Equal(2, result.SuccessCount); } [Fact] @@ -264,47 +264,47 @@ public async Task ParallelAsync_ToleratedFailureCount_AllowsUpToThreshold() } [Fact] - public async Task ParallelAsync_ToleratedFailureCount_ExceededThrows() + public async Task ParallelAsync_ToleratedFailureCount_ExceededResolvesFailureTolerance() { var (context, _, _, _) = CreateContext(); - var ex = await Assert.ThrowsAsync(() => - context.ParallelAsync( - new Func>[] - { - async (_, _) => { await Task.Yield(); throw new InvalidOperationException("fail-1"); }, - async (_, _) => { await Task.Yield(); throw new InvalidOperationException("fail-2"); }, - async (_, _) => { await Task.Yield(); return 3; }, - }, - config: new ParallelConfig - { - CompletionConfig = new CompletionConfig { ToleratedFailureCount = 1 } - })); + var result = await context.ParallelAsync( + new Func>[] + { + async (_, _) => { await Task.Yield(); throw new InvalidOperationException("fail-1"); }, + async (_, _) => { await Task.Yield(); throw new InvalidOperationException("fail-2"); }, + async (_, _) => { await Task.Yield(); return 3; }, + }, + config: new ParallelConfig + { + CompletionConfig = new CompletionConfig { ToleratedFailureCount = 1 } + }); - Assert.Equal(CompletionReason.FailureToleranceExceeded, ex.CompletionReason); + Assert.Equal(CompletionReason.FailureToleranceExceeded, result.CompletionReason); + Assert.True(result.HasFailure); } [Fact] - public async Task ParallelAsync_ToleratedFailurePercentage_ExceededThrows() + public async Task ParallelAsync_ToleratedFailurePercentage_ExceededResolvesFailureTolerance() { var (context, _, _, _) = CreateContext(); // 4 branches, 3 fail (75%) > 0.5 (50%) → exceeded. - var ex = await Assert.ThrowsAsync(() => - context.ParallelAsync( - new Func>[] - { - async (_, _) => { await Task.Yield(); throw new InvalidOperationException("f1"); }, - async (_, _) => { await Task.Yield(); throw new InvalidOperationException("f2"); }, - async (_, _) => { await Task.Yield(); throw new InvalidOperationException("f3"); }, - async (_, _) => { await Task.Yield(); return 4; }, - }, - config: new ParallelConfig - { - CompletionConfig = new CompletionConfig { ToleratedFailurePercentage = 0.5 } - })); + var result = await context.ParallelAsync( + new Func>[] + { + async (_, _) => { await Task.Yield(); throw new InvalidOperationException("f1"); }, + async (_, _) => { await Task.Yield(); throw new InvalidOperationException("f2"); }, + async (_, _) => { await Task.Yield(); throw new InvalidOperationException("f3"); }, + async (_, _) => { await Task.Yield(); return 4; }, + }, + config: new ParallelConfig + { + CompletionConfig = new CompletionConfig { ToleratedFailurePercentage = 0.5 } + }); - Assert.Equal(CompletionReason.FailureToleranceExceeded, ex.CompletionReason); + Assert.Equal(CompletionReason.FailureToleranceExceeded, result.CompletionReason); + Assert.True(result.HasFailure); } [Fact] @@ -946,7 +946,7 @@ public async Task ParallelAsync_NestingTypeFlat_ReplaySucceeded_RebuildsFromInli } [Fact] - public async Task ParallelAsync_NestingTypeFlat_ReplayFailed_ThrowsWithInlineError() + public async Task ParallelAsync_NestingTypeFlat_ReplayFailed_ResolvesWithInlineError() { var parentOpId = IdAt(1); @@ -976,20 +976,20 @@ public async Task ParallelAsync_NestingTypeFlat_ReplayFailed_ThrowsWithInlineErr } }); - var ex = await Assert.ThrowsAsync(() => - context.ParallelAsync( - new Func>[] - { - async (_, _) => { await Task.Yield(); return 1; }, - async (_, _) => { await Task.Yield(); return 2; }, - }, - name: "fanout", - config: new ParallelConfig { NestingType = NestingType.Flat })); + var result = await context.ParallelAsync( + new Func>[] + { + async (_, _) => { await Task.Yield(); return 1; }, + async (_, _) => { await Task.Yield(); return 2; }, + }, + name: "fanout", + config: new ParallelConfig { NestingType = NestingType.Flat }); - Assert.Equal(CompletionReason.FailureToleranceExceeded, ex.CompletionReason); - var typed = (IBatchResult)ex.Result!; - Assert.Equal(1, typed.FailureCount); - Assert.Contains("flat branch 0 failed", typed.GetErrors()[0].Message); + // Replay reconstructs the frozen FailureToleranceExceeded result from the + // inline payload and returns it — no throw (JS parity). + Assert.Equal(CompletionReason.FailureToleranceExceeded, result.CompletionReason); + Assert.Equal(1, result.FailureCount); + Assert.Contains("flat branch 0 failed", result.GetErrors()[0].Message); } // ────────────────────────────────────────────────────────────────────── @@ -1062,7 +1062,7 @@ public async Task ParallelAsync_ReplaySucceeded_RebuildsResultFromCheckpoints() } [Fact] - public async Task ParallelAsync_ReplayFailed_ThrowsParallelException() + public async Task ParallelAsync_ReplayFailed_ResolvesFailureTolerance() { var parentOpId = IdAt(1); var b0 = ChildIdAt(parentOpId, 1); @@ -1123,21 +1123,19 @@ public async Task ParallelAsync_ReplayFailed_ThrowsParallelException() } }); - var ex = await Assert.ThrowsAsync(() => - context.ParallelAsync( - new Func>[] - { - async (_, _) => { await Task.Yield(); return 1; }, - async (_, _) => { await Task.Yield(); return 2; }, - }, - name: "fanout")); - - Assert.Equal(CompletionReason.FailureToleranceExceeded, ex.CompletionReason); - Assert.NotNull(ex.Result); + var result = await context.ParallelAsync( + new Func>[] + { + async (_, _) => { await Task.Yield(); return 1; }, + async (_, _) => { await Task.Yield(); return 2; }, + }, + name: "fanout"); - var typed = (IBatchResult)ex.Result!; - Assert.Equal(2, typed.FailureCount); - Assert.Contains("branch 0 failed", typed.GetErrors()[0].Message); + // Replay reconstructs the frozen FailureToleranceExceeded result and + // returns it — no throw (JS parity). + Assert.Equal(CompletionReason.FailureToleranceExceeded, result.CompletionReason); + Assert.Equal(2, result.FailureCount); + Assert.Contains("branch 0 failed", result.GetErrors()[0].Message); } [Fact] @@ -1489,14 +1487,14 @@ await context.ParallelAsync( } [Fact] - public async Task ParallelAsync_FirstSuccessful_AllFail_AggregatesAsParallelException() + public async Task ParallelAsync_FirstSuccessful_AllFail_ResolvesAllCompleted() { - // FirstSuccessful() aliases MinSuccessful=1 with no explicit failure - // tolerance. When every branch fails, MinSuccessful is unreachable - // AND there is no failure-tolerance threshold, so the run completes - // as AllCompleted with HasFailure=true. Calling ThrowIfError surfaces - // the first failure; without explicit failure tolerance the parallel - // does NOT throw on its own (matches Python). + // FirstSuccessful() aliases MinSuccessful=1. Setting any completion + // criterion opts out of the fail-fast default, so there is no + // failure-tolerance threshold here. When every branch fails, + // MinSuccessful is unreachable, so the run completes as AllCompleted with + // HasFailure=true. The parallel never throws on its own (JS parity); + // ThrowIfError surfaces the first failure only when the caller asks. var (context, _, _, _) = CreateContext(); var result = await context.ParallelAsync(