diff --git a/.autover/changes/a3f6c9d2-7b41-4e08-9c5a-2d6f1e0b8a74.json b/.autover/changes/a3f6c9d2-7b41-4e08-9c5a-2d6f1e0b8a74.json new file mode 100644 index 000000000..7c46fb327 --- /dev/null +++ b/.autover/changes/a3f6c9d2-7b41-4e08-9c5a-2d6f1e0b8a74.json @@ -0,0 +1,18 @@ +{ + "Projects": [ + { + "Name": "Amazon.Lambda.DurableExecution.LocalEmulation", + "Type": "Patch", + "ChangelogMessages": [ + "Add a shared local-emulation kernel for durable execution: the in-memory operation store and checkpoint state machine that drive durable workflows locally. Consumed by Amazon.Lambda.DurableExecution.Testing (and the Lambda Test Tool) via a transport-neutral OperationUpdateInput so the two cannot drift. Implementation detail; not intended for direct use." + ] + }, + { + "Name": "Amazon.Lambda.DurableExecution.Testing", + "Type": "Patch", + "ChangelogMessages": [ + "Refactor the in-memory checkpoint state machine and operation store into the shared Amazon.Lambda.DurableExecution.LocalEmulation package. No public API or behavior change." + ] + } + ] +} diff --git a/Libraries/Amazon.Lambda.DurableExecution.slnf b/Libraries/Amazon.Lambda.DurableExecution.slnf index 645c64888..02bf76083 100644 --- a/Libraries/Amazon.Lambda.DurableExecution.slnf +++ b/Libraries/Amazon.Lambda.DurableExecution.slnf @@ -7,9 +7,11 @@ "src\\Amazon.Lambda.Serialization.SystemTextJson\\Amazon.Lambda.Serialization.SystemTextJson.csproj", "src\\Amazon.Lambda.TestUtilities\\Amazon.Lambda.TestUtilities.csproj", "src\\Amazon.Lambda.DurableExecution\\Amazon.Lambda.DurableExecution.csproj", + "src\\Amazon.Lambda.DurableExecution.LocalEmulation\\Amazon.Lambda.DurableExecution.LocalEmulation.csproj", "src\\Amazon.Lambda.DurableExecution.Testing\\Amazon.Lambda.DurableExecution.Testing.csproj", "test\\Amazon.Lambda.DurableExecution.Tests\\Amazon.Lambda.DurableExecution.Tests.csproj", "test\\Amazon.Lambda.DurableExecution.Testing.Tests\\Amazon.Lambda.DurableExecution.Testing.Tests.csproj", + "test\\Amazon.Lambda.DurableExecution.LocalEmulation.Tests\\Amazon.Lambda.DurableExecution.LocalEmulation.Tests.csproj", "test\\Amazon.Lambda.DurableExecution.IntegrationTests\\Amazon.Lambda.DurableExecution.IntegrationTests.csproj", "test\\Amazon.Lambda.DurableExecution.AotPublishTest\\Amazon.Lambda.DurableExecution.AotPublishTest.csproj" ] diff --git a/Libraries/Libraries.sln b/Libraries/Libraries.sln index 89a88dd60..acd72ddf0 100644 --- a/Libraries/Libraries.sln +++ b/Libraries/Libraries.sln @@ -175,6 +175,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TestDurableServerlessApp", EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TestDurableServerlessApp.IntegrationTests", "test\TestDurableServerlessApp.IntegrationTests\TestDurableServerlessApp.IntegrationTests.csproj", "{A89D56F9-63CC-4B47-BFED-693E3410993D}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Amazon.Lambda.DurableExecution.LocalEmulation", "src\Amazon.Lambda.DurableExecution.LocalEmulation\Amazon.Lambda.DurableExecution.LocalEmulation.csproj", "{22AB1C70-26CB-443E-9ABD-3A6BF94743EB}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Amazon.Lambda.DurableExecution.LocalEmulation.Tests", "test\Amazon.Lambda.DurableExecution.LocalEmulation.Tests\Amazon.Lambda.DurableExecution.LocalEmulation.Tests.csproj", "{3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -1109,6 +1113,30 @@ Global {A89D56F9-63CC-4B47-BFED-693E3410993D}.Release|x64.Build.0 = Release|Any CPU {A89D56F9-63CC-4B47-BFED-693E3410993D}.Release|x86.ActiveCfg = Release|Any CPU {A89D56F9-63CC-4B47-BFED-693E3410993D}.Release|x86.Build.0 = Release|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Debug|Any CPU.Build.0 = Debug|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Debug|x64.ActiveCfg = Debug|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Debug|x64.Build.0 = Debug|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Debug|x86.ActiveCfg = Debug|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Debug|x86.Build.0 = Debug|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Release|Any CPU.ActiveCfg = Release|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Release|Any CPU.Build.0 = Release|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Release|x64.ActiveCfg = Release|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Release|x64.Build.0 = Release|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Release|x86.ActiveCfg = Release|Any CPU + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB}.Release|x86.Build.0 = Release|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Debug|Any CPU.Build.0 = Debug|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Debug|x64.ActiveCfg = Debug|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Debug|x64.Build.0 = Debug|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Debug|x86.ActiveCfg = Debug|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Debug|x86.Build.0 = Debug|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Release|Any CPU.ActiveCfg = Release|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Release|Any CPU.Build.0 = Release|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Release|x64.ActiveCfg = Release|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Release|x64.Build.0 = Release|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Release|x86.ActiveCfg = Release|Any CPU + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -1195,6 +1223,8 @@ Global {0460CF9D-B5CC-47C5-9B30-CEA84695AB3B} = {1DE4EE60-45BA-4EF7-BE00-B9EB861E4C69} {616E108A-9ED5-4282-B1D4-7FD49BC6DA2E} = {1DE4EE60-45BA-4EF7-BE00-B9EB861E4C69} {A89D56F9-63CC-4B47-BFED-693E3410993D} = {1DE4EE60-45BA-4EF7-BE00-B9EB861E4C69} + {22AB1C70-26CB-443E-9ABD-3A6BF94743EB} = {AAB54E74-20B1-42ED-BC3D-CE9F7BC7FD12} + {3D323A2C-AD83-4CE1-A663-6FCC7AA44FC1} = {1DE4EE60-45BA-4EF7-BE00-B9EB861E4C69} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {503678A4-B8D1-4486-8915-405A3E9CF0EB} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/Amazon.Lambda.DurableExecution.LocalEmulation.csproj b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/Amazon.Lambda.DurableExecution.LocalEmulation.csproj new file mode 100644 index 000000000..0796171af --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/Amazon.Lambda.DurableExecution.LocalEmulation.csproj @@ -0,0 +1,52 @@ + + + + + + $(DefaultPackageTargets) + Shared in-memory emulation kernel for Amazon Lambda Durable Execution - the operation store and checkpoint state machine used to drive durable workflows locally (by the .Testing package and the Lambda Test Tool). Implementation detail; not intended for direct use. + Amazon.Lambda.DurableExecution.LocalEmulation + 1.0.0 + Amazon.Lambda.DurableExecution.LocalEmulation + Amazon.Lambda.DurableExecution.LocalEmulation + AWS;Amazon;Lambda;Durable;Workflow;Testing + enable + enable + true + + + + + + <_Parameter1>Amazon.Lambda.DurableExecution.LocalEmulation.Tests, PublicKey="0024000004800000940000000602000000240000525341310004000001000100db5f59f098d27276c7833875a6263a3cc74ab17ba9a9df0b52aedbe7252745db7274d5271fd79c1f08f668ecfa8eaab5626fa76adc811d3c8fc55859b0d09d3bc0a84eecd0ba891f2b8a2fc55141cdcc37c2053d53491e650a479967c3622762977900eddbf1252ed08a2413f00a28f3a0752a81203f03ccb7f684db373518b4" + + + <_Parameter1>Amazon.Lambda.DurableExecution.Testing, PublicKey="0024000004800000940000000602000000240000525341310004000001000100db5f59f098d27276c7833875a6263a3cc74ab17ba9a9df0b52aedbe7252745db7274d5271fd79c1f08f668ecfa8eaab5626fa76adc811d3c8fc55859b0d09d3bc0a84eecd0ba891f2b8a2fc55141cdcc37c2053d53491e650a479967c3622762977900eddbf1252ed08a2413f00a28f3a0752a81203f03ccb7f684db373518b4" + + + + <_Parameter1>Amazon.Lambda.DurableExecution.Testing.Tests, PublicKey="0024000004800000940000000602000000240000525341310004000001000100db5f59f098d27276c7833875a6263a3cc74ab17ba9a9df0b52aedbe7252745db7274d5271fd79c1f08f668ecfa8eaab5626fa76adc811d3c8fc55859b0d09d3bc0a84eecd0ba891f2b8a2fc55141cdcc37c2053d53491e650a479967c3622762977900eddbf1252ed08a2413f00a28f3a0752a81203f03ccb7f684db373518b4" + + + <_Parameter1>Amazon.Lambda.TestTool, PublicKey="0024000004800000940000000602000000240000525341310004000001000100db5f59f098d27276c7833875a6263a3cc74ab17ba9a9df0b52aedbe7252745db7274d5271fd79c1f08f668ecfa8eaab5626fa76adc811d3c8fc55859b0d09d3bc0a84eecd0ba891f2b8a2fc55141cdcc37c2053d53491e650a479967c3622762977900eddbf1252ed08a2413f00a28f3a0752a81203f03ccb7f684db373518b4" + + + + + + + + + diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/CheckpointProcessor.cs b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/CheckpointProcessor.cs similarity index 68% rename from Libraries/src/Amazon.Lambda.DurableExecution.Testing/CheckpointProcessor.cs rename to Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/CheckpointProcessor.cs index 8d709d776..34c220ce6 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/CheckpointProcessor.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/CheckpointProcessor.cs @@ -1,44 +1,53 @@ // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 -using Amazon.Lambda.Model; -using SdkOperationUpdate = Amazon.Lambda.Model.OperationUpdate; - -namespace Amazon.Lambda.DurableExecution.Testing; +namespace Amazon.Lambda.DurableExecution.LocalEmulation; /// -/// Processes checkpoint updates against the in-memory operation store. -/// Handles action-to-status mapping, callback ID minting, time skipping, -/// and producing the "new operations" response the runtime expects. +/// 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. /// +/// +/// This is the shared local-emulation state machine used by both the durable-execution testing +/// package (driving the workflow as an in-process delegate) and the Lambda Test Tool (driving a +/// real, separately-running function over the Runtime API + HTTP data plane). Both feed it the +/// same transport-neutral ; the state transitions below are the +/// single source of truth for how a local checkpoint mutates recorded operations. +/// internal sealed class CheckpointProcessor { private readonly InMemoryOperationStore _store; - private readonly bool _skipTime; + // Read on each checkpoint (not captured) so a consumer's time-skip toggle can be flipped at + // runtime — e.g. the Test Tool's UI switch — and take effect for subsequent checkpoints. + private readonly Func _skipTimeProvider; private readonly object _pendingGate = new(); private readonly List _pendingInvokes = new(); + /// Creates a processor whose time-skip mode is fixed for the run. public CheckpointProcessor(InMemoryOperationStore store, bool skipTime) + : this(store, () => skipTime) + { + } + + /// Creates a processor whose time-skip mode is read live on each checkpoint. + public CheckpointProcessor(InMemoryOperationStore store, Func skipTimeProvider) { _store = store; - _skipTime = skipTime; + _skipTimeProvider = skipTimeProvider; } /// - /// A chained-invoke (ctx.InvokeAsync) that has been 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; in the - /// local harness the - /// drains these between invocations and resolves them via the - /// . The target function name lives only on the - /// wire-format OperationUpdate.ChainedInvokeOptions, so it is captured - /// here rather than on the persisted . + /// A chained-invoke (ctx.InvokeAsync) that has been 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; the local drivers drain these between invocations and resolve them + /// (the testing package via its function registry, the Test Tool by starting a nested durable + /// execution). The target function name lives only on the update, not on the persisted + /// , so it is captured here. /// internal readonly record struct PendingInvoke(string OperationId, string FunctionName, string? Payload); - /// - /// Returns and clears the chained-invokes started since the last drain. - /// + /// Returns and clears the chained-invokes started since the last drain. public IReadOnlyList DrainPendingInvokes() { lock (_pendingGate) @@ -52,14 +61,13 @@ public IReadOnlyList DrainPendingInvokes() } /// - /// Processes a batch of updates and returns the new checkpoint token - /// and any operations that were created or modified (to feed back to - /// the runtime's onNewOperations callback). + /// 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) + IReadOnlyList updates) { var newOperations = new List(); @@ -73,29 +81,28 @@ public IReadOnlyList DrainPendingInvokes() return (newToken, newOperations); } - private Operation ApplyUpdate(string arn, SdkOperationUpdate update) + private Operation ApplyUpdate(string arn, OperationUpdateInput update) { - var existing = _store.GetOperation(arn, update.Id); + var existing = _store.GetOperation(arn, update.Id!); var operation = existing ?? new Operation { Id = update.Id }; - operation.Type = update.Type?.Value ?? operation.Type; + 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?.Value; + var action = update.Action; ApplyAction(operation, action, update); - if (_skipTime) + if (_skipTimeProvider()) ApplyTimeSkipping(operation, action); - // A chained-invoke START suspends the workflow until an external system - // resolves it. Record it so the orchestrator 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. + // A chained-invoke START suspends the workflow until an external system resolves it. + // Record it so a driver can run the sibling and stamp the result/error before the next + // replay. The function name is only carried on the update, not on the Operation. if (action == "START" && operation.Type == OperationTypes.ChainedInvoke - && update.ChainedInvokeOptions?.FunctionName is { } functionName) + && update.ChainedInvokeFunctionName is { } functionName) { lock (_pendingGate) { @@ -107,7 +114,7 @@ private Operation ApplyUpdate(string arn, SdkOperationUpdate update) return operation; } - private static void ApplyAction(Operation operation, string? action, SdkOperationUpdate update) + private static void ApplyAction(Operation operation, string? action, OperationUpdateInput update) { switch (action) { @@ -141,23 +148,23 @@ private static void ApplyAction(Operation operation, string? action, SdkOperatio } } - private static void ApplyStartDetails(Operation operation, SdkOperationUpdate update) + private static void ApplyStartDetails(Operation operation, OperationUpdateInput 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. + // 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) + if (update.WaitSeconds is { } seconds) { operation.WaitDetails.ScheduledEndTimestamp = DateTimeOffset.UtcNow.AddSeconds(seconds).ToUnixTimeMilliseconds(); @@ -183,7 +190,7 @@ private static void ApplyStartDetails(Operation operation, SdkOperationUpdate up } } - private static void ApplySucceedDetails(Operation operation, SdkOperationUpdate update) + private static void ApplySucceedDetails(Operation operation, OperationUpdateInput update) { var payload = update.Payload; switch (operation.Type) @@ -214,9 +221,9 @@ private static void ApplySucceedDetails(Operation operation, SdkOperationUpdate } } - private static void ApplyFailDetails(Operation operation, SdkOperationUpdate update) + private static void ApplyFailDetails(Operation operation, OperationUpdateInput update) { - var error = MapSdkError(update.Error); + var error = update.Error; switch (operation.Type) { case OperationTypes.Step: @@ -241,26 +248,26 @@ private static void ApplyFailDetails(Operation operation, SdkOperationUpdate upd } } - private static void ApplyRetryDetails(Operation operation, SdkOperationUpdate update) + private static void ApplyRetryDetails(Operation operation, OperationUpdateInput 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. + // 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) + if (update.NextAttemptDelaySeconds is { } delaySeconds) { operation.StepDetails.NextAttemptTimestamp = DateTimeOffset.UtcNow.AddSeconds(delaySeconds).ToUnixTimeMilliseconds(); } - operation.StepDetails.Error = MapSdkError(update.Error); + operation.StepDetails.Error = 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). + // 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) @@ -283,9 +290,9 @@ private void ApplyTimeSkipping(Operation operation, string? action) } } - // 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. + // 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; @@ -296,16 +303,4 @@ private void ApplyTimeSkipping(Operation operation, string? action) } } } - - private static ErrorObject? MapSdkError(Amazon.Lambda.Model.ErrorObject? sdkError) - { - if (sdkError == null) return null; - return new ErrorObject - { - ErrorType = sdkError.ErrorType, - ErrorMessage = sdkError.ErrorMessage, - ErrorData = sdkError.ErrorData, - StackTrace = sdkError.StackTrace - }; - } } diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/DurableEmulationHelpers.cs b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/DurableEmulationHelpers.cs new file mode 100644 index 000000000..c5d5ffcfd --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/DurableEmulationHelpers.cs @@ -0,0 +1,77 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +namespace Amazon.Lambda.DurableExecution.LocalEmulation; + +/// +/// Shared helpers for the local drive loops: seeding the top-level EXECUTION operation and +/// computing when a suspended workflow can next make progress. Both the testing package's +/// orchestrator and the Test Tool's driver use these so the semantics stay identical. +/// +internal static class DurableEmulationHelpers +{ + /// The operation id of the seeded top-level EXECUTION op. + public const string ExecutionOperationId = "exec-0"; + + /// + /// Seeds the top-level EXECUTION operation carrying the user input payload. Idempotent: + /// re-driving (e.g. a replay pass, or WaitForResultAsync after a callback) must not reset the + /// op back to Started or clobber recorded state. + /// + public static void SeedExecution(InMemoryOperationStore store, string arn, string? inputPayload) + { + if (store.GetOperation(arn, ExecutionOperationId) is not null) + return; + + store.Upsert(arn, new Operation + { + Id = ExecutionOperationId, + Type = OperationTypes.Execution, + Status = OperationStatuses.Started, + ExecutionDetails = new ExecutionDetails { InputPayload = inputPayload } + }); + } + + /// + /// Computes how long to wait before the workflow can next make progress, based on the earliest + /// future ScheduledEndTimestamp (pending WAIT) or NextAttemptTimestamp (pending + /// STEP retry backoff) across all operations. Returns false when no operation carries a future + /// resume time. + /// + public static bool TryGetNextResumeDelay(InMemoryOperationStore store, string arn, out TimeSpan delay) + => TryGetNextResumeDelay(store.GetAllOperations(arn), out delay); + + /// + /// Operation-list overload of + /// for callers that hold a snapshot rather than the store (e.g. the Test Tool driver, which + /// does not expose the underlying store). + /// + public static bool TryGetNextResumeDelay(IReadOnlyList operations, out TimeSpan delay) + { + long? earliest = null; + foreach (var op in operations) + { + long? resumeAt = op.Type switch + { + OperationTypes.Wait when op.Status == OperationStatuses.Started + => op.WaitDetails?.ScheduledEndTimestamp, + OperationTypes.Step when op.Status == OperationStatuses.Pending + => op.StepDetails?.NextAttemptTimestamp, + _ => null, + }; + + if (resumeAt is { } ts && (earliest is null || ts < earliest)) + earliest = ts; + } + + if (earliest is null) + { + delay = TimeSpan.Zero; + return false; + } + + var deltaMs = earliest.Value - DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + delay = deltaMs > 0 ? TimeSpan.FromMilliseconds(deltaMs) : TimeSpan.Zero; + return true; + } +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/InMemoryOperationStore.cs b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/InMemoryOperationStore.cs similarity index 72% rename from Libraries/src/Amazon.Lambda.DurableExecution.Testing/InMemoryOperationStore.cs rename to Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/InMemoryOperationStore.cs index 56ebe7dbe..6604a25cd 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/InMemoryOperationStore.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/InMemoryOperationStore.cs @@ -1,20 +1,21 @@ // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 -namespace Amazon.Lambda.DurableExecution.Testing; +namespace Amazon.Lambda.DurableExecution.LocalEmulation; /// -/// In-memory store for operations recorded during a test execution. -/// Each execution (keyed by ARN) maintains its own isolated operation set. +/// In-memory store for operations recorded during a local durable execution. Each execution +/// (keyed by ARN) maintains its own isolated operation set and checkpoint token. Operates on the +/// public wire type so consumers 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. + // 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) @@ -24,9 +25,8 @@ public string CurrentToken(string arn) } /// - /// 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. + /// 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) { @@ -79,6 +79,13 @@ public int OperationCount(string arn) 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)) diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/OperationUpdateInput.cs b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/OperationUpdateInput.cs new file mode 100644 index 000000000..08220bc5c --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution.LocalEmulation/OperationUpdateInput.cs @@ -0,0 +1,61 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +namespace Amazon.Lambda.DurableExecution.LocalEmulation; + +/// +/// A transport-neutral checkpoint update consumed by . +/// +/// +/// The two local-emulation consumers receive checkpoint updates in different shapes: +/// the testing package gets the AWSSDK Amazon.Lambda.Model.OperationUpdate (whose +/// Type/Action/SubType are ConstantClass values), and the Test Tool gets a plain-STJ +/// wire DTO deserialized from the HTTP data plane. Rather than make the state machine generic +/// over those two shapes, each consumer flattens its own representation into this one record at +/// the boundary — unwrapping enum-like members to plain strings and mapping nested option/error +/// types up front. That keeps every branch of the shared state machine identical and free of +/// ?.Value ceremony. +/// +/// Only the members the checkpoint state machine actually reads are carried here; option groups +/// are flattened to their single relevant field (e.g. from +/// WaitOptions). +/// +internal sealed class OperationUpdateInput +{ + /// Operation id the update targets. + public string? Id { get; init; } + + /// Parent operation id (null for top-level operations). + public string? ParentId { get; init; } + + /// Human-readable operation name. + public string? Name { get; init; } + + /// Operation type — compared against OperationTypes constants. + public string? Type { get; init; } + + /// Operation sub-type — compared against OperationSubTypes constants. + public string? SubType { get; init; } + + /// Lifecycle action: START, SUCCEED, FAIL, RETRY, or CANCEL. + public string? Action { get; init; } + + /// Serialized result / carried state, depending on the action and operation type. + public string? Payload { get; init; } + + /// Failure detail for a FAIL/RETRY action. + public ErrorObject? Error { get; init; } + + /// Retry backoff (from StepOptions.NextAttemptDelaySeconds) for a STEP RETRY. + public int? NextAttemptDelaySeconds { get; init; } + + /// Timer duration (from WaitOptions.WaitSeconds) for a WAIT START. + public long? WaitSeconds { get; init; } + + /// + /// Target function of a CHAINED_INVOKE START (from ChainedInvokeOptions.FunctionName). + /// Carried on the update only — never persisted on the — so the driver + /// captures it here to resolve the sibling. + /// + public string? ChainedInvokeFunctionName { get; init; } +} diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/Amazon.Lambda.DurableExecution.Testing.csproj b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/Amazon.Lambda.DurableExecution.Testing.csproj index eeae78e25..3c8f5f266 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/Amazon.Lambda.DurableExecution.Testing.csproj +++ b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/Amazon.Lambda.DurableExecution.Testing.csproj @@ -33,6 +33,7 @@ --> + diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/DurableTestRunner.cs b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/DurableTestRunner.cs index 3e7f74fc7..1c3d97db2 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/DurableTestRunner.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/DurableTestRunner.cs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 using Amazon.Lambda.Core; +using Amazon.Lambda.DurableExecution.LocalEmulation; using Amazon.Lambda.Serialization.SystemTextJson; using Amazon.Lambda.TestUtilities; diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/ExecutionOrchestrator.cs b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/ExecutionOrchestrator.cs index 71bb42d7d..a9092f7b7 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/ExecutionOrchestrator.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/ExecutionOrchestrator.cs @@ -3,6 +3,7 @@ using System.Text; using Amazon.Lambda.Core; +using Amazon.Lambda.DurableExecution.LocalEmulation; using Amazon.Lambda.DurableExecution.Services; namespace Amazon.Lambda.DurableExecution.Testing; @@ -120,7 +121,7 @@ public async Task> DriveToTerminalAsync( // would actually elapse. If nothing has a scheduled resume time the // workflow cannot progress on its own, so re-drive immediately and let // MaxInvocations / the timeout surface the stall. - if (TryGetNextResumeDelay(arn, out var delay) && delay > TimeSpan.Zero) + if (DurableEmulationHelpers.TryGetNextResumeDelay(_store, arn, out var delay) && delay > TimeSpan.Zero) { // Task.Delay surfaces cancellation as TaskCanceledException; swallow // it so the loop-top ThrowIfCancellationRequested throws the canonical @@ -136,41 +137,6 @@ public async Task> DriveToTerminalAsync( } } - /// - /// Computes how long to wait before the workflow can next make progress, based - /// on the earliest future ScheduledEndTimestamp (pending WAIT) or - /// NextAttemptTimestamp (pending STEP retry backoff) across all operations. - /// Returns false when no operation carries a future resume time. - /// - private bool TryGetNextResumeDelay(string arn, out TimeSpan delay) - { - long? earliest = null; - foreach (var op in _store.GetAllOperations(arn)) - { - long? resumeAt = op.Type switch - { - OperationTypes.Wait when op.Status == OperationStatuses.Started - => op.WaitDetails?.ScheduledEndTimestamp, - OperationTypes.Step when op.Status == OperationStatuses.Pending - => op.StepDetails?.NextAttemptTimestamp, - _ => null, - }; - - if (resumeAt is { } ts && (earliest is null || ts < earliest)) - earliest = ts; - } - - if (earliest is null) - { - delay = TimeSpan.Zero; - return false; - } - - var deltaMs = earliest.Value - DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); - delay = deltaMs > 0 ? TimeSpan.FromMilliseconds(deltaMs) : TimeSpan.Zero; - return true; - } - /// /// Runs any chained-invoke siblings started during the last invocation through /// the and stamps the result/error onto the @@ -231,19 +197,10 @@ private bool HasPendingCallback(string arn) private void SeedExecutionOperation(string arn, TInput input) { - // Idempotent: re-driving (e.g. WaitForResultAsync after a callback) must - // not reset the EXECUTION op back to Started or clobber recorded state. - if (_store.GetOperation(arn, "exec-0") is not null) - return; - - var serializedInput = SerializeToString(input); - _store.Upsert(arn, new Operation - { - Id = "exec-0", - Type = OperationTypes.Execution, - Status = OperationStatuses.Started, - ExecutionDetails = new ExecutionDetails { InputPayload = serializedInput } - }); + // Idempotent seeding of the top-level EXECUTION op (exec-0) is owned by the shared kernel + // so the orchestrator and the Test Tool driver stay in lock-step; we only supply the + // serialized input payload. + DurableEmulationHelpers.SeedExecution(_store, arn, SerializeToString(input)); } private DurableExecutionInvocationInput BuildInvocationInput(string arn) diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/InMemoryDurableServiceClient.cs b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/InMemoryDurableServiceClient.cs index a49f6c2b1..c0e2da9ba 100644 --- a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/InMemoryDurableServiceClient.cs +++ b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/InMemoryDurableServiceClient.cs @@ -1,6 +1,7 @@ // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 +using Amazon.Lambda.DurableExecution.LocalEmulation; using Amazon.Lambda.DurableExecution.Services; using SdkOperationUpdate = Amazon.Lambda.Model.OperationUpdate; @@ -32,7 +33,8 @@ public InMemoryDurableServiceClient(InMemoryOperationStore store, CheckpointProc if (pendingOperations.Count == 0) return Task.FromResult(checkpointToken); - var (newToken, newOps) = _processor.Process(durableExecutionArn, checkpointToken, pendingOperations); + var inputs = SdkOperationUpdateMapper.ToInputs(pendingOperations); + var (newToken, newOps) = _processor.Process(durableExecutionArn, checkpointToken, inputs); if (onNewOperations is not null && newOps.Count > 0) onNewOperations(newOps); diff --git a/Libraries/src/Amazon.Lambda.DurableExecution.Testing/SdkOperationUpdateMapper.cs b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/SdkOperationUpdateMapper.cs new file mode 100644 index 000000000..ccbc4571c --- /dev/null +++ b/Libraries/src/Amazon.Lambda.DurableExecution.Testing/SdkOperationUpdateMapper.cs @@ -0,0 +1,51 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using Amazon.Lambda.DurableExecution.LocalEmulation; +using SdkOperationUpdate = Amazon.Lambda.Model.OperationUpdate; + +namespace Amazon.Lambda.DurableExecution.Testing; + +/// +/// Maps the AWSSDK Amazon.Lambda.Model.OperationUpdate the in-process runtime emits into the +/// transport-neutral the shared checkpoint state machine consumes. +/// This is where the SDK's ConstantClass enum-like members are unwrapped to plain strings and +/// nested option/error types are flattened; the state machine itself stays representation-agnostic. +/// +internal static class SdkOperationUpdateMapper +{ + public static IReadOnlyList ToInputs(IReadOnlyList updates) + { + var result = new List(updates.Count); + foreach (var update in updates) + result.Add(ToInput(update)); + return result; + } + + public static OperationUpdateInput ToInput(SdkOperationUpdate update) => new() + { + Id = update.Id, + ParentId = update.ParentId, + Name = update.Name, + Type = update.Type?.Value, + SubType = update.SubType, + Action = update.Action?.Value, + Payload = update.Payload, + Error = MapSdkError(update.Error), + NextAttemptDelaySeconds = update.StepOptions?.NextAttemptDelaySeconds, + WaitSeconds = update.WaitOptions?.WaitSeconds, + ChainedInvokeFunctionName = update.ChainedInvokeOptions?.FunctionName, + }; + + private static ErrorObject? MapSdkError(Amazon.Lambda.Model.ErrorObject? sdkError) + { + if (sdkError == null) return null; + return new ErrorObject + { + ErrorType = sdkError.ErrorType, + ErrorMessage = sdkError.ErrorMessage, + ErrorData = sdkError.ErrorData, + StackTrace = sdkError.StackTrace + }; + } +} diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/Amazon.Lambda.DurableExecution.LocalEmulation.Tests.csproj b/Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/Amazon.Lambda.DurableExecution.LocalEmulation.Tests.csproj new file mode 100644 index 000000000..4f5b9e2cd --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/Amazon.Lambda.DurableExecution.LocalEmulation.Tests.csproj @@ -0,0 +1,30 @@ + + + + + + $(DefaultPackageTargets) + Amazon.Lambda.DurableExecution.LocalEmulation.Tests + Amazon.Lambda.DurableExecution.LocalEmulation.Tests + true + ..\..\..\buildtools\public.snk + true + enable + enable + $(NoWarn);CS1591 + true + + + + + + + + + + + + + + + diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/CheckpointProcessorTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/CheckpointProcessorTests.cs similarity index 64% rename from Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/CheckpointProcessorTests.cs rename to Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/CheckpointProcessorTests.cs index 5ebc2a6ca..2bc84822b 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/CheckpointProcessorTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/CheckpointProcessorTests.cs @@ -1,12 +1,11 @@ // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 -using Amazon.Lambda.DurableExecution.Testing; -using Amazon.Lambda.Model; +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.DurableExecution.LocalEmulation; using Xunit; -using SdkOperationUpdate = Amazon.Lambda.Model.OperationUpdate; -namespace Amazon.Lambda.DurableExecution.Testing.Tests; +namespace Amazon.Lambda.DurableExecution.LocalEmulation.Tests; public class CheckpointProcessorTests { @@ -17,9 +16,9 @@ public void Process_StepStart_CreatesOperation() { var (store, processor) = Create(skipTime: false); - var updates = new List + var updates = new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START, Name = "validate", SubType = OperationSubTypes.Step } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "START", Name = "validate", SubType = OperationSubTypes.Step } }; var (token, newOps) = processor.Process(Arn, null, updates); @@ -37,14 +36,14 @@ public void Process_StepStart_CreatesOperation() public void Process_StepSucceed_UpdatesStatus() { var (store, processor) = Create(skipTime: false); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START, Name = "step1" } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "START", Name = "step1" } }); - processor.Process(Arn, "1", new List + processor.Process(Arn, "1", new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.SUCCEED, Payload = """{"x":1}""" } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "SUCCEED", Payload = """{"x":1}""" } }); var op = store.GetOperation(Arn, "op-1"); @@ -58,17 +57,17 @@ public void Process_StepSucceed_UpdatesStatus() public void Process_StepFail_SetsError() { var (store, processor) = Create(skipTime: false); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "START" } }); - processor.Process(Arn, "1", new List + processor.Process(Arn, "1", new List { new() { - Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.FAIL, - Error = new Amazon.Lambda.Model.ErrorObject { ErrorType = "TestEx", ErrorMessage = "boom" } + Id = "op-1", Type = OperationTypes.Step, Action = "FAIL", + Error = new ErrorObject { ErrorType = "TestEx", ErrorMessage = "boom" } } }); @@ -82,18 +81,18 @@ public void Process_StepFail_SetsError() public void Process_StepRetry_SetsPendingWithDelay() { var (store, processor) = Create(skipTime: false); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "START" } }); - processor.Process(Arn, "1", new List + processor.Process(Arn, "1", new List { new() { - Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.RETRY, - StepOptions = new StepOptions { NextAttemptDelaySeconds = 5 }, - Error = new Amazon.Lambda.Model.ErrorObject { ErrorType = "TransientEx" } + Id = "op-1", Type = OperationTypes.Step, Action = "RETRY", + NextAttemptDelaySeconds = 5, + Error = new ErrorObject { ErrorType = "TransientEx" } } }); @@ -107,17 +106,17 @@ public void Process_StepRetry_SetsPendingWithDelay() public void Process_StepRetry_WithSkipTime_SetsReady() { var (store, processor) = Create(skipTime: true); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "START" } }); - processor.Process(Arn, "1", new List + processor.Process(Arn, "1", new List { new() { - Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.RETRY, - StepOptions = new StepOptions { NextAttemptDelaySeconds = 60 } + Id = "op-1", Type = OperationTypes.Step, Action = "RETRY", + NextAttemptDelaySeconds = 60 } }); @@ -131,12 +130,12 @@ public void Process_WaitStart_SetsScheduledEnd() { var (store, processor) = Create(skipTime: false); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { new() { - Id = "op-1", Type = OperationTypes.Wait, Action = OperationAction.START, - WaitOptions = new WaitOptions { WaitSeconds = 300 } + Id = "op-1", Type = OperationTypes.Wait, Action = "START", + WaitSeconds = 300 } }); @@ -152,12 +151,12 @@ public void Process_WaitStart_WithSkipTime_ImmediatelySucceeds() { var (store, processor) = Create(skipTime: true); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { new() { - Id = "op-1", Type = OperationTypes.Wait, Action = OperationAction.START, - WaitOptions = new WaitOptions { WaitSeconds = 86400 } + Id = "op-1", Type = OperationTypes.Wait, Action = "START", + WaitSeconds = 86400 } }); @@ -171,9 +170,9 @@ public void Process_CallbackStart_MintsCallbackId() { var (store, processor) = Create(skipTime: false); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-cb-1", Type = OperationTypes.Callback, Action = OperationAction.START, Name = "approval" } + new() { Id = "op-cb-1", Type = OperationTypes.Callback, Action = "START", Name = "approval" } }); var op = store.GetOperation(Arn, "op-cb-1"); @@ -186,13 +185,13 @@ public void Process_IncreasesTokenEachCall() { var (store, processor) = Create(skipTime: false); - var (t1, _) = processor.Process(Arn, null, new List + var (t1, _) = processor.Process(Arn, null, new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "START" } }); - var (t2, _) = processor.Process(Arn, t1, new List + var (t2, _) = processor.Process(Arn, t1, new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.SUCCEED } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "SUCCEED" } }); Assert.NotEqual(t1, t2); @@ -203,10 +202,10 @@ public void Process_ReturnsNewOperations() { var (_, processor) = Create(skipTime: false); - var (_, newOps) = processor.Process(Arn, null, new List + var (_, newOps) = processor.Process(Arn, null, new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START, Name = "s1" }, - new() { Id = "op-2", Type = OperationTypes.Wait, Action = OperationAction.START, WaitOptions = new WaitOptions { WaitSeconds = 10 } } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "START", Name = "s1" }, + new() { Id = "op-2", Type = OperationTypes.Wait, Action = "START", WaitSeconds = 10 } }); Assert.Equal(2, newOps.Count); @@ -219,9 +218,9 @@ public void Process_ContextStart_SetsContextDetails() { var (store, processor) = Create(skipTime: false); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-ctx", Type = OperationTypes.Context, Action = OperationAction.START, Name = "parallel_batch", SubType = "Parallel" } + new() { Id = "op-ctx", Type = OperationTypes.Context, Action = "START", Name = "parallel_batch", SubType = "Parallel" } }); var op = store.GetOperation(Arn, "op-ctx"); @@ -235,9 +234,9 @@ public void Process_ChainedInvokeStart_SetsDetails() { var (store, processor) = Create(skipTime: false); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-inv", Type = OperationTypes.ChainedInvoke, Action = OperationAction.START, Name = "process-payment" } + new() { Id = "op-inv", Type = OperationTypes.ChainedInvoke, Action = "START", Name = "process-payment" } }); var op = store.GetOperation(Arn, "op-inv"); @@ -245,16 +244,41 @@ public void Process_ChainedInvokeStart_SetsDetails() Assert.NotNull(op.ChainedInvokeDetails); } + [Fact] + public void Process_ChainedInvokeStart_WithFunctionName_RecordsPendingInvoke() + { + var (_, processor) = Create(skipTime: false); + + processor.Process(Arn, null, new List + { + new() + { + Id = "op-inv", Type = OperationTypes.ChainedInvoke, Action = "START", + Name = "process-payment", Payload = """{"amount":10}""", + ChainedInvokeFunctionName = "payment-fn" + } + }); + + var pending = processor.DrainPendingInvokes(); + var invoke = Assert.Single(pending); + Assert.Equal("op-inv", invoke.OperationId); + Assert.Equal("payment-fn", invoke.FunctionName); + Assert.Equal("""{"amount":10}""", invoke.Payload); + + // Draining is one-shot: a second drain returns nothing. + Assert.Empty(processor.DrainPendingInvokes()); + } + [Fact] public void Process_MultipleUpdatesInBatch_AllApplied() { var (store, processor) = Create(skipTime: true); - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START, Name = "a" }, - new() { Id = "op-2", Type = OperationTypes.Step, Action = OperationAction.START, Name = "b" }, - new() { Id = "op-3", Type = OperationTypes.Wait, Action = OperationAction.START, WaitOptions = new WaitOptions { WaitSeconds = 60 } } + new() { Id = "op-1", Type = OperationTypes.Step, Action = "START", Name = "a" }, + new() { Id = "op-2", Type = OperationTypes.Step, Action = "START", Name = "b" }, + new() { Id = "op-3", Type = OperationTypes.Wait, Action = "START", WaitSeconds = 60 } }); Assert.Equal(3, store.OperationCount(Arn)); @@ -271,9 +295,9 @@ public void Process_WaitForCondition_Retry_WithSkipTime_SetsReady() // SubType=WaitForCondition (see WaitForConditionOperation.OperationType). // A STEP START is NOT time-skipped to Succeeded — only WAIT timers are — // so the op stays Started and the condition is genuinely re-evaluated. - processor.Process(Arn, null, new List + processor.Process(Arn, null, new List { - new() { Id = "op-wfc", Type = OperationTypes.Step, Action = OperationAction.START, SubType = OperationSubTypes.WaitForCondition } + new() { Id = "op-wfc", Type = OperationTypes.Step, Action = "START", SubType = OperationSubTypes.WaitForCondition } }); var op = store.GetOperation(Arn, "op-wfc"); @@ -281,15 +305,39 @@ public void Process_WaitForCondition_Retry_WithSkipTime_SetsReady() // A RETRY (condition not yet met) becomes immediately READY under SkipTime, // so the next replay re-runs the check without waiting for the poll delay. - processor.Process(Arn, "1", new List + processor.Process(Arn, "1", new List { - new() { Id = "op-wfc", Type = OperationTypes.Step, Action = OperationAction.RETRY, SubType = OperationSubTypes.WaitForCondition, StepOptions = new StepOptions { NextAttemptDelaySeconds = 10 } } + new() { Id = "op-wfc", Type = OperationTypes.Step, Action = "RETRY", SubType = OperationSubTypes.WaitForCondition, NextAttemptDelaySeconds = 10 } }); op = store.GetOperation(Arn, "op-wfc"); Assert.Equal(OperationStatuses.Ready, op!.Status); } + [Fact] + public void Process_LiveSkipTimeProvider_TogglesAtRuntime() + { + // The Func overload is read on each checkpoint, so flipping the flag between + // checkpoints changes whether subsequent WAIT starts are folded to Succeeded — this is + // the behavior the Test Tool's runtime time-skip toggle depends on. + var store = new InMemoryOperationStore(); + var skipTime = false; + var processor = new CheckpointProcessor(store, () => skipTime); + + processor.Process(Arn, null, new List + { + new() { Id = "wait-1", Type = OperationTypes.Wait, Action = "START", WaitSeconds = 3600 } + }); + Assert.Equal(OperationStatuses.Started, store.GetOperation(Arn, "wait-1")!.Status); + + skipTime = true; + processor.Process(Arn, "1", new List + { + new() { Id = "wait-2", Type = OperationTypes.Wait, Action = "START", WaitSeconds = 3600 } + }); + Assert.Equal(OperationStatuses.Succeeded, store.GetOperation(Arn, "wait-2")!.Status); + } + private static (InMemoryOperationStore Store, CheckpointProcessor Processor) Create(bool skipTime) { var store = new InMemoryOperationStore(); diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/InMemoryOperationStoreTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/InMemoryOperationStoreTests.cs similarity index 91% rename from Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/InMemoryOperationStoreTests.cs rename to Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/InMemoryOperationStoreTests.cs index c49fecf4a..9060b328c 100644 --- a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/InMemoryOperationStoreTests.cs +++ b/Libraries/test/Amazon.Lambda.DurableExecution.LocalEmulation.Tests/InMemoryOperationStoreTests.cs @@ -1,10 +1,11 @@ // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 -using Amazon.Lambda.DurableExecution.Testing; +using Amazon.Lambda.DurableExecution; +using Amazon.Lambda.DurableExecution.LocalEmulation; using Xunit; -namespace Amazon.Lambda.DurableExecution.Testing.Tests; +namespace Amazon.Lambda.DurableExecution.LocalEmulation.Tests; public class InMemoryOperationStoreTests { @@ -59,6 +60,17 @@ public void GetOperation_ReturnsNull_WhenNotFound() Assert.Null(store.GetOperation("arn:test", "nonexistent")); } + [Fact] + public void Exists_TrueOnceExecutionSeen() + { + var store = new InMemoryOperationStore(); + Assert.False(store.Exists("arn:test")); + + store.Upsert("arn:test", new Operation { Id = "op-1", Type = OperationTypes.Step }); + + Assert.True(store.Exists("arn:test")); + } + [Fact] public void IncrementToken_IncrementsCounter() { diff --git a/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/SdkOperationUpdateMapperTests.cs b/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/SdkOperationUpdateMapperTests.cs new file mode 100644 index 000000000..4ab11fe0e --- /dev/null +++ b/Libraries/test/Amazon.Lambda.DurableExecution.Testing.Tests/SdkOperationUpdateMapperTests.cs @@ -0,0 +1,101 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +using Amazon.Lambda.DurableExecution.Testing; +using Amazon.Lambda.Model; +using Xunit; + +namespace Amazon.Lambda.DurableExecution.Testing.Tests; + +/// +/// Covers the boundary adapter that flattens the AWSSDK OperationUpdate into the neutral +/// OperationUpdateInput the shared kernel consumes. The state-machine behavior itself is +/// tested in the LocalEmulation kernel test project; here we only assert the field-by-field mapping +/// (ConstantClass unwrapping, nested option flattening, error conversion). +/// +public class SdkOperationUpdateMapperTests +{ + [Fact] + public void ToInput_UnwrapsConstantClassMembers() + { + var update = new OperationUpdate + { + Id = "op-1", + ParentId = "parent-1", + Name = "validate", + Type = OperationTypes.Step, + SubType = OperationSubTypes.WaitForCondition, + Action = OperationAction.RETRY, + Payload = """{"x":1}""" + }; + + var input = SdkOperationUpdateMapper.ToInput(update); + + Assert.Equal("op-1", input.Id); + Assert.Equal("parent-1", input.ParentId); + Assert.Equal("validate", input.Name); + Assert.Equal(OperationTypes.Step, input.Type); + Assert.Equal(OperationSubTypes.WaitForCondition, input.SubType); + Assert.Equal("RETRY", input.Action); + Assert.Equal("""{"x":1}""", input.Payload); + } + + [Fact] + public void ToInput_FlattensNestedOptions() + { + var update = new OperationUpdate + { + Id = "op-1", + Type = OperationTypes.Wait, + Action = OperationAction.START, + StepOptions = new StepOptions { NextAttemptDelaySeconds = 7 }, + WaitOptions = new WaitOptions { WaitSeconds = 300 }, + ChainedInvokeOptions = new ChainedInvokeOptions { FunctionName = "payment-fn" } + }; + + var input = SdkOperationUpdateMapper.ToInput(update); + + Assert.Equal(7, input.NextAttemptDelaySeconds); + Assert.Equal(300, input.WaitSeconds); + Assert.Equal("payment-fn", input.ChainedInvokeFunctionName); + } + + [Fact] + public void ToInput_MapsError() + { + var update = new OperationUpdate + { + Id = "op-1", + Type = OperationTypes.Step, + Action = OperationAction.FAIL, + Error = new Amazon.Lambda.Model.ErrorObject + { + ErrorType = "TestEx", + ErrorMessage = "boom", + ErrorData = "detail", + StackTrace = new List { "frame-1", "frame-2" } + } + }; + + var input = SdkOperationUpdateMapper.ToInput(update); + + Assert.NotNull(input.Error); + Assert.Equal("TestEx", input.Error!.ErrorType); + Assert.Equal("boom", input.Error.ErrorMessage); + Assert.Equal("detail", input.Error.ErrorData); + Assert.Equal(new[] { "frame-1", "frame-2" }, input.Error.StackTrace); + } + + [Fact] + public void ToInput_NullOptionalsMapToNull() + { + var update = new OperationUpdate { Id = "op-1", Type = OperationTypes.Step, Action = OperationAction.START }; + + var input = SdkOperationUpdateMapper.ToInput(update); + + Assert.Null(input.Error); + Assert.Null(input.NextAttemptDelaySeconds); + Assert.Null(input.WaitSeconds); + Assert.Null(input.ChainedInvokeFunctionName); + } +}