Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions .autover/changes/durable-concurrent-js-parity.json
Original file line number Diff line number Diff line change
@@ -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<TItem>) so ItemNamer is typed Func<TItem, int, string> instead of Func<object, int, string>. Preview."
]
}
]
}
Original file line number Diff line number Diff line change
Expand Up @@ -90,20 +90,29 @@ public double? ToleratedFailurePercentage
}

/// <summary>
/// All items must succeed. Equivalent to
/// <see cref="ToleratedFailureCount"/> = 0. The default for
/// <see cref="ParallelConfig.CompletionConfig"/>.
/// All items must succeed — any single failure resolves the batch with
/// <see cref="CompletionReason.FailureToleranceExceeded"/>. Equivalent to
/// <see cref="ToleratedFailureCount"/> = 0, and to a default (empty)
/// <see cref="CompletionConfig"/>. The default for both
/// <see cref="ParallelConfig.CompletionConfig"/> and
/// <see cref="MapConfig{TItem}.CompletionConfig"/>, matching the JS/Python SDKs.
/// </summary>
public static CompletionConfig AllSuccessful() => new() { ToleratedFailureCount = 0 };

/// <summary>
/// Run every branch regardless of failures; surface failures per-item via
/// <see cref="IBatchResult{T}.Failed"/>. Resolution does not auto-throw —
/// the caller can inspect the result and call
/// <see cref="IBatchResult{T}.ThrowIfError"/> if they want strict-success
/// behavior.
/// <see cref="IBatchResult{T}.Failed"/>. Never resolves with
/// <see cref="CompletionReason.FailureToleranceExceeded"/>. The caller
/// inspects the result and can call <see cref="IBatchResult{T}.ThrowIfError"/>
/// if it wants strict-success behavior.
/// </summary>
public static CompletionConfig AllCompleted() => new();
/// <remarks>
/// Sets <see cref="ToleratedFailureCount"/> to <see cref="int.MaxValue"/> so it
/// stays lenient. An <em>empty</em> <see cref="CompletionConfig"/> is fail-fast
/// (equivalent to <see cref="AllSuccessful"/>), so leniency must be opted into
/// explicitly.
/// </remarks>
public static CompletionConfig AllCompleted() => new() { ToleratedFailureCount = int.MaxValue };

/// <summary>
/// Resolve once at least one branch has succeeded. Branches that were not
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,11 @@ public enum CompletionReason

/// <summary>
/// <see cref="CompletionConfig.ToleratedFailureCount"/> or
/// <see cref="CompletionConfig.ToleratedFailurePercentage"/> was exceeded.
/// The batch is considered failed and surfaces a
/// <see cref="ParallelException"/> when awaited.
/// <see cref="CompletionConfig.ToleratedFailurePercentage"/> was exceeded (a
/// default/empty <see cref="CompletionConfig"/> is fail-fast, so any failure
/// trips this). The batch is considered failed: <see cref="IBatchResult.HasFailure"/>
/// is <c>true</c> and <see cref="IBatchResult{T}.ThrowIfError"/> surfaces the
/// first item error. The operation itself does not throw.
/// </summary>
FailureToleranceExceeded
}
Original file line number Diff line number Diff line change
Expand Up @@ -258,21 +258,21 @@ public Task<IBatchResult<TResult>> MapAsync<TItem, TResult>(
IReadOnlyList<TItem> items,
Func<IDurableContext, TItem, int, IReadOnlyList<TItem>, CancellationToken, Task<TResult>> func,
string? name = null,
MapConfig? config = null,
MapConfig<TItem>? config = null,
CancellationToken cancellationToken = default)
=> RunMap(items, func, name, config, cancellationToken);

