Add MapAsync for concurrent collection processing (#2408)

Author: GarrettBeatty

.autover/changes/durable-mapasync.json

@@ -0,0 +1,12 @@
+{
+  "Projects": [
+    {
+      "Name": "Amazon.Lambda.DurableExecution",
+      "Type": "Patch",
+      "ChangelogMessages": [
+        "Add `MapAsync` to `IDurableContext` for processing a collection in parallel with one child context per item and automatic checkpointing.",
+        "Breaking: renamed the concurrent-operation parent checkpoint payload field from `Branches` (shipped `ParallelAsync` format) to `Units`. The deserializer only binds `Units`, so durable executions checkpointed by earlier SDK versions will deserialize with an empty unit list and lose their per-unit names/statuses (including drift detection) on replay. In-flight workflows started on an earlier version should be drained before upgrading."
+      ]
+    }
+  ]
+}

Docs/durable-execution-design.md

@@ -1162,7 +1162,7 @@ public interface IDurableContext
     /// </summary>
     Task<IBatchResult<TResult>> MapAsync<TItem, TResult>(
         IReadOnlyList<TItem> items,
-        Func<IDurableContext, TItem, int, IReadOnlyList<TItem>, Task<TResult>> func,
+        Func<IDurableContext, TItem, int, IReadOnlyList<TItem>, CancellationToken, Task<TResult>> func,
         string? name = null,
         MapConfig? config = null,
         CancellationToken cancellationToken = default);
@@ -1443,22 +1443,21 @@ public class MapConfig
     public int? MaxConcurrency { get; set; }
 
     /// <summary>
-    /// When to consider the operation complete.
+    /// When to consider the operation complete. Defaults to AllCompleted() —
+    /// every item runs regardless of per-item failures, which surface via
+    /// IBatchResult&lt;T&gt;.Failed rather than throwing. This permissive default
+    /// matches the Python and Java SDKs' map operation. It differs intentionally
+    /// from ParallelConfig.CompletionConfig, which defaults to AllSuccessful()
+    /// (fail-fast). For fail-fast map behavior, set this to
+    /// CompletionConfig.AllSuccessful() or call IBatchResult&lt;T&gt;.ThrowIfError().
     /// </summary>
-    public CompletionConfig CompletionConfig { get; set; } = CompletionConfig.AllSuccessful();
+    public CompletionConfig CompletionConfig { get; set; } = CompletionConfig.AllCompleted();
 
     /// <summary>
     /// How item branches are represented in the checkpoint graph.
     /// </summary>
     public NestingType NestingType { get; set; } = NestingType.Nested;
 
-    /// <summary>
-    /// Optional batching configuration for grouping items before processing.
-    /// When set, items are grouped into batches and each batch is processed as a unit.
-    /// Reduces checkpoint overhead for large collections.
-    /// </summary>
-    public ItemBatcher? Batcher { get; set; }
-
     /// <summary>
     /// Optional function to generate a custom name for each item's branch.
     /// Improves observability in execution traces. Receives the item and its index.
@@ -1467,23 +1466,6 @@ public class MapConfig
     public Func<object, int, string>? ItemNamer { get; set; }
 }
 
-/// <summary>
-/// Groups items into batches for map operations to reduce checkpoint overhead.
-/// At least one of MaxItemsPerBatch or MaxBytesPerBatch must be set.
-/// </summary>
-public class ItemBatcher
-{
-    /// <summary>
-    /// Maximum number of items per batch. Null = no count limit.
-    /// </summary>
-    public int? MaxItemsPerBatch { get; set; }
-
-    /// <summary>
-    /// Maximum serialized size (bytes) per batch. Null = no size limit.
-    /// </summary>
-    public int? MaxBytesPerBatch { get; set; }
-}
-
 /// <summary>
 /// Defines completion criteria for parallel/map operations.
 /// </summary>
@@ -2222,7 +2204,6 @@ All four SDKs expose the same core operations. The differences are naming conven
 | Jitter strategy | `JitterStrategy` enum on `Exponential()` | `jitter_strategy` on `RetryStrategyConfig` | `jitter` on `createRetryStrategy()` |
 | Retry presets | `RetryStrategy.None/Default/Transient` | `RetryPresets.none()/default()/transient()` | `retryPresets.default/linear/noRetry` |
 | Nesting type | `NestingType` on `ParallelConfig`/`MapConfig` | `NestingType` on parallel/map config | `NestingType` on parallel/map config |
