Skip to content

Fix durable Map/Parallel replay: MapIteration subtype + inline per-unit results #2476

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
GarrettBeatty merged 1 commit into from
Jul 14, 2026
Merged

Fix durable Map/Parallel replay: MapIteration subtype + inline per-unit results #2476

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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-map-iteration-subtype-and-inline-results.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
{
"Projects": [
{
"Name": "Amazon.Lambda.DurableExecution",
"Type": "Patch",
"ChangelogMessages": [
"Fix two durable Map/Parallel conformance issues found via cross-SDK testing. (1) Per-item map iteration child contexts now use the SubType \"MapIteration\" (was \"MapItem\"), matching the JS/Python/Java SDKs. (2) Persist each unit's result/error inline on the parent Parallel/Map SUCCEED payload for Nested nesting (previously only Flat), so a batch that completes and then suspends (e.g. a wait after the map) reconstructs its per-item results correctly on replay — the service collapses completed per-unit child contexts out of the resumed state, so the inline copy is the authoritative source. Large aggregate results still overflow to the ReplayChildren path. Preview."
]
}
]
}
29 changes: 17 additions & 12 deletions Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummary.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,15 @@ namespace Amazon.Lambda.DurableExecution.Internal;
/// rebuilt without depending on user T shape.
/// </summary>
/// <remarks>
/// Under <see cref="NestingType.Nested"/> per-unit results live on the children's
/// own CONTEXT checkpoints and only <see cref="BatchUnitSummary.Status"/> (plus
/// index/name) is recorded here. Under <see cref="NestingType.Flat"/> the
/// children emit no checkpoint, so each unit's serialized result
/// (<see cref="BatchUnitSummary.Result"/>) or error
/// (<see cref="BatchUnitSummary.Error"/>) is recorded inline here and read back
/// on replay.
/// Each dispatched unit's serialized result (<see cref="BatchUnitSummary.Result"/>)
/// or error (<see cref="BatchUnitSummary.Error"/>) is recorded inline here for
/// BOTH <see cref="NestingType.Nested"/> and <see cref="NestingType.Flat"/>. Flat
/// children emit no checkpoint of their own, and the service collapses completed
/// Nested per-unit child contexts out of the state returned on a later resume, so
/// the inline copy is the authoritative source for rebuilding the
/// <see cref="IBatchResult{T}"/> on replay. If the inline payload would exceed the
/// checkpoint size limit it is stripped (statuses only) and <c>ReplayChildren</c>
/// is set, so replay re-executes the units to recover their values instead.
/// </remarks>
internal sealed class BatchSummary
{
Expand All @@ -40,16 +42,19 @@ internal sealed class BatchUnitSummary
public string? Status { get; set; }

/// <summary>
/// Serialized per-unit result, recorded inline only for
/// <see cref="NestingType.Flat"/> succeeded units (where no child checkpoint
/// exists to read it from). <c>null</c> under <see cref="NestingType.Nested"/>.
/// Serialized per-unit result, recorded inline for succeeded units under both
/// <see cref="NestingType.Nested"/> and <see cref="NestingType.Flat"/>.
/// <c>null</c> for non-succeeded units, or when the summary was stripped on
/// overflow (results recovered by re-executing units on replay).
/// </summary>
[JsonPropertyName("Result")]
public string? Result { get; set; }

/// <summary>
/// Per-unit error, recorded inline only for <see cref="NestingType.Flat"/>
/// failed units. <c>null</c> under <see cref="NestingType.Nested"/>.
/// Per-unit error, recorded inline for failed units under both
/// <see cref="NestingType.Nested"/> and <see cref="NestingType.Flat"/>.
/// <c>null</c> for non-failed units, or when the summary was stripped on
/// overflow.
/// </summary>
[JsonPropertyName("Error")]
public ErrorObject? Error { get; set; }
Expand Down
68 changes: 39 additions & 29 deletions Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ protected ConcurrentOperation(
/// <summary>Parent CONTEXT sub-type label (e.g. Parallel / Map).</summary>
protected abstract string ParentSubType { get; }

/// <summary>Per-unit child-context sub-type label (e.g. ParallelBranch / MapItem).</summary>
/// <summary>Per-unit child-context sub-type label (e.g. ParallelBranch / MapIteration).</summary>
protected abstract string ChildSubType { get; }

/// <summary>Singular operation noun used in messages (e.g. "Parallel" / "Map").</summary>
Expand Down Expand Up @@ -720,7 +720,15 @@ BatchSummary BuildSummary(bool includeInline)
Name = item?.Name ?? unitName,
Status = SerializeStatus(item?.Status ?? BatchItemStatus.Started)
};
if (includeInline && _isVirtual && item != null)
// Persist each unit's result/error inline on the parent summary —
// for BOTH Nested and Flat units. The service collapses completed
// per-unit child contexts out of the state returned on a later
// (post-operation) resume, so replay cannot recover a Nested unit's
// value from its child checkpoint; the inline copy is the only
// durable source. This mirrors the JS SDK, whose default
// BatchResult serdes serializes the whole `all` array (results
// included) into the parent payload.
if (includeInline && item != null)
{
if (item.Status == BatchItemStatus.Succeeded)
unit.Result = SerializeResult(item.Result);
Expand All @@ -735,11 +743,12 @@ BatchSummary BuildSummary(bool includeInline)
var summary = BuildSummary(includeInline: true);
var payload = JsonSerializer.Serialize(summary, BatchJsonContext.Default.BatchSummary);

// Flat overflow: the inline per-unit results pushed the summary over the
// Overflow: the inline per-unit results pushed the summary over the
// checkpoint limit. Re-emit a stripped summary (statuses only) and flag
// ReplayChildren so replay reconstructs the values by re-executing units.
var overflow = _isVirtual
&& Encoding.UTF8.GetByteCount(payload) > DurableConstants.MaxOperationCheckpointBytes;
// Applies to both Nested and Flat now that both inline their results.
var overflow =
Encoding.UTF8.GetByteCount(payload) > DurableConstants.MaxOperationCheckpointBytes;
if (overflow)
{
summary = BuildSummary(includeInline: false);
Expand Down Expand Up @@ -796,38 +805,39 @@ private IBatchResult<T> ReconstructFromCheckpoints(Operation parent)
}
var resolvedName = checkpointedName ?? unitName;

// A never-dispatched unit (Started status with no per-unit child
// checkpoint) is excluded from the reconstructed All, matching the
// fresh-run view and the JS SDK. A dispatched-then-bailed unit is also
// Started but HAS a child op, so it stays. The name-drift check above
// still runs over the complete summary before we skip.
// A never-dispatched unit (Started, with neither an inline result/error
// on the summary nor a surviving per-unit child checkpoint) is excluded
// from the reconstructed All, matching the fresh-run view and the JS
// SDK. A dispatched-then-bailed unit is also Started but HAS a child op,
// so it stays. The name-drift check above still runs over the complete
// summary before we skip.
if (status == BatchItemStatus.Started && childOp == null)
continue;

T? unitResult = default;
DurableExecutionException? unitError = null;

// Flat (virtual) units have no child checkpoint — their result/error
// was recorded inline on this summary. Nested units read from the
// child's own CONTEXT checkpoint. A unit is "inline" when the summary
// entry carries a Result/Error, which only Flat writes.
if (_isVirtual && summaryEntry != null)
// Prefer the result/error recorded INLINE on the parent summary (both
// Nested and Flat units write it). This is the only durable source on a
// post-operation resume, since the service collapses completed per-unit
// child contexts out of the returned state. Fall back to the child's
// own CONTEXT checkpoint only when the summary carries no inline copy
// (e.g. state hydrated from an older checkpoint, or a dispatched unit
// whose child op is still present in state).
if (status == BatchItemStatus.Succeeded && summaryEntry?.Result != null)
{
if (status == BatchItemStatus.Succeeded && summaryEntry.Result != null)
{
unitResult = DeserializeResult(summaryEntry.Result);
}
else if (status == BatchItemStatus.Failed && summaryEntry.Error != null)
unitResult = DeserializeResult(summaryEntry.Result);
}
else if (status == BatchItemStatus.Failed && summaryEntry?.Error != null)
{
var err = summaryEntry.Error;
unitError = new ChildContextException(err.ErrorMessage ?? "Unit failed")
{
var err = summaryEntry.Error;
unitError = new ChildContextException(err.ErrorMessage ?? "Unit failed")
{
SubType = ChildSubType,
ErrorType = err.ErrorType,
ErrorData = err.ErrorData,
OriginalStackTrace = err.StackTrace
};
}
SubType = ChildSubType,
ErrorType = err.ErrorType,
ErrorData = err.ErrorData,
OriginalStackTrace = err.StackTrace
};
}
else if (status == BatchItemStatus.Succeeded && childOp?.ContextDetails?.Result != null)
{
Expand Down
2 changes: 1 addition & 1 deletion Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ public MapOperation(

protected override int UnitCount => _items.Count;
protected override string ParentSubType => OperationSubTypes.Map;
protected override string ChildSubType => OperationSubTypes.MapItem;
protected override string ChildSubType => OperationSubTypes.MapIteration;
protected override string OperationNoun => "Map";

protected override (string? Name, Func<IDurableContext, CancellationToken, Task<TResult>> Func) GetUnit(int index)
Expand Down
4 changes: 2 additions & 2 deletions Libraries/src/Amazon.Lambda.DurableExecution/Operation.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,8 +216,8 @@ public static class OperationSubTypes
/// <summary>Map parent sub-type.</summary>
public const string Map = "Map";

/// <summary>Map item (per-item child-context) sub-type.</summary>
public const string MapItem = "MapItem";
/// <summary>Map iteration (per-item child-context) sub-type.</summary>
public const string MapIteration = "MapIteration";
}

/// <summary>
Expand Down
99 changes: 89 additions & 10 deletions Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ await context.MapAsync(
var secondItemId = ChildIdAt(parentOpId, 2);

var itemStarts = recorder.Flushed
.Where(o => o.Type == "CONTEXT" && o.SubType == "MapItem" && o.Action == "START")
.Where(o => o.Type == "CONTEXT" && o.SubType == "MapIteration" && o.Action == "START")
.ToArray();
Assert.Equal(2, itemStarts.Length);
Assert.Contains(itemStarts, o => o.Id == firstItemId);
Expand Down Expand Up @@ -155,7 +155,7 @@ public async Task MapAsync_ItemNamer_PropagatesNameToCheckpointAndItem()
await recorder.Batcher.DrainAsync();

var itemSucceeds = recorder.Flushed
.Where(o => o.Type == "CONTEXT" && o.SubType == "MapItem" && o.Action == "SUCCEED")
.Where(o => o.Type == "CONTEXT" && o.SubType == "MapIteration" && o.Action == "SUCCEED")
.ToArray();
Assert.Contains(itemSucceeds, o => o.Name == "Order-order-1");
Assert.Contains(itemSucceeds, o => o.Name == "Order-order-2");
Expand Down Expand Up @@ -410,7 +410,7 @@ public async Task MapAsync_NestingTypeFlat_SuppressesPerItemContextOps()
Assert.Equal(new[] { "START", "SUCCEED" }, parentActions);

Assert.Empty(recorder.Flushed.Where(o =>
o.Type == "CONTEXT" && o.SubType == "MapItem"));
o.Type == "CONTEXT" && o.SubType == "MapIteration"));
}