private Task<IBatchResult<TResult>> RunMap<TItem, TResult>(
IReadOnlyList<TItem> items,
Func<IDurableContext, TItem, int, IReadOnlyList<TItem>, CancellationToken, Task<TResult>> func,
string? name,
MapConfig? config,
MapConfig<TItem>? 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<TItem>();

var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -98,69 +98,3 @@ public ChildContextException(string message) : base(message) { }
/// <summary>Creates a <see cref="ChildContextException"/> wrapping an inner exception.</summary>
public ChildContextException(string message, Exception innerException) : base(message, innerException) { }
}

/// <summary>
/// Thrown when a parallel operation resolves with
/// <see cref="CompletionReason.FailureToleranceExceeded"/>. The aggregate
/// <see cref="IBatchResult"/> is preserved on <see cref="Result"/> so callers
/// can inspect per-branch outcomes.
/// </summary>
/// <remarks>
/// This is the base type for parallel failures. Subclasses may be added in
/// future releases (for example, a dedicated
/// <c>ParallelFailureToleranceExceededException</c>); catching
/// <see cref="ParallelException"/> remains forward-compatible.
/// </remarks>
public class ParallelException : DurableExecutionException
{
/// <summary>
/// The aggregate result of the parallel operation. Type-erased — cast to
/// <c>IBatchResult&lt;T&gt;</c> if the per-branch result type is known.
/// </summary>
public IBatchResult? Result { get; init; }

/// <summary>
/// Why the parallel operation resolved.
/// </summary>
public CompletionReason CompletionReason { get; init; }

/// <summary>Creates an empty <see cref="ParallelException"/>.</summary>
public ParallelException() { }
/// <summary>Creates a <see cref="ParallelException"/> with the given message.</summary>
public ParallelException(string message) : base(message) { }
/// <summary>Creates a <see cref="ParallelException"/> wrapping an inner exception.</summary>
public ParallelException(string message, Exception innerException) : base(message, innerException) { }
}

/// <summary>
/// Thrown when a map operation resolves with
/// <see cref="CompletionReason.FailureToleranceExceeded"/>. The aggregate
/// <see cref="IBatchResult"/> is preserved on <see cref="Result"/> so callers
/// can inspect per-item outcomes.
/// </summary>
/// <remarks>
/// This is the base type for map failures. Subclasses may be added in future
/// releases; catching <see cref="MapException"/> remains forward-compatible.
/// A dedicated type (rather than reusing <see cref="ParallelException"/>) lets
/// callers pattern-match which concurrent operation failed.
/// </remarks>
public class MapException : DurableExecutionException
{
/// <summary>
/// The aggregate result of the map operation. Type-erased — cast to
/// <c>IBatchResult&lt;T&gt;</c> if the per-item result type is known.
/// </summary>
public IBatchResult? Result { get; init; }

/// <summary>
/// Why the map operation resolved.
/// </summary>
public CompletionReason CompletionReason { get; init; }

/// <summary>Creates an empty <see cref="MapException"/>.</summary>
public MapException() { }
/// <summary>Creates a <see cref="MapException"/> with the given message.</summary>
public MapException(string message) : base(message) { }
/// <summary>Creates a <see cref="MapException"/> wrapping an inner exception.</summary>
public MapException(string message, Exception innerException) : base(message, innerException) { }
}
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,9 @@
namespace Amazon.Lambda.DurableExecution;

