Skip to content

Commit fa13eef

Browse files
GarrettBeattyclaude
andcommitted
Add ParallelAsync for concurrent branch execution (DOTNET-8662)
Adds parallel branch execution to the .NET Durable Execution SDK. ParallelAsync runs N branches concurrently with configurable concurrency limits and completion policies, returning an IBatchResult<T> with per-branch status and error information. Public surface: - IDurableContext.ParallelAsync<T> (4 overloads: reflection x 2 for Func[] vs DurableBranch<T>[]; AOT-safe x 2 same) - DurableBranch<T> record (Name + Func) - ParallelConfig (MaxConcurrency, CompletionConfig, NestingType) - CompletionConfig with factories AllSuccessful() / FirstSuccessful() / AllCompleted(); ToleratedFailureCount / ToleratedFailurePercentage (validated 0.0-1.0) - IBatchResult<T> with All / Succeeded / Failed / Started accessors, GetResults, GetErrors, ThrowIfError, HasFailure, CompletionReason, count properties - IBatchItem<T> with Index, Name, Status, Result, Error - BatchItemStatus { Succeeded, Failed, Started } - CompletionReason { AllCompleted, MinSuccessfulReached, FailureToleranceExceeded } - NestingType (Nested default; Flat throws NotSupportedException - reserved) - ParallelException (carries IBatchResult; future-subclassable) Internal: - ParallelOperation<T> orchestrator dispatches branches with optional semaphore-bounded concurrency. Each branch runs as a ChildContextOperation<T> with deterministic ID via OperationIdGenerator.CreateChild. - Branch failures aggregated as IBatchItem<T> entries; orchestrator throws ParallelException only when CompletionConfig signals FailureToleranceExceeded. - Parent CONTEXT checkpoint records summary (CompletionReason + per-branch index/name/status); branch results live on per-branch CONTEXT checkpoints. - ExecutionState now thread-safe (lock around reads/writes of _operations, _visitedOperations, _isReplaying). Required for concurrent branch replay; affects all operations but no regressions. - ParallelOperation awaits Task.WhenAll(inFlight) before disposing the semaphore so cancellation/exception during dispatch lets in-flight branches settle cleanly. - Reuses OperationSubTypes.Parallel / OperationSubTypes.ParallelBranch from Wave 0. Adds 33 unit tests + 6 integration tests covering CompletionConfig matrix, MaxConcurrency, FirstSuccessful short-circuit, replay determinism, mixed-status replay, cancellation, and concurrency stress. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent d308c3b commit fa13eef

44 files changed

Lines changed: 3597 additions & 52 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.
Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
{
2+
"Projects": [
3+
{
4+
"Name": "Amazon.Lambda.DurableExecution",
5+
"Type": "Patch",
6+
"ChangelogMessages": [
7+
"Added IDurableContext.ParallelAsync for executing multiple branches concurrently with configurable concurrency limits and completion criteria. Introduces IBatchResult<T>, IBatchItem<T>, BatchItemStatus, CompletionReason, NestingType, DurableBranch<T>, ParallelConfig, CompletionConfig, and ParallelException."
8+
]
9+
}
10+
]
11+
}