-| Item batching | `ItemBatcher` on `MapConfig` | `ItemBatcher` on `MapConfig` | *(checkpoint manager handles batching)* |
 | Item namer | `ItemNamer` on `MapConfig` | Item naming function on `MapConfig` | `itemNamer` on `MapConfig` |
 | Error mapping | `ErrorMapping` on `ChildContextConfig` | *(typed exception wrapping)* | `errorMapping` on child context config |
 | Message-based retry filter | `retryableMessagePatterns` (regex) | `retryable_errors` (regex) | `retryableErrors` (RegExp[]) |

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

@@ -260,6 +260,41 @@ private Task<IBatchResult<T>> RunParallel<T>(
         return op.ExecuteAsync(cancellationToken);
     }
 
+    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,
+        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,
+        CancellationToken cancellationToken)
+    {
+        if (items == null) throw new ArgumentNullException(nameof(items));
+        if (func == null) throw new ArgumentNullException(nameof(func));
+
+        var effectiveConfig = config ?? new MapConfig();
+        if (effectiveConfig.NestingType == NestingType.Flat)
+        {
+            throw new NotSupportedException(
+                "NestingType.Flat is not yet supported in the .NET Durable Execution SDK. " +
+                "Use NestingType.Nested (the default) for now.");
+        }
+
+        var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);
+
+        var operationId = _idGenerator.NextId();
+        var op = new Internal.MapOperation<TItem, TResult>(
+            operationId, name, _idGenerator.ParentId, items, func, effectiveConfig, serializer, MakeChildFactory(),
+            _state, _terminationManager, _workflowCancellation, _durableExecutionArn, _batcher);
+        return op.ExecuteAsync(cancellationToken);
+    }
+
     public Task<T> WaitForCallbackAsync<T>(
         Func<string, IWaitForCallbackContext, CancellationToken, Task> submitter,
         string? name = null,

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

@@ -131,3 +131,36 @@ 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) { }
+}

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

@@ -444,6 +444,38 @@ Task<IBatchResult<T>> ParallelAsync<T>(
         string? name = null,
         ParallelConfig? config = null,
         CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// 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"/>.
+    /// </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
+    /// 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
+    /// <see cref="IBatchResult{TResult}.ThrowIfError"/> for explicit
+    /// strict-success semantics. Per-item results are serialized to checkpoints
+    /// using the <see cref="ILambdaSerializer"/> registered on
+    /// <see cref="ILambdaContext.Serializer"/>.
+    /// </remarks>
+    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,
+        CancellationToken cancellationToken = default);
 }
 
 /// <summary>

Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchJsonContext.cs

@@ -0,0 +1,15 @@
+using System.Text.Json.Serialization;
+
+namespace Amazon.Lambda.DurableExecution.Internal;
+
+/// <summary>
+/// AOT-friendly <see cref="JsonSerializerContext"/> for the internal
+/// <see cref="BatchSummary"/> payload stored on a concurrent operation's parent
+/// CONTEXT checkpoint (parallel or map). Only this internal type — never user T —
+/// flows through here, so the source-generated metadata is sufficient.
+/// </summary>
+[JsonSerializable(typeof(BatchSummary))]
+[JsonSerializable(typeof(BatchUnitSummary))]
+internal sealed partial class BatchJsonContext : JsonSerializerContext
+{
+}

Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummary.cs

@@ -0,0 +1,33 @@
+using System.Text.Json.Serialization;
+
+namespace Amazon.Lambda.DurableExecution.Internal;
+
+/// <summary>
+/// Internal payload shape stored on a concurrent operation's parent CONTEXT
+/// checkpoint (as <c>ContextDetails.Result</c>) and reconstructed on replay.
+/// Shared by both <see cref="ParallelOperation{T}"/> and
+/// <see cref="MapOperation{TItem, TResult}"/>: carries the completion reason and
+/// the per-unit index → status map so the <see cref="IBatchResult{T}"/> can be
+/// rebuilt without depending on user T shape — per-unit results live on the
+/// children's own checkpoints.
+/// </summary>
+internal sealed class BatchSummary
+{
+    [JsonPropertyName("CompletionReason")]
+    public string? CompletionReason { get; set; }
+
+    [JsonPropertyName("Units")]
+    public IList<BatchUnitSummary> Units { get; set; } = new List<BatchUnitSummary>();
+}
+
+internal sealed class BatchUnitSummary
+{
+    [JsonPropertyName("Index")]
+    public int Index { get; set; }
+
+    [JsonPropertyName("Name")]
+    public string? Name { get; set; }
+
+    [JsonPropertyName("Status")]
+    public string? Status { get; set; }
+}