Add durable execution Step + Wait end-to-end #2360
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| { | ||
| "Projects": [ | ||
| { | ||
| "Name": "Amazon.Lambda.RuntimeSupport", | ||
| "Type": "Minor", | ||
| "ChangelogMessages": [ | ||
| "Add LambdaBootstrapBuilder.Create(Func<Stream, ILambdaContext, Task<Stream>>, ILambdaSerializer) overload (and matching HandlerWrapper.GetHandlerWrapper) so stream-stream handlers can expose a serializer via ILambdaContext.Serializer. Enables frameworks that own envelope (de)serialization but delegate inner-payload (de)serialization to a user-supplied serializer." | ||
| ] | ||
| } | ||
| ] | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -41,3 +41,6 @@ global.json | |
|
|
||
| **/cdk.out/** | ||
| **/.DS_Store | ||
|
|
||
| # JetBrains Rider per-project cache | ||
| **/*.lscache | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| using Amazon.Lambda.Core; | ||
| using Amazon.Lambda.DurableExecution.Internal; | ||
| using Microsoft.Extensions.Logging; | ||
| using Microsoft.Extensions.Logging.Abstractions; | ||
|
|
||
| namespace Amazon.Lambda.DurableExecution; | ||
|
|
||
| /// <summary> | ||
| /// Implementation of <see cref="IDurableContext"/>. Constructs and dispatches | ||
| /// per-operation classes (<see cref="StepOperation{T}"/>, <see cref="WaitOperation"/>); | ||
| /// the replay logic lives in those classes. | ||
| /// </summary> | ||
| internal sealed class DurableContext : IDurableContext | ||
| { | ||
| private readonly ExecutionState _state; | ||
| private readonly TerminationManager _terminationManager; | ||
| private readonly OperationIdGenerator _idGenerator; | ||
| private readonly string _durableExecutionArn; | ||
| private readonly CheckpointBatcher? _batcher; | ||
|
|
||
| public DurableContext( | ||
| ExecutionState state, | ||
| TerminationManager terminationManager, | ||
| OperationIdGenerator idGenerator, | ||
| string durableExecutionArn, | ||
| ILambdaContext lambdaContext, | ||
| CheckpointBatcher? batcher = null) | ||
| { | ||
| _state = state; | ||
| _terminationManager = terminationManager; | ||
| _idGenerator = idGenerator; | ||
| _durableExecutionArn = durableExecutionArn; | ||
| _batcher = batcher; | ||
| LambdaContext = lambdaContext; | ||
| } | ||
|
|
||
| // Replay-safe logger ships in a follow-up PR; see IDurableContext.Logger doc. | ||
| public ILogger Logger => NullLogger.Instance; | ||
| public IExecutionContext ExecutionContext => new DurableExecutionContext(_durableExecutionArn); | ||
| public ILambdaContext LambdaContext { get; } | ||
|
GarrettBeatty marked this conversation as resolved.
|
||
|
|
||
| public Task<T> StepAsync<T>( | ||
| Func<IStepContext, Task<T>> func, | ||
| string? name = null, | ||
| StepConfig? config = null, | ||
| CancellationToken cancellationToken = default) | ||
| => RunStep(func, name, config, cancellationToken); | ||
|
|
||
| public async Task StepAsync( | ||
| Func<IStepContext, Task> func, | ||
| string? name = null, | ||
| StepConfig? config = null, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| // Void steps don't carry a meaningful payload — wrap with an object?-typed | ||
| // step that always returns null. The serializer isn't actually invoked | ||
| // with a non-null value, so any registered ILambdaSerializer suffices. | ||
| await RunStep<object?>( | ||
| async (ctx) => { await func(ctx); return null; }, | ||
| name, config, cancellationToken); | ||
| } | ||
|
|
||
| private Task<T> RunStep<T>( | ||
| Func<IStepContext, Task<T>> func, | ||
| string? name, | ||
| StepConfig? config, | ||
| CancellationToken cancellationToken) | ||
| { | ||
| var serializer = LambdaContext.Serializer | ||
| ?? throw new InvalidOperationException( | ||
| "No ILambdaSerializer is registered on ILambdaContext.Serializer. " + | ||
| "In the class library programming model, register one with " + | ||
| "[assembly: LambdaSerializer(typeof(...))]. In an executable / custom " + | ||
| "runtime, pass it to LambdaBootstrapBuilder.Create(handler, serializer). " + | ||
| "In tests, set TestLambdaContext.Serializer."); | ||
|
|
||
| var operationId = _idGenerator.NextId(); | ||
| var op = new StepOperation<T>( | ||
| operationId, name, func, config, serializer, Logger, | ||
| _state, _terminationManager, _durableExecutionArn, _batcher); | ||
| return op.ExecuteAsync(cancellationToken); | ||
| } | ||
|
|
||
| public Task WaitAsync( | ||
| TimeSpan duration, | ||
| string? name = null, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| // Service timer granularity is 1 second; sub-second waits would round to 0. | ||
| // WaitOptions.WaitSeconds is integer in [1, 31_622_400] (1 second to ~1 year). | ||
| if (duration < TimeSpan.FromSeconds(1)) | ||
| throw new ArgumentOutOfRangeException(nameof(duration), duration, "Wait duration must be at least 1 second."); | ||
|
|
||
| if (duration > TimeSpan.FromSeconds(31_622_400)) | ||
|
Contributor
Author
There was a problem hiding this comment. should we be validating this on our end? |
||
| throw new ArgumentOutOfRangeException(nameof(duration), duration, "Wait duration must be at most 31,622,400 seconds (~1 year)."); | ||
|
|
||
| cancellationToken.ThrowIfCancellationRequested(); | ||
|
|
||
| var operationId = _idGenerator.NextId(); | ||
| var waitSeconds = (int)Math.Max(1, Math.Ceiling(duration.TotalSeconds)); | ||
|
GarrettBeatty marked this conversation as resolved.
|
||
| var op = new WaitOperation( | ||
| operationId, name, waitSeconds, | ||
| _state, _terminationManager, _durableExecutionArn, _batcher); | ||
| return op.ExecuteAsync(cancellationToken); | ||
| } | ||
| } | ||
|
|
||
| internal sealed class DurableExecutionContext : IExecutionContext | ||
| { | ||
| public DurableExecutionContext(string durableExecutionArn) | ||
| { | ||
| DurableExecutionArn = durableExecutionArn; | ||
| } | ||
|
|
||
| public string DurableExecutionArn { get; } | ||
| } | ||
|
|
||
| internal sealed class StepContext : IStepContext | ||
| { | ||
| public StepContext(string operationId, int attemptNumber, ILogger logger) | ||
| { | ||
| OperationId = operationId; | ||
| AttemptNumber = attemptNumber; | ||
| Logger = logger; | ||
| } | ||
|
|
||
| public ILogger Logger { get; } | ||
| public int AttemptNumber { get; } | ||
| public string OperationId { get; } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| using System.IO; | ||
| using System.Text; | ||
| using System.Text.Json; | ||
| using System.Threading; | ||
| using Amazon.Lambda; | ||
| using Amazon.Lambda.Core; | ||
| using Amazon.Lambda.DurableExecution.Internal; | ||
| using Amazon.Lambda.DurableExecution.Services; | ||
| using Amazon.Lambda.Model; | ||
| using Amazon.Runtime; | ||
|
|
||
| namespace Amazon.Lambda.DurableExecution; | ||
|
|
||
| /// <summary> | ||
| /// AOT-friendly entry point for a durable workflow. Owns (de)serialization of | ||
| /// the wire envelope so users only register their own POCO types in their | ||
| /// <c>JsonSerializerContext</c> — the library's <see cref="DurableEnvelopeJsonContext"/> | ||
| /// handles envelope JSON, the user's <see cref="ILambdaSerializer"/> (read from | ||
| /// <see cref="ILambdaContext.Serializer"/>) handles only <typeparamref name="TInput"/> | ||
| /// and <typeparamref name="TOutput"/>. | ||
| /// </summary> | ||
| /// <typeparam name="TInput">The workflow's input payload type.</typeparam> | ||
| /// <typeparam name="TOutput">The workflow's return type.</typeparam> | ||
| /// <example> | ||
| /// <code> | ||
| /// private static readonly DurableEntryPoint<OrderEvent, OrderResult> _entry = new(MyWorkflow); | ||
| /// | ||
| /// static async Task Main() | ||
| /// { | ||
| /// await LambdaBootstrapBuilder | ||
| /// .Create(_entry.InvokeAsync, new SourceGeneratorLambdaJsonSerializer<MyJsonContext>()) | ||
| /// .Build() | ||
| /// .RunAsync(); | ||
| /// } | ||
| /// </code> | ||
| /// </example> | ||
| public sealed class DurableEntryPoint<TInput, TOutput> | ||
| { | ||
| private static readonly Lazy<IAmazonLambda> _cachedLambdaClient = | ||
| new(() => new AmazonLambdaClient(), LazyThreadSafetyMode.ExecutionAndPublication); | ||
|
|
||
| private readonly Func<TInput, IDurableContext, Task<TOutput>> _workflow; | ||
| private readonly IAmazonLambda _lambdaClient; | ||
|
|
||
| /// <summary> | ||
| /// Creates an entry point that uses a default <see cref="AmazonLambdaClient"/>, | ||
| /// constructed lazily and cached process-wide. | ||
| /// </summary> | ||
| public DurableEntryPoint(Func<TInput, IDurableContext, Task<TOutput>> workflow) | ||
| : this(workflow, _cachedLambdaClient.Value) | ||
| { | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Creates an entry point that uses the supplied <see cref="IAmazonLambda"/> client | ||
| /// for checkpoint and state-fetch calls. | ||
| /// </summary> | ||
| public DurableEntryPoint(Func<TInput, IDurableContext, Task<TOutput>> workflow, IAmazonLambda lambdaClient) | ||
| { | ||
| _workflow = workflow ?? throw new ArgumentNullException(nameof(workflow)); | ||
| _lambdaClient = lambdaClient ?? throw new ArgumentNullException(nameof(lambdaClient)); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Lambda handler entry point. Register this method with <c>LambdaBootstrapBuilder</c> | ||
| /// alongside an <see cref="ILambdaSerializer"/> that knows how to (de)serialize | ||
| /// <typeparamref name="TInput"/> / <typeparamref name="TOutput"/>. | ||
| /// </summary> | ||
| public async Task<Stream> InvokeAsync(Stream input, ILambdaContext context) | ||
| { | ||
| var output = await DurableEntryPointCore.InvokeAsync(_workflow, input, context, _lambdaClient); | ||
| var ms = new MemoryStream(); | ||
| JsonSerializer.Serialize(ms, output, DurableEnvelopeJsonContext.Default.DurableExecutionInvocationOutput); | ||
| ms.Position = 0; | ||
| return ms; | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// AOT-friendly entry point for a void durable workflow. | ||
| /// See <see cref="DurableEntryPoint{TInput,TOutput}"/> for details. | ||
| /// </summary> | ||
| public sealed class DurableEntryPoint<TInput> | ||
| { | ||
| private readonly DurableEntryPoint<TInput, object?> _inner; | ||
|
|
||
| /// <summary> | ||
| /// Creates an entry point that uses a default <see cref="AmazonLambdaClient"/>, | ||
| /// constructed lazily and cached process-wide. | ||
| /// </summary> | ||
| public DurableEntryPoint(Func<TInput, IDurableContext, Task> workflow) | ||
| { | ||
| if (workflow == null) throw new ArgumentNullException(nameof(workflow)); | ||
| _inner = new DurableEntryPoint<TInput, object?>(async (i, c) => { await workflow(i, c); return null; }); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Creates an entry point that uses the supplied <see cref="IAmazonLambda"/> client | ||
| /// for checkpoint and state-fetch calls. | ||
| /// </summary> | ||
| public DurableEntryPoint(Func<TInput, IDurableContext, Task> workflow, IAmazonLambda lambdaClient) | ||
| { | ||
| if (workflow == null) throw new ArgumentNullException(nameof(workflow)); | ||
| _inner = new DurableEntryPoint<TInput, object?>( | ||
| async (i, c) => { await workflow(i, c); return null; }, | ||
| lambdaClient); | ||
| } | ||
|
|
||
| /// <inheritdoc cref="DurableEntryPoint{TInput,TOutput}.InvokeAsync"/> | ||
| public Task<Stream> InvokeAsync(Stream input, ILambdaContext context) | ||
| => _inner.InvokeAsync(input, context); | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.