[Fact]
Expand Down Expand Up @@ -548,7 +548,7 @@ public async Task MapAsync_ReplaySucceeded_RebuildsResultFromCheckpoints()
Id = i0,
Type = OperationTypes.Context,
Status = OperationStatuses.Succeeded,
SubType = OperationSubTypes.MapItem,
SubType = OperationSubTypes.MapIteration,
Name = "0",
ContextDetails = new ContextDetails { Result = "100" }
},
Expand All @@ -557,7 +557,7 @@ public async Task MapAsync_ReplaySucceeded_RebuildsResultFromCheckpoints()
Id = i1,
Type = OperationTypes.Context,
Status = OperationStatuses.Succeeded,
SubType = OperationSubTypes.MapItem,
SubType = OperationSubTypes.MapIteration,
Name = "1",
ContextDetails = new ContextDetails { Result = "200" }
}
Expand All @@ -579,6 +579,85 @@ public async Task MapAsync_ReplaySucceeded_RebuildsResultFromCheckpoints()
Assert.Empty(recorder.Flushed);
}

[Fact]
public async Task MapAsync_NestedSucceeded_InlinesPerItemResultsOnParentPayload()
{
// A Nested map must persist each item's result INLINE on the parent
// SUCCEED payload (not only on the per-item child checkpoints). The
// service collapses completed per-item child contexts out of the state
// returned on a later resume, so the inline copy is the only durable
// source for reconstructing results on replay (see the 9-17 conformance
// test: a wait after a successful map).
var (context, recorder, _, _) = CreateContext();

var result = await context.MapAsync(
new[] { "a", "b" },
async (ctx, item, index, all, _) => { await Task.Yield(); return item.ToUpperInvariant(); },
name: "then-wait");

Assert.Equal(new[] { "A", "B" }, result.GetResults());

await recorder.Batcher.DrainAsync();

var parentSucceed = Assert.Single(recorder.Flushed.Where(o =>
o.Type == "CONTEXT" && o.SubType == "Map" && $"{o.Action}" == "SUCCEED"));

// The parent payload carries the per-item results inline (the serializer
// stores each item's serialized value on the summary unit).
var summary = System.Text.Json.JsonSerializer.Deserialize<BatchSummary>(parentSucceed.Payload!);
Assert.NotNull(summary);
Assert.Equal("\"A\"", summary!.Units[0].Result);
Assert.Equal("\"B\"", summary.Units[1].Result);
}