Docs/durable-execution-design.md

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -548,7 +548,7 @@ For better observability, you can name individual branches (matching the JS SDK
548548
```csharp
549549
// Named branches for easier debugging and testing
550550
var results = await context.ParallelAsync(
551-
new NamedBranch<object>[]
551+
new DurableBranch<object>[]
552552
{
553553
new("fetch_user", async (ctx) => await ctx.StepAsync(async (step) => await FetchUserData(userId))),
554554
new("fetch_orders", async (ctx) => await ctx.StepAsync(async (step) => await FetchOrderHistory(userId))),
@@ -1416,6 +1416,13 @@ public class CompletionConfig
14161416
{
14171417
public int? MinSuccessful { get; set; }
14181418
public int? ToleratedFailureCount { get; set; }
1419+
/// <summary>
1420+
/// Maximum tolerated failure ratio, expressed as a value in the range
1421+
/// <c>0.0</c> to <c>1.0</c> (inclusive). For example, <c>0.25</c> means
1422+
/// "tolerate up to 25% failures; fail when the failure ratio strictly
1423+
/// exceeds 25%". <c>null</c> = no ratio-based threshold. Validated by the
1424+
/// setter; out-of-range values throw <see cref="ArgumentOutOfRangeException"/>.
1425+
/// </summary>
14191426
public double? ToleratedFailurePercentage { get; set; }
14201427

14211428
public static CompletionConfig AllSuccessful() => new() { ToleratedFailureCount = 0 };
Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
namespace Amazon.Lambda.DurableExecution;
2+
3+
/// <summary>
4+
/// Status of an individual item in a <see cref="IBatchResult{T}"/>.
5+
/// </summary>
6+
/// <remarks>
7+
/// Mirrors the wire-state of the per-branch checkpoint at the moment the batch
8+
/// resolved. Items that finished produce <see cref="Succeeded"/> or
9+
/// <see cref="Failed"/>; items still in flight when the batch's
10+
/// <see cref="CompletionConfig"/> short-circuits remain in <see cref="Started"/>.
11+
/// </remarks>
12+
public enum BatchItemStatus
13+
{
14+
/// <summary>
15+
/// The branch ran to completion and produced a result.
16+
/// </summary>
17+
Succeeded,
18+
19+
/// <summary>
20+
/// The branch ran to completion and threw.
21+
/// </summary>
22+
Failed,
23+
24+
/// <summary>
25+
/// The branch was still in flight when the batch's <see cref="CompletionConfig"/>
26+
/// resolved (e.g., <see cref="CompletionConfig.FirstSuccessful"/> returned
27+
/// before this branch finished).
28+
/// </summary>
29+
Started
30+
}
Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
namespace Amazon.Lambda.DurableExecution;
2+
3+
/// <summary>
4+
/// Why a batch operation (<see cref="IDurableContext.ParallelAsync{T}(IReadOnlyList{System.Func{IDurableContext, System.Threading.Tasks.Task{T}}}, string?, ParallelConfig?, System.Threading.CancellationToken)"/>
5+
/// or future Map) resolved.
6+
/// </summary>
7+
public enum CompletionReason
8+
{
9+
/// <summary>
10+
/// Every branch finished — no <see cref="CompletionConfig"/> short-circuit
11+
/// was triggered. Branches may be a mix of <see cref="BatchItemStatus.Succeeded"/>
12+
/// and <see cref="BatchItemStatus.Failed"/>.
13+
/// </summary>
14+
AllCompleted,
15+
16+
/// <summary>
17+
/// <see cref="CompletionConfig.MinSuccessful"/> branches succeeded; remaining
18+
/// branches were left in <see cref="BatchItemStatus.Started"/>.
19+
/// </summary>
20+
MinSuccessfulReached,
21+
22+
/// <summary>
23+
/// <see cref="CompletionConfig.ToleratedFailureCount"/> or
24+
/// <see cref="CompletionConfig.ToleratedFailurePercentage"/> was exceeded.
25+
/// The batch is considered failed and surfaces a
26+
/// <see cref="ParallelException"/> when awaited.
27+
/// </summary>
28+
FailureToleranceExceeded
29+
}
Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
namespace Amazon.Lambda.DurableExecution;
2+
3+
/// <summary>
4+
/// Defines completion criteria for parallel/map operations.
5+
/// </summary>
6+
/// <remarks>
7+
/// Construct via the static factories (<see cref="AllSuccessful"/>,
8+
/// <see cref="AllCompleted"/>, <see cref="FirstSuccessful"/>) or set the
9+
/// individual properties directly. Multiple criteria combine: the operation
10+
/// resolves as soon as any criterion is met (success short-circuit) or violated
11+
/// (failure short-circuit).
12+
/// </remarks>
13+
public sealed class CompletionConfig
14+
{
15+
private double? _toleratedFailurePercentage;
16+
17+
/// <summary>
18+
/// Minimum number of <see cref="BatchItemStatus.Succeeded"/> items required
19+
/// before the operation resolves successfully. <c>null</c> = no minimum.
20+
/// </summary>
21+
public int? MinSuccessful { get; set; }
22+
23+
/// <summary>
24+
/// Maximum tolerated <see cref="BatchItemStatus.Failed"/> count. When the
25+
/// failure count <i>strictly exceeds</i> this value, the operation resolves
26+
/// with <see cref="CompletionReason.FailureToleranceExceeded"/>.
27+
/// <c>null</c> = no count-based failure threshold.
28+
/// </summary>
29+
public int? ToleratedFailureCount { get; set; }
30+
31+
/// <summary>
32+
/// Maximum tolerated failure ratio, expressed as a value in the range
33+
/// <c>0.0</c> to <c>1.0</c> (inclusive). For example, <c>0.25</c> means
34+
/// "tolerate up to 25% failures; fail when the failure ratio strictly
35+
/// exceeds 25%". <c>null</c> = no ratio-based failure threshold.
36+
/// </summary>
37+
/// <exception cref="System.ArgumentOutOfRangeException">
38+
/// Thrown by the setter if the value is outside <c>[0.0, 1.0]</c>.
39+
/// </exception>
40+
public double? ToleratedFailurePercentage
41+
{
42+
get => _toleratedFailurePercentage;
43+
set
44+
{
45+
if (value is { } v && (v < 0.0 || v > 1.0))
46+
{
47+
throw new ArgumentOutOfRangeException(nameof(value), v,
48+
"ToleratedFailurePercentage must be a ratio in [0.0, 1.0].");
49+
}
50+
_toleratedFailurePercentage = value;
51+
}
52+
}
53+
54+
/// <summary>
55+
/// All items must succeed. Equivalent to
56+
/// <see cref="ToleratedFailureCount"/> = 0. The default for
57+
/// <see cref="ParallelConfig.CompletionConfig"/>.
58+
/// </summary>
59+
public static CompletionConfig AllSuccessful() => new() { ToleratedFailureCount = 0 };
60+
61+
/// <summary>
62+
/// Run every branch regardless of failures; surface failures per-item via
63+
/// <see cref="IBatchResult{T}.Failed"/>. Resolution does not auto-throw —
64+
/// the caller can inspect the result and call
65+
/// <see cref="IBatchResult{T}.ThrowIfError"/> if they want strict-success
66+
/// behavior.
67+
/// </summary>
68+
public static CompletionConfig AllCompleted() => new();
69+
70+
/// <summary>
71+
/// Resolve as soon as one branch succeeds. Remaining in-flight branches are
72+
/// reported as <see cref="BatchItemStatus.Started"/>.
73+
/// </summary>
74+
public static CompletionConfig FirstSuccessful() => new() { MinSuccessful = 1 };
75+
}
Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
namespace Amazon.Lambda.DurableExecution;
2+
3+
/// <summary>
4+
/// Configuration for
5+
/// <see cref="IDurableContext.ParallelAsync{T}(IReadOnlyList{System.Func{IDurableContext, System.Threading.Tasks.Task{T}}}, string?, ParallelConfig?, System.Threading.CancellationToken)"/>.
6+
/// </summary>
7+
/// <remarks>
8+
/// Per-branch and aggregate serializers are supplied via the AOT-safe
9+
/// <c>ParallelAsync</c> overloads that take an
10+
/// <see cref="ICheckpointSerializer{T}"/>; this config does not expose a
11+
/// serializer slot.
12+
/// </remarks>
13+
public sealed class ParallelConfig
14+
{
15+
private int? _maxConcurrency;
16+
17+
/// <summary>
18+
/// Maximum number of branches running concurrently. <c>null</c> (default) =
19+
/// unlimited. Must be at least 1 when set.
20+
/// </summary>
21+
/// <exception cref="System.ArgumentOutOfRangeException">
22+
/// Thrown by the setter if the value is less than or equal to 0.
23+
/// </exception>
24+
public int? MaxConcurrency
25+
{
26+
get => _maxConcurrency;
27+
set
28+
{
29+
if (value is { } v && v <= 0)
30+
{
31+
throw new ArgumentOutOfRangeException(nameof(value), v,
32+
"MaxConcurrency must be at least 1, or null for unlimited.");
33+
}
34+
_maxConcurrency = value;
35+
}
36+
}
37+
38+
/// <summary>
39+
/// When the parallel operation is considered complete. Defaults to
40+
/// <see cref="CompletionConfig.AllSuccessful"/> — any single branch failure
41+
/// surfaces as a <see cref="ParallelException"/> when the parallel result
42+
/// is awaited.
43+
/// </summary>
44+
public CompletionConfig CompletionConfig { get; set; } = CompletionConfig.AllSuccessful();
45+
46+
/// <summary>
47+
/// How branches are represented in the checkpoint graph. Defaults to
48+
/// <see cref="NestingType.Nested"/>.
49+
/// </summary>
50+
/// <remarks>
51+
/// <see cref="NestingType.Flat"/> is not yet supported in the .NET SDK and
52+
/// will throw <see cref="System.NotSupportedException"/> when the parallel
53+
/// operation is invoked.
54+
/// </remarks>
55+
public NestingType NestingType { get; set; } = NestingType.Nested;
56+
}
Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
namespace Amazon.Lambda.DurableExecution;
2+
3+
/// <summary>
4+
/// A named branch for
5+
/// <see cref="IDurableContext.ParallelAsync{T}(IReadOnlyList{DurableBranch{T}}, string?, ParallelConfig?, System.Threading.CancellationToken)"/>.
6+
/// Names appear in execution traces and on the wire <c>OperationUpdate.Name</c>
7+
/// field, and surface on <see cref="IBatchItem{T}.Name"/>.
8+
/// </summary>
9+
/// <typeparam name="T">The branch's result type.</typeparam>
10+
/// <param name="Name">Human-readable branch name. Required.</param>
11+
/// <param name="Func">The user function executed inside the branch's
12+
/// child context.</param>
13+
public sealed record DurableBranch<T>(string Name, Func<IDurableContext, Task<T>> Func);

Libraries/src/Amazon.Lambda.DurableExecution/DurableContext.cs

Lines changed: 100 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -149,19 +149,110 @@ private Task<T> RunChildContext<T>(
149149
{
150150
var operationId = _idGenerator.NextId();
151151

152-
// Capture this DurableContext's collaborators; the child shares state,
153-
// termination, batcher, ARN, and Lambda context — but uses a child
154-
// OperationIdGenerator so its operation IDs are deterministically
155-
// namespaced under the parent op ID.
156-
IDurableContext ChildFactory(string parentOpId) => new DurableContext(
157-
_state, _terminationManager, _idGenerator.CreateChild(parentOpId),
158-
_durableExecutionArn, LambdaContext, _batcher);
159-
160152
var op = new ChildContextOperation<T>(
161-
operationId, name, func, config, serializer, ChildFactory,
153+
operationId, name, func, config, serializer, MakeChildFactory(),
162154
_state, _terminationManager, _durableExecutionArn, _batcher);
163155
return op.ExecuteAsync(cancellationToken);
164156
}
157+
158+
[RequiresUnreferencedCode("Reflection-based JSON for T. Use the ICheckpointSerializer<T> overload for AOT/trimmed deployments.")]
159+
[RequiresDynamicCode("Reflection-based JSON for T. Use the ICheckpointSerializer<T> overload for AOT/trimmed deployments.")]
160+
public Task<IBatchResult<T>> ParallelAsync<T>(
161+
IReadOnlyList<Func<IDurableContext, Task<T>>> branches,
162+
string? name = null,
163+
ParallelConfig? config = null,
164+
CancellationToken cancellationToken = default)
165+
=> RunParallel(WrapToDurableBranches(branches), new ReflectionJsonCheckpointSerializer<T>(), name, config, cancellationToken);
166+
167+
[RequiresUnreferencedCode("Reflection-based JSON for T. Use the ICheckpointSerializer<T> overload for AOT/trimmed deployments.")]
168+
[RequiresDynamicCode("Reflection-based JSON for T. Use the ICheckpointSerializer<T> overload for AOT/trimmed deployments.")]
169+
public Task<IBatchResult<T>> ParallelAsync<T>(
170+
IReadOnlyList<DurableBranch<T>> branches,
171+
string? name = null,
172+
ParallelConfig? config = null,
173+
CancellationToken cancellationToken = default)
174+
=> RunParallel(branches, new ReflectionJsonCheckpointSerializer<T>(), name, config, cancellationToken);
175+
176+
public Task<IBatchResult<T>> ParallelAsync<T>(
177+
IReadOnlyList<Func<IDurableContext, Task<T>>> branches,
178+
ICheckpointSerializer<T> serializer,
179+
string? name = null,
180+
ParallelConfig? config = null,
181+
CancellationToken cancellationToken = default)
182+
=> RunParallel(WrapToDurableBranches(branches), serializer, name, config, cancellationToken);
183+
184+
public Task<IBatchResult<T>> ParallelAsync<T>(
185+
IReadOnlyList<DurableBranch<T>> branches,
186+
ICheckpointSerializer<T> serializer,
187+
string? name = null,
188+
ParallelConfig? config = null,
189+
CancellationToken cancellationToken = default)
190+
=> RunParallel(branches, serializer, name, config, cancellationToken);
191+
192+
private static IReadOnlyList<DurableBranch<T>> WrapToDurableBranches<T>(
193+
IReadOnlyList<Func<IDurableContext, Task<T>>> branches)
194+
{
195+
if (branches == null) throw new ArgumentNullException(nameof(branches));
196+
197+
var result = new DurableBranch<T>[branches.Count];
198+
for (var i = 0; i < branches.Count; i++)
199+
{
200+
var func = branches[i];
201+
if (func == null)
202+
throw new ArgumentException($"Branch at index {i} is null.", nameof(branches));
203+
// Default name is the index — surfaces in execution traces and on
204+
// IBatchItem<T>.Name. Users wanting custom names use the
205+
// DurableBranch<T> overload.
206+
result[i] = new DurableBranch<T>(i.ToString(System.Globalization.CultureInfo.InvariantCulture), func);
207+
}
208+
return result;
209+
}
210+
211+
private Task<IBatchResult<T>> RunParallel<T>(
212+
IReadOnlyList<DurableBranch<T>> branches,
213+
ICheckpointSerializer<T> serializer,
214+
string? name,
215+
ParallelConfig? config,
216+
CancellationToken cancellationToken)
217+
{
218+
if (branches == null) throw new ArgumentNullException(nameof(branches));
219+
for (var i = 0; i < branches.Count; i++)
220+
{
221+
if (branches[i] == null)
222+
throw new ArgumentException($"Branch at index {i} is null.", nameof(branches));
223+
if (branches[i].Func == null)
224+
throw new ArgumentException($"Branch at index {i} has a null Func.", nameof(branches));
225+
}
226+
227+
var effectiveConfig = config ?? new ParallelConfig();
228+
if (effectiveConfig.NestingType == NestingType.Flat)
229+
{
230+
throw new NotSupportedException(
231+
"NestingType.Flat is not yet supported in the .NET Durable Execution SDK. " +
232+
"Use NestingType.Nested (the default) for now.");
233+
}
234+
235+
var operationId = _idGenerator.NextId();
236+
var op = new Internal.ParallelOperation<T>(
237+
operationId, name, branches, effectiveConfig, serializer, MakeChildFactory(),
238+
_state, _terminationManager, _durableExecutionArn, _batcher);
239+
return op.ExecuteAsync(cancellationToken);
240+
}
241+
242+
/// <summary>
243+
/// Builds the factory used by <see cref="ChildContextOperation{T}"/> (and
244+
/// each <see cref="Internal.ParallelOperation{T}"/> branch) to construct
245+
/// the inner <see cref="IDurableContext"/>. The child shares state,
246+
/// termination, batcher, ARN, and Lambda context — but uses a child
247+
/// <see cref="OperationIdGenerator"/> so its operation IDs are
248+
/// deterministically namespaced under the parent op ID.
249+
/// </summary>
250+
private Func<string, IDurableContext> MakeChildFactory()
251+
{
252+
return parentOpId => new DurableContext(
253+
_state, _terminationManager, _idGenerator.CreateChild(parentOpId),
254+
_durableExecutionArn, LambdaContext, _batcher);
255+
}
165256
}
166257

167258
/// <summary>

Libraries/src/Amazon.Lambda.DurableExecution/Exceptions/DurableExecutionException.cs

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,3 +75,36 @@ public ChildContextException(string message) : base(message) { }
7575
/// <summary>Creates a <see cref="ChildContextException"/> wrapping an inner exception.</summary>
7676
public ChildContextException(string message, Exception innerException) : base(message, innerException) { }
7777
}
78+
79+
/// <summary>
80+
/// Thrown when a parallel operation resolves with
81+
/// <see cref="CompletionReason.FailureToleranceExceeded"/>. The aggregate
82+
/// <see cref="IBatchResult"/> is preserved on <see cref="Result"/> so callers
83+
/// can inspect per-branch outcomes.
84+
/// </summary>
85+
/// <remarks>
86+
/// This is the base type for parallel failures. Subclasses may be added in
87+
/// future releases (for example, a dedicated
88+
/// <c>ParallelFailureToleranceExceededException</c>); catching
89+
/// <see cref="ParallelException"/> remains forward-compatible.
90+
/// </remarks>
91+
public class ParallelException : DurableExecutionException
92+
{
93+
/// <summary>
94+
/// The aggregate result of the parallel operation. Type-erased — cast to
95+
/// <c>IBatchResult&lt;T&gt;</c> if the per-branch result type is known.
96+
/// </summary>
97+
public IBatchResult? Result { get; init; }
98+
99+
/// <summary>
100+
/// Why the parallel operation resolved.
101+
/// </summary>
102+
public CompletionReason CompletionReason { get; init; }
103+
104+
/// <summary>Creates an empty <see cref="ParallelException"/>.</summary>
105+
public ParallelException() { }
106+
/// <summary>Creates a <see cref="ParallelException"/> with the given message.</summary>
107+
public ParallelException(string message) : base(message) { }
108+
/// <summary>Creates a <see cref="ParallelException"/> wrapping an inner exception.</summary>
109+
public ParallelException(string message, Exception innerException) : base(message, innerException) { }
110+
}

0 commit comments

Comments
 (0)