Skip to content

Commit f01d9f7

Browse files
GarrettBeattyGarrettBeatty
committed
Nested
1 parent 18c63ec commit f01d9f7

23 files changed

Lines changed: 1123 additions & 91 deletions

‎.autover/changes/1086291e-5286-4ea4-b9c1-af4eb1d0314d.json‎

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": "Minor",
6+
"ChangelogMessages": [
7+
"Implement NestingType.Flat for ParallelAsync and MapAsync (previously threw NotSupportedException). Under Flat, each branch/item runs in a virtual context that emits no per-branch CONTEXT checkpoint; per-branch results and errors are recorded inline on the parent operation's payload, reducing checkpoint volume. Operations inside a flat branch (steps, waits) still checkpoint, re-parented to the parallel/map operation. NestingType.Nested remains the default."
8+
]
9+
}
10+
]
11+
}

‎Libraries/src/Amazon.Lambda.DurableExecution/CLAUDE.md‎

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
# CLAUDE.md
2+
3+
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
4+
5+
## What this is
6+
7+
`Amazon.Lambda.DurableExecution` is the .NET SDK (preview, 0.x) for resilient, long-running AWS Lambda
8+
workflows that checkpoint progress after each step and resume after failures or waits. A workflow can run
9+
for up to ~1 year (the WAIT cap is 31,622,400 seconds) and is only billed for active compute. The SDK is
10+
client-side glue: the *durable execution service* (part of Lambda) owns the checkpoint store, fires timers,
11+
and re-invokes the function; this library re-derives in-memory workflow position from the checkpoint history
12+
the service sends on each invocation. See sibling SDKs (Python/JS/Java) listed in `README.md` for the shared
13+
model — this SDK deliberately mirrors their semantics.
14+
15+
## Build & test
16+
17+
Targets `net8.0;net10.0` (`DefaultPackageTargets` in `buildtools/common.props`). `TreatWarningsAsErrors` is on
18+
everywhere, and the main library is `IsTrimmable` with the trim analyzer enabled — keep new code AOT/trim-clean.
19+
20+
```bash
21+
# Build the library (run from this directory)
22+
dotnet build
23+
24+
# Unit tests (fast, no AWS). Project: Libraries/test/Amazon.Lambda.DurableExecution.Tests
25+
dotnet test ../../test/Amazon.Lambda.DurableExecution.Tests/Amazon.Lambda.DurableExecution.Tests.csproj
26+
27+
# A single test
28+
dotnet test ../../test/Amazon.Lambda.DurableExecution.Tests/Amazon.Lambda.DurableExecution.Tests.csproj \
29+
--filter "FullyQualifiedName~StepOperationTests"
30+
31+
# Coverage report (requires reportgenerator tool)
32+
../../test/Amazon.Lambda.DurableExecution.Tests/coverage.sh
33+
```
34+
35+
Unit tests reach `internal` types via `InternalsVisibleTo` (declared in the `.csproj`). They use
36+
`Amazon.Lambda.TestUtilities` (`TestLambdaContext`) and the real `SourceGeneratorLambdaJsonSerializer` —
37+
set `TestLambdaContext.Serializer` so `LambdaSerializerHelper.GetRequired` finds one.
38+
39+
### Integration tests (expensive, real AWS)
40+
41+
`Libraries/test/Amazon.Lambda.DurableExecution.IntegrationTests` deploys real Lambdas. Each test builds a
42+
`TestFunctions/<X>/` project into a container image via **`dotnet publish` + `docker build`**, pushes to ECR,
43+
creates an IAM role + Lambda (`DurableFunctionDeployment`), invokes it, and tears everything down on dispose.
44+
Requires Docker, AWS creds (us-east-1), and is slow. Every behavior in `docs/` should have a paired
45+
integration test under that project. Prefix AWS commands with `unset AWS_PROFILE` to use `[default]` creds.
46+
47+
**Run integration tests against `net10.0`.** The project multi-targets `net8.0;net10.0`; `dotnet test`
48+
without a framework spins up one testhost per TFW and runs them concurrently, which races two processes on
49+
the same `TestFunctions/<X>/` build dir. Pin the framework:
50+
51+
```bash
52+
dotnet test ../../test/Amazon.Lambda.DurableExecution.IntegrationTests/Amazon.Lambda.DurableExecution.IntegrationTests.csproj \
53+
-f net10.0 --filter "FullyQualifiedName~MultipleStepsTest"
54+
```
55+
56+
## Architecture: the replay model
57+
58+
This is the part you must understand before changing anything. Read these together:
59+
`DurableFunction.cs`, `DurableExecutionHandler.cs`, `DurableContext.cs`, `Internal/DurableOperation.cs`,
60+
`Internal/ExecutionState.cs`, `Internal/OperationIdGenerator.cs`, `Internal/TerminationManager.cs`.
61+
62+
**Entry point.** The user's Lambda handler delegates to `DurableFunction.WrapAsync<TInput,TOutput>`, which:
63+
hydrates `ExecutionState` from `invocationInput.InitialExecutionState` (paging the service via `NextMarker`),
64+
extracts the user payload from the `EXECUTION`-type op, builds a `CheckpointBatcher` + `DurableContext`, runs
65+
the workflow through `DurableExecutionHandler.RunAsync`, drains checkpoints, and maps the result to a
66+
`DurableExecutionInvocationOutput` with status **Succeeded / Failed / Pending**.
67+
68+
**Each operation runs the same workflow code every invocation.** There is no persisted program counter.
69+
On re-invocation the user function executes from the top again; each durable call (`StepAsync`, `WaitAsync`,
70+
etc.) looks up its own checkpoint and either replays the cached result or runs fresh. This is why workflow
71+
code **must be deterministic** — same operations, same order, same names across deployments.
72+
73+
**Deterministic operation IDs** (`OperationIdGenerator`). Each durable call gets an ID = SHA-256 of
74+
`"<parentPrefix>-<counter>"`, where the counter is per-context and pre-incremented. The same workflow position
75+
yields the same opaque ID across replays, so a checkpoint correlates to a call by *position*, not by name —
76+
renaming a step does **not** break replay (the human name rides separately on `OperationUpdate.Name`).
77+
Reordering or adding/removing calls *does* break it. `ValidateReplayConsistency` enforces this and throws
78+
`NonDeterministicExecutionException` on type/name drift.
79+
80+
**Suspension is implemented by never completing a Task** (`TerminationManager` + `DurableExecutionHandler`).
81+
When an op must suspend (wait timer, scheduled retry, pending callback/invoke) it calls
82+
`Termination.SuspendAndAwait<T>()`, which trips a one-shot signal and returns a Task that *never resolves*.
83+
`RunAsync` runs the user code via `Task.Run` and races it against `TerminationTask` with `Task.WhenAny`:
84+
- user task wins → **Succeeded** (or **Failed** if it threw)
85+
- termination wins → **Pending**; the abandoned user task is GC'd, checkpoints flush, the service fires the
86+
timer and re-invokes. On replay the suspended op sees its now-terminal checkpoint and returns normally.
87+
88+
**Operation classes** (`Internal/*Operation.cs`) all extend `DurableOperation<TResult>`. The base's
89+
`ExecuteAsync` does: `ValidateReplayConsistency` → `TrackReplay` → look up checkpoint → dispatch to
90+
`StartAsync` (no prior checkpoint) or `ReplayAsync` (checkpoint exists). `StepOperation` is the canonical
91+
example — read its class doc comment for the full status decision table (Succeeded→cached, Failed→rethrow,
92+
Pending→re-suspend if retry timer hasn't fired, Started→crash-recovery under `AtMostOncePerRetry`,
93+
Ready→run next attempt). `DurableContext` is a thin dispatcher: it allocates the op ID, pulls the serializer
94+
off `ILambdaContext.Serializer`, constructs the right `*Operation`, and calls `ExecuteAsync`.
95+
96+
**Checkpointing** (`CheckpointBatcher`). Outbound `OperationUpdate`s (START/SUCCEED/FAIL/RETRY) are enqueued
97+
to a background channel worker that batches and flushes them via `LambdaDurableServiceClient` (which wraps
98+
the `AWSSDK.Lambda` `Checkpoint`/`GetExecutionState` calls). `EnqueueAsync` awaits its batch's flush
99+
(sync semantics); fire-and-forget callers (e.g. the START checkpoint under the default
100+
`AtLeastOncePerRetry`) don't await but must observe the Task's exception. Flush errors become a terminal
101+
error rethrown by the next `EnqueueAsync`/`DrainAsync`. `DurableFunction.IsTerminalCheckpointError`
102+
classifies SDK errors on the final drain: 4xx (except 429 and stale-token) → **Failed** envelope; 429/5xx/
103+
network → let it escape so Lambda retries the whole invocation.
104+
105+
**Replay-mode tracking** (`ExecutionState`). `IsReplaying` starts true iff any completed non-`EXECUTION` op
106+
exists; `TrackReplay` decrements as each is visited and flips to false once the workflow catches up to the
107+
frontier. `ReplayAwareLogger` uses this to suppress log lines emitted during replay so a 30-step workflow
108+
re-invoked 30 times logs each line once — **always use `ctx.Logger`**, never `Console.WriteLine`.
109+
`ExecutionState` is lock-guarded because the batcher worker thread and concurrent parallel/map branches all
110+
touch it.
111+
112+
### Operations surface (`IDurableContext`)
113+
114+
`StepAsync` (checkpointed code + retries), `WaitAsync` (1s–~1yr timer), `RunInChildContextAsync` (isolated
115+
sub-workflow checkpointed as one `CONTEXT` op), `CreateCallbackAsync` / `WaitForCallbackAsync` (external
116+
events; `WaitForCallback` is *composed* from child-context + callback + submitter step — see
117+
`DurableContext.RunWaitForCallback`), `InvokeAsync` (durable-to-durable chained invoke, qualified ARN
118+
required), and `ParallelAsync` / `MapAsync` (concurrent branches → `IBatchResult<T>`).
119+
120+
**Nesting (`NestingType`)** matters for parallel/map. `Nested` (default) gives each branch a full `CONTEXT`
121+
checkpoint. `Flat` runs branches in *virtual* contexts that emit no `CONTEXT` op — inner ops re-parent to the
122+
parallel/map op via `OperationIdGenerator.CreateVirtualChild(operationId, reportedParentId)`, trading trace
123+
granularity for fewer checkpoints. The `idPrefix` vs `reportedParentId` split is the subtle part: inner IDs
124+
always derive from the branch's own op ID (so siblings never collide), but are *reported* under the nearest
125+
non-virtual ancestor (so they reference a parent that actually exists in the checkpoint store).
126+
127+
### Wire format (`Operation.cs`)
128+
129+
`Operation` and its `*Details` types mirror the service envelope JSON exactly (`[JsonPropertyName]`).
130+
String constants live in `OperationTypes` (STEP/WAIT/CALLBACK/CHAINED_INVOKE/CONTEXT/EXECUTION),
131+
`OperationStatuses` (STARTED/SUCCEEDED/FAILED/PENDING/READY/CANCELLED/STOPPED/TIMED_OUT), and
132+
`OperationSubTypes` (PascalCase finer classifier). Plural type names (`OperationTypes`, not `OperationType`)
133+
intentionally avoid collision with `AWSSDK.Lambda` model enums.
134+
135+
## Conventions
136+
137+
- **Programming model:** preview supports only the *executable* model — `Main` builds a `LambdaBootstrap`
138+
with a handler wrapper and an `ILambdaSerializer`. The serializer is read off `ILambdaContext.Serializer`
139+
(a preview API; the project-wide `AWSLAMBDA001` suppression in the `.csproj` is intentional for that
140+
reason). All step/result/payload (de)serialization flows through that one registered serializer, so AOT
141+
and reflection callers share a single code path — there is no per-call `JsonSerializerContext` argument.
142+
- **Errors:** durable exceptions carry `ErrorType`/`ErrorData`/`OriginalStackTrace` so a failure can be
143+
reconstructed on replay when the live exception object is gone. `StepException`, `ChildContextException`,
144+
`CallbackFailedException`/`CallbackTimeoutException`/`CallbackSubmitterException`, `ParallelException`,
145+
`MapException`, and `NonDeterministicExecutionException` all derive from `DurableExecutionException`.
146+
When adding error-mapping logic, handle *both* the fresh path (`InnerException` is the live exception) and
147+
the replay path (`InnerException` is null, `ErrorType` carries the type string) — see
148+
`DurableContext.MapWaitForCallbackException` for the pattern.
149+
- **Public config types** (`StepConfig`, `WaitForCallbackConfig`, `ParallelConfig`, `MapConfig`,
150+
`CompletionConfig`, etc.) are nullable optional args; resolve to an effective config inside the dispatcher.
151+
- Inclusive language is enforced repo-wide (see the user's global rules): no master/slave, whitelist/blacklist.

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

Lines changed: 24 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -240,12 +240,6 @@ private Task<IBatchResult<T>> RunParallel<T>(
240240
}
241241

242242
var effectiveConfig = config ?? new ParallelConfig();
243-
if (effectiveConfig.NestingType == NestingType.Flat)
244-
{
245-
throw new NotSupportedException(
246-
"NestingType.Flat is not yet supported in the .NET Durable Execution SDK. " +
247-
"Use NestingType.Nested (the default) for now.");
248-
}
249243

250244
var serializer = LambdaContext.Serializer
251245
?? throw new InvalidOperationException(
@@ -279,12 +273,6 @@ private Task<IBatchResult<TResult>> RunMap<TItem, TResult>(
279273
if (func == null) throw new ArgumentNullException(nameof(func));
280274

281275
var effectiveConfig = config ?? new MapConfig();
282-
if (effectiveConfig.NestingType == NestingType.Flat)
283-
{
284-
throw new NotSupportedException(
285-
"NestingType.Flat is not yet supported in the .NET Durable Execution SDK. " +
286-
"Use NestingType.Nested (the default) for now.");
287-
}
288276

289277
var serializer = LambdaSerializerHelper.GetRequired(LambdaContext);
290278

@@ -526,10 +514,31 @@ private Task<TResult> RunInvoke<TPayload, TResult>(
526514
/// but uses a child <see cref="OperationIdGenerator"/> so its operation IDs
527515
/// are deterministically namespaced under the parent op ID.
528516
/// </summary>
529-
private Func<string, IDurableContext> MakeChildFactory()
517+
/// <summary>
518+
/// Builds the factory each operation uses to create the inner
519+
/// <see cref="DurableContext"/> its user function runs against.
520+
/// </summary>
521+
/// <remarks>
522+
/// The delegate takes <c>(operationId, reportedParentId, isVirtual)</c>:
523+
/// <list type="bullet">
524+
/// <item><c>isVirtual == false</c> (the default child-context case): the
525+
/// inner context's ID space and reported parent both root at
526+
/// <c>operationId</c> via <see cref="OperationIdGenerator.CreateChild"/>;
527+
/// <c>reportedParentId</c> is ignored.</item>
528+
/// <item><c>isVirtual == true</c> (a <see cref="NestingType.Flat"/> branch):
529+
/// inner-op IDs still root at <c>operationId</c> (so sibling branches
530+
/// never collide), but inner ops report <c>reportedParentId</c> — the
531+
/// parallel/map operation — as their parent, since the virtual branch
532+
/// emits no CONTEXT checkpoint to reference.</item>
533+
/// </list>
534+
/// </remarks>
535+
private Func<string, string?, bool, IDurableContext> MakeChildFactory()
530536
{
531-
return parentOpId => new DurableContext(
532-
_state, _terminationManager, _workflowCancellation, _idGenerator.CreateChild(parentOpId),
537+
return (operationId, reportedParentId, isVirtual) => new DurableContext(
538+
_state, _terminationManager, _workflowCancellation,
539+
isVirtual
540+
? _idGenerator.CreateVirtualChild(operationId, reportedParentId)
541+
: _idGenerator.CreateChild(operationId),
533542
_durableExecutionArn, LambdaContext, _batcher);
534543
}
535544
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ namespace Amazon.Lambda.DurableExecution.Internal;
1010
/// </summary>
1111
[JsonSerializable(typeof(BatchSummary))]
1212
[JsonSerializable(typeof(BatchUnitSummary))]
13+
[JsonSerializable(typeof(ErrorObject))]
1314
internal sealed partial class BatchJsonContext : JsonSerializerContext
1415
{
1516
}

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

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,17 @@ namespace Amazon.Lambda.DurableExecution.Internal;
88
/// Shared by both <see cref="ParallelOperation{T}"/> and
99
/// <see cref="MapOperation{TItem, TResult}"/>: carries the completion reason and
1010
/// the per-unit index → status map so the <see cref="IBatchResult{T}"/> can be
11-
/// rebuilt without depending on user T shape — per-unit results live on the
12-
/// children's own checkpoints.
11+
/// rebuilt without depending on user T shape.
1312
/// </summary>
13+
/// <remarks>
14+
/// Under <see cref="NestingType.Nested"/> per-unit results live on the children's
15+
/// own CONTEXT checkpoints and only <see cref="BatchUnitSummary.Status"/> (plus
16+
/// index/name) is recorded here. Under <see cref="NestingType.Flat"/> the
17+
/// children emit no checkpoint, so each unit's serialized result
18+
/// (<see cref="BatchUnitSummary.Result"/>) or error
19+
/// (<see cref="BatchUnitSummary.Error"/>) is recorded inline here and read back
20+
/// on replay.
21+
/// </remarks>
1422
internal sealed class BatchSummary
1523
{
1624
[JsonPropertyName("CompletionReason")]
@@ -30,4 +38,19 @@ internal sealed class BatchUnitSummary
3038

3139
[JsonPropertyName("Status")]
3240
public string? Status { get; set; }
41+
42+
/// <summary>
43+
/// Serialized per-unit result, recorded inline only for
44+
/// <see cref="NestingType.Flat"/> succeeded units (where no child checkpoint
45+
/// exists to read it from). <c>null</c> under <see cref="NestingType.Nested"/>.
46+
/// </summary>
47+
[JsonPropertyName("Result")]
48+
public string? Result { get; set; }
49+
50+
/// <summary>
51+
/// Per-unit error, recorded inline only for <see cref="NestingType.Flat"/>
52+
/// failed units. <c>null</c> under <see cref="NestingType.Nested"/>.
53+
/// </summary>
54+
[JsonPropertyName("Error")]
55+
public ErrorObject? Error { get; set; }
3356
}

0 commit comments

Comments
 (0)