[Fact]
public async Task MapAsync_ReplaySucceeded_RebuildsResultFromInlineSummaryWithoutChildOps()
{
// Reproduces the 9-17 conformance scenario: on a post-map resume the
// service returns only the parent Map op (plus EXECUTION / the trailing
// wait) — the per-item MapIteration child ops are collapsed away. Results
// must be recovered from the inline summary alone, WITHOUT re-executing
// the item callback and WITHOUT the child ops present in state.
var parentOpId = IdAt(1);

var summaryJson = """
{"CompletionReason":"ALL_COMPLETED","Units":[
{"Index":0,"Name":"0","Status":"SUCCEEDED","Result":"\"A\""},
{"Index":1,"Name":"1","Status":"SUCCEEDED","Result":"\"B\""}
]}
""";

var (context, recorder, _, _) = CreateContext(new InitialExecutionState
{
Operations = new List<Operation>
{
new()
{
Id = parentOpId,
Type = OperationTypes.Context,
Status = OperationStatuses.Succeeded,
SubType = OperationSubTypes.Map,
Name = "then-wait",
ContextDetails = new ContextDetails { Result = summaryJson }
}
// NOTE: no per-item child ops — mirrors the pruned resume state.
}
});

var calls = 0;
var result = await context.MapAsync(
new[] { "a", "b" },
async (ctx, item, index, all, _) => { calls++; await Task.Yield(); return "SHOULD-NOT-RUN"; },
name: "then-wait");

Assert.Equal(0, calls);
Assert.Equal(2, result.SuccessCount);
Assert.Equal(new[] { "A", "B" }, result.GetResults());

await recorder.Batcher.DrainAsync();
Assert.Empty(recorder.Flushed);
}