/// <summary>
/// Non-generic marker for <see cref="IBatchResult{T}"/>. Used by
/// <see cref="ParallelException.Result"/> so callers can hold a reference to
/// the aggregate result without knowing the per-branch type at compile time.
/// Non-generic marker for <see cref="IBatchResult{T}"/>. Lets callers hold a
/// reference to the aggregate result, or read the completion bookkeeping, without
/// knowing the per-branch type at compile time.
/// </summary>
public interface IBatchResult
{
Expand Down
28 changes: 17 additions & 11 deletions Libraries/src/Amazon.Lambda.DurableExecution/IDurableContext.cs
Original file line number Diff line number Diff line change
Expand Up @@ -377,9 +377,11 @@ Task<TState> WaitForConditionAsync<TState>(
/// <remarks>
/// On per-branch failure (a branch's user function throws), the failure is
/// captured on the corresponding <see cref="IBatchItem{T}"/> instead of
/// aborting the parallel. The parallel only throws
/// <see cref="ParallelException"/> when <see cref="CompletionConfig"/>
/// criteria are violated. Use
/// aborting the parallel. The parallel NEVER throws on failure — it always
/// returns an <see cref="IBatchResult{T}"/>. By default
/// (<see cref="CompletionConfig.AllSuccessful"/>, fail-fast) any branch failure
/// resolves it with <see cref="CompletionReason.FailureToleranceExceeded"/>;
/// failures surface via <see cref="IBatchResult.HasFailure"/>. Use
/// <see cref="IBatchResult{T}.ThrowIfError"/> for explicit strict-success
/// semantics. Per-branch results are serialized to checkpoints using the
/// <see cref="ILambdaSerializer"/> registered on
Expand Down Expand Up @@ -449,22 +451,26 @@ Task<IBatchResult<T>> ParallelAsync<T>(
/// Process a collection of items concurrently, running <paramref name="func"/>
/// once per item. Each item runs inside its own child context; per-item
/// results are aggregated into an <see cref="IBatchResult{TResult}"/>. Items
/// are dispatched up to <see cref="MapConfig.MaxConcurrency"/>; the aggregate
/// resolves according to <see cref="MapConfig.CompletionConfig"/>.
/// are dispatched up to <see cref="MapConfig{TItem}.MaxConcurrency"/>; the aggregate
/// resolves according to <see cref="MapConfig{TItem}.CompletionConfig"/>.
/// </summary>
/// <remarks>
/// The per-item function receives the durable context, the item, its
/// zero-based index, the full source list, and a
/// <see cref="CancellationToken"/> linking the
/// caller-supplied token with the SDK's workflow-shutdown signal (it is also
/// tripped cooperatively when a sibling item satisfies the
/// <see cref="MapConfig.CompletionConfig"/> and the map short-circuits). On
/// <see cref="MapConfig{TItem}.CompletionConfig"/> and the map short-circuits). On
/// per-item failure (the user function throws), the failure is captured on
/// the corresponding <see cref="IBatchItem{TResult}"/> instead of aborting
/// the map. By default (<see cref="CompletionConfig.AllCompleted"/>) every
/// item runs and failures surface via <see cref="IBatchResult{TResult}.Failed"/>;
/// the map throws <see cref="MapException"/> only when
/// <see cref="CompletionConfig"/> criteria are violated. Use
/// the map. The map NEVER throws on failure — it always returns an
/// <see cref="IBatchResult{TResult}"/>. By default
/// (<see cref="CompletionConfig.AllSuccessful"/>, fail-fast) any item failure
/// resolves the map with
/// <see cref="CompletionReason.FailureToleranceExceeded"/>; use
/// <see cref="CompletionConfig.AllCompleted"/> to run every item regardless.
/// Failures surface via <see cref="IBatchResult{TResult}.Failed"/> /
/// <see cref="IBatchResult.HasFailure"/>; call
/// <see cref="IBatchResult{TResult}.ThrowIfError"/> for explicit
/// strict-success semantics. Per-item results are serialized to checkpoints
/// using the <see cref="ILambdaSerializer"/> registered on
Expand All @@ -474,7 +480,7 @@ Task<IBatchResult<TResult>> MapAsync<TItem, TResult>(
IReadOnlyList<TItem> items,
Func<IDurableContext, TItem, int, IReadOnlyList<TItem>, CancellationToken, Task<TResult>> func,
string? name = null,
MapConfig? config = null,
MapConfig<TItem>? config = null,
CancellationToken cancellationToken = default);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

/// <summary>
Expand Down Expand Up @@ -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;

Expand Down
Loading
Loading