Skip to content

Commit c19d6e8

Browse files
committed
address review: consistent serializer resolution, config headers, serializer unit coverage
- DurableContext: Parallel now resolves the effective serializer via LambdaSerializerHelper.GetRequired(LambdaContext) like every other operation, instead of an inline 'LambdaContext.Serializer ?? throw' (consistent error path). - Add the standard copyright/SPDX header to MapConfig.cs and WaitForConditionConfig.cs. - Add unit tests: Invoke fresh (non-replay) path serializes the request payload via the per-op serializer; Map/Parallel ItemSerializer is used to deserialize cached per-item/per-branch results on replay.
1 parent 508ae2e commit c19d6e8

4 files changed

Lines changed: 178 additions & 5 deletions

File tree

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

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -246,11 +246,7 @@ private Task<IBatchResult<T>> RunParallel<T>(
246246
// globally-registered serializer. This is the only serializer ConcurrentOperation
247247
// uses (per-unit child results + inline summary results); the aggregate batch
248248
// envelope is a source-generated structure and is unaffected.
249-
var serializer = effectiveConfig.ItemSerializer ?? LambdaContext.Serializer
250-
?? throw new InvalidOperationException(
251-
"No ILambdaSerializer is registered on ILambdaContext.Serializer. " +
252-
"Register a serializer via LambdaBootstrapBuilder.Create(handler, serializer) " +
253-
"(or in tests, set TestLambdaContext.Serializer).");
249+
var serializer = effectiveConfig.ItemSerializer ?? LambdaSerializerHelper.GetRequired(LambdaContext);
254250

255251
var operationId = _idGenerator.NextId();
256252
var op = new Internal.ParallelOperation<T>(

‎Libraries/src/Amazon.Lambda.DurableExecution/MapConfig.cs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,6 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
14
using Amazon.Lambda.Core;
25

36
namespace Amazon.Lambda.DurableExecution;

‎Libraries/src/Amazon.Lambda.DurableExecution/WaitForConditionConfig.cs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,6 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
14
using Amazon.Lambda.Core;
25

36
namespace Amazon.Lambda.DurableExecution;

‎Libraries/test/Amazon.Lambda.DurableExecution.Tests/PerOperationSerializerTests.cs‎

Lines changed: 171 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,9 @@ public class PerOperationSerializerTests
2222

2323
private static string IdAt(int position) => OperationIdGenerator.HashOperationId(position.ToString());
2424

25+
private static string ChildIdAt(string parentOpId, int position) =>
26+
OperationIdGenerator.HashOperationId($"{parentOpId}-{position}");
27+
2528
private static DurableContext CreateContext(ILambdaSerializer globalSerializer, InitialExecutionState? initialState = null)
2629
{
2730
var state = new ExecutionState();
@@ -178,6 +181,41 @@ public async Task Invoke_Replay_UsesPerOpSerializerToDeserialize()
178181
Assert.Equal(0, global.DeserializeCount);
179182
}
180183

184+
[Fact]
185+
public async Task Invoke_FreshExecution_UsesPerOpSerializerToSerializeRequestPayload()
186+
{
187+
// Comment (Copilot): InvokeConfig.Serializer applies to the outbound request
188+
// payload on the initial (non-replay) execution as well as to the replay result.
189+
// This covers the outbound-serialize path (the replay-deserialize path is above).
190+
var global = new SpySerializer();
191+
var perOp = new SpySerializer();
192+
193+
var state = new ExecutionState();
194+
state.LoadFromCheckpoint(null);
195+
var tm = new TerminationManager();
196+
var idGen = new OperationIdGenerator();
197+
var lambdaContext = new TestLambdaContext { Serializer = global };
198+
var recorder = new RecordingBatcher();
199+
var ctx = new DurableContext(
200+
state, tm, new WorkflowCancellation(tm), idGen, TestArn, lambdaContext, recorder.Batcher);
201+
202+
// Fresh chained invoke: the request payload is serialized (via the per-op
203+
// serializer), the CHAINED_INVOKE START is flushed, then the workflow suspends.
204+
// The returned task never completes on the fresh path.
205+
var task = ctx.InvokeAsync<string, string>(
206+
"arn:aws:lambda:us-east-1:123:function:callee:1",
207+
"payload",
208+
name: "inv",
209+
config: new InvokeConfig { Serializer = perOp });
210+
211+
await tm.WaitForTerminationAsync();
212+
Assert.False(task.IsCompleted);
213+
214+
// The outbound request payload used the per-op serializer, not the global one.
215+
Assert.Equal(1, perOp.SerializeCount);
216+
Assert.Equal(0, global.SerializeCount);
217+
}
218+
181219
// ---------------------------------------------------------------- ChildContext
182220

183221
[Fact]
@@ -284,4 +322,137 @@ public async Task Parallel_WithItemSerializer_UsesItForBranchResults()
284322
Assert.True(itemSer.SerializeCount >= 2); // each branch result serialized via the item serializer
285323
Assert.Equal(0, global.SerializeCount);
286324
}
325+
326+
// ---------------------------------------------------------------- Map / Parallel replay (deserialize)
327+
328+
[Fact]
329+
public async Task Map_Replay_UsesItemSerializerToDeserializeItemResults()
330+
{
331+
// Comment (Copilot): also cover that ItemSerializer is used for replay
332+
// deserialization of cached per-item results (not just fresh serialization).
333+
var global = new SpySerializer();
334+
var itemSer = new SpySerializer();
335+
336+
var parentOpId = IdAt(1);
337+
var i0 = ChildIdAt(parentOpId, 1);
338+
var i1 = ChildIdAt(parentOpId, 2);
339+
340+
var summaryJson =
341+
"{\"CompletionReason\":\"ALL_COMPLETED\",\"Units\":[" +
342+
"{\"Index\":0,\"Name\":\"0\",\"Status\":\"SUCCEEDED\"}," +
343+
"{\"Index\":1,\"Name\":\"1\",\"Status\":\"SUCCEEDED\"}]}";
344+
345+
var ctx = CreateContext(global, new InitialExecutionState
346+
{
347+
Operations = new List<Operation>
348+
{
349+
new()
350+
{
351+
Id = parentOpId,
352+
Type = OperationTypes.Context,
353+
Status = OperationStatuses.Succeeded,
354+
SubType = OperationSubTypes.Map,
355+
Name = "map",
356+
ContextDetails = new ContextDetails { Result = summaryJson }
357+
},
358+
new()
359+
{
360+
Id = i0,
361+
Type = OperationTypes.Context,
362+
Status = OperationStatuses.Succeeded,
363+
SubType = OperationSubTypes.MapIteration,
364+
Name = "0",
365+
ContextDetails = new ContextDetails { Result = "\"A\"" }
366+
},
367+
new()
368+
{
369+
Id = i1,
370+
Type = OperationTypes.Context,
371+
Status = OperationStatuses.Succeeded,
372+
SubType = OperationSubTypes.MapIteration,
373+
Name = "1",
374+
ContextDetails = new ContextDetails { Result = "\"B\"" }
375+
}
376+
}
377+
});
378+
379+
var calls = 0;
380+
var result = await ctx.MapAsync(
381+
new[] { "a", "b" },
382+
async (_, item, _, _, _) => { calls++; await Task.CompletedTask; return item.ToUpperInvariant(); },
383+
name: "map",
384+
config: new MapConfig<string> { ItemSerializer = itemSer });
385+
386+
Assert.Equal(0, calls); // cached — callback not re-run
387+
Assert.Equal(new[] { "A", "B" }, result.GetResults());
388+
Assert.True(itemSer.DeserializeCount >= 2); // each cached item result deserialized via the item serializer
389+
Assert.Equal(0, global.DeserializeCount); // aggregate summary uses source-gen, not the ILambdaSerializer
390+
}
391+
392+
[Fact]
393+
public async Task Parallel_Replay_UsesItemSerializerToDeserializeBranchResults()
394+
{
395+
var global = new SpySerializer();
396+
var itemSer = new SpySerializer();
397+
398+
var parentOpId = IdAt(1);
399+
var b0 = ChildIdAt(parentOpId, 1);
400+
var b1 = ChildIdAt(parentOpId, 2);
401+
402+
var summaryJson =
403+
"{\"CompletionReason\":\"ALL_COMPLETED\",\"Units\":[" +
404+
"{\"Index\":0,\"Name\":\"0\",\"Status\":\"SUCCEEDED\"}," +
405+
"{\"Index\":1,\"Name\":\"1\",\"Status\":\"SUCCEEDED\"}]}";
406+
407+
var ctx = CreateContext(global, new InitialExecutionState
408+
{
409+
Operations = new List<Operation>
410+
{
411+
new()
412+
{
413+
Id = parentOpId,
414+
Type = OperationTypes.Context,
415+
Status = OperationStatuses.Succeeded,
416+
SubType = OperationSubTypes.Parallel,
417+
Name = "par",
418+
ContextDetails = new ContextDetails { Result = summaryJson }
419+
},
420+
new()
421+
{
422+
Id = b0,
423+
Type = OperationTypes.Context,
424+
Status = OperationStatuses.Succeeded,
425+
SubType = OperationSubTypes.ParallelBranch,
426+
Name = "0",
427+
ContextDetails = new ContextDetails { Result = "\"x\"" }
428+
},
429+
new()
430+
{
431+
Id = b1,
432+
Type = OperationTypes.Context,
433+
Status = OperationStatuses.Succeeded,
434+
SubType = OperationSubTypes.ParallelBranch,
435+
Name = "1",
436+
ContextDetails = new ContextDetails { Result = "\"y\"" }
437+
}
438+
}
439+
});
440+
441+
var executed = false;
442+
var branches = new Func<IDurableContext, CancellationToken, Task<string>>[]
443+
{
444+
async (_, _) => { executed = true; await Task.CompletedTask; return "x"; },
445+
async (_, _) => { executed = true; await Task.CompletedTask; return "y"; },
446+
};
447+
448+
var result = await ctx.ParallelAsync(
449+
branches,
450+
name: "par",
451+
config: new ParallelConfig { ItemSerializer = itemSer });
452+
453+
Assert.False(executed); // cached — branches not re-run
454+
Assert.Equal(new[] { "x", "y" }, result.GetResults());
455+
Assert.True(itemSer.DeserializeCount >= 2); // each cached branch result deserialized via the item serializer
456+
Assert.Equal(0, global.DeserializeCount);
457+
}
287458
}

0 commit comments

Comments
 (0)