From cc2fbf842aa0f25f7d780b69cac7398ecd396767 Mon Sep 17 00:00:00 2001 From: Garrett Beatty Date: Wed, 22 Jul 2026 16:32:14 -0400 Subject: [PATCH 1/6] Add durable-execution emulator to Lambda Test Tool v2 (Phases 1-2) Adds a local durable-execution service emulator to the Test Tool, enabled with --durable-execution. A durable function whose AWS_ENDPOINT_URL_LAMBDA is pointed at the Test Tool can now run a full workflow locally. Phase 1 (data plane): CheckpointDurableExecution and GetDurableExecutionState endpoints hosted alongside the Runtime API, backed by the operation store + checkpoint processor ported from Amazon.Lambda.DurableExecution.Testing and retyped to plain System.Text.Json request DTOs (the AWSSDK model types use ConstantClass and can't be STJ-deserialized). Response timestamps are emitted as unix seconds to match the AWSSDK rest-json unmarshaller. Phase 2 (control plane): invoking with X-Amz-Durable-Execution-Name starts an execution that DurableExecutionDriver drives to completion, re-invoking the function over the Runtime API across the replay cycle. Timers are time-skipped by default (--durable-time-skip). Chained durable invokes and external callbacks are not yet supported and fail fast with a clear message. Includes end-to-end tests exercising both the data plane and the full start-hook/drive-loop cycle through a real AmazonLambdaClient and Runtime API. --- .../50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json | 12 + .../docs/design/testtool-integration-plan.md | 263 ++++++++++++++ .../Amazon.Lambda.TestTool.csproj | 5 + .../Commands/Settings/RunCommandSettings.cs | 19 + .../Processes/TestToolProcess.cs | 14 + .../DurableExecution/CheckpointProcessor.cs | 314 +++++++++++++++++ .../DurableExecutionDriver.cs | Bin 0 -> 14120 bytes .../DurableExecution/DurableExecutionStore.cs | 144 ++++++++ .../DurableExecution/DurableRestJsonModels.cs | 301 ++++++++++++++++ .../DurableExecution/DurableServiceApi.cs | 180 ++++++++++ .../InMemoryOperationStore.cs | 112 ++++++ .../Services/LambdaRuntimeAPI.cs | 48 +++ .../DurableExecutionDriverTests.cs | 201 +++++++++++ .../DurableServiceApiTests.cs | 324 ++++++++++++++++++ 14 files changed, 1937 insertions(+) create mode 100644 .autover/changes/50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json create mode 100644 Libraries/src/Amazon.Lambda.DurableExecution/docs/design/testtool-integration-plan.md create mode 100644 Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/CheckpointProcessor.cs create mode 100644 Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableExecutionDriver.cs create mode 100644 Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableExecutionStore.cs create mode 100644 Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableRestJsonModels.cs create mode 100644 Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableServiceApi.cs create mode 100644 Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/InMemoryOperationStore.cs create mode 100644 Tools/LambdaTestTool-v2/tests/Amazon.Lambda.TestTool.UnitTests/DurableExecutionDriverTests.cs create mode 100644 Tools/LambdaTestTool-v2/tests/Amazon.Lambda.TestTool.UnitTests/DurableServiceApiTests.cs diff --git a/.autover/changes/50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json b/.autover/changes/50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json new file mode 100644 index 000000000..ccd76c6ae --- /dev/null +++ b/.autover/changes/50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json @@ -0,0 +1,12 @@ +{ + "Projects": [ + { + "Name": "Amazon.Lambda.TestTool", + "Type": "Minor", + "ChangelogMessages": [ + "Add durable-execution service emulator (preview): with --durable-execution, the Test Tool hosts CheckpointDurableExecution and GetDurableExecutionState endpoints alongside the Lambda Runtime API so durable workflows can checkpoint and read state locally. Point the function's AWS_ENDPOINT_URL_LAMBDA at the Test Tool to use it. Timers are time-skipped by default (--durable-time-skip).", + "Support running a durable workflow locally end to end: invoking a function with the X-Amz-Durable-Execution-Name header starts a durable execution that the Test Tool drives to completion, re-invoking the function across the replay cycle (steps, waits, retries, child contexts, parallel/map). Chained durable-to-durable invokes and external callbacks are not yet supported." + ] + } + ] +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/docs/design/testtool-integration-plan.md b/Libraries/src/Amazon.Lambda.DurableExecution/docs/design/testtool-integration-plan.md new file mode 100644 index 000000000..93ae4cde4 --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution/docs/design/testtool-integration-plan.md @@ -0,0 +1,263 @@ +# Durable Execution support in Lambda Test Tool v2 — Design + +**Status:** Proposed · **Phase 0 spike:** ✅ complete (endpoint redirection verified) +**Audience:** contributors to `Tools/LambdaTestTool-v2` and `Amazon.Lambda.DurableExecution*` + +## 1. Goal + +Let a developer run a **real** durable Lambda function locally (`dotnet run`, debugger attached) and +exercise a full multi-invocation durable workflow — steps, waits, retries, callbacks, child contexts, +parallel/map — without deploying to AWS and without a real durable-execution service. The Test Tool +already emulates the Lambda Runtime API for ordinary functions; this adds the **durable-execution +service data plane** as a sibling emulator on the same host. + +Two experiences result: +- **Headless** — invoke a durable workflow and watch it drive to completion (time-skipped by default). +- **Interactive** — step through the replay cycle in a debugger; inspect the operation timeline and fire + callbacks from the Blazor UI. + +The in-process `DurableTestRunner` (in `Amazon.Lambda.DurableExecution.Testing`) already covers automated +unit/integration testing. It is **not** a substitute for this: it drives an in-process C# delegate, so it +cannot run the developer's real, separately-launched executable under a debugger — which is the entire +point of the Test Tool. See §3 for why the orchestrator cannot simply be reused. + +## 2. How the two systems work today (verified) + +### 2.1 Test Tool v2 (`Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool`) + +- Accepts SDK invokes at `POST /2015-03-31/functions/{functionName}/invocations` and a fixed + `/functions/function/invocations` default variant (`LambdaRuntimeAPI.cs:25-26`). +- Queues events into a per-function `RuntimeApiDataStore`, partitioned **by URL path** through + `RuntimeApiDataStoreManager` (a `ConcurrentDictionary` keyed by function name, `:40-52`). The partition + is selected by the path, **not** by `AWS_LAMBDA_FUNCTION_NAME`. +- Exposes the Runtime API a local bootstrap polls: `GET /{fn}/2018-06-01/runtime/invocation/next` + (`GetNextInvocation`, polls `TryActivateEvent` every 100 ms, `:120`) and + `POST .../invocation/{awsRequestId}/response` (`PostInvocationResponse` → `ReportSuccess`, `:189`). +- `EventContainer` tracks a request/response pair and blocks the caller in RequestResponse mode via + `WaitForCompletion` — a **fail-safe 15-minute bounded wait**, not indefinite (`EventContainer.cs:86-96`). + RequestResponse is the default unless `X-Amz-Invocation-Type` says otherwise (`LambdaRuntimeAPI.cs:56-60`). +- **Hosting model:** `TestToolProcess` is a plain class with a static `Startup(...)` factory (not a + `BackgroundService`, not a base class); endpoints register from `TestToolProcess.cs:108`. Only the SQS / + DynamoDB event sources use real `BackgroundService`. The API Gateway emulator uses a + `app.Map("/{**catchAll}", ...)` catch-all (`ApiGatewayEmulatorProcess.cs:86`). Subsystems are wired into + the run from `RunCommand.cs:49-97` — which starts **only the tool's own web app and a browser**; it does + **not** launch the user's function. + +### 2.2 Durable Execution SDK (`Libraries/src/Amazon.Lambda.DurableExecution`) + +- The function's handler delegates to `DurableFunction.WrapAsync`, which hydrates `ExecutionState` from the + invocation envelope's `InitialExecutionState.Operations` (+ `NextMarker` paging), runs the workflow, and + returns a `DurableExecutionInvocationOutput` with status **Succeeded / Failed / Pending**. +- During a run it calls exactly **two** data-plane RPCs through `IDurableServiceClient` + (`Services/IDurableServiceClient.cs`): `CheckpointAsync` (flush `OperationUpdate`s) and + `GetExecutionStateAsync` (page state). The production adapter `LambdaDurableServiceClient` calls + `IAmazonLambda.CheckpointDurableExecutionAsync` / `GetDurableExecutionStateAsync`. +- **The SDK owns no timers.** Suspension = returning a never-completing Task and emitting `Pending` + (`TerminationManager`). Timers, retry backoff, and re-invocation are the **service's** responsibility. +- By default `DurableFunction.cs:30-31` constructs `new AmazonLambdaClient()` with **no config**, cached in + a `Lazy`. There is also a `WrapAsync(..., IAmazonLambda lambdaClient)` overload (`DurableFunction.cs:46-52`) + for supplying a custom client. + +### 2.3 Testing package (`Amazon.Lambda.DurableExecution.Testing`) + +- `ExecutionOrchestrator` plays the service role **in-process**: it repeatedly calls + `DurableFunction.WrapAsync` with an in-memory `IDurableServiceClient`, interprets `Pending`, and drives + re-invocation until terminal (`ExecutionOrchestrator.cs:83-136`). +- Time-skipping lives in `CheckpointProcessor.ApplyTimeSkipping` (`CheckpointProcessor.cs:273-298`), which + mutates stored ops (`WAIT → Succeeded`, retry `STEP → Ready`) **at checkpoint-application time**. +- The reusable backend trio — `InMemoryOperationStore`, `CheckpointProcessor`, `InMemoryDurableServiceClient` + — are all `internal sealed`, wired at `DurableTestRunner.cs:60-62`. They already traffic in + `Amazon.Lambda.Model.OperationUpdate` / `Operation`. + +### 2.4 Wire format (`aws-sdk-net` `generator/ServiceModels/lambda/lambda-2015-03-31.normal.json`) + +- REST-JSON with path templates. `CheckpointDurableExecution` → + `POST /2025-12-01/durable-executions/{DurableExecutionArn}/checkpoint`; `GetDurableExecutionState` → + `GET /2025-12-01/durable-executions/{DurableExecutionArn}/state`; callbacks → + `POST /2025-12-01/durable-execution-callbacks/{CallbackId}/{succeed|fail|heartbeat}`. +- **There is no `StartDurableExecution` operation** and **no named `DurableExecutionInvocationInput/Output` + wire shape.** An execution starts as an ordinary `Invoke` carrying the `X-Amz-Durable-Execution-Name` + header; the response returns `X-Amz-Durable-Execution-Arn`. The `InvocationInput/Output` envelope is an + **SDK-side POCO** that rides the opaque Runtime-API event/response. +- `DurableExecutionArn` **contains slashes** (`arn:...:function:NAME:QUALIFIER/durable-execution/GROUP/NAME`), + so an emulator router must use a catch-all and handle URL-encoded slashes. +- Payload caps: 6 MB max `OperationPayload`/`InputPayload` string length; effective caps are 256 KB + (CONTEXT/STEP/WAIT/CALLBACK) and 1 MB (async/chained). `ReplayChildren` exists on both checkpoint updates + (`ContextOptions.ReplayChildren`) and state responses (`ContextDetails.ReplayChildren`). +- **ID allocation is split:** the **server (emulator)** allocates the `DurableExecutionArn` and the rotating + `CheckpointToken`; the **SDK (client)** allocates operation IDs and callback IDs. The emulator only mints + ARNs and tokens. + +## 3. Corrections to the original investigation + +| Claim | Verdict | Correction | +|---|---|---| +| Reuse `ExecutionOrchestrator` in the Test Tool | **Rejected** | It holds a generic in-process `Func>` (`ExecutionOrchestrator.cs:16`, invoked `:95`) with no serialization boundary. It cannot drive a separate process. Reuse the **backend trio**; **re-write** the ~50-line drive loop over the wire. | +| Test Tool injects env vars into the function process | **False** | The Test Tool never launches the function (`RunCommand.cs:49-97`). The **user** sets the env vars on their own `dotnet run`, matching today's `AWS_LAMBDA_RUNTIME_API` UX. | +| Redirect the durable client via a ServiceURL hook in the SDK | **Resolved (see §4)** | No bespoke SDK hook; relies on stock AWSSDK `AWS_ENDPOINT_URL_LAMBDA` resolution. **Verified working** by the Phase-0 spike. | +| Feed `CheckpointDurableExecutionRequest` straight into `CheckpointProcessor.Process` | **False as written** | `Process` takes `Amazon.Lambda.Model.OperationUpdate`, whose `Action`/`Type` are AWSSDK `ConstantClass` types with **no System.Text.Json contract**. The emulator must define its own rest-json DTOs and **hand-map** them into the model types. This mapper is the bulk of the data-plane work. | +| `TestToolProcess`-style background services | **Partly** | `TestToolProcess` is not a `BackgroundService` and not a base class. The durable driver should follow the Runtime-API model — a per-execution `Task` — not `AddHostedService`. | +| Time-skipping incompatible with a wall-clock re-invoke loop | **Resolved** | True only against the **real** service. Here the Test Tool **owns** the store the function reads back on replay, so `ApplyTimeSkipping` is authoritative and is the correct fast default. | +| IVT already lets the Test Tool reuse the backend trio | **False** | The trio is `internal sealed`; `.Testing` grants IVT only to `.Testing.Tests`. A **new IVT entry (or public facade)** plus matching strong-name is required. Prefer a small public facade to avoid coupling to 0.x internals. | + +## 4. Phase 0 spike — endpoint redirection (COMPLETE ✅) + +**Question:** does the bundled `AWSSDK.Lambda 4.0.13.1` route `CheckpointDurableExecutionAsync` to +`AWS_ENDPOINT_URL_LAMBDA` when the client is `new AmazonLambdaClient()` with no config — i.e. exactly the +`DurableFunction` default path? + +**Method** (`C:\tmp\durable-endpoint-spike`): a local `HttpListener` stub; set only +`AWS_ENDPOINT_URL_LAMBDA`, `AWS_REGION`, and dummy creds; construct `new AmazonLambdaClient()`; call +`CheckpointDurableExecutionAsync` and observe where the request lands. + +**Result: SUCCESS.** The SDK resolved `client.Config.ServiceURL` to the stub and sent: +``` +POST /2025-12-01/durable-executions/arn%3Aaws%3Alambda%3A...%2Fdurable-execution%2Fg%2Fe/checkpoint +User-Agent: aws-sdk-dotnet-coreclr/4.0.13.1 ... api/Lambda#4.0.13.1 +``` +Confirmed in AWSSDK.Core: `ClientConfig.ServiceURL` reads the service-specific +`AWS_ENDPOINT_URL_LAMBDA` (then global `AWS_ENDPOINT_URL`) whenever `IgnoreConfiguredEndpointUrls == false` +(default) and `ServiceId != null` (`ClientConfig.cs:340-395`). + +**Implications:** +- The **zero-user-code-change** story holds: the developer sets env vars, no SDK modification needed. +- Note the ARN's slashes arrive **URL-encoded** (`%2F`) — the emulator router must decode them. +- The user must also set `AWS_REGION` + dummy `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY`, or SigV4 signing + throws before the request leaves the process; the emulator ignores the signature. +- **Fallback (not needed):** if a future SDK build regressed this, the `WrapAsync(IAmazonLambda)` overload + (`DurableFunction.cs:46-52`) with a caller-built `AmazonLambdaConfig { ServiceURL }` would work but require + a user code change. + +## 5. Recommended architecture + +Single Kestrel `WebApplication` (the existing `TestToolProcess`), one port, three route groups, one +out-of-process function. + +**Env vars the user sets on their `dotnet run`:** +- `AWS_LAMBDA_RUNTIME_API=127.0.0.1:{port}/{functionName}` — the bootstrap poll target. **Must include the + function-name path prefix** so the bootstrap's `BaseUrl` hits the named partition (`RuntimeApiDataStoreManager`). +- `AWS_ENDPOINT_URL_LAMBDA=http://127.0.0.1:{port}` — redirects checkpoint/state/callback traffic (verified §4). +- `AWS_REGION` + dummy `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` — required for signing; ignored by the emulator. + +**Route group A — Runtime API (existing, unchanged).** Transports the durable envelope in/out via the +event queue + `EventContainer`. + +**Route group B — Durable data plane (NEW).** Registered via `SetupDurableServiceEndpoints(WebApplication)` +called next to the Runtime-API setup at `TestToolProcess.cs:108`: +- `POST /2025-12-01/durable-executions/{**path}` (suffix `/checkpoint`, `/stop`) +- `GET /2025-12-01/durable-executions/{**path}` (bare, `/state`, `/history`) +- `POST /2025-12-01/durable-execution-callbacks/{callbackId}/{succeed|fail|heartbeat}` + +Uses a catch-all `{**path}` + suffix parse (the `ApiGatewayEmulatorProcess.cs:86` pattern) because the ARN +contains encoded slashes. Each handler deserializes into **emulator-owned STJ DTOs**, hand-maps to +`Amazon.Lambda.Model.OperationUpdate`, calls `CheckpointProcessor.Process(arn, token, updates)` over a +shared `InMemoryOperationStore`, and returns `NewExecutionState { Operations, NextMarker }` with a +**freshly minted CheckpointToken**. + +**Route group C — Durable start (NEW).** Extend `PostEvent` (`LambdaRuntimeAPI.cs:51`) to branch on the +`X-Amz-Durable-Execution-Name` header. It does **not** take the blocking `EventContainer` path; it allocates +the ARN + initial token, seeds an `EXECUTION`-type op with the input payload, launches the drive loop as a +detached `Task`, and returns **202 immediately** with `X-Amz-Durable-Execution-Arn`. Idempotency: same +name + same payload → existing ARN; differing payload → `DurableExecutionAlreadyStartedException`. + +**Drive loop (`DurableExecutionDriver`, re-expressing `ExecutionOrchestrator.cs:83-136` over the wire):** +1. Read the **current** CheckpointToken from the store (rotates each checkpoint — never reuse the prior). +2. Build the invocation envelope with a **bounded first page** of `InitialExecutionState.Operations` + + `NextMarker` (honoring `ReplayChildren`). **Do not inline full history** — it breaches the 6 MB + queued-event cap (`LambdaRuntimeAPI.cs:14,65`); page the rest via `GET .../state`. +3. `GetLambdaRuntimeDataStore(fn).QueueEvent(inputJson, isRequestResponseMode:true)` + (`RuntimeApiDataStore.cs:92`), then block on `EventContainer.WaitForCompletion` — the **driver** blocks, + not the end caller. +4. While the function runs, its redirected client hits route group B; with time-skip on, + `ApplyTimeSkipping` flips `WAIT → Succeeded` / retry `STEP → Ready` in the store **at checkpoint time**, + before the function posts its Pending response — so on `WaitForCompletion` return the store shows no + pending timer. +5. Parse `DurableExecutionInvocationOutput.Status`: `Succeeded`/`Failed` → terminal, store result/error, + stop. `Pending` → re-drive from step 1. **Serialize passes per ARN** so the abandoned-event resend + (`LambdaRuntimeAPI.cs:120-134`) can't double-deliver. + +**Timers.** Default: time-skip. Optional wall-clock mode: `Task.Delay` to the min pending +`WaitDetails.ScheduledEndTimestamp` (mirrors `ExecutionOrchestrator.cs:123-135`) to watch real backoff. + +**Callbacks.** Inbound endpoints are keyed by `{CallbackId}`, **not** ARN. Maintain a `CallbackId → ARN` +reverse map to wake the correct parked driver. The `onNewOperations` cb-id feedback survives the HTTP hop as +long as the checkpoint response's `NewExecutionState.Operations` is populated — `LambdaDurableServiceClient` +already forwards minted cb-ids back into `ExecutionState`. + +## 6. Reused vs new + +**Reused verbatim:** +- Runtime-API transport: `LambdaRuntimeAPI.cs:26-38,120,189`, `RuntimeApiDataStore.QueueEvent` (`:92`), + `EventContainer.WaitForCompletion` (`:86-96`), `RuntimeApiDataStoreManager` (`:40-52`). +- Durable state machine: `CheckpointProcessor` (incl. `ApplyTimeSkipping` `:273-298`, cb minting `:167-170`), + `InMemoryOperationStore`, `InMemoryDurableServiceClient`. +- Public STJ envelope POCOs: `DurableExecutionInvocationInput/Output`, `InitialExecutionState`, `Operation`. +- Hosting/DI seams: `TestToolProcess.cs:43-84,108`; catch-all routing `ApiGatewayEmulatorProcess.cs:86`; + task-list wiring `RunCommand.cs:49-97`. +- Stock AWSSDK.Lambda `AWS_ENDPOINT_URL_LAMBDA` resolution (verified §4). + +**New:** +- `Services/DurableExecution/DurableServiceApi.cs` — `SetupDurableServiceEndpoints`, route group B. +- `Services/DurableExecution/DurableRestJsonDtos.cs` + mapper — emulator-owned STJ DTOs ⇄ + `Amazon.Lambda.Model.OperationUpdate/Operation` (**the largest new component**; must track + `Action{START/SUCCEED/FAIL/RETRY/CANCEL}`, `Type`, `Status` against `lambda-2015-03-31.normal.json`). +- `Services/DurableExecution/DurableExecutionDriver` (+ `IDurableExecutionDriver`) — start + per-ARN drive + loop, `ConcurrentDictionary`, callback reverse-map, `Stop`. +- `DurableExecutionStore` singleton — owns `CheckpointProcessor` + `InMemoryOperationStore` keyed by ARN; + mints ARNs + rotates tokens. +- Extension of `PostEvent` (`LambdaRuntimeAPI.cs:51`) for the `X-Amz-Durable-Execution-Name` branch. +- `RunCommandSettings` flags: `--durable-execution`, `--durable-time-skip` (default true). +- New **IVT entry / public facade** on the `.Testing` assembly for the Test Tool (+ strong-name match). +- (Phase 3) `Components/Pages/DurableExecutionPanel.razor` — timeline + callback buttons + time control, + using the existing `StateChange → StateHasChanged` pattern (`Home.razor.cs:114,218`). + +## 7. Risks & open questions + +1. ~~**Endpoint redirection**~~ — **resolved by the Phase-0 spike (§4).** +2. **rest-json ⇄ SDK-model mapping fidelity.** `ConstantClass` blocks STJ; the hand-map across every nested + option (Step/Wait/Callback/ChainedInvoke/Context, Error, `ReplayChildren`) is the real cost and must stay + in sync with the service model. **The most under-estimated piece of work.** +3. **Chained durable-to-durable invokes are not v1.** Out-of-process, `ctx.Invoke` would loop forever on + `InvokePending`. **Detect and fail fast** with a clear "chained invoke not yet supported" error until Phase 4. +4. **IVT + strong-naming coupling.** Locks the Test Tool version to the 0.x `.Testing` package. Prefer a + small public facade over raw IVT. +5. **Checkpoint-before-suspend ordering.** The driver time-skips on store state, so `WrapAsyncCore` must + flush the WAIT/retry op over HTTP **before** posting Pending, not on async dispose after. Verify. +6. **15-min `WaitForCompletion` cap** at a breakpoint — dev-acceptable, but the driver needs a clean + abort/re-attach path. +7. **Payload caps / `ReplayChildren`** tiered enforcement (6 MB / 1 MB / 256 KB) — fidelity, deferrable past + first ship but required before "high-fidelity" is claimed. + +## 8. Phased plan + +- **Phase 0 — Spike: endpoint redirection (S).** ✅ **Done.** Verified §4. +- **Phase 1 — Data plane + backend reuse (M).** ✅ **Done.** Chose to **copy** `CheckpointProcessor` + + `InMemoryOperationStore` into the Test Tool (retyped to plain-STJ DTOs) rather than IVT/facade, keeping + it decoupled from the 0.x preview internals. `DurableServiceApi` catch-all endpoints; `DurableExecutionStore` + over the copied backend. Delivered under `Services/DurableExecution/`, gated by `--durable-execution`. + Tests (`DurableServiceApiTests`) exercise checkpoint/state round-trips through a real `AmazonLambdaClient`. + **Bug found & fixed:** the checkpoint/state *response* is read by the AWSSDK rest-json unmarshaller (not + System.Text.Json), whose `timestamp` members are unix **seconds** while the `Operation` POCO emits epoch + **millis** — the direct-POCO serialization overflowed `DateTime`. Fixed with dedicated `WireOperation` + response DTOs that convert millis→seconds (validates §7 risk #2). +- **Phase 2 — Start hook + drive loop (M).** ✅ **Done.** `X-Amz-Durable-Execution-Name` branch in + `PostEvent`; `DurableExecutionDriver` (QueueEvent + WaitForCompletion, per-ARN idempotency by name + + payload hash, paged envelope, time-skip default, fail-fast on chained invokes, park on pending callback). + Registered when `--durable-execution` is set. `DurableExecutionDriverTests` drives a wait-then-step + workflow end to end (start → Pending → time-skip → Succeeded) through the real Runtime API + SDK wire. + **Bug found & fixed:** the ARN's internal slashes arrive percent-encoded (`%2F`) and ASP.NET's catch-all + preserves them un-decoded, so the checkpoint handler keyed the store under the encoded ARN while the driver + reads by the raw ARN — the two never met. Fixed by `Uri.UnescapeDataString` on the captured ARN in + `DurableServiceApi.TrySplit`. (Phase 1 tests didn't catch it because checkpoint+getstate were both + SDK-encoded and thus internally consistent.) **First genuinely useful ship** — single-function workflows + headless, debugger-attached. +- **Phase 3 — Callbacks + Blazor UI (M).** Callback endpoints + `CallbackId → ARN` map + driver wake; + `DurableExecutionPanel.razor` timeline, Send-Callback buttons, time control. +- **Phase 4 — Chained invokes + fidelity polish (L).** Out-of-process nested drive for `CHAINED_INVOKE` + across registered sibling functions; tiered payload caps + `ReplayChildren`; idempotency; `StopDurableExecution`. + +## Appendix — Phase-0 spike source + +Kept at `C:\tmp\durable-endpoint-spike\` (throwaway; not checked in). `Program.cs` stands up an +`HttpListener`, sets only `AWS_ENDPOINT_URL_LAMBDA` + region + dummy creds, constructs +`new AmazonLambdaClient()`, and calls `CheckpointDurableExecutionAsync`. Reproduce with `dotnet run -c Release`. diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Amazon.Lambda.TestTool.csproj b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Amazon.Lambda.TestTool.csproj index b2ae28208..16c990bbb 100644 --- a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Amazon.Lambda.TestTool.csproj +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Amazon.Lambda.TestTool.csproj @@ -23,6 +23,11 @@ + + diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Commands/Settings/RunCommandSettings.cs b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Commands/Settings/RunCommandSettings.cs index f4ab3d57f..315775b6c 100644 --- a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Commands/Settings/RunCommandSettings.cs +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Commands/Settings/RunCommandSettings.cs @@ -97,6 +97,25 @@ public sealed class RunCommandSettings : CommandSettings [Description("The absolute path used to save global settings and saved requests. You will need to specify a path in order to enable saving global settings and requests.")] public string? ConfigStoragePath { get; set; } + /// + /// Enables the durable-execution service emulator: HTTP endpoints emulating the Lambda + /// durable-execution data plane (checkpoint / get-state) are hosted alongside the Runtime + /// API. Point the function's AWS_ENDPOINT_URL_LAMBDA at the emulator to run a durable + /// workflow locally. + /// + [CommandOption("--durable-execution")] + [Description("Enable the durable-execution service emulator (checkpoint/get-state endpoints) hosted alongside the Lambda Runtime API.")] + public bool DurableExecution { get; set; } + + /// + /// When the durable-execution emulator is enabled, resolves timers (WaitAsync, retry + /// backoff) immediately instead of waiting for wall-clock time. Defaults to true so + /// local workflows advance without real delays. + /// + [CommandOption("--durable-time-skip")] + [Description("When durable execution is enabled, resolve timers/retry backoff immediately rather than waiting for wall-clock time. Default: true.")] + public bool DurableTimeSkip { get; set; } = true; + /// /// Validate that is an absolute path. /// diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Processes/TestToolProcess.cs b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Processes/TestToolProcess.cs index 7dfe0f022..a9d488225 100644 --- a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Processes/TestToolProcess.cs +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Processes/TestToolProcess.cs @@ -5,6 +5,7 @@ using Amazon.Lambda.TestTool.Components; using Amazon.Lambda.TestTool.Models; using Amazon.Lambda.TestTool.Services; +using Amazon.Lambda.TestTool.Services.DurableExecution; using Amazon.Lambda.TestTool.Services.IO; using Amazon.Lambda.TestTool.Utilities; using Microsoft.Extensions.FileProviders; @@ -71,6 +72,14 @@ public static TestToolProcess Startup(RunCommandSettings settings, CancellationT builder.Services.AddSingleton(); builder.Services.AddSingleton(); + + if (settings.DurableExecution) + { + builder.Services.AddSingleton(new DurableExecutionStore(settings.DurableTimeSkip)); + builder.Services.AddSingleton(sp => new DurableExecutionDriver( + sp.GetRequiredService(), + sp.GetRequiredService())); + } builder.Services.AddSingleton(); builder.Services.AddSingleton(); @@ -161,6 +170,11 @@ public static TestToolProcess Startup(RunCommandSettings settings, CancellationT LambdaRuntimeApi.SetupLambdaRuntimeApiEndpoints(app); + if (settings.DurableExecution) + { + DurableServiceApi.SetupDurableServiceEndpoints(app); + } + var runTask = app.RunAsync(cancellationToken); var startup = new TestToolProcess diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/CheckpointProcessor.cs b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/CheckpointProcessor.cs new file mode 100644 index 000000000..1899b0994 --- /dev/null +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/CheckpointProcessor.cs @@ -0,0 +1,314 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using Amazon.Lambda.DurableExecution; + +namespace Amazon.Lambda.TestTool.Services.DurableExecution; + +/// +/// Processes checkpoint updates against the in-memory operation store. Handles +/// action-to-status mapping, callback ID minting, time skipping, and producing the +/// "new operations" the runtime expects back in the checkpoint response. +/// +/// +/// Ported from Amazon.Lambda.DurableExecution.Testing.CheckpointProcessor (which is +/// internal). The state-machine logic is intentionally identical; the only change is +/// the input type — a plain-STJ deserialized from the wire, +/// instead of the AWSSDK Amazon.Lambda.Model.OperationUpdate whose ConstantClass +/// members can't be deserialized by System.Text.Json. This drops the ?.Value access on +/// the enum-like members but leaves every branch and status transition unchanged. +/// +internal sealed class CheckpointProcessor +{ + private readonly InMemoryOperationStore _store; + private readonly bool _skipTime; + private readonly object _pendingGate = new(); + private readonly List _pendingInvokes = new(); + + public CheckpointProcessor(InMemoryOperationStore store, bool skipTime) + { + _store = store; + _skipTime = skipTime; + } + + /// + /// A chained-invoke (ctx.InvokeAsync) started by the workflow but not yet resolved. + /// The runtime suspends after emitting the START and expects an external system to run the + /// target function. Phase 1 records these but does not resolve them (chained durable + /// invokes are a Phase 4 feature); the driver detects a non-empty pending list and fails + /// fast. The target function name lives only on the wire-format + /// ChainedInvokeOptions, so it is captured here rather than on the persisted + /// . + /// + internal readonly record struct PendingInvoke(string OperationId, string FunctionName, string? Payload); + + /// Returns and clears the chained-invokes started since the last drain. + public IReadOnlyList DrainPendingInvokes() + { + lock (_pendingGate) + { + if (_pendingInvokes.Count == 0) + return Array.Empty(); + var drained = _pendingInvokes.ToArray(); + _pendingInvokes.Clear(); + return drained; + } + } + + /// + /// Processes a batch of updates and returns the new checkpoint token and any operations + /// created or modified (to feed back to the runtime's onNewOperations mechanism). + /// + public (string NewToken, IReadOnlyList NewOperations) Process( + string arn, + string? currentToken, + IReadOnlyList updates) + { + var newOperations = new List(); + + foreach (var update in updates) + { + var operation = ApplyUpdate(arn, update); + newOperations.Add(operation); + } + + var newToken = _store.IncrementToken(arn); + return (newToken, newOperations); + } + + private Operation ApplyUpdate(string arn, WireOperationUpdate update) + { + var existing = _store.GetOperation(arn, update.Id!); + var operation = existing ?? new Operation { Id = update.Id }; + + operation.Type = update.Type ?? operation.Type; + operation.Name = update.Name ?? operation.Name; + operation.ParentId = update.ParentId ?? operation.ParentId; + operation.SubType = update.SubType ?? operation.SubType; + + var action = update.Action; + ApplyAction(operation, action, update); + + if (_skipTime) + ApplyTimeSkipping(operation, action); + + // A chained-invoke START suspends the workflow until an external system resolves it. + // Record it so a future driver can run the registered sibling and stamp the + // result/error before the next replay. The function name is only carried on the + // wire-format update, not on the Operation. + if (action == "START" + && operation.Type == OperationTypes.ChainedInvoke + && update.ChainedInvokeOptions?.FunctionName is { } functionName) + { + lock (_pendingGate) + { + _pendingInvokes.Add(new PendingInvoke(operation.Id!, functionName, update.Payload)); + } + } + + _store.Upsert(arn, operation); + return operation; + } + + private static void ApplyAction(Operation operation, string? action, WireOperationUpdate update) + { + switch (action) + { + case "START": + operation.Status = OperationStatuses.Started; + operation.StartTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + ApplyStartDetails(operation, update); + break; + + case "SUCCEED": + operation.Status = OperationStatuses.Succeeded; + operation.EndTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + ApplySucceedDetails(operation, update); + break; + + case "FAIL": + operation.Status = OperationStatuses.Failed; + operation.EndTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + ApplyFailDetails(operation, update); + break; + + case "RETRY": + operation.Status = OperationStatuses.Pending; + ApplyRetryDetails(operation, update); + break; + + case "CANCEL": + operation.Status = OperationStatuses.Cancelled; + operation.EndTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + break; + } + } + + private static void ApplyStartDetails(Operation operation, WireOperationUpdate update) + { + switch (operation.Type) + { + case OperationTypes.Step: + operation.StepDetails ??= new StepDetails(); + // A plain step re-emits START before every attempt, so START owns the attempt + // count. WaitForCondition (Type=STEP, SubType=WaitForCondition) emits START + // only once and advances the count on each RETRY instead, so it must NOT + // increment here. + if (operation.SubType != OperationSubTypes.WaitForCondition) + operation.StepDetails.Attempt = (operation.StepDetails.Attempt ?? 0) + 1; + break; + + case OperationTypes.Wait: + operation.WaitDetails ??= new WaitDetails(); + if (update.WaitOptions?.WaitSeconds is { } seconds) + { + operation.WaitDetails.ScheduledEndTimestamp = + DateTimeOffset.UtcNow.AddSeconds(seconds).ToUnixTimeMilliseconds(); + } + break; + + case OperationTypes.Callback: + operation.CallbackDetails ??= new CallbackDetails(); + operation.CallbackDetails.CallbackId = $"cb-{operation.Id}"; + break; + + case OperationTypes.ChainedInvoke: + operation.ChainedInvokeDetails ??= new ChainedInvokeDetails(); + break; + + case OperationTypes.Context: + operation.ContextDetails ??= new ContextDetails(); + break; + + case OperationTypes.Execution: + operation.ExecutionDetails ??= new ExecutionDetails(); + break; + } + } + + private static void ApplySucceedDetails(Operation operation, WireOperationUpdate update) + { + var payload = update.Payload; + switch (operation.Type) + { + case OperationTypes.Step: + operation.StepDetails ??= new StepDetails(); + operation.StepDetails.Result = payload; + operation.StepDetails.Error = null; + break; + + case OperationTypes.ChainedInvoke: + operation.ChainedInvokeDetails ??= new ChainedInvokeDetails(); + operation.ChainedInvokeDetails.Result = payload; + operation.ChainedInvokeDetails.Error = null; + break; + + case OperationTypes.Context: + operation.ContextDetails ??= new ContextDetails(); + operation.ContextDetails.Result = payload; + operation.ContextDetails.Error = null; + break; + + case OperationTypes.Callback: + operation.CallbackDetails ??= new CallbackDetails(); + operation.CallbackDetails.Result = payload; + operation.CallbackDetails.Error = null; + break; + } + } + + private static void ApplyFailDetails(Operation operation, WireOperationUpdate update) + { + var error = MapWireError(update.Error); + switch (operation.Type) + { + case OperationTypes.Step: + operation.StepDetails ??= new StepDetails(); + operation.StepDetails.Error = error; + break; + + case OperationTypes.ChainedInvoke: + operation.ChainedInvokeDetails ??= new ChainedInvokeDetails(); + operation.ChainedInvokeDetails.Error = error; + break; + + case OperationTypes.Context: + operation.ContextDetails ??= new ContextDetails(); + operation.ContextDetails.Error = error; + break; + + case OperationTypes.Callback: + operation.CallbackDetails ??= new CallbackDetails(); + operation.CallbackDetails.Error = error; + break; + } + } + + private static void ApplyRetryDetails(Operation operation, WireOperationUpdate update) + { + // Both retried steps and WaitForCondition polls are wire-encoded as Type=STEP + // (WaitForCondition uses SubType=WaitForCondition); the runtime never emits a + // WAIT-typed RETRY, so this single STEP branch covers both. + if (operation.Type == OperationTypes.Step) + { + operation.StepDetails ??= new StepDetails(); + if (update.StepOptions?.NextAttemptDelaySeconds is { } delaySeconds) + { + operation.StepDetails.NextAttemptTimestamp = + DateTimeOffset.UtcNow.AddSeconds(delaySeconds).ToUnixTimeMilliseconds(); + } + operation.StepDetails.Error = MapWireError(update.Error); + + // WaitForCondition emits START once and advances per RETRY: it carries the next + // poll state in Payload and relies on the persistence layer to own the attempt + // count. Persist both so the next replay resumes from the latest state with an + // advanced attempt number (a plain step RETRY carries no Payload and owns its count + // via START, so leave it alone). + if (operation.SubType == OperationSubTypes.WaitForCondition) + { + if (update.Payload is not null) + operation.StepDetails.Result = update.Payload; + operation.StepDetails.Attempt = (operation.StepDetails.Attempt ?? 0) + 1; + } + } + } + + private void ApplyTimeSkipping(Operation operation, string? action) + { + if (action == "START" && operation.Type == OperationTypes.Wait) + { + operation.Status = OperationStatuses.Succeeded; + operation.EndTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + if (operation.WaitDetails != null) + { + operation.WaitDetails.ScheduledEndTimestamp = + DateTimeOffset.UtcNow.AddMilliseconds(-1).ToUnixTimeMilliseconds(); + } + } + + // A retried step (or WaitForCondition poll, also Type=STEP) becomes immediately READY + // under time-skipping so the next replay runs the next attempt without waiting for the + // backoff/poll delay. + if (action == "RETRY" && operation.Type == OperationTypes.Step) + { + operation.Status = OperationStatuses.Ready; + if (operation.StepDetails != null) + { + operation.StepDetails.NextAttemptTimestamp = + DateTimeOffset.UtcNow.AddMilliseconds(-1).ToUnixTimeMilliseconds(); + } + } + } + + private static ErrorObject? MapWireError(WireErrorObject? wireError) + { + if (wireError == null) return null; + return new ErrorObject + { + ErrorType = wireError.ErrorType, + ErrorMessage = wireError.ErrorMessage, + ErrorData = wireError.ErrorData, + StackTrace = wireError.StackTrace + }; + } +} diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableExecutionDriver.cs b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableExecutionDriver.cs new file mode 100644 index 0000000000000000000000000000000000000000..0c1157b2804ccd0916770fa7ca2b8c056dcb71f3 GIT binary patch literal 14120 zcmcIr>uwv{b?$FJ#g2LqkuW)OoU|xxUBF0kjuKl^O}Y+R6vl|NDNZ$V zimOtEi$$EoVWrDJO_D@?=IBy=)}=0PbrihX+f&oalRu9?#&eyPdVCh?w2Bw8F5an0 z9?q9~{5E*=>Q!CF>9v}!%Sx|;<19(^yo$55#P@Vw7X@a$?+%*j`)!YcKbBeQhFY>e zhO4V63{L7Iyh`-xy`I-xeEX=GE~{CVCBdgG(noN(~rK*-%6mtPD5O(PW*@vWAvSE|aeNTrKaoQ4U< zS6W@IRiSf@uTipAaeABGPMySsYXs>S~-V#(9xJLOffT zWY@Lu5<7VuToiy|SruWG72Fp`lIcLTkOrCFgx7ZS8P>&OL@^i=#fyaoIOqDE*|D{h zBIgYucUf_>NU}S18;2kR8+f9Q|6T!R`WoCE$?5KvSTN)~oI*qaRU?m$NHF?VCt0ow zNE`!`5fG4vAYM0OR!+o})qmEyCWpgR*2na;!itbAHF&?&TFnc+IM_L}0-EIUNmzwb zutV?c9jd`+jWx;&S}jsgD`>`0L>S(MvGB+V-$_ci;i<^85?ft?k9b~W_4QKTw*V0+ zGM49@hty4_Q*%tAdC;P=uf=D6&RLq*6$mz@fs#6@eqj1VUHSQirWOGQ&@RY1{)L7^ zU0xhtlp}SwjDb~1=WZFA4OXF(4p5YWOGTy1yD(Qcco}1=09RA-_H$@L?nW#Nkeii1!KpihZyM`uP zYf)um`_)5f01cQ|_W{{=qbKD$ofG@u9nPL+6*erh1U6g|kLrK^>%S>7*~5Y|(iM!f zwG*@(i!io$_XY*e%1RB)p??R;ta^)X5Dz5}KvbBen~Ag)iJ*jPB+!9n$TI*JyfF*@ zc8GOSQd+~T#FaA--XMsD5yJ&V!BOT(2s_Y6&I(U_Gy4&%eQw z(2BQ>gnjqKY7m0@whHeXMWweSt+3^PsKX*EjUyV1cZegs%CibKbuA_a8bz4EWG)g$ z7ki61jmssZ>A_M4+t1O&4kxjAJUG{V7<%a3NO|-BScWvl3jHs^? zV7D)MgY$t5A^ZF94Kf56BR}XlEIbW`W|j!fa|5{(nX0t z2L3N-#VsUiyK(ZuHLw&pJa}p3MzRGQa$EpT2Fz+wr0PLk>*_sLjq(4-@6GsNV?aUA z^HXoL1f~e?&y!}XulfH$?{NU9&PcOizNOiadc@oSc5vhMX;Eaw#npenr5q{wu`$H0 z@Bx0NF7#(aNetr2OMX0Wti_&%<;`)n$`b^bu-#c*%=NxMgVXwEggWp`aXA_QSpo{o z1+01J`CwNN^wjdW6xWHMCe7+HSq}|87D`^kI;~edMnPbfRU>r?AE={?^w_$(k($=? zxz>@6MvBLUsO^%@KY9CJi1{KC>UpL5CIV3e%^eYfK+$}^4A)5(Mn8w;66DIEyXgmy z&nL1Nye*SGX8w9I3@v+G1R@h!*OU|-_Nzg0yc*3+ti7%|6VM1 z^vYMD_uuwuRjW&f5oRthoE{fAtUabe6#fIt#{yvU^bA89ZgJCesrZR0X zRPP!FaOXy^xrg1JkcG*o&eoOYuYn`1JDxQCNv|>Xr=5rPIUoP-f$tu7EbxsWyZ&9I zA*Xt6BxCmg?%T+WO?C)izV|XA9Np-+6dBtc5T|MaRwC(DSRszzfbI*yV-o#vj(C@g z^FtmvkQ}(EAozo>E{aJM4JYF++u;8QzKiw@6!jE9O_}}0YtQVA^7wvEawx_sDJCn*gA>f59Y=R(1 z%}+a%Emn-DjnuBe%&yc_9GdXdeLW|lc7IG`cUnxerk$j~>V6cfDORmzGxa?L4%>@z z7;;kt;=o-a|5ohUNZs)hLL4EjgJ_?Uy_NVB9>*KNL0fb!Vo4?8X8n#t%8hF?ZWX=_ zn*vI*#)o#Sao}qcJ8^dd#7VjsDmow=5V#1iANyIW%%g6G zBww-rhpcEK6LWg&y}?prg~>DUr!D$fv4+n7OPJK!hz}L8mav+6oO@(Nh!U#{k^&Oc z5y(3ujt@n)I(4pYgW^qXqAftahssHG0kBG>~|k<(#PFk zhw>JhIzh^@DGECnO0_fP#1U-?{k``+EY;kk*y-9>137>CkL z^tU<0iMr`rcNNatT!~K^S=)1s5GZ(d13D)7793&ic!lsC8Q(*1qsP{nQ4U0OMunSO zeL^&(cGPQsYHqppvm-_8x~viYuhmMUgeW_^Vjuy`k{rs65?~bBHELc6%tJg*$1KFL zK+k|=uG}Q#qHF-rnC-#*p?JvHdkB4czm+0mA~)Z)oRcG-!glppR=yzV-`6NhHWi+B zU5u8UZB?Y3vtlnDBwOVvaWcC7flv z&@zQ%sY`W$l$xY5%VZh12+WJX=DjaPH~7ok!UZWvP!KP}XAKaF*69)zTc(vuu+K!M zBf(gc%YB%feLOvZQDMG5++;2f(9LSL;v5xDm2HkfF+U}lVCyY3mY#z zvvq2a31A3m8Oj6mds8RLU_0fB0+pxW*~#sos8~-myI&x{#h_2kWrDc*Zi@|l8Mi2K z9sMAyq%gN!J< z&+6ojA*6u@>s(*X)@12nIao6=<$t$_t@NA5=vuXPDsz#y)3mIkL9VNTgUb-*p zow7yBOBQUg+G`ff{Gb$&GHxk6<-~r#K>?N%CuT_^r31P>G-QZJqVm*DY<33Bx5|OH@(Nr%& zm?@*?e;ssfLeK&(h#5SCNyhzK!Wo|H$544e%XJsy4g(d?B^%68tl|=exH5#?vGCUx z$ct>T2-H^=u=#FW-o!a9-}MH&flRC*75m8a%!k6}QnHvBDogcqmGMQEph-feN}rWj zC0$a%MDaRPcY50n=wO)ITC1h7jXETKixDsAgllW{xq8a07bQ9yQ4R0BeSJ|P?9cJi z^q{p#cQFsAmZ!&;U!-ibL=wYM-!uHwEmN5|$NmgQke-vWZmCCz9jagd+LC>!X5jZU z57XdZb&>r)sqq&MF*Xlg88p#lB{aGkI zVm%amx|lp(>iLb_-I^g_Gcuryjvc@qDfE4 z2tqN-`deVe>;m-x=OirYzf6l|6GraA6WGv(#zmr!F<(PyaPQl6S^2-+lu*C<3}-B)cVz|&G^J@=6z zJ*Y>8UIlb?;6-mM!0Io5;j4X*iQ%S#`Z)&}+E!72IBw`i*_E4hx+0c%9tOox +/// Root of the durable-execution service emulator. Owns one operation store and one checkpoint +/// processor per durable execution (keyed by ARN), mints execution ARNs, and serves paged +/// state reads. +/// +/// +/// Registered as a singleton in TestToolProcess so the checkpoint/state HTTP endpoints +/// (see ) and — in a later phase — the re-invocation driver +/// share a single backing store. Phase 1 exposes only the data-plane operations the running +/// function calls (CheckpointDurableExecution / GetDurableExecutionState); the +/// start hook and drive loop arrive in Phase 2. +/// +internal sealed class DurableExecutionStore +{ + /// + /// Max operations returned in a single GetDurableExecutionState page. The real + /// service pages large histories; a small page here exercises the SDK's paging path + /// (NextMarker follow-up calls) during local testing. + /// + internal const int StatePageSize = 100; + + private readonly bool _skipTime; + private readonly ConcurrentDictionary _executions = new(); + private long _executionCounter; + + public DurableExecutionStore(bool skipTime) + { + _skipTime = skipTime; + } + + private sealed class ExecutionContext + { + public required InMemoryOperationStore Store { get; init; } + public required CheckpointProcessor Processor { get; init; } + } + + private ExecutionContext GetOrCreate(string arn) + { + return _executions.GetOrAdd(arn, _ => + { + var store = new InMemoryOperationStore(); + return new ExecutionContext + { + Store = store, + Processor = new CheckpointProcessor(store, _skipTime) + }; + }); + } + + /// True once an execution with this ARN has been checkpointed at least once. + public bool Exists(string arn) => _executions.ContainsKey(arn); + + /// + /// Mints a synthetic durable-execution ARN for a newly-started execution. The shape mirrors + /// the service's (arn:…:function:NAME:QUALIFIER/durable-execution/GROUP/NAME) so the + /// SDK's ARN handling and the emulator's own routing exercise the real slash-bearing form. + /// + public string MintArn(string functionName, string executionName, string? qualifier = null) + { + var n = Interlocked.Increment(ref _executionCounter); + var qual = string.IsNullOrEmpty(qualifier) ? "$LATEST" : qualifier; + var group = "local"; + var name = string.IsNullOrEmpty(executionName) ? $"exec-{n}" : executionName; + return $"arn:aws:lambda:us-east-1:000000000000:function:{functionName}:{qual}/durable-execution/{group}/{name}"; + } + + /// + /// Applies a batch of checkpoint updates. Returns the rotated checkpoint token and the + /// operations created/modified by this batch (the NewExecutionState.Operations the + /// SDK merges back into its in-memory state — carrying, e.g., freshly minted callback IDs). + /// + public (string NewToken, IReadOnlyList NewOperations) Checkpoint( + string arn, + string? checkpointToken, + IReadOnlyList updates) + { + var ctx = GetOrCreate(arn); + return ctx.Processor.Process(arn, checkpointToken, updates); + } + + /// + /// Returns one page of the execution's operation history starting at + /// (an integer offset encoded as a string; null/empty = start). The returned NextMarker is + /// null when the page reaches the end of the history. + /// + public (IReadOnlyList Operations, string? NextMarker) GetState(string arn, string? marker) + { + var ctx = GetOrCreate(arn); + var all = ctx.Store.GetAllOperations(arn); + + var offset = 0; + if (!string.IsNullOrEmpty(marker) && int.TryParse(marker, out var parsed) && parsed > 0) + offset = parsed; + + if (offset >= all.Count) + return (Array.Empty(), null); + + var page = all.Skip(offset).Take(StatePageSize).ToList(); + var nextOffset = offset + page.Count; + var nextMarker = nextOffset < all.Count ? nextOffset.ToString() : null; + return (page, nextMarker); + } + + /// Current checkpoint token for the execution (creates an empty execution if new). + public string CurrentToken(string arn) => GetOrCreate(arn).Store.CurrentToken(arn); + + /// + /// Seeds the top-level EXECUTION operation carrying the user input payload. Idempotent: + /// re-driving (e.g. a replay pass) must not reset the op or clobber recorded state. Mirrors + /// ExecutionOrchestrator.SeedExecutionOperation (operation id exec-0). + /// + public void SeedExecution(string arn, string? inputPayload) + { + var ctx = GetOrCreate(arn); + if (ctx.Store.GetOperation(arn, ExecutionOperationId) is not null) + return; + + ctx.Store.Upsert(arn, new Operation + { + Id = ExecutionOperationId, + Type = OperationTypes.Execution, + Status = OperationStatuses.Started, + ExecutionDetails = new ExecutionDetails { InputPayload = inputPayload } + }); + } + + /// The operation id of the seeded top-level EXECUTION op. + internal const string ExecutionOperationId = "exec-0"; + + /// Snapshot of all operations recorded for the execution. + public IReadOnlyList GetAllOperations(string arn) => GetOrCreate(arn).Store.GetAllOperations(arn); + + /// Chained-invokes started but not resolved (Phase 4). Non-empty ⇒ unsupported workflow. + public IReadOnlyList DrainPendingInvokes(string arn) + => GetOrCreate(arn).Processor.DrainPendingInvokes(); +} diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableRestJsonModels.cs b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableRestJsonModels.cs new file mode 100644 index 000000000..906259882 --- /dev/null +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableRestJsonModels.cs @@ -0,0 +1,301 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using System.Text.Json.Serialization; +using SdkOperation = Amazon.Lambda.DurableExecution.Operation; + +namespace Amazon.Lambda.TestTool.Services.DurableExecution; + +// Plain System.Text.Json models for the durable-execution data-plane request bodies. +// +// The AWSSDK.Lambda types (Amazon.Lambda.Model.OperationUpdate et al.) can't be used to +// deserialize the inbound wire JSON: their Action/Type/Status properties are AWSSDK +// ConstantClass values with no STJ contract. So the emulator owns these plain DTOs for the +// *inbound* direction. Property names mirror lambda-2015-03-31.normal.json exactly. +// +// Responses go the other way: we serialize the public Amazon.Lambda.DurableExecution.Operation +// type, whose JsonPropertyName attributes already match the wire, so no response DTOs are needed. + +/// Wire body for POST /2025-12-01/durable-executions/{arn}/checkpoint. +internal sealed class CheckpointRequestBody +{ + [JsonPropertyName("CheckpointToken")] + public string? CheckpointToken { get; set; } + + [JsonPropertyName("ClientToken")] + public string? ClientToken { get; set; } + + [JsonPropertyName("Updates")] + public List Updates { get; set; } = new(); + + // DurableExecutionArn rides in the URL path, not the body. +} + +/// Wire body for POST /2025-12-01/durable-execution-callbacks/{callbackId}/{succeed|fail}. +internal sealed class CallbackRequestBody +{ + [JsonPropertyName("Payload")] + public string? Payload { get; set; } + + [JsonPropertyName("Error")] + public WireErrorObject? Error { get; set; } +} + +/// +/// Plain-STJ mirror of Amazon.Lambda.Model.OperationUpdate. Enum-like members +/// (, , ) are plain strings — +/// the emulator's checkpoint processor compares them against the +/// Amazon.Lambda.DurableExecution.OperationTypes/OperationStatuses constants. +/// +internal sealed class WireOperationUpdate +{ + [JsonPropertyName("Id")] + public string? Id { get; set; } + + [JsonPropertyName("ParentId")] + public string? ParentId { get; set; } + + [JsonPropertyName("Name")] + public string? Name { get; set; } + + [JsonPropertyName("Type")] + public string? Type { get; set; } + + [JsonPropertyName("SubType")] + public string? SubType { get; set; } + + [JsonPropertyName("Action")] + public string? Action { get; set; } + + [JsonPropertyName("Payload")] + public string? Payload { get; set; } + + [JsonPropertyName("Error")] + public WireErrorObject? Error { get; set; } + + [JsonPropertyName("ContextOptions")] + public WireContextOptions? ContextOptions { get; set; } + + [JsonPropertyName("StepOptions")] + public WireStepOptions? StepOptions { get; set; } + + [JsonPropertyName("WaitOptions")] + public WireWaitOptions? WaitOptions { get; set; } + + [JsonPropertyName("CallbackOptions")] + public WireCallbackOptions? CallbackOptions { get; set; } + + [JsonPropertyName("ChainedInvokeOptions")] + public WireChainedInvokeOptions? ChainedInvokeOptions { get; set; } +} + +internal sealed class WireStepOptions +{ + [JsonPropertyName("NextAttemptDelaySeconds")] + public int? NextAttemptDelaySeconds { get; set; } +} + +internal sealed class WireWaitOptions +{ + [JsonPropertyName("WaitSeconds")] + public long? WaitSeconds { get; set; } +} + +internal sealed class WireCallbackOptions +{ + [JsonPropertyName("TimeoutSeconds")] + public long? TimeoutSeconds { get; set; } + + [JsonPropertyName("HeartbeatTimeoutSeconds")] + public long? HeartbeatTimeoutSeconds { get; set; } +} + +internal sealed class WireChainedInvokeOptions +{ + [JsonPropertyName("FunctionName")] + public string? FunctionName { get; set; } + + [JsonPropertyName("TenantId")] + public string? TenantId { get; set; } +} + +internal sealed class WireContextOptions +{ + [JsonPropertyName("ReplayChildren")] + public bool? ReplayChildren { get; set; } +} + +internal sealed class WireErrorObject +{ + [JsonPropertyName("ErrorType")] + public string? ErrorType { get; set; } + + [JsonPropertyName("ErrorMessage")] + public string? ErrorMessage { get; set; } + + [JsonPropertyName("ErrorData")] + public string? ErrorData { get; set; } + + [JsonPropertyName("StackTrace")] + public List? StackTrace { get; set; } + + public static WireErrorObject? From(Amazon.Lambda.DurableExecution.ErrorObject? error) + { + if (error is null) return null; + return new WireErrorObject + { + ErrorType = error.ErrorType, + ErrorMessage = error.ErrorMessage, + ErrorData = error.ErrorData, + StackTrace = error.StackTrace?.ToList() + }; + } +} + +// ---- Response bodies ------------------------------------------------------------------------- +// Serialized back to the SDK, which unmarshals them with the AWSSDK rest-json reader (NOT +// System.Text.Json). Two consequences drive these DTOs rather than reusing the public +// Amazon.Lambda.DurableExecution.Operation POCO: +// 1. rest-json `timestamp` members are unix SECONDS (a JSON number), whereas the POCO +// serializes timestamps as epoch-MILLIS longs — the SDK would read 1.7e12 as seconds and +// overflow DateTime. So timestamps here are doubles carrying seconds. +// 2. Everything else (enum-like strings, payloads, nested details) marshals cleanly by name. +// WireOperation.From maps the emulator's millis-based Operation into this wire shape. + +/// Response body for CheckpointDurableExecution. +internal sealed class CheckpointResponseBody +{ + [JsonPropertyName("CheckpointToken")] + public string? CheckpointToken { get; set; } + + [JsonPropertyName("NewExecutionState")] + public NewExecutionStateBody? NewExecutionState { get; set; } +} + +/// The CheckpointUpdatedExecutionState shape returned inside a checkpoint response. +internal sealed class NewExecutionStateBody +{ + [JsonPropertyName("Operations")] + public List Operations { get; set; } = new(); + + [JsonPropertyName("NextMarker")] + public string? NextMarker { get; set; } +} + +/// Response body for GetDurableExecutionState. +internal sealed class StateResponseBody +{ + [JsonPropertyName("Operations")] + public List Operations { get; set; } = new(); + + [JsonPropertyName("NextMarker")] + public string? NextMarker { get; set; } +} + +/// +/// Wire shape of an Operation for outbound (response) serialization. Mirrors +/// Amazon.Lambda.Model.Operation so the AWSSDK rest-json unmarshaller reads it faithfully; +/// timestamps are unix seconds (doubles), converted from the emulator's epoch-millis store. +/// +internal sealed class WireOperation +{ + [JsonPropertyName("Id")] public string? Id { get; set; } + [JsonPropertyName("Type")] public string? Type { get; set; } + [JsonPropertyName("Status")] public string? Status { get; set; } + [JsonPropertyName("Name")] public string? Name { get; set; } + [JsonPropertyName("ParentId")] public string? ParentId { get; set; } + [JsonPropertyName("SubType")] public string? SubType { get; set; } + [JsonPropertyName("StartTimestamp")] public double? StartTimestamp { get; set; } + [JsonPropertyName("EndTimestamp")] public double? EndTimestamp { get; set; } + [JsonPropertyName("StepDetails")] public WireStepDetails? StepDetails { get; set; } + [JsonPropertyName("WaitDetails")] public WireWaitDetails? WaitDetails { get; set; } + [JsonPropertyName("ExecutionDetails")] public WireExecutionDetails? ExecutionDetails { get; set; } + [JsonPropertyName("CallbackDetails")] public WireCallbackDetails? CallbackDetails { get; set; } + [JsonPropertyName("ChainedInvokeDetails")] public WireChainedInvokeDetails? ChainedInvokeDetails { get; set; } + [JsonPropertyName("ContextDetails")] public WireContextDetails? ContextDetails { get; set; } + + // Epoch millis (store) → unix seconds (wire). rest-json timestamps are seconds. + private static double? ToSeconds(long? millis) => millis.HasValue ? millis.Value / 1000.0 : null; + + public static WireOperation From(SdkOperation op) => new() + { + Id = op.Id, + Type = op.Type, + Status = op.Status, + Name = op.Name, + ParentId = op.ParentId, + SubType = op.SubType, + StartTimestamp = ToSeconds(op.StartTimestamp), + EndTimestamp = ToSeconds(op.EndTimestamp), + StepDetails = op.StepDetails is null ? null : new WireStepDetails + { + Result = op.StepDetails.Result, + Error = WireErrorObject.From(op.StepDetails.Error), + Attempt = op.StepDetails.Attempt, + NextAttemptTimestamp = ToSeconds(op.StepDetails.NextAttemptTimestamp) + }, + WaitDetails = op.WaitDetails is null ? null : new WireWaitDetails + { + ScheduledEndTimestamp = ToSeconds(op.WaitDetails.ScheduledEndTimestamp) + }, + ExecutionDetails = op.ExecutionDetails is null ? null : new WireExecutionDetails + { + InputPayload = op.ExecutionDetails.InputPayload + }, + CallbackDetails = op.CallbackDetails is null ? null : new WireCallbackDetails + { + CallbackId = op.CallbackDetails.CallbackId, + Result = op.CallbackDetails.Result, + Error = WireErrorObject.From(op.CallbackDetails.Error) + }, + ChainedInvokeDetails = op.ChainedInvokeDetails is null ? null : new WireChainedInvokeDetails + { + Result = op.ChainedInvokeDetails.Result, + Error = WireErrorObject.From(op.ChainedInvokeDetails.Error) + }, + ContextDetails = op.ContextDetails is null ? null : new WireContextDetails + { + Result = op.ContextDetails.Result, + Error = WireErrorObject.From(op.ContextDetails.Error), + ReplayChildren = op.ContextDetails.ReplayChildren + } + }; +} + +internal sealed class WireStepDetails +{ + [JsonPropertyName("Result")] public string? Result { get; set; } + [JsonPropertyName("Error")] public WireErrorObject? Error { get; set; } + [JsonPropertyName("Attempt")] public int? Attempt { get; set; } + [JsonPropertyName("NextAttemptTimestamp")] public double? NextAttemptTimestamp { get; set; } +} + +internal sealed class WireWaitDetails +{ + [JsonPropertyName("ScheduledEndTimestamp")] public double? ScheduledEndTimestamp { get; set; } +} + +internal sealed class WireExecutionDetails +{ + [JsonPropertyName("InputPayload")] public string? InputPayload { get; set; } +} + +internal sealed class WireCallbackDetails +{ + [JsonPropertyName("CallbackId")] public string? CallbackId { get; set; } + [JsonPropertyName("Result")] public string? Result { get; set; } + [JsonPropertyName("Error")] public WireErrorObject? Error { get; set; } +} + +internal sealed class WireChainedInvokeDetails +{ + [JsonPropertyName("Result")] public string? Result { get; set; } + [JsonPropertyName("Error")] public WireErrorObject? Error { get; set; } +} + +internal sealed class WireContextDetails +{ + [JsonPropertyName("Result")] public string? Result { get; set; } + [JsonPropertyName("Error")] public WireErrorObject? Error { get; set; } + [JsonPropertyName("ReplayChildren")] public bool? ReplayChildren { get; set; } +} diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableServiceApi.cs b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableServiceApi.cs new file mode 100644 index 000000000..faf417e1f --- /dev/null +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/DurableServiceApi.cs @@ -0,0 +1,180 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using System.Net; +using System.Text; +using System.Text.Json; + +namespace Amazon.Lambda.TestTool.Services.DurableExecution; + +/// +/// Emulates the AWS Lambda durable-execution data plane as HTTP endpoints hosted alongside the +/// Lambda Runtime API on the same web app. A durable function whose IAmazonLambda client +/// is redirected here (via AWS_ENDPOINT_URL_LAMBDA) checkpoints and reads execution state +/// against these routes instead of the real service. +/// +/// +/// Routes mirror lambda-2015-03-31.normal.json (API version segment 2025-12-01): +/// +/// POST /2025-12-01/durable-executions/{arn}/checkpoint +/// GET /2025-12-01/durable-executions/{arn}/state +/// +/// The DurableExecutionArn contains slashes +/// (arn:…:function:NAME:QUALIFIER/durable-execution/GROUP/NAME) and arrives URL-encoded, +/// so a catch-all route captures the whole tail and the suffix (/checkpoint, /state) +/// is split off in code — mirroring the API Gateway emulator's {**catchAll} pattern. +/// +internal sealed class DurableServiceApi +{ + internal const string ApiVersion = "2025-12-01"; + private const string RoutePrefix = "/" + ApiVersion + "/durable-executions"; + + // Case-insensitive to match the SDK's PascalCase members regardless of any future casing drift. + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web) + { + PropertyNameCaseInsensitive = true, + DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull + }; + + private readonly DurableExecutionStore _store; + + private DurableServiceApi(WebApplication app) + { + _store = app.Services.GetRequiredService(); + + // A single catch-all captures "{arn-with-slashes}/{suffix}". ASP.NET Core decodes %2F in + // {**tail} back to '/', so the raw ARN is reconstructed transparently. + app.MapPost(RoutePrefix + "/{**tail}", HandleDurableExecutionPost); + app.MapGet(RoutePrefix + "/{**tail}", HandleDurableExecutionGet); + } + + public static void SetupDurableServiceEndpoints(WebApplication app) + { + _ = new DurableServiceApi(app); + } + + private async Task HandleDurableExecutionPost(HttpContext ctx, string tail) + { + if (!TrySplit(tail, "checkpoint", out var arn)) + { + await WriteErrorAsync(ctx, HttpStatusCode.NotFound, "ResourceNotFoundException", + $"Unsupported durable-executions POST path: '{tail}'."); + return; + } + + await HandleCheckpointAsync(ctx, arn); + } + + private async Task HandleDurableExecutionGet(HttpContext ctx, string tail) + { + if (!TrySplit(tail, "state", out var arn)) + { + await WriteErrorAsync(ctx, HttpStatusCode.NotFound, "ResourceNotFoundException", + $"Unsupported durable-executions GET path: '{tail}'."); + return; + } + + await HandleGetStateAsync(ctx, arn); + } + + private async Task HandleCheckpointAsync(HttpContext ctx, string arn) + { + using var reader = new StreamReader(ctx.Request.Body); + var body = await reader.ReadToEndAsync(); + + CheckpointRequestBody? request; + try + { + request = JsonSerializer.Deserialize(body, JsonOptions); + } + catch (JsonException ex) + { + await WriteErrorAsync(ctx, HttpStatusCode.BadRequest, "InvalidRequestContentException", + $"Could not parse checkpoint request body: {ex.Message}"); + return; + } + + if (request is null) + { + await WriteErrorAsync(ctx, HttpStatusCode.BadRequest, "InvalidRequestContentException", + "Checkpoint request body was empty."); + return; + } + + var (newToken, newOps) = _store.Checkpoint(arn, request.CheckpointToken, request.Updates); + + var response = new CheckpointResponseBody + { + CheckpointToken = newToken, + NewExecutionState = new NewExecutionStateBody + { + Operations = newOps.Select(WireOperation.From).ToList(), + NextMarker = null + } + }; + + await WriteJsonAsync(ctx, HttpStatusCode.OK, response); + } + + private async Task HandleGetStateAsync(HttpContext ctx, string arn) + { + var marker = ctx.Request.Query.TryGetValue("Marker", out var m) ? m.ToString() : null; + + var (operations, nextMarker) = _store.GetState(arn, marker); + + var response = new StateResponseBody + { + Operations = operations.Select(WireOperation.From).ToList(), + NextMarker = nextMarker + }; + + await WriteJsonAsync(ctx, HttpStatusCode.OK, response); + } + + /// + /// Splits a catch-all tail of the form "{arn}/{suffix}" into its ARN and validates the + /// suffix. The ARN itself contains slashes, so we match on the LAST segment. + /// + private static bool TrySplit(string tail, string expectedSuffix, out string arn) + { + arn = string.Empty; + if (string.IsNullOrEmpty(tail)) + return false; + + var lastSlash = tail.LastIndexOf('/'); + if (lastSlash <= 0 || lastSlash == tail.Length - 1) + return false; + + var suffix = tail[(lastSlash + 1)..]; + if (!string.Equals(suffix, expectedSuffix, StringComparison.Ordinal)) + return false; + + // The ARN's own slashes arrive percent-encoded (%2F) and ASP.NET's catch-all preserves + // them un-decoded, whereas the driver keys the store by the raw ARN. Unescape so both + // sides agree on the key. The suffix separator itself is a genuine '/', so the split + // point above is unaffected. + arn = Uri.UnescapeDataString(tail[..lastSlash]); + return true; + } + + private static async Task WriteJsonAsync(HttpContext ctx, HttpStatusCode status, T payload) + { + var json = JsonSerializer.Serialize(payload, JsonOptions); + var bytes = Encoding.UTF8.GetBytes(json); + ctx.Response.StatusCode = (int)status; + ctx.Response.ContentType = "application/json"; + ctx.Response.ContentLength = bytes.Length; + await ctx.Response.Body.WriteAsync(bytes); + } + + private static async Task WriteErrorAsync(HttpContext ctx, HttpStatusCode status, string errorType, string message) + { + var bytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(new { message }, JsonOptions)); + ctx.Response.StatusCode = (int)status; + ctx.Response.ContentType = "application/json"; + // AWSSDK reads the error type from this header (rest-json protocol). + ctx.Response.Headers["X-Amzn-Errortype"] = errorType; + ctx.Response.ContentLength = bytes.Length; + await ctx.Response.Body.WriteAsync(bytes); + } +} diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/InMemoryOperationStore.cs b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/InMemoryOperationStore.cs new file mode 100644 index 000000000..43ebc81da --- /dev/null +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/DurableExecution/InMemoryOperationStore.cs @@ -0,0 +1,112 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using Amazon.Lambda.DurableExecution; + +namespace Amazon.Lambda.TestTool.Services.DurableExecution; + +/// +/// In-memory store for operations recorded during a durable execution. Each execution +/// (keyed by ARN) maintains its own isolated operation set and checkpoint token. +/// +/// +/// Ported from Amazon.Lambda.DurableExecution.Testing.InMemoryOperationStore (which is +/// internal). The logic is intentionally identical; it operates on the public +/// wire type so the emulator can serialize state responses directly. +/// +internal sealed class InMemoryOperationStore +{ + private readonly Dictionary _executions = new(); + + // A single lock guards both the per-ARN dictionary and each ExecutionData's collections. + // Writes are normally serialized by the runtime's single-reader checkpoint batcher, but + // parallel/map workflows and the snapshot reads below can interleave, so we lock to keep + // the Dictionary/List internally consistent. + private readonly object _gate = new(); + + public string CurrentToken(string arn) + { + lock (_gate) + return GetOrCreate(arn).CheckpointToken; + } + + /// + /// Returns a snapshot copy of the operations for the execution. The copy is detached from + /// the backing list so callers can iterate safely while the store continues to mutate. + /// + public IReadOnlyList GetAllOperations(string arn) + { + lock (_gate) + return GetOrCreate(arn).Operations.ToList(); + } + + public Operation? GetOperation(string arn, string operationId) + { + lock (_gate) + { + var data = GetOrCreate(arn); + return data.OperationMap.TryGetValue(operationId, out var op) ? op : null; + } + } + + public void Upsert(string arn, Operation operation) + { + lock (_gate) + { + var data = GetOrCreate(arn); + if (data.OperationMap.TryGetValue(operation.Id!, out var existing)) + { + var index = data.Operations.IndexOf(existing); + data.Operations[index] = operation; + data.OperationMap[operation.Id!] = operation; + } + else + { + data.Operations.Add(operation); + data.OperationMap[operation.Id!] = operation; + } + } + } + + public string IncrementToken(string arn) + { + lock (_gate) + { + var data = GetOrCreate(arn); + data.TokenCounter++; + data.CheckpointToken = data.TokenCounter.ToString(); + return data.CheckpointToken; + } + } + + public int OperationCount(string arn) + { + lock (_gate) + return GetOrCreate(arn).Operations.Count; + } + + /// True once an execution with this ARN has been created. + public bool Exists(string arn) + { + lock (_gate) + return _executions.ContainsKey(arn); + } + + private ExecutionData GetOrCreate(string arn) + { + if (!_executions.TryGetValue(arn, out var data)) + { + data = new ExecutionData(); + _executions[arn] = data; + } + return data; + } + + private sealed class ExecutionData + { + public readonly List Operations = new(); + public readonly Dictionary OperationMap = new(); + public string CheckpointToken = "0"; + public int TokenCounter; + } +} diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/LambdaRuntimeAPI.cs b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/LambdaRuntimeAPI.cs index b05bb85ea..69916fa25 100644 --- a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/LambdaRuntimeAPI.cs +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Services/LambdaRuntimeAPI.cs @@ -3,6 +3,7 @@ using System.Text; using Amazon.Lambda.TestTool.Models; +using Amazon.Lambda.TestTool.Services.DurableExecution; using Microsoft.AspNetCore.Mvc; namespace Amazon.Lambda.TestTool.Services; @@ -14,11 +15,19 @@ public class LambdaRuntimeApi private const int MaxRequestSize = 6 * 1024 * 1024; private const int MaxResponseSize = 6 * 1024 * 1024; + // Header the SDK's Invoke sets to start a durable execution, and the header it reads the + // resulting ARN back from (rest-json locationName; see lambda-2015-03-31.normal.json). + private const string DurableExecutionNameHeader = "X-Amz-Durable-Execution-Name"; + private const string DurableExecutionArnHeader = "X-Amz-Durable-Execution-Arn"; + private readonly IRuntimeApiDataStoreManager _runtimeApiDataStoreManager; + // Present only when the durable-execution emulator is enabled (--durable-execution). + private readonly DurableExecutionDriver? _durableDriver; internal LambdaRuntimeApi(WebApplication app) { _runtimeApiDataStoreManager = app.Services.GetRequiredService(); + _durableDriver = app.Services.GetService(); app.MapGet("/lambda-runtime-api/healthcheck", () => "health"); @@ -73,6 +82,17 @@ public async Task PostEvent(HttpContext ctx, string functionName) return; } + // Durable-execution start: when the invoke carries X-Amz-Durable-Execution-Name and the + // emulator is enabled, the durable service (not the function directly) owns this call. It + // starts the execution asynchronously and returns the ARN; the driver then invokes the + // function itself, replay-style, until the workflow reaches a terminal state. + if (_durableDriver is not null + && ctx.Request.Headers.TryGetValue(DurableExecutionNameHeader, out var durableName)) + { + await StartDurableExecution(ctx, functionName, durableName.ToString(), testEvent); + return; + } + var evnt = runtimeDataStore.QueueEvent(testEvent, isRequestResponseMode); if (isRequestResponseMode) @@ -112,6 +132,34 @@ public async Task PostEvent(HttpContext ctx, string functionName) } } + /// + /// Handles a durable-execution start (an Invoke carrying the durable-execution-name header). + /// Mirrors the service's async start: launch the drive loop and return 202 with the minted + /// ARN in the X-Amz-Durable-Execution-Arn response header, or the + /// DurableExecutionAlreadyStartedException error if the name was reused with a + /// different payload. + /// + private async Task StartDurableExecution(HttpContext ctx, string functionName, string? executionName, string payload) + { + try + { + var arn = _durableDriver!.Start(functionName, executionName, payload); + ctx.Response.Headers[DurableExecutionArnHeader] = arn; + // The real service returns the started execution asynchronously (202); the driver + // invokes the function on its own thread from here. + ctx.Response.StatusCode = 202; + } + catch (DurableExecutionAlreadyStartedException ex) + { + ctx.Response.StatusCode = 409; + ctx.Response.Headers.ContentType = "application/json"; + ctx.Response.Headers["X-Amzn-Errortype"] = nameof(DurableExecutionAlreadyStartedException); + var body = Encoding.UTF8.GetBytes(System.Text.Json.JsonSerializer.Serialize(new { message = ex.Message })); + ctx.Response.Headers.ContentLength = body.Length; + await ctx.Response.Body.WriteAsync(body); + } + } + public Task GetNextInvocationDefaultFunction(HttpContext ctx) { return GetNextInvocation(ctx, DefaultFunctionName); diff --git a/Tools/LambdaTestTool-v2/tests/Amazon.Lambda.TestTool.UnitTests/DurableExecutionDriverTests.cs b/Tools/LambdaTestTool-v2/tests/Amazon.Lambda.TestTool.UnitTests/DurableExecutionDriverTests.cs new file mode 100644 index 000000000..e04dd9fd1 --- /dev/null +++ b/Tools/LambdaTestTool-v2/tests/Amazon.Lambda.TestTool.UnitTests/DurableExecutionDriverTests.cs @@ -0,0 +1,201 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using System.Net.Http.Json; +using System.Text.Json; +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.Model; +using Amazon.Lambda.TestTool.Commands.Settings; +using Amazon.Lambda.TestTool.Processes; +using Amazon.Lambda.TestTool.Tests.Common.Helpers; +using Amazon.Runtime; +using Xunit; +using Environment = System.Environment; +using DurableOperation = Amazon.Lambda.DurableExecution.Operation; +using DurableErrorObject = Amazon.Lambda.DurableExecution.ErrorObject; + +namespace Amazon.Lambda.TestTool.UnitTests; + +/// +/// End-to-end tests for the Phase 2 durable-execution driver: the start hook +/// (X-Amz-Durable-Execution-Name), the re-invocation drive loop, and time-skipping — +/// all exercised together against a durable "function" that speaks the real wire protocol. +/// +/// The fake function polls the Runtime API (next/response) exactly as a real Lambda bootstrap +/// does, and checkpoints via a real pointed at the Test Tool. +/// It deliberately models a wait-then-step workflow so the driver must re-invoke: invocation 1 +/// starts a WAIT and returns Pending; the driver time-skips the wait and re-invokes; invocation 2 +/// sees the wait completed, runs a step, and returns Succeeded. This proves the multi-invocation +/// replay cycle without depending on the DurableExecution SDK's internal engine (tested in its +/// own package) or its process-global cached checkpoint client. +/// +public class DurableExecutionDriverTests +{ + private const string FunctionName = "DurableDriverFoo"; + + private static readonly JsonSerializerOptions Json = new() + { + DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull + }; + + private static (TestToolProcess Process, RunCommandSettings Options, CancellationTokenSource Cts) StartTool(bool skipTime = true) + { + var lambdaPort = TestHelpers.GetNextLambdaRuntimePort(); + var cts = new CancellationTokenSource(); + cts.CancelAfter(60_000); + var options = new RunCommandSettings + { + LambdaEmulatorPort = lambdaPort, + DurableExecution = true, + DurableTimeSkip = skipTime + }; + Environment.SetEnvironmentVariable("ASPNETCORE_ENVIRONMENT", "Development"); + var process = TestToolProcess.Startup(options, cts.Token); + return (process, options, cts); + } + + private static IAmazonLambda ConstructLambdaServiceClient(string url) + { + var config = new AmazonLambdaConfig { ServiceURL = url, MaxErrorRetry = 0 }; + var credentials = new BasicAWSCredentials("accessKeyId", "secretKey"); + return new AmazonLambdaClient(credentials, config); + } + + /// + /// Runs a fake durable function loop: polls the Runtime API, checkpoints a wait-then-step + /// workflow, and returns Pending then Succeeded. Returns the number of invocations processed. + /// + private static async Task RunFakeDurableFunctionAsync( + string serviceUrl, RunCommandSettings options, CancellationToken ct) + { + // Note: no HttpClient.BaseAddress — a leading-slash relative path would replace the + // {FunctionName} prefix and poll the default-function partition instead of the one the + // driver queues into. Build absolute URLs against the named runtime-API partition. + var funcBase = $"http://{options.LambdaEmulatorHost}:{options.LambdaEmulatorPort}/{FunctionName}"; + using var http = new HttpClient(); + using var lambda = ConstructLambdaServiceClient(serviceUrl); + + var invocations = 0; + while (!ct.IsCancellationRequested) + { + // GET next invocation (long-polls until the driver queues an envelope). + HttpResponseMessage next; + try + { + next = await http.GetAsync($"{funcBase}/2018-06-01/runtime/invocation/next", ct); + } + catch (OperationCanceledException) { break; } + + if (!next.IsSuccessStatusCode) + continue; + + var requestId = next.Headers.TryGetValues("Lambda-Runtime-Aws-Request-Id", out var ids) + ? ids.First() : null; + if (requestId is null) + continue; + + var envelopeJson = await next.Content.ReadAsStringAsync(ct); + var input = JsonSerializer.Deserialize(envelopeJson, Json)!; + var ops = input.InitialExecutionState?.Operations ?? new List(); + invocations++; + + DurableExecutionInvocationOutput output; + + var waitOp = ops.FirstOrDefault(o => o.Id == "w1"); + if (waitOp is null) + { + // Invocation 1: start a 1-hour wait, then suspend (Pending). + await lambda.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = input.DurableExecutionArn, + CheckpointToken = input.CheckpointToken, + Updates = new List + { + new() + { + Id = "w1", + Type = OperationType.WAIT, + Action = OperationAction.START, + WaitOptions = new WaitOptions { WaitSeconds = 3600 } + } + } + }, ct); + output = new DurableExecutionInvocationOutput { Status = InvocationStatus.Pending }; + } + else if (waitOp.Status == "SUCCEEDED") + { + // Invocation 2: the driver time-skipped the wait. Run a step and finish. + await lambda.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = input.DurableExecutionArn, + CheckpointToken = input.CheckpointToken, + Updates = new List + { + new() { Id = "s1", Type = OperationType.STEP, Action = OperationAction.START, Name = "finalize" }, + new() { Id = "s1", Type = OperationType.STEP, Action = OperationAction.SUCCEED, Payload = "\"done\"" } + } + }, ct); + output = new DurableExecutionInvocationOutput + { + Status = InvocationStatus.Succeeded, + Result = "\"done\"" + }; + } + else + { + // Unexpected replay state — fail loudly so the test surfaces it. + output = new DurableExecutionInvocationOutput + { + Status = InvocationStatus.Failed, + Error = new DurableErrorObject { ErrorType = "UnexpectedReplayState", ErrorMessage = $"wait status={waitOp.Status}" } + }; + } + + // POST the response back to the Runtime API, unblocking the driver's WaitForCompletion. + await http.PostAsync( + $"{funcBase}/2018-06-01/runtime/invocation/{requestId}/response", + new StringContent(JsonSerializer.Serialize(output, Json)), + ct); + + if (output.Status != InvocationStatus.Pending) + break; + } + + return invocations; + } + + [Fact] + public async Task StartHook_DrivesWaitThenStepWorkflow_ToSucceeded() + { + var (process, options, cts) = StartTool(skipTime: true); + try + { + Assert.True(await TestHelpers.WaitForApiToStartAsync($"{process.ServiceUrl}/lambda-runtime-api/healthcheck")); + + // Start the fake durable function loop in the background. + var functionTask = RunFakeDurableFunctionAsync(process.ServiceUrl, options, cts.Token); + + // Start the durable execution via the SDK Invoke + durable-execution-name header. + var client = ConstructLambdaServiceClient(process.ServiceUrl); + var invoke = await client.InvokeAsync(new InvokeRequest + { + FunctionName = FunctionName, + Payload = "{\"OrderId\":\"o-1\"}", + DurableExecutionName = "exec-alpha" + }, cts.Token); + + // The start hook returns the minted ARN in the X-Amz-Durable-Execution-Arn header, + // which the SDK surfaces on the modeled InvokeResponse.DurableExecutionArn property. + Assert.Equal(System.Net.HttpStatusCode.Accepted, invoke.HttpStatusCode); + Assert.False(string.IsNullOrEmpty(invoke.DurableExecutionArn)); + + // The driver should reach a terminal state; the function processed exactly 2 invocations. + var invocations = await functionTask; + Assert.Equal(2, invocations); + } + finally + { + await cts.CancelAsync(); + } + } +} diff --git a/Tools/LambdaTestTool-v2/tests/Amazon.Lambda.TestTool.UnitTests/DurableServiceApiTests.cs b/Tools/LambdaTestTool-v2/tests/Amazon.Lambda.TestTool.UnitTests/DurableServiceApiTests.cs new file mode 100644 index 000000000..23a90d7fe --- /dev/null +++ b/Tools/LambdaTestTool-v2/tests/Amazon.Lambda.TestTool.UnitTests/DurableServiceApiTests.cs @@ -0,0 +1,324 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using Amazon.Lambda.Model; +using Amazon.Lambda.TestTool.Commands.Settings; +using Amazon.Lambda.TestTool.Processes; +using Amazon.Lambda.TestTool.Tests.Common.Helpers; +using Amazon.Runtime; +using Xunit; +using Environment = System.Environment; + +namespace Amazon.Lambda.TestTool.UnitTests; + +/// +/// End-to-end tests for the durable-execution service emulator (Phase 1: data plane). +/// A real is pointed at the Test Tool host and calls the actual +/// AWSSDK durable operations (CheckpointDurableExecution / GetDurableExecutionState), +/// so the wire serialization, catch-all ARN routing, and state machine are all exercised together. +/// +public class DurableServiceApiTests +{ + private const string FunctionName = "DurableFoo"; + // A realistic slash-bearing durable-execution ARN — the SDK URL-encodes the slashes and the + // emulator's catch-all route must reconstruct it intact. + private const string Arn = + "arn:aws:lambda:us-east-1:000000000000:function:DurableFoo:$LATEST/durable-execution/local/exec-1"; + + private static (TestToolProcess Process, CancellationTokenSource Cts) StartTool(bool skipTime = true) + { + var lambdaPort = TestHelpers.GetNextLambdaRuntimePort(); + var cts = new CancellationTokenSource(); + cts.CancelAfter(30_000); + var options = new RunCommandSettings + { + LambdaEmulatorPort = lambdaPort, + DurableExecution = true, + DurableTimeSkip = skipTime + }; + Environment.SetEnvironmentVariable("ASPNETCORE_ENVIRONMENT", "Development"); + var process = TestToolProcess.Startup(options, cts.Token); + return (process, cts); + } + + private static IAmazonLambda ConstructLambdaServiceClient(string url) + { + var config = new AmazonLambdaConfig + { + ServiceURL = url, + MaxErrorRetry = 0 + }; + var credentials = new BasicAWSCredentials("accessKeyId", "secretKey"); + return new AmazonLambdaClient(credentials, config); + } + + [Fact] + public async Task Checkpoint_StepStartThenSucceed_RoundTripsThroughState() + { + var (process, cts) = StartTool(); + try + { + Assert.True(await TestHelpers.WaitForApiToStartAsync($"{process.ServiceUrl}/lambda-runtime-api/healthcheck")); + var client = ConstructLambdaServiceClient(process.ServiceUrl); + + // START a step. + var startResp = await client.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = Arn, + CheckpointToken = "0", + Updates = new List + { + new() + { + Id = "op-step-1", + Type = OperationType.STEP, + Action = OperationAction.START, + Name = "my-step" + } + } + }, cts.Token); + + // Server rotates the token and echoes the new op back in NewExecutionState. + Assert.NotEqual("0", startResp.CheckpointToken); + Assert.NotNull(startResp.NewExecutionState); + var startedOp = Assert.Single(startResp.NewExecutionState.Operations); + Assert.Equal("op-step-1", startedOp.Id); + Assert.Equal("STARTED", startedOp.Status.Value); + Assert.Equal(1, startedOp.StepDetails.Attempt); + + // SUCCEED the step with a result payload, using the rotated token. + var succeedResp = await client.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = Arn, + CheckpointToken = startResp.CheckpointToken, + Updates = new List + { + new() + { + Id = "op-step-1", + Type = OperationType.STEP, + Action = OperationAction.SUCCEED, + Payload = "\"step-result\"" + } + } + }, cts.Token); + + var succeededOp = Assert.Single(succeedResp.NewExecutionState.Operations); + Assert.Equal("SUCCEEDED", succeededOp.Status.Value); + Assert.Equal("\"step-result\"", succeededOp.StepDetails.Result); + + // GetState returns the full (single-op) history. + var state = await client.GetDurableExecutionStateAsync(new GetDurableExecutionStateRequest + { + DurableExecutionArn = Arn, + CheckpointToken = succeedResp.CheckpointToken, + Marker = "" + }, cts.Token); + + var op = Assert.Single(state.Operations); + Assert.Equal("op-step-1", op.Id); + Assert.Equal("SUCCEEDED", op.Status.Value); + Assert.Equal("\"step-result\"", op.StepDetails.Result); + Assert.Null(state.NextMarker); + } + finally + { + await cts.CancelAsync(); + } + } + + [Fact] + public async Task Checkpoint_WaitStart_WithTimeSkip_IsImmediatelySucceeded() + { + var (process, cts) = StartTool(skipTime: true); + try + { + Assert.True(await TestHelpers.WaitForApiToStartAsync($"{process.ServiceUrl}/lambda-runtime-api/healthcheck")); + var client = ConstructLambdaServiceClient(process.ServiceUrl); + + var resp = await client.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = Arn, + CheckpointToken = "0", + Updates = new List + { + new() + { + Id = "op-wait-1", + Type = OperationType.WAIT, + Action = OperationAction.START, + WaitOptions = new WaitOptions { WaitSeconds = 3600 } + } + } + }, cts.Token); + + // Time-skip flips a WAIT START straight to SUCCEEDED so the next replay proceeds. + var op = Assert.Single(resp.NewExecutionState.Operations); + Assert.Equal("op-wait-1", op.Id); + Assert.Equal("SUCCEEDED", op.Status.Value); + } + finally + { + await cts.CancelAsync(); + } + } + + [Fact] + public async Task Checkpoint_WaitStart_WithoutTimeSkip_StaysStartedWithScheduledEnd() + { + var (process, cts) = StartTool(skipTime: false); + try + { + Assert.True(await TestHelpers.WaitForApiToStartAsync($"{process.ServiceUrl}/lambda-runtime-api/healthcheck")); + var client = ConstructLambdaServiceClient(process.ServiceUrl); + + var resp = await client.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = Arn, + CheckpointToken = "0", + Updates = new List + { + new() + { + Id = "op-wait-1", + Type = OperationType.WAIT, + Action = OperationAction.START, + WaitOptions = new WaitOptions { WaitSeconds = 3600 } + } + } + }, cts.Token); + + var op = Assert.Single(resp.NewExecutionState.Operations); + Assert.Equal("STARTED", op.Status.Value); + Assert.NotNull(op.WaitDetails.ScheduledEndTimestamp); + } + finally + { + await cts.CancelAsync(); + } + } + + [Fact] + public async Task Checkpoint_CallbackStart_MintsCallbackId() + { + var (process, cts) = StartTool(); + try + { + Assert.True(await TestHelpers.WaitForApiToStartAsync($"{process.ServiceUrl}/lambda-runtime-api/healthcheck")); + var client = ConstructLambdaServiceClient(process.ServiceUrl); + + var resp = await client.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = Arn, + CheckpointToken = "0", + Updates = new List + { + new() + { + Id = "op-cb-1", + Type = OperationType.CALLBACK, + Action = OperationAction.START + } + } + }, cts.Token); + + var op = Assert.Single(resp.NewExecutionState.Operations); + Assert.Equal("cb-op-cb-1", op.CallbackDetails.CallbackId); + } + finally + { + await cts.CancelAsync(); + } + } + + [Fact] + public async Task GetState_PagesHistory_WithNextMarker() + { + var (process, cts) = StartTool(); + try + { + Assert.True(await TestHelpers.WaitForApiToStartAsync($"{process.ServiceUrl}/lambda-runtime-api/healthcheck")); + var client = ConstructLambdaServiceClient(process.ServiceUrl); + + // Write more operations than one page (StatePageSize = 100). + const int total = 150; + var updates = new List(); + for (var i = 0; i < total; i++) + { + updates.Add(new OperationUpdate + { + Id = $"op-{i}", + Type = OperationType.STEP, + Action = OperationAction.START, + Name = $"step-{i}" + }); + } + await client.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = Arn, + CheckpointToken = "0", + Updates = updates + }, cts.Token); + + // First page: 100 ops + a NextMarker. + var page1 = await client.GetDurableExecutionStateAsync(new GetDurableExecutionStateRequest + { + DurableExecutionArn = Arn, + CheckpointToken = "1", + Marker = "" + }, cts.Token); + Assert.Equal(100, page1.Operations.Count); + Assert.False(string.IsNullOrEmpty(page1.NextMarker)); + + // Second page: remaining 50 ops, no NextMarker. + var page2 = await client.GetDurableExecutionStateAsync(new GetDurableExecutionStateRequest + { + DurableExecutionArn = Arn, + CheckpointToken = "1", + Marker = page1.NextMarker + }, cts.Token); + Assert.Equal(50, page2.Operations.Count); + Assert.True(string.IsNullOrEmpty(page2.NextMarker)); + } + finally + { + await cts.CancelAsync(); + } + } + + [Fact] + public async Task DurableEndpoints_NotRegistered_WhenFlagDisabled() + { + var lambdaPort = TestHelpers.GetNextLambdaRuntimePort(); + var cts = new CancellationTokenSource(); + cts.CancelAfter(30_000); + var options = new RunCommandSettings + { + LambdaEmulatorPort = lambdaPort, + DurableExecution = false + }; + Environment.SetEnvironmentVariable("ASPNETCORE_ENVIRONMENT", "Development"); + var process = TestToolProcess.Startup(options, cts.Token); + try + { + Assert.True(await TestHelpers.WaitForApiToStartAsync($"{process.ServiceUrl}/lambda-runtime-api/healthcheck")); + var client = ConstructLambdaServiceClient(process.ServiceUrl); + + // With the emulator off, the durable route is absent → the SDK gets a 404. + await Assert.ThrowsAsync(async () => + await client.CheckpointDurableExecutionAsync(new CheckpointDurableExecutionRequest + { + DurableExecutionArn = Arn, + CheckpointToken = "0", + Updates = new List + { + new() { Id = "op-1", Type = OperationType.STEP, Action = OperationAction.START } + } + }, cts.Token)); + } + finally + { + await cts.CancelAsync(); + } + } +} From 433a09fe3b61436acafbacc700541db4bb85d684 Mon Sep 17 00:00:00 2001 From: Garrett Beatty Date: Wed, 22 Jul 2026 16:57:49 -0400 Subject: [PATCH 2/6] Add durable-execution callbacks and web UI to Test Tool v2 (Phase 3) Callbacks: SendDurableExecutionCallback{Success,Failure,Heartbeat} endpoints resolve a pending CALLBACK operation by its minted id and wake the parked drive loop. The driver now parks on a pending callback (instead of failing) and resumes when the callback is resolved; park/resume is lock-guarded so a resolve that races ahead of the park is not lost. Web UI: a new Durable Execution page lists local executions, renders each one's operation timeline (id/type/status/detail), and can send a callback from the browser. The page shows a friendly hint when --durable-execution is not enabled. Includes an end-to-end test covering the full start -> park-on-callback -> SendDurableExecutionCallbackSuccess -> resume -> Succeeded cycle. --- .../50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json | 3 +- .../docs/design/testtool-integration-plan.md | 10 +- .../Components/Layout/MainLayout.razor | 8 + .../Components/Pages/DurableExecution.razor | 143 ++++++++++++++++++ .../Pages/DurableExecution.razor.cs | 126 +++++++++++++++ .../DurableExecutionDriver.cs | Bin 14120 -> 17804 bytes .../DurableExecution/DurableExecutionStore.cs | 61 +++++++- .../DurableExecution/DurableServiceApi.cs | 81 ++++++++++ .../DurableExecutionDriverTests.cs | 131 ++++++++++++++++ 9 files changed, 558 insertions(+), 5 deletions(-) create mode 100644 Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Components/Pages/DurableExecution.razor create mode 100644 Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Components/Pages/DurableExecution.razor.cs diff --git a/.autover/changes/50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json b/.autover/changes/50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json index ccd76c6ae..6545ea65e 100644 --- a/.autover/changes/50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json +++ b/.autover/changes/50f64d33-d1b0-445c-aac3-b2c95ae10ff6.json @@ -5,7 +5,8 @@ "Type": "Minor", "ChangelogMessages": [ "Add durable-execution service emulator (preview): with --durable-execution, the Test Tool hosts CheckpointDurableExecution and GetDurableExecutionState endpoints alongside the Lambda Runtime API so durable workflows can checkpoint and read state locally. Point the function's AWS_ENDPOINT_URL_LAMBDA at the Test Tool to use it. Timers are time-skipped by default (--durable-time-skip).", - "Support running a durable workflow locally end to end: invoking a function with the X-Amz-Durable-Execution-Name header starts a durable execution that the Test Tool drives to completion, re-invoking the function across the replay cycle (steps, waits, retries, child contexts, parallel/map). Chained durable-to-durable invokes and external callbacks are not yet supported." + "Support running a durable workflow locally end to end: invoking a function with the X-Amz-Durable-Execution-Name header starts a durable execution that the Test Tool drives to completion, re-invoking the function across the replay cycle (steps, waits, retries, child contexts, parallel/map). Chained durable-to-durable invokes are not yet supported.", + "Add external callback support and a Durable Execution web UI: SendDurableExecutionCallbackSuccess/Failure/Heartbeat endpoints resume a workflow parked on a callback, and a new Durable Execution page lists executions, shows each operation timeline, and can send a callback from the browser." ] } ] diff --git a/Libraries/src/Amazon.Lambda.DurableExecution/docs/design/testtool-integration-plan.md b/Libraries/src/Amazon.Lambda.DurableExecution/docs/design/testtool-integration-plan.md index 93ae4cde4..b1ec03bfc 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution/docs/design/testtool-integration-plan.md +++ b/Libraries/src/Amazon.Lambda.DurableExecution/docs/design/testtool-integration-plan.md @@ -251,10 +251,14 @@ already forwards minted cb-ids back into `ExecutionState`. `DurableServiceApi.TrySplit`. (Phase 1 tests didn't catch it because checkpoint+getstate were both SDK-encoded and thus internally consistent.) **First genuinely useful ship** — single-function workflows headless, debugger-attached. -- **Phase 3 — Callbacks + Blazor UI (M).** Callback endpoints + `CallbackId → ARN` map + driver wake; - `DurableExecutionPanel.razor` timeline, Send-Callback buttons, time control. +- **Phase 3 — Callbacks + Blazor UI (M).** ✅ **Done.** Callback endpoints keyed by `{CallbackId}` + + a `CallbackId → (arn, opId)` index; the driver parks on a pending callback and resumes on + `NotifyCallbackResolved` (lock-guarded against a resolve racing the park). `DurableExecution.razor` lists + executions, shows the operation timeline, and offers a Send-Callback action. E2E test covers start → park + → SendCallbackSuccess → resume → Succeeded. - **Phase 4 — Chained invokes + fidelity polish (L).** Out-of-process nested drive for `CHAINED_INVOKE` - across registered sibling functions; tiered payload caps + `ReplayChildren`; idempotency; `StopDurableExecution`. + across registered sibling functions; tiered payload caps + `ReplayChildren`; `StopDurableExecution`. + (Idempotency via `DurableExecutionAlreadyStartedException` already landed in Phase 2.) ## Appendix — Phase-0 spike source diff --git a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Components/Layout/MainLayout.razor b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Components/Layout/MainLayout.razor index dd2e997fc..873af3d88 100644 --- a/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Components/Layout/MainLayout.razor +++ b/Tools/LambdaTestTool-v2/src/Amazon.Lambda.TestTool/Components/Layout/MainLayout.razor @@ -25,6 +25,14 @@ Function Tester +