[Fact]
public async Task MapAsync_ReplayMixedStatus_PreservesStartedShortCircuited()
{
Expand Down Expand Up @@ -612,7 +691,7 @@ public async Task MapAsync_ReplayMixedStatus_PreservesStartedShortCircuited()
Id = i0,
Type = OperationTypes.Context,
Status = OperationStatuses.Succeeded,
SubType = OperationSubTypes.MapItem,
SubType = OperationSubTypes.MapIteration,
Name = "0",
ContextDetails = new ContextDetails { Result = "10" }
},
Expand All @@ -621,7 +700,7 @@ public async Task MapAsync_ReplayMixedStatus_PreservesStartedShortCircuited()
Id = i1,
Type = OperationTypes.Context,
Status = OperationStatuses.Succeeded,
SubType = OperationSubTypes.MapItem,
SubType = OperationSubTypes.MapIteration,
Name = "1",
ContextDetails = new ContextDetails { Result = "20" }
}
Expand Down Expand Up @@ -680,7 +759,7 @@ public async Task MapAsync_ReplayFailed_RebuildsResultWithoutThrowing()
Id = i0,
Type = OperationTypes.Context,
Status = OperationStatuses.Failed,
SubType = OperationSubTypes.MapItem,
SubType = OperationSubTypes.MapIteration,
Name = "0",
ContextDetails = new ContextDetails
{
Expand Down Expand Up @@ -734,7 +813,7 @@ public async Task MapAsync_ReplayWithDriftedItemName_ThrowsNonDeterministic()
Id = i0,
Type = OperationTypes.Context,
Status = OperationStatuses.Succeeded,
SubType = OperationSubTypes.MapItem,
SubType = OperationSubTypes.MapIteration,
Name = "alpha",
ContextDetails = new ContextDetails { Result = "10" }
}
Expand Down Expand Up @@ -768,7 +847,7 @@ string[] IdsFromRun()
async (ctx, item, index, all, _) => { await Task.Yield(); return item; }).GetAwaiter().GetResult();
recorder.Batcher.DrainAsync().GetAwaiter().GetResult();
return recorder.Flushed
.Where(o => o.Type == "CONTEXT" && o.SubType == "MapItem" && o.Action == "START")
.Where(o => o.Type == "CONTEXT" && o.SubType == "MapIteration" && o.Action == "START")
.Select(o => o.Id)
.OrderBy(id => id)
.ToArray();
Expand Down
Loading
Loading