diff --git a/.autover/changes/durable-map-iteration-subtype-and-inline-results.json b/.autover/changes/durable-map-iteration-subtype-and-inline-results.json new file mode 100644 index 000000000..4d8612982 --- /dev/null +++ b/.autover/changes/durable-map-iteration-subtype-and-inline-results.json @@ -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." + ] + } + ] +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummary.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummary.cs index b118ce558..35a16638e 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummary.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/BatchSummary.cs @@ -11,13 +11,15 @@ namespace Amazon.Lambda.DurableExecution.Internal; /// rebuilt without depending on user T shape. /// /// -/// Under per-unit results live on the children's -/// own CONTEXT checkpoints and only (plus -/// index/name) is recorded here. Under the -/// children emit no checkpoint, so each unit's serialized result -/// () or error -/// () is recorded inline here and read back -/// on replay. +/// Each dispatched unit's serialized result () +/// or error () is recorded inline here for +/// BOTH and . 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 +/// on replay. If the inline payload would exceed the +/// checkpoint size limit it is stripped (statuses only) and ReplayChildren +/// is set, so replay re-executes the units to recover their values instead. /// internal sealed class BatchSummary { @@ -40,16 +42,19 @@ internal sealed class BatchUnitSummary public string? Status { get; set; } /// - /// Serialized per-unit result, recorded inline only for - /// succeeded units (where no child checkpoint - /// exists to read it from). null under . + /// Serialized per-unit result, recorded inline for succeeded units under both + /// and . + /// null for non-succeeded units, or when the summary was stripped on + /// overflow (results recovered by re-executing units on replay). /// [JsonPropertyName("Result")] public string? Result { get; set; } /// - /// Per-unit error, recorded inline only for - /// failed units. null under . + /// Per-unit error, recorded inline for failed units under both + /// and . + /// null for non-failed units, or when the summary was stripped on + /// overflow. /// [JsonPropertyName("Error")] public ErrorObject? Error { get; set; } diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs index 51c7d3352..44658262f 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/ConcurrentOperation.cs @@ -101,7 +101,7 @@ protected ConcurrentOperation( /// Parent CONTEXT sub-type label (e.g. Parallel / Map). protected abstract string ParentSubType { get; } - /// Per-unit child-context sub-type label (e.g. ParallelBranch / MapItem). + /// Per-unit child-context sub-type label (e.g. ParallelBranch / MapIteration). protected abstract string ChildSubType { get; } /// Singular operation noun used in messages (e.g. "Parallel" / "Map"). @@ -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); @@ -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); @@ -796,38 +805,39 @@ private IBatchResult 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) { diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs index 21299a3cf..7e5322d74 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Internal/MapOperation.cs @@ -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> Func) GetUnit(int index) diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/Operation.cs b/Libraries/src/Amazon.Lambda.DurableExecution/Operation.cs index 412dd2111..770fe600a 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/Operation.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution/Operation.cs @@ -216,8 +216,8 @@ public static class OperationSubTypes /// Map parent sub-type. public const string Map = "Map"; - /// Map item (per-item child-context) sub-type. - public const string MapItem = "MapItem"; + /// Map iteration (per-item child-context) sub-type. + public const string MapIteration = "MapIteration"; } /// diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs index 891dd063f..005232eb0 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/MapOperationTests.cs @@ -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); @@ -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"); @@ -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] @@ -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" } }, @@ -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" } } @@ -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(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 + { + 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() { @@ -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" } }, @@ -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" } } @@ -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 { @@ -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" } } @@ -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(); diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/ParallelOperationTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/ParallelOperationTests.cs index e37862662..ab6962898 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Tests/ParallelOperationTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Tests/ParallelOperationTests.cs @@ -1065,6 +1065,88 @@ public async Task ParallelAsync_ReplaySucceeded_RebuildsResultFromCheckpoints() Assert.Empty(recorder.Flushed); } + [Fact] + public async Task ParallelAsync_NestedSucceeded_InlinesPerBranchResultsOnParentPayload() + { + // A Nested parallel must persist each branch's result INLINE on the parent + // SUCCEED payload (not only on the per-branch child checkpoints). The + // service collapses completed per-branch 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 (e.g. a wait after the parallel). + var (context, recorder, _, _) = CreateContext(); + + var result = await context.ParallelAsync( + new Func>[] + { + async (_, _) => { await Task.Yield(); return 100; }, + async (_, _) => { await Task.Yield(); return 200; }, + }, + name: "fanout"); + + Assert.Equal(new[] { 100, 200 }, result.GetResults()); + + await recorder.Batcher.DrainAsync(); + + var parentSucceed = Assert.Single(recorder.Flushed.Where(o => + o.Type == "CONTEXT" && o.SubType == "Parallel" && $"{o.Action}" == "SUCCEED")); + + // The parent payload carries the per-branch results inline. + var summary = System.Text.Json.JsonSerializer.Deserialize(parentSucceed.Payload!); + Assert.NotNull(summary); + Assert.Equal("100", summary!.Units[0].Result); + Assert.Equal("200", summary.Units[1].Result); + } + + [Fact] + public async Task ParallelAsync_ReplaySucceeded_RebuildsResultFromInlineSummaryWithoutChildOps() + { + // On a post-parallel resume the service returns only the parent Parallel op + // — the per-branch child ops are collapsed away. Results must be recovered + // from the inline summary alone, WITHOUT re-executing the branch bodies 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":"100"}, + {"Index":1,"Name":"1","Status":"SUCCEEDED","Result":"200"} + ]} + """; + + var (context, recorder, _, _) = CreateContext(new InitialExecutionState + { + Operations = new List + { + new() + { + Id = parentOpId, + Type = OperationTypes.Context, + Status = OperationStatuses.Succeeded, + SubType = OperationSubTypes.Parallel, + Name = "fanout", + ContextDetails = new ContextDetails { Result = summaryJson } + } + // NOTE: no per-branch child ops — mirrors the pruned resume state. + } + }); + + var executed = false; + var result = await context.ParallelAsync( + new Func>[] + { + async (_, _) => { executed = true; await Task.Yield(); return 999; }, + async (_, _) => { executed = true; await Task.Yield(); return 999; }, + }, + name: "fanout"); + + Assert.False(executed); + Assert.Equal(new[] { 100, 200 }, result.GetResults()); + Assert.Equal(CompletionReason.AllCompleted, result.CompletionReason); + + await recorder.Batcher.DrainAsync(); + Assert.Empty(recorder.Flushed); + } + [Fact] public async Task ParallelAsync_ReplayFailed_ResolvesFailureTolerance() {