diff --git a/Directory.Packages.props b/Directory.Packages.props index 8388030fa..7a24d9b7f 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -23,6 +23,7 @@ + @@ -33,6 +34,11 @@ + + + + + diff --git a/Microsoft.DurableTask.sln b/Microsoft.DurableTask.sln index 26c2e80de..383f2e158 100644 --- a/Microsoft.DurableTask.sln +++ b/Microsoft.DurableTask.sln @@ -93,6 +93,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ScheduleWebApp", "samples\S EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ScheduledTasks.Tests", "test\ScheduledTasks.Tests\ScheduledTasks.Tests.csproj", "{D2779F32-A548-44F8-B60A-6AC018966C79}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "LargePayloadConsoleApp", "samples\LargePayloadConsoleApp\LargePayloadConsoleApp.csproj", "{6EB9D002-62C8-D6C1-62A8-14C54CA6DBBC}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "AzureBlobPayloads", "src\Extensions\AzureBlobPayloads\AzureBlobPayloads.csproj", "{FE1DA748-D6DB-E168-BC42-6DBBCEAF229C}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -247,6 +251,14 @@ Global {D2779F32-A548-44F8-B60A-6AC018966C79}.Debug|Any CPU.Build.0 = Debug|Any CPU {D2779F32-A548-44F8-B60A-6AC018966C79}.Release|Any CPU.ActiveCfg = Release|Any CPU {D2779F32-A548-44F8-B60A-6AC018966C79}.Release|Any CPU.Build.0 = Release|Any CPU + {6EB9D002-62C8-D6C1-62A8-14C54CA6DBBC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {6EB9D002-62C8-D6C1-62A8-14C54CA6DBBC}.Debug|Any CPU.Build.0 = Debug|Any CPU + {6EB9D002-62C8-D6C1-62A8-14C54CA6DBBC}.Release|Any CPU.ActiveCfg = Release|Any CPU + {6EB9D002-62C8-D6C1-62A8-14C54CA6DBBC}.Release|Any CPU.Build.0 = Release|Any CPU + {FE1DA748-D6DB-E168-BC42-6DBBCEAF229C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {FE1DA748-D6DB-E168-BC42-6DBBCEAF229C}.Debug|Any CPU.Build.0 = Debug|Any CPU + {FE1DA748-D6DB-E168-BC42-6DBBCEAF229C}.Release|Any CPU.ActiveCfg = Release|Any CPU + {FE1DA748-D6DB-E168-BC42-6DBBCEAF229C}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -293,6 +305,8 @@ Global {A89B766C-987F-4C9F-8937-D0AB9FE640C8} = {EFF7632B-821E-4CFC-B4A0-ED4B24296B17} {100348B5-4D97-4A3F-B777-AB14F276F8FE} = {EFF7632B-821E-4CFC-B4A0-ED4B24296B17} {D2779F32-A548-44F8-B60A-6AC018966C79} = {E5637F81-2FB9-4CD7-900D-455363B142A7} + {6EB9D002-62C8-D6C1-62A8-14C54CA6DBBC} = {EFF7632B-821E-4CFC-B4A0-ED4B24296B17} + {FE1DA748-D6DB-E168-BC42-6DBBCEAF229C} = {8AFC9781-F6F1-4696-BB4A-9ED7CA9D612B} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {AB41CB55-35EA-4986-A522-387AB3402E71} diff --git a/samples/LargePayloadConsoleApp/LargePayloadConsoleApp.csproj b/samples/LargePayloadConsoleApp/LargePayloadConsoleApp.csproj new file mode 100644 index 000000000..b0f2914c7 --- /dev/null +++ b/samples/LargePayloadConsoleApp/LargePayloadConsoleApp.csproj @@ -0,0 +1,24 @@ + + + + Exe + net8.0 + enable + + + + + + + + + + + + + + + + + + diff --git a/samples/LargePayloadConsoleApp/Program.cs b/samples/LargePayloadConsoleApp/Program.cs new file mode 100644 index 000000000..87258e1ea --- /dev/null +++ b/samples/LargePayloadConsoleApp/Program.cs @@ -0,0 +1,204 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Client.Entities; +using Microsoft.DurableTask.Entities; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; + +// Demonstrates Large Payload Externalization using Azure Blob Storage. +// This sample uses Azurite/emulator by default via UseDevelopmentStorage=true. + +HostApplicationBuilder builder = Host.CreateApplicationBuilder(args); + +// Connection string for Durable Task Scheduler +string schedulerConnectionString = builder.Configuration.GetValue("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? throw new InvalidOperationException("Missing required configuration 'DURABLE_TASK_SCHEDULER_CONNECTION_STRING'"); + +// Configure Durable Task client with Durable Task Scheduler and externalized payloads +builder.Services.AddDurableTaskClient(b => +{ + b.UseDurableTaskScheduler(schedulerConnectionString); + // Ensure entity APIs are enabled for the client + b.Configure(o => { o.EnableEntitySupport = true; o.EnableLargePayloadSupport = true; }); + b.UseExternalizedPayloads(opts => + { + // Keep threshold small to force externalization for demo purposes + opts.ExternalizeThresholdBytes = 1024; // 1KB + opts.ConnectionString = builder.Configuration.GetValue("DURABLETASK_STORAGE") ?? "UseDevelopmentStorage=true"; + opts.ContainerName = builder.Configuration.GetValue("DURABLETASK_PAYLOAD_CONTAINER"); + }); +}); + +// Configure Durable Task worker with tasks and externalized payloads +builder.Services.AddDurableTaskWorker(b => +{ + b.UseDurableTaskScheduler(schedulerConnectionString); + b.AddTasks(tasks => + { + // Orchestrator: call activity first, return its output (should equal original input) + tasks.AddOrchestratorFunc("LargeInputEcho", async (ctx, input) => + { + string echoed = await ctx.CallActivityAsync("Echo", input); + return echoed; + }); + + // Activity: validate it receives raw input (not token) and return it + tasks.AddActivityFunc("Echo", (ctx, value) => + { + if (value is null) + { + return string.Empty; + } + + // If we ever see a token in the activity, externalization is not being resolved correctly. + if (value.StartsWith("blob:v1:", StringComparison.Ordinal)) + { + throw new InvalidOperationException("Activity received a payload token instead of raw input."); + } + + return value; + }); + + // Entity samples + // 1) Large entity operation input (worker externalizes input; entity receives resolved payload) + tasks.AddOrchestratorFunc( + "LargeEntityOperationInput", + (ctx, _) => ctx.Entities.CallEntityAsync( + new EntityInstanceId(nameof(EchoLengthEntity), "1"), + operationName: "EchoLength", + input: new string('E', 700 * 1024))); + tasks.AddEntity(nameof(EchoLengthEntity)); + + // 2) Large entity operation output (worker externalizes output; orchestrator reads resolved payload) + tasks.AddOrchestratorFunc( + "LargeEntityOperationOutput", + async (ctx, _) => (await ctx.Entities.CallEntityAsync( + new EntityInstanceId(nameof(LargeResultEntity), "1"), + operationName: "Produce", + input: 850 * 1024)).Length); + tasks.AddEntity(nameof(LargeResultEntity)); + + // 3) Large entity state (worker externalizes state; client resolves on query) + tasks.AddOrchestratorFunc( + "LargeEntityState", + async (ctx, _) => + { + await ctx.Entities.CallEntityAsync( + new EntityInstanceId(nameof(StateEntity), "1"), + operationName: "Set", + input: new string('S', 900 * 1024)); + return null; + }); + tasks.AddEntity(nameof(StateEntity)); + }); + b.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // mirror client + opts.ConnectionString = builder.Configuration.GetValue("DURABLETASK_STORAGE") ?? "UseDevelopmentStorage=true"; + opts.ContainerName = builder.Configuration.GetValue("DURABLETASK_PAYLOAD_CONTAINER"); + }); + // Ensure entity APIs are enabled for the worker + b.Configure(o => { o.EnableEntitySupport = true; o.EnableLargePayloadSupport = true; }); +}); + +IHost host = builder.Build(); +await host.StartAsync(); + +await using DurableTaskClient client = host.Services.GetRequiredService(); + +// Option A: Directly pass an oversized input to orchestration to trigger externalization +string largeInput = new string('B', 1024 * 1024); // 1MB +string instanceId = await client.ScheduleNewOrchestrationInstanceAsync("LargeInputEcho", largeInput); +Console.WriteLine($"Started orchestration with direct large input. Instance: {instanceId}"); + + +using CancellationTokenSource cts = new CancellationTokenSource(TimeSpan.FromSeconds(120)); +OrchestrationMetadata result = await client.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + cts.Token); + +Console.WriteLine($"RuntimeStatus: {result.RuntimeStatus}"); +string deserializedInput = result.ReadInputAs() ?? string.Empty; +string deserializedOutput = result.ReadOutputAs() ?? string.Empty; + +Console.WriteLine($"SerializedInput: {result.SerializedInput}"); +Console.WriteLine($"SerializedOutput: {result.SerializedOutput}"); +Console.WriteLine($"Deserialized input equals original: {deserializedInput == largeInput}"); +Console.WriteLine($"Deserialized output equals original: {deserializedOutput == largeInput}"); +Console.WriteLine($"Deserialized input length: {deserializedInput.Length}"); + + + +// Run entity samples +Console.WriteLine(); +Console.WriteLine("Running LargeEntityOperationInput..."); +string largeEntityInput = new string('E', 700 * 1024); // 700KB +string entityInputInstance = await client.ScheduleNewOrchestrationInstanceAsync("LargeEntityOperationInput"); +OrchestrationMetadata entityInputResult = await client.WaitForInstanceCompletionAsync(entityInputInstance, getInputsAndOutputs: true, cts.Token); +int entityInputLength = entityInputResult.ReadOutputAs(); +Console.WriteLine($"Status: {entityInputResult.RuntimeStatus}, Output length: {entityInputLength}"); +Console.WriteLine($"Deserialized input length equals original: {entityInputLength == largeEntityInput.Length}"); + +Console.WriteLine(); +Console.WriteLine("Running LargeEntityOperationOutput..."); +int largeEntityOutputLength = 850 * 1024; // 850KB +string entityOutputInstance = await client.ScheduleNewOrchestrationInstanceAsync("LargeEntityOperationOutput"); +OrchestrationMetadata entityOutputResult = await client.WaitForInstanceCompletionAsync(entityOutputInstance, getInputsAndOutputs: true, cts.Token); +int entityOutputLength = entityOutputResult.ReadOutputAs(); +Console.WriteLine($"Status: {entityOutputResult.RuntimeStatus}, Output length: {entityOutputLength}"); +Console.WriteLine($"Deserialized output length equals original: {entityOutputLength == largeEntityOutputLength}"); + +Console.WriteLine(); +Console.WriteLine("Running LargeEntityState and querying state..."); +string largeEntityState = new string('S', 900 * 1024); // 900KB +string entityStateInstance = await client.ScheduleNewOrchestrationInstanceAsync("LargeEntityState"); +OrchestrationMetadata entityStateOrch = await client.WaitForInstanceCompletionAsync(entityStateInstance, getInputsAndOutputs: true, cts.Token); +Console.WriteLine($"Status: {entityStateOrch.RuntimeStatus}"); +EntityMetadata? state = await client.Entities.GetEntityAsync(new EntityInstanceId(nameof(StateEntity), "1"), includeState: true); +int stateLength = state?.State?.Length ?? 0; +Console.WriteLine($"State length: {stateLength}"); +Console.WriteLine($"Deserialized state equals original: {state?.State == largeEntityState}"); + + + + + +public class EchoLengthEntity : TaskEntity +{ + public int EchoLength(string input) + { + return input.Length; + } +} + +public class LargeResultEntity : TaskEntity +{ + public string Produce(int length) + { + return new string('R', length); + } +} + +public class StateEntity : TaskEntity +{ + protected override string? InitializeState(TaskEntityOperation entityOperation) + { + // Avoid Activator.CreateInstance() which throws; start as null (no state) + return null; + } + + public void Set(string value) + { + this.State = value; + } +} + + + diff --git a/samples/LargePayloadConsoleApp/README.md b/samples/LargePayloadConsoleApp/README.md new file mode 100644 index 000000000..812098efd --- /dev/null +++ b/samples/LargePayloadConsoleApp/README.md @@ -0,0 +1,29 @@ +# Large Payload Externalization Sample + +This sample demonstrates configuring Durable Task to externalize large payloads to Azure Blob Storage using `UseExternalizedPayloads` on both client and worker, connecting via Durable Task Scheduler (no local sidecar). + +- Defaults to Azurite/Storage Emulator via `UseDevelopmentStorage=true`. +- Threshold is set to 1KB for demo, so even modest inputs are externalized. + +## Prerequisites + +- A Durable Task Scheduler connection string (e.g., from Azure portal) in `DURABLE_TASK_SCHEDULER_CONNECTION_STRING`. +- Optional: Run Azurite (if not using real Azure Storage) for payload storage tokens. + +## Configure + +Environment variables (optional): + +- `DURABLETASK_STORAGE`: Azure Storage connection string. Defaults to `UseDevelopmentStorage=true`. +- `DURABLETASK_PAYLOAD_CONTAINER`: Blob container name. Defaults to `durabletask-payloads`. + +## Run + +```bash +# from repo root +dotnet run --project samples/LargePayloadConsoleApp/LargePayloadConsoleApp.csproj +``` + +The app starts an orchestration with a 1MB input, which is externalized by the client and resolved by the worker. The console shows a token-like serialized input and a deserialized input length. + + diff --git a/samples/LargePayloadConsoleApp/run.ps1 b/samples/LargePayloadConsoleApp/run.ps1 new file mode 100644 index 000000000..466f07d72 --- /dev/null +++ b/samples/LargePayloadConsoleApp/run.ps1 @@ -0,0 +1,81 @@ +Param( + [Parameter(Mandatory = $true)] + [string]$SchedulerConnectionString, + + [string]$StorageConnectionString = "UseDevelopmentStorage=true", + + [string]$PayloadContainer = "durabletask-payloads", + + [switch]$StartAzurite, + + [switch]$VerboseLogging +) + +$ErrorActionPreference = "Stop" + +function Write-Info($msg) { + Write-Host "[info] $msg" +} + +function Start-AzuriteDocker { + param( + [string]$ContainerName = "durabletask-azurite" + ) + + if (-not (Get-Command docker -ErrorAction SilentlyContinue)) { + Write-Info "Docker not found; skipping Azurite startup." + return $false + } + + try { + $existing = (docker ps -a --filter "name=$ContainerName" --format "{{.ID}}") + if ($existing) { + Write-Info "Starting existing Azurite container '$ContainerName'..." + docker start $ContainerName | Out-Null + return $true + } + + Write-Info "Launching Azurite in Docker as '$ContainerName' on ports 10000-10002..." + docker run -d -p 10000:10000 -p 10001:10001 -p 10002:10002 --name $ContainerName mcr.microsoft.com/azure-storage/azurite | Out-Null + Start-Sleep -Seconds 2 + return $true + } + catch { + Write-Warning "Failed to start Azurite via Docker: $_" + return $false + } +} + +try { + # Set required/optional environment variables for the sample + $env:DURABLE_TASK_SCHEDULER_CONNECTION_STRING = $SchedulerConnectionString + $env:DURABLETASK_STORAGE = $StorageConnectionString + $env:DURABLETASK_PAYLOAD_CONTAINER = $PayloadContainer + + Write-Info "DURABLE_TASK_SCHEDULER_CONNECTION_STRING is set." + Write-Info "DURABLETASK_STORAGE = '$($env:DURABLETASK_STORAGE)'" + Write-Info "DURABLETASK_PAYLOAD_CONTAINER = '$($env:DURABLETASK_PAYLOAD_CONTAINER)'" + + if ($StartAzurite) { + $started = Start-AzuriteDocker + if ($started) { + Write-Info "Azurite is running (Docker)." + } + } + + $projectPath = Join-Path $PSScriptRoot "LargePayloadConsoleApp.csproj" + if (-not (Test-Path $projectPath)) { + throw "Project file not found at $projectPath" + } + + Write-Info "Running sample..." + $argsList = @("run", "--project", $projectPath) + if ($VerboseLogging) { $argsList += @("-v", "detailed") } + + & dotnet @argsList +} +catch { + Write-Error $_ + exit 1 +} + diff --git a/src/Abstractions/Converters/IPayloadStore.cs b/src/Abstractions/Converters/IPayloadStore.cs new file mode 100644 index 000000000..c2a6ff6cb --- /dev/null +++ b/src/Abstractions/Converters/IPayloadStore.cs @@ -0,0 +1,34 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Converters; + +/// +/// Abstraction for storing and retrieving large payloads out-of-band. +/// +public interface IPayloadStore +{ + /// + /// Uploads a payload and returns an opaque reference token that can be embedded in orchestration messages. + /// + /// The payload bytes. + /// Cancellation token. + /// Opaque reference token. + Task UploadAsync(ReadOnlyMemory payloadBytes, CancellationToken cancellationToken); + + /// + /// Downloads the payload referenced by the token. + /// + /// The opaque reference token. + /// Cancellation token. + /// Payload string. + Task DownloadAsync(string token, CancellationToken cancellationToken); + + /// + /// Returns true if the specified value appears to be a token understood by this store. + /// Implementations should not throw for unknown tokens. + /// + /// The value to check. + /// true if the value is a token issued by this store; otherwise, false. + bool IsKnownPayloadToken(string value); +} diff --git a/src/Abstractions/Converters/LargePayloadStorageOptions.cs b/src/Abstractions/Converters/LargePayloadStorageOptions.cs new file mode 100644 index 000000000..af005854b --- /dev/null +++ b/src/Abstractions/Converters/LargePayloadStorageOptions.cs @@ -0,0 +1,50 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +// Intentionally no DataAnnotations to avoid extra package requirements in minimal hosts. +namespace Microsoft.DurableTask.Converters; + +/// +/// Options for externalized payload storage, used by SDKs to store large payloads out-of-band. +/// +public sealed class LargePayloadStorageOptions +{ + /// + /// Initializes a new instance of the class. + /// Parameterless constructor required for options activation. + /// + public LargePayloadStorageOptions() + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The Azure Storage connection string to the customer's storage account. + public LargePayloadStorageOptions(string connectionString) + { + Check.NotNullOrEmpty(connectionString, nameof(connectionString)); + this.ConnectionString = connectionString; + } + + /// + /// Gets or sets the threshold in bytes at which payloads are externalized. Default is 900_000 bytes. + /// + public int ExternalizeThresholdBytes { get; set; } = 900_000; // leave headroom below 1MB + + /// + /// Gets or sets the Azure Storage connection string to the customer's storage account. Required. + /// + public string ConnectionString { get; set; } = string.Empty; + + /// + /// Gets or sets the blob container name to use for payloads. Defaults to "durabletask-payloads". + /// + public string ContainerName { get; set; } = "durabletask-payloads"; + + /// + /// Gets or sets a value indicating whether payloads should be gzip-compressed when stored. + /// Defaults to true for reduced storage and bandwidth. + /// + public bool CompressPayloads { get; set; } = true; +} diff --git a/src/Abstractions/DataConverter.cs b/src/Abstractions/DataConverter.cs index 3248761ac..78369f35b 100644 --- a/src/Abstractions/DataConverter.cs +++ b/src/Abstractions/DataConverter.cs @@ -13,6 +13,8 @@ namespace Microsoft.DurableTask; /// uses the JSON serializer from the System.Text.Json namespace. Currently only strings are supported as /// the serialized representation of data. Byte array payloads and streams are not supported by this abstraction. /// Note that these methods all accept null values, in which case the return value should also be null. +/// Implementations may choose to return a pointer or reference (such as an external token) to the data +/// instead of the actual serialized data itself. /// public abstract class DataConverter { @@ -47,4 +49,46 @@ public abstract class DataConverter /// [return: NotNullIfNotNull("data")] public virtual T? Deserialize(string? data) => (T?)(this.Deserialize(data, typeof(T)) ?? default); + + /// + /// Asynchronously serializes into a text string. + /// Default implementation delegates to . + /// + /// The value to be serialized. + /// Cancellation token. + /// A task whose result is the serialized string or null. + public virtual ValueTask SerializeAsync(object? value, CancellationToken cancellationToken = default) + { + return new ValueTask(this.Serialize(value)); + } + + /// + /// Asynchronously deserializes into an object of type . + /// Default implementation delegates to . + /// + /// The text data to be deserialized. + /// The type to deserialize to. + /// Cancellation token. + /// A task whose result is the deserialized value or null. + public virtual ValueTask DeserializeAsync( + string? data, + Type targetType, + CancellationToken cancellationToken = default) + { + return new ValueTask(this.Deserialize(data, targetType)); + } + + /// + /// Asynchronously deserializes into an object of type . + /// + /// The type to deserialize to. + /// The text data to be deserialized. + /// Cancellation token. + /// A task whose result is the deserialized value or null. + public virtual ValueTask DeserializeAsync( + string? data, + CancellationToken cancellationToken = default) + { + return new ValueTask(this.Deserialize(data)); + } } diff --git a/src/Abstractions/Entities/TaskEntityContext.cs b/src/Abstractions/Entities/TaskEntityContext.cs index 9de1ff46e..6ec68e5fd 100644 --- a/src/Abstractions/Entities/TaskEntityContext.cs +++ b/src/Abstractions/Entities/TaskEntityContext.cs @@ -53,4 +53,37 @@ public virtual string ScheduleNewOrchestration(TaskName name, StartOrchestration /// The instance id for the new orchestration. public abstract string ScheduleNewOrchestration( TaskName name, object? input = null, StartOrchestrationOptions? options = null); + + /// + /// Starts an orchestration. + /// + /// The name of the orchestration to start. + /// The input for the orchestration. + /// The options for starting the orchestration. + /// A task representing the asynchronous operation. + public virtual ValueTask ScheduleNewOrchestrationAsync( + TaskName name, + object? input = null, + StartOrchestrationOptions? options = null) + { + return new ValueTask(this.ScheduleNewOrchestration(name, input, options)); + } + + /// + /// Signals an entity operation asynchronously. + /// + /// The entity to signal. + /// The operation name. + /// The operation input. + /// The options to signal with. + /// A task representing the asynchronous operation. + public virtual ValueTask SignalEntityAsync( + EntityInstanceId id, + string operationName, + object? input = null, + SignalEntityOptions? options = null) + { + this.SignalEntity(id, operationName, input, options); + return default; + } } diff --git a/src/Abstractions/Entities/TaskEntityOperation.cs b/src/Abstractions/Entities/TaskEntityOperation.cs index 4f6963130..5ee438df6 100644 --- a/src/Abstractions/Entities/TaskEntityOperation.cs +++ b/src/Abstractions/Entities/TaskEntityOperation.cs @@ -42,6 +42,16 @@ public abstract class TaskEntityOperation /// The deserialized input type. public abstract object? GetInput(Type inputType); + /// + /// Gets the input for this operation. + /// + /// The type to deserialize the input as. + /// The deserialized input type. + public virtual Task GetInputAsync(Type inputType) + { + return Task.FromResult(this.GetInput(inputType)); + } + /// public override string ToString() { diff --git a/src/Abstractions/Entities/TaskEntityState.cs b/src/Abstractions/Entities/TaskEntityState.cs index 6f1a61f8e..9ee407579 100644 --- a/src/Abstractions/Entities/TaskEntityState.cs +++ b/src/Abstractions/Entities/TaskEntityState.cs @@ -35,6 +35,17 @@ public abstract class TaskEntityState return defaultValue; } + /// + /// Asynchronously gets the current state of the entity. This will return null if no state is present, regardless if + /// is a value-type or not. + /// + /// The type to retrieve the state as. + /// The entity state. + public virtual Task GetStateAsync(Type type) + { + return Task.FromResult(this.GetState(type)); + } + /// /// Gets the current state of the entity. This will return null if no state is present, regardless if /// is a value-type or not. @@ -48,4 +59,15 @@ public abstract class TaskEntityState /// /// The state to set. public abstract void SetState(object? state); + + /// + /// Asynchronously sets the entity state. Setting of null will delete entity state. + /// + /// The state to set. + /// A task representing the asynchronous operation. + public virtual Task SetStateAsync(object? state) + { + this.SetState(state); + return Task.CompletedTask; + } } diff --git a/src/Client/Core/DurableTaskClientOptions.cs b/src/Client/Core/DurableTaskClientOptions.cs index 05f19e7bd..575768bd6 100644 --- a/src/Client/Core/DurableTaskClientOptions.cs +++ b/src/Client/Core/DurableTaskClientOptions.cs @@ -12,6 +12,7 @@ public class DurableTaskClientOptions { DataConverter dataConverter = JsonDataConverter.Default; bool enableEntitySupport; + bool enableLargePayloadSupport; /// /// Gets or sets the version of orchestrations that will be created. @@ -69,6 +70,20 @@ public bool EnableEntitySupport } } + /// + /// Gets or sets a value indicating whether this client should support large payloads using async serialization/deserialization. + /// When enabled, the client will use async methods for serialization and deserialization to support externalized payloads. + /// + public bool EnableLargePayloadSupport + { + get => this.enableLargePayloadSupport; + set + { + this.enableLargePayloadSupport = value; + this.LargePayloadSupportExplicitlySet = true; + } + } + /// /// Gets a value indicating whether was explicitly set or not. /// @@ -85,6 +100,11 @@ public bool EnableEntitySupport /// internal bool EntitySupportExplicitlySet { get; private set; } + /// + /// Gets a value indicating whether was explicitly set or not. + /// + internal bool LargePayloadSupportExplicitlySet { get; private set; } + /// /// Applies these option values to another. /// @@ -104,6 +124,11 @@ internal void ApplyTo(DurableTaskClientOptions other) other.EnableEntitySupport = this.EnableEntitySupport; } + if (!other.LargePayloadSupportExplicitlySet) + { + other.EnableLargePayloadSupport = this.EnableLargePayloadSupport; + } + if (!string.IsNullOrWhiteSpace(this.DefaultVersion)) { other.DefaultVersion = this.DefaultVersion; diff --git a/src/Client/Core/OrchestrationMetadata.cs b/src/Client/Core/OrchestrationMetadata.cs index a1cf3d9fa..8955978a8 100644 --- a/src/Client/Core/OrchestrationMetadata.cs +++ b/src/Client/Core/OrchestrationMetadata.cs @@ -80,6 +80,12 @@ public OrchestrationMetadata(string name, string instanceId) /// The serialized custom status or null. public string? SerializedCustomStatus { get; init; } + /// + /// Gets a value indicating whether large payload support is enabled. + /// + /// true if large payload support is enabled; false otherwise. + public bool EnableLargePayloadSupport { get; init; } + /// /// Gets the tags associated with the orchestration instance. /// @@ -141,6 +147,7 @@ public OrchestrationMetadata(string name, string instanceId) "that are fetched with the option to include input data."); } + Check.ThrowIfLargePayloadEnabled(this.EnableLargePayloadSupport, nameof(this.ReadInputAs)); return this.DataConverter.Deserialize(this.SerializedInput); } @@ -167,6 +174,7 @@ public OrchestrationMetadata(string name, string instanceId) "that are fetched with the option to include output data."); } + Check.ThrowIfLargePayloadEnabled(this.EnableLargePayloadSupport, nameof(this.ReadOutputAs)); return this.DataConverter.Deserialize(this.SerializedOutput); } @@ -193,9 +201,106 @@ public OrchestrationMetadata(string name, string instanceId) + " objects that are fetched with the option to include input and output data."); } + Check.ThrowIfLargePayloadEnabled(this.EnableLargePayloadSupport, nameof(this.ReadCustomStatusAs)); return this.DataConverter.Deserialize(this.SerializedCustomStatus); } + /// + /// Asynchronously deserializes the orchestration's input into an object of the specified type. + /// + /// + /// This method can only be used when inputs and outputs are explicitly requested from the + /// or + /// method that produced + /// this object. + /// + /// The type to deserialize the orchestration input into. + /// Cancellation token. + /// Returns the deserialized input value. + /// + /// Thrown if this metadata object was fetched without the option to read inputs and outputs. + /// + public async ValueTask ReadInputAsAsync(CancellationToken cancellationToken = default) + { + if (!this.RequestedInputsAndOutputs) + { + throw new InvalidOperationException( + $"The {nameof(this.ReadInputAsAsync)} method can only be used on {nameof(OrchestrationMetadata)} objects " + + "that are fetched with the option to include input data."); + } + + if (this.EnableLargePayloadSupport) + { + return await this.DataConverter.DeserializeAsync(this.SerializedInput, cancellationToken); + } + + return this.DataConverter.Deserialize(this.SerializedInput); + } + + /// + /// Asynchronously deserializes the orchestration's output into an object of the specified type. + /// + /// + /// This method can only be used when inputs and outputs are explicitly requested from the + /// or + /// method that produced + /// this object. + /// + /// The type to deserialize the orchestration output into. + /// Cancellation token. + /// Returns the deserialized output value. + /// + /// Thrown if this metadata object was fetched without the option to read inputs and outputs. + /// + public async ValueTask ReadOutputAsAsync(CancellationToken cancellationToken = default) + { + if (!this.RequestedInputsAndOutputs) + { + throw new InvalidOperationException( + $"The {nameof(this.ReadOutputAsAsync)} method can only be used on {nameof(OrchestrationMetadata)} objects " + + "that are fetched with the option to include output data."); + } + + if (this.EnableLargePayloadSupport) + { + return await this.DataConverter.DeserializeAsync(this.SerializedOutput, cancellationToken); + } + + return this.DataConverter.Deserialize(this.SerializedOutput); + } + + /// + /// Asynchronously deserializes the orchestration's custom status value into an object of the specified type. + /// + /// + /// This method can only be used when inputs and outputs are explicitly requested from the + /// or + /// method that produced + /// this object. + /// + /// The type to deserialize the orchestration' custom status into. + /// Cancellation token. + /// Returns the deserialized custom status value. + /// + /// Thrown if this metadata object was fetched without the option to read inputs and outputs. + /// + public async ValueTask ReadCustomStatusAsAsync(CancellationToken cancellationToken = default) + { + if (!this.RequestedInputsAndOutputs) + { + throw new InvalidOperationException( + $"The {nameof(this.ReadCustomStatusAsAsync)} method can only be used on {nameof(OrchestrationMetadata)}" + + " objects that are fetched with the option to include input and output data."); + } + + if (this.EnableLargePayloadSupport) + { + return await this.DataConverter.DeserializeAsync(this.SerializedCustomStatus, cancellationToken); + } + + return await this.DataConverter.DeserializeAsync(this.SerializedCustomStatus, cancellationToken); + } + /// /// Generates a user-friendly string representation of the current metadata object. /// diff --git a/src/Client/Core/SerializedData.cs b/src/Client/Core/SerializedData.cs index 7bd2c127d..0dc792910 100644 --- a/src/Client/Core/SerializedData.cs +++ b/src/Client/Core/SerializedData.cs @@ -20,6 +20,11 @@ public sealed class SerializedData(string data, DataConverter? converter = null) /// public string Value { get; } = Check.NotNull(data); + /// + /// Gets a value indicating whether large payload support is enabled. + /// + public bool EnableLargePayloadSupport { get; init; } + /// /// Gets the data converter. /// @@ -30,17 +35,47 @@ public sealed class SerializedData(string data, DataConverter? converter = null) /// /// The type to deserialize into. /// The deserialized type. - public T ReadAs() => this.Converter.Deserialize(this.Value); + public T ReadAs() + { + Check.ThrowIfLargePayloadEnabled(this.EnableLargePayloadSupport, nameof(this.ReadAs)); + return this.Converter.Deserialize(this.Value); + } + + /// + /// Deserializes the data into . + /// + /// The type to deserialize into. + /// The deserialized type. + public async Task ReadAsAsync() => await this.Converter.DeserializeAsync(this.Value); /// /// Creates a new instance of from the specified data. /// /// The data to serialize. /// The data converter. + /// Whether to use async serialization for large payloads. /// Serialized data. - internal static SerializedData Create(object data, DataConverter? converter = null) + internal static SerializedData Create(object data, DataConverter? converter = null, bool enableLargePayloadSupport = false) { converter ??= JsonDataConverter.Default; - return new SerializedData(converter.Serialize(data), converter); + return new SerializedData(converter.Serialize(data), converter) + { + EnableLargePayloadSupport = enableLargePayloadSupport, + }; + } + + /// + /// Creates a new instance of from the specified data. + /// + /// The data to serialize. + /// The data converter. + /// Whether to use async serialization for large payloads. + /// Serialized data. + internal static async Task CreateAsync(object data, DataConverter converter, bool enableLargePayloadSupport = false) + { + return new SerializedData(await converter.SerializeAsync(data), converter) + { + EnableLargePayloadSupport = enableLargePayloadSupport, + }; } } diff --git a/src/Client/Grpc/GrpcDurableEntityClient.cs b/src/Client/Grpc/GrpcDurableEntityClient.cs index fd8e63926..826b26857 100644 --- a/src/Client/Grpc/GrpcDurableEntityClient.cs +++ b/src/Client/Grpc/GrpcDurableEntityClient.cs @@ -19,6 +19,7 @@ class GrpcDurableEntityClient : DurableEntityClient readonly TaskHubSidecarServiceClient sidecarClient; readonly DataConverter dataConverter; readonly ILogger logger; + readonly bool enableLargePayloadSupport; /// /// Initializes a new instance of the class. @@ -27,13 +28,19 @@ class GrpcDurableEntityClient : DurableEntityClient /// The data converter. /// The client for the GRPC connection to the sidecar. /// The logger for logging client requests. + /// Whether to use async serialization for large payloads. public GrpcDurableEntityClient( - string name, DataConverter dataConverter, TaskHubSidecarServiceClient sidecarClient, ILogger logger) + string name, + DataConverter dataConverter, + TaskHubSidecarServiceClient sidecarClient, + ILogger logger, + bool enableLargePayloadSupport = false) : base(name) { this.dataConverter = dataConverter; this.sidecarClient = sidecarClient; this.logger = logger; + this.enableLargePayloadSupport = enableLargePayloadSupport; } /// @@ -54,11 +61,13 @@ public override async Task SignalEntityAsync( InstanceId = id.ToString(), RequestId = requestId.ToString(), Name = operationName, - Input = this.dataConverter.Serialize(input), - ScheduledTime = scheduledTime?.ToTimestamp(), + Input = this.enableLargePayloadSupport + ? await this.dataConverter.SerializeAsync(input, cancellation) + : this.dataConverter.Serialize(input), + ScheduledTime = scheduledTime?.ToTimestamp(), RequestTime = DateTimeOffset.UtcNow.ToTimestamp(), - }; - + }; + if (Activity.Current is { } activity) { request.ParentTraceContext ??= new P.TraceContext(); @@ -170,6 +179,36 @@ public override async Task CleanEntityStorageAsync( } } + async Task GetEntityCoreAsync( + EntityInstanceId id, + bool includeState, + Func> select, + CancellationToken cancellation) + where TMetadata : class + { + Check.NotNullOrEmpty(id.Name); + Check.NotNull(id.Key); + + P.GetEntityRequest request = new() + { + InstanceId = id.ToString(), + IncludeState = includeState, + }; + + try + { + P.GetEntityResponse response = await this.sidecarClient + .GetEntityAsync(request, cancellationToken: cancellation); + + return response.Exists ? await select(response.Entity, includeState) : null; + } + catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled) + { + throw new OperationCanceledException( + $"The {nameof(this.GetEntityAsync)} operation was canceled.", e, cancellation); + } + } + AsyncPageable GetAllEntitiesCoreAsync( EntityQuery? filter, Func select) where TMetadata : class @@ -216,13 +255,64 @@ AsyncPageable GetAllEntitiesCoreAsync( }); } + AsyncPageable GetAllEntitiesCoreAsync( + EntityQuery? filter, Func> select) + where TMetadata : class + { + bool includeState = filter?.IncludeState ?? true; + bool includeTransient = filter?.IncludeTransient ?? false; + string startsWith = filter?.InstanceIdStartsWith ?? string.Empty; + DateTimeOffset? lastModifiedFrom = filter?.LastModifiedFrom; + DateTimeOffset? lastModifiedTo = filter?.LastModifiedTo; + + return Pageable.Create(async (continuation, pageSize, cancellation) => + { + pageSize ??= filter?.PageSize; + + try + { + P.QueryEntitiesResponse response = await this.sidecarClient.QueryEntitiesAsync( + new P.QueryEntitiesRequest + { + Query = new P.EntityQuery + { + InstanceIdStartsWith = startsWith, + LastModifiedFrom = lastModifiedFrom?.ToTimestamp(), + LastModifiedTo = lastModifiedTo?.ToTimestamp(), + IncludeState = includeState, + IncludeTransient = includeTransient, + PageSize = pageSize, + ContinuationToken = continuation ?? filter?.ContinuationToken, + }, + }, + cancellationToken: cancellation); + + List values = new(); + foreach (var entity in response.Entities) + { + values.Add(await select(entity, includeState)); + } + + return new Page(values, response.ContinuationToken); + } + catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled) + { + throw new OperationCanceledException( + $"The {nameof(this.GetAllEntitiesAsync)} operation was canceled.", e, cancellation); + } + }); + } + EntityMetadata ToEntityMetadata(P.EntityMetadata metadata, bool includeState) { var coreEntityId = DTCore.Entities.EntityId.FromString(metadata.InstanceId); EntityInstanceId entityId = new(coreEntityId.Name, coreEntityId.Key); bool hasState = metadata.SerializedState != null; - SerializedData? data = (includeState && hasState) ? new(metadata.SerializedState!, this.dataConverter) : null; + SerializedData? data = (includeState && hasState) + ? new SerializedData(metadata.SerializedState!, this.dataConverter) + { EnableLargePayloadSupport = this.enableLargePayloadSupport } + : null; return new EntityMetadata(entityId, data) { LastModifiedTime = metadata.LastModifiedTime.ToDateTimeOffset(), @@ -231,7 +321,7 @@ EntityMetadata ToEntityMetadata(P.EntityMetadata metadata, bool includeState) }; } - EntityMetadata ToEntityMetadata(P.EntityMetadata metadata, bool includeState) + async ValueTask> ToEntityMetadata(P.EntityMetadata metadata, bool includeState) { var coreEntityId = DTCore.Entities.EntityId.FromString(metadata.InstanceId); EntityInstanceId entityId = new(coreEntityId.Name, coreEntityId.Key); @@ -240,7 +330,11 @@ EntityMetadata ToEntityMetadata(P.EntityMetadata metadata, bool includeSta if (includeState && hasState) { - T? data = includeState ? this.dataConverter.Deserialize(metadata.SerializedState) : default; + T? data = includeState + ? (this.enableLargePayloadSupport + ? await this.dataConverter.DeserializeAsync(metadata.SerializedState, CancellationToken.None) + : this.dataConverter.Deserialize(metadata.SerializedState)) + : default; return new EntityMetadata(entityId, data) { LastModifiedTime = lastModified, diff --git a/src/Client/Grpc/GrpcDurableTaskClient.cs b/src/Client/Grpc/GrpcDurableTaskClient.cs index c57b22f96..6aab2efd2 100644 --- a/src/Client/Grpc/GrpcDurableTaskClient.cs +++ b/src/Client/Grpc/GrpcDurableTaskClient.cs @@ -54,7 +54,7 @@ public GrpcDurableTaskClient(string name, GrpcDurableTaskClientOptions options, if (this.options.EnableEntitySupport) { - this.entityClient = new GrpcDurableEntityClient(this.Name, this.DataConverter, this.sidecarClient, logger); + this.entityClient = new GrpcDurableEntityClient(this.Name, this.DataConverter, this.sidecarClient, logger, this.options.EnableLargePayloadSupport); } } @@ -90,12 +90,16 @@ public override async Task ScheduleNewOrchestrationInstanceAsync( version = this.options.DefaultVersion; } + string instanceId = options?.InstanceId ?? Guid.NewGuid().ToString("N"); + var request = new P.CreateInstanceRequest { Name = orchestratorName.Name, Version = version, - InstanceId = options?.InstanceId ?? Guid.NewGuid().ToString("N"), - Input = this.DataConverter.Serialize(input), + InstanceId = instanceId, + Input = this.options.EnableLargePayloadSupport + ? await this.DataConverter.SerializeAsync(input, cancellation) + : this.DataConverter.Serialize(input), RequestTime = DateTimeOffset.UtcNow.ToTimestamp(), }; @@ -109,11 +113,17 @@ public override async Task ScheduleNewOrchestrationInstanceAsync( } DateTimeOffset? startAt = options?.StartAt; - this.logger.SchedulingOrchestration( - request.InstanceId, - orchestratorName, - sizeInBytes: request.Input != null ? Encoding.UTF8.GetByteCount(request.Input) : 0, - startAt.GetValueOrDefault(DateTimeOffset.UtcNow)); + string name = orchestratorName.Name ?? string.Empty; + string? serializedInput = request.Input; + int sizeInBytes = 0; + if (!string.IsNullOrEmpty(serializedInput)) + { + sizeInBytes = Encoding.UTF8.GetByteCount(serializedInput!); + } + + DateTimeOffset startTime = startAt.GetValueOrDefault(DateTimeOffset.UtcNow); + + this.logger.SchedulingOrchestration(instanceId, name, sizeInBytes, startTime); if (startAt.HasValue) { @@ -141,7 +151,9 @@ public override async Task RaiseEventAsync( { InstanceId = instanceId, Name = eventName, - Input = this.DataConverter.Serialize(eventPayload), + Input = this.options.EnableLargePayloadSupport + ? await this.DataConverter.SerializeAsync(eventPayload, cancellation) + : this.DataConverter.Serialize(eventPayload), }; using Activity? traceActivity = TraceHelper.StartActivityForNewEventRaisedFromClient(request, instanceId); @@ -161,7 +173,9 @@ public override async Task TerminateInstanceAsync( this.logger.TerminatingInstance(instanceId); - string? serializedOutput = this.DataConverter.Serialize(output); + string? serializedOutput = this.options.EnableLargePayloadSupport + ? await this.DataConverter.SerializeAsync(output, cancellation) + : this.DataConverter.Serialize(output); await this.sidecarClient.TerminateInstanceAsync( new P.TerminateRequest { @@ -502,6 +516,7 @@ OrchestrationMetadata CreateMetadata(P.OrchestrationState state, bool includeInp SerializedCustomStatus = state.CustomStatus, FailureDetails = state.FailureDetails.ToTaskFailureDetails(), DataConverter = includeInputsAndOutputs ? this.DataConverter : null, + EnableLargePayloadSupport = this.options.EnableLargePayloadSupport, Tags = new Dictionary(state.Tags), }; diff --git a/src/Client/OrchestrationServiceClientShim/ShimDurableEntityClient.cs b/src/Client/OrchestrationServiceClientShim/ShimDurableEntityClient.cs index 4d798479d..88d941050 100644 --- a/src/Client/OrchestrationServiceClientShim/ShimDurableEntityClient.cs +++ b/src/Client/OrchestrationServiceClientShim/ShimDurableEntityClient.cs @@ -3,7 +3,7 @@ using System.Diagnostics; using DurableTask.Core; -using DurableTask.Core.Entities; +using DurableTask.Core.Entities; using DurableTask.Core.Tracing; using Microsoft.DurableTask.Client.Entities; using Microsoft.DurableTask.Entities; @@ -67,7 +67,7 @@ public override AsyncPageable> GetAllEntitiesAsync(EntityQu /// public override async Task?> GetEntityAsync( EntityInstanceId id, bool includeState = true, CancellationToken cancellation = default) - => this.Convert(await this.Queries.GetEntityAsync( + => await this.Convert(await this.Queries.GetEntityAsync( id.ConvertToCore(), includeState, false, cancellation)); /// @@ -82,7 +82,9 @@ public override async Task SignalEntityAsync( Check.NotNull(id.Key); DateTimeOffset? scheduledTime = options?.SignalTime; - string? serializedInput = this.Converter.Serialize(input); + string? serializedInput = this.options.EnableLargePayloadSupport + ? await this.Converter.SerializeAsync(input, cancellation) + : this.Converter.Serialize(input); EntityMessageEvent eventToSend = ClientEntityHelpers.EmitOperationSignal( new OrchestrationInstance() { InstanceId = id.ToString() }, @@ -92,9 +94,9 @@ public override async Task SignalEntityAsync( EntityMessageEvent.GetCappedScheduledTime( DateTime.UtcNow, this.options.Entities.MaxSignalDelayTimeOrDefault, - scheduledTime?.UtcDateTime), - Activity.Current is { } activity ? new DistributedTraceContext(activity.Id!, activity.TraceStateString) : null, - requestTime: DateTimeOffset.UtcNow, + scheduledTime?.UtcDateTime), + Activity.Current is { } activity ? new DistributedTraceContext(activity.Id!, activity.TraceStateString) : null, + requestTime: DateTimeOffset.UtcNow, createTrace: true); await this.options.Client!.SendTaskOrchestrationMessageAsync(eventToSend.AsTaskMessage()); @@ -132,11 +134,48 @@ AsyncPageable GetAllEntitiesAsync( }); } - EntityMetadata Convert(EntityBackendQueries.EntityMetadata metadata) + AsyncPageable GetAllEntitiesAsync( + Func> selectAsync, + EntityQuery? filter) + where TMetadata : notnull { + bool includeState = filter?.IncludeState ?? true; + bool includeTransient = filter?.IncludeTransient ?? false; + string startsWith = filter?.InstanceIdStartsWith ?? string.Empty; + DateTime? lastModifiedFrom = filter?.LastModifiedFrom?.UtcDateTime; + DateTime? lastModifiedTo = filter?.LastModifiedTo?.UtcDateTime; + + return Pageable.Create(async (continuation, size, cancellation) => + { + continuation ??= filter?.ContinuationToken; + size ??= filter?.PageSize; + EntityBackendQueries.EntityQueryResult result = await this.Queries.QueryEntitiesAsync( + new EntityBackendQueries.EntityQuery() + { + InstanceIdStartsWith = startsWith, + LastModifiedFrom = lastModifiedFrom, + LastModifiedTo = lastModifiedTo, + IncludeTransient = includeTransient, + IncludeState = includeState, + ContinuationToken = continuation, + PageSize = size, + }, + cancellation); + + TMetadata[] items = await Task.WhenAll(result.Results.Select(selectAsync)); + return new Page(items, result.ContinuationToken); + }); + } + + async Task> Convert(EntityBackendQueries.EntityMetadata metadata) + { + T? state = this.options.EnableLargePayloadSupport + ? await this.Converter.DeserializeAsync(metadata.SerializedState) + : this.Converter.Deserialize(metadata.SerializedState); + return new( metadata.EntityId.ConvertFromCore(), - this.Converter.Deserialize(metadata.SerializedState)) + state) { LastModifiedTime = metadata.LastModifiedTime, BacklogQueueSize = metadata.BacklogQueueSize, @@ -144,19 +183,24 @@ EntityMetadata Convert(EntityBackendQueries.EntityMetadata metadata) }; } - EntityMetadata? Convert(EntityBackendQueries.EntityMetadata? metadata) + async Task?> Convert(EntityBackendQueries.EntityMetadata? metadata) { if (metadata is null) { return null; } - return this.Convert(metadata.Value); + return await this.Convert(metadata.Value); } EntityMetadata Convert(EntityBackendQueries.EntityMetadata metadata) { - SerializedData? data = metadata.SerializedState is null ? null : new(metadata.SerializedState, this.Converter); + SerializedData? data = metadata.SerializedState is null + ? null + : new SerializedData(metadata.SerializedState, this.Converter) + { + EnableLargePayloadSupport = this.options.EnableLargePayloadSupport, + }; return new(new EntityInstanceId(metadata.EntityId.Name, metadata.EntityId.Key), data) { LastModifiedTime = metadata.LastModifiedTime, diff --git a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs index 4fbf828d0..d378e7f6e 100644 --- a/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs +++ b/src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs @@ -67,6 +67,11 @@ public override DurableEntityClient Entities DataConverter DataConverter => this.options.DataConverter; + /// + /// Gets a value indicating whether the DataConverter supports async operations (LargePayload enabled). + /// + bool SupportsAsyncSerialization => this.options.EnableLargePayloadSupport; + IOrchestrationServiceClient Client => this.options.Client!; IOrchestrationServicePurgeClient PurgeClient => this.CastClient(); @@ -141,14 +146,16 @@ public override async Task PurgeAllInstancesAsync( } /// - public override Task RaiseEventAsync( + public override async Task RaiseEventAsync( string instanceId, string eventName, object? eventPayload = null, CancellationToken cancellation = default) { Check.NotNullOrEmpty(instanceId); Check.NotNullOrEmpty(eventName); - string? serializedInput = this.DataConverter.Serialize(eventPayload); - return this.SendInstanceMessageAsync( + string? serializedInput = this.SupportsAsyncSerialization + ? await this.DataConverter.SerializeAsync(eventPayload, cancellation) + : this.DataConverter.Serialize(eventPayload); + await this.SendInstanceMessageAsync( instanceId, new EventRaisedEvent(-1, serializedInput) { Name = eventName }, cancellation); } @@ -167,7 +174,9 @@ public override async Task ScheduleNewOrchestrationInstanceAsync( ExecutionId = Guid.NewGuid().ToString("N"), }; - string? serializedInput = this.DataConverter.Serialize(input); + string? serializedInput = this.SupportsAsyncSerialization + ? await this.DataConverter.SerializeAsync(input, cancellation) + : this.DataConverter.Serialize(input); var tags = new Dictionary(); if (options?.Tags != null) @@ -207,16 +216,18 @@ public override Task ResumeInstanceAsync( => this.SendInstanceMessageAsync(instanceId, new ExecutionResumedEvent(-1, reason), cancellation); /// - public override Task TerminateInstanceAsync( + public override async Task TerminateInstanceAsync( string instanceId, TerminateInstanceOptions? options = null, CancellationToken cancellation = default) { object? output = options?.Output; Check.NotNullOrEmpty(instanceId); cancellation.ThrowIfCancellationRequested(); - string? reason = this.DataConverter.Serialize(output); + string? reason = this.SupportsAsyncSerialization + ? await this.DataConverter.SerializeAsync(output, cancellation) + : this.DataConverter.Serialize(output); // TODO: Support recursive termination of sub-orchestrations - return this.Client.ForceTerminateTaskOrchestrationAsync(instanceId, reason); + await this.Client.ForceTerminateTaskOrchestrationAsync(instanceId, reason); } /// @@ -325,6 +336,7 @@ public override async Task RestartAsync( SerializedOutput = state.Output, SerializedCustomStatus = state.Status, FailureDetails = state.FailureDetails?.ConvertFromCore(), + EnableLargePayloadSupport = this.SupportsAsyncSerialization, }; } diff --git a/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj new file mode 100644 index 000000000..338022860 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj @@ -0,0 +1,28 @@ + + + + netstandard2.0;net6.0 + Azure Blob Storage externalized payload support for Durable Task. + Microsoft.DurableTask.Extensions.AzureBlobPayloads + Microsoft.DurableTask + true + + + + + + + + + + + + + + + + + + + + diff --git a/src/Extensions/AzureBlobPayloads/Converters/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/Converters/BlobPayloadStore.cs new file mode 100644 index 000000000..1f101288d --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/Converters/BlobPayloadStore.cs @@ -0,0 +1,215 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Globalization; +using System.IO.Compression; +using System.Text; +using Azure; +using Azure.Core; +using Azure.Storage.Blobs; +using Azure.Storage.Blobs.Models; + +namespace Microsoft.DurableTask.Converters; + +/// +/// Azure Blob Storage implementation of . +/// Stores payloads as blobs and returns opaque tokens in the form "blob:v1:<container>:<blobName>". +/// +public sealed class BlobPayloadStore : IPayloadStore +{ + const string TokenPrefix = "blob:v1:"; + const string ContentEncodingGzip = "gzip"; + const int DefaultCopyBufferSize = 81920; + const int MaxRetryAttempts = 8; + const int BaseDelayMs = 250; + const int MaxDelayMs = 10_000; + const int MaxJitterMs = 100; + + // Jitter RNG for retry backoff + static readonly object RandomLock = new object(); + static readonly Random SharedRandom = new Random(); + readonly BlobContainerClient containerClient; + readonly LargePayloadStorageOptions options; + + /// + /// Initializes a new instance of the class. + /// + /// The options for the blob payload store. + /// Thrown when is null. + /// Thrown when is null or empty. + public BlobPayloadStore(LargePayloadStorageOptions options) + { + this.options = options ?? throw new ArgumentNullException(nameof(options)); + + Check.NotNullOrEmpty(options.ConnectionString, nameof(options.ConnectionString)); + Check.NotNullOrEmpty(options.ContainerName, nameof(options.ContainerName)); + + BlobClientOptions clientOptions = new() + { + Retry = + { + Mode = RetryMode.Exponential, + MaxRetries = MaxRetryAttempts, + Delay = TimeSpan.FromMilliseconds(BaseDelayMs), + MaxDelay = TimeSpan.FromSeconds(10), + NetworkTimeout = TimeSpan.FromMinutes(2), + }, + }; + BlobServiceClient serviceClient = new(options.ConnectionString, clientOptions); + this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName); + } + + /// + public async Task UploadAsync(ReadOnlyMemory payloadBytes, CancellationToken cancellationToken) + { + // One blob per payload using GUID-based name for uniqueness (stable across retries) + string timestamp = DateTimeOffset.UtcNow.ToString("yyyy/MM/dd/HH/mm/ss", CultureInfo.InvariantCulture); + string blobName = $"{timestamp}/{Guid.NewGuid():N}"; + BlobClient blob = this.containerClient.GetBlobClient(blobName); + + byte[] payloadBuffer = payloadBytes.ToArray(); + + string token = await WithTransientRetryAsync( + async ct => + { + // Ensure container exists (idempotent) + await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, default, default, ct); + + if (this.options.CompressPayloads) + { + BlobOpenWriteOptions writeOptions = new() + { + HttpHeaders = new BlobHttpHeaders { ContentEncoding = ContentEncodingGzip }, + }; + using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, ct); + using GZipStream compressedBlobStream = new(blobStream, CompressionLevel.Optimal, leaveOpen: true); + using MemoryStream payloadStream = new(payloadBuffer, writable: false); + + await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, ct); + await compressedBlobStream.FlushAsync(ct); + await blobStream.FlushAsync(ct); + } + else + { + using Stream blobStream = await blob.OpenWriteAsync(true, default, ct); + using MemoryStream payloadStream = new(payloadBuffer, writable: false); + await payloadStream.CopyToAsync(blobStream, bufferSize: DefaultCopyBufferSize, ct); + await blobStream.FlushAsync(ct); + } + + return EncodeToken(this.containerClient.Name, blobName); + }, + cancellationToken); + + return token; + } + + /// + public async Task DownloadAsync(string token, CancellationToken cancellationToken) + { + (string container, string name) = DecodeToken(token); + if (!string.Equals(container, this.containerClient.Name, StringComparison.Ordinal)) + { + throw new ArgumentException("Token container does not match configured container.", nameof(token)); + } + + BlobClient blob = this.containerClient.GetBlobClient(name); + + return await WithTransientRetryAsync( + async ct => + { + using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: ct); + Stream contentStream = result.Content; + bool isGzip = string.Equals( + result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase); + + if (isGzip) + { + using GZipStream decompressed = new(contentStream, CompressionMode.Decompress); + using StreamReader reader = new(decompressed, Encoding.UTF8); + return await reader.ReadToEndAsync(); + } + + using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8); + return await uncompressedReader.ReadToEndAsync(); + }, + cancellationToken); + } + + /// + public bool IsKnownPayloadToken(string value) + { + if (string.IsNullOrEmpty(value)) + { + return false; + } + + return value.StartsWith(TokenPrefix, StringComparison.Ordinal); + } + + static string EncodeToken(string container, string name) => $"blob:v1:{container}:{name}"; + + static (string Container, string Name) DecodeToken(string token) + { + if (!token.StartsWith("blob:v1:", StringComparison.Ordinal)) + { + throw new ArgumentException("Invalid external payload token.", nameof(token)); + } + + string rest = token.Substring("blob:v1:".Length); + int sep = rest.IndexOf(':'); + if (sep <= 0 || sep >= rest.Length - 1) + { + throw new ArgumentException("Invalid external payload token format.", nameof(token)); + } + + return (rest.Substring(0, sep), rest.Substring(sep + 1)); + } + + static async Task WithTransientRetryAsync(Func> operation, CancellationToken cancellationToken) + { + const int maxAttempts = MaxRetryAttempts; + TimeSpan baseDelay = TimeSpan.FromMilliseconds(BaseDelayMs); + int attempt = 0; + + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + try + { + return await operation(cancellationToken); + } + catch (RequestFailedException ex) when (IsTransient(ex) && attempt < maxAttempts - 1) + { + attempt++; + TimeSpan delay = ComputeBackoff(baseDelay, attempt); + await Task.Delay(delay, cancellationToken); + } + catch (IOException) when (attempt < maxAttempts - 1) + { + attempt++; + TimeSpan delay = ComputeBackoff(baseDelay, attempt); + await Task.Delay(delay, cancellationToken); + } + } + } + + static bool IsTransient(RequestFailedException ex) + { + return ex.Status == 503 || ex.Status == 502 || ex.Status == 500 || ex.Status == 429 || + string.Equals(ex.ErrorCode, "ServerBusy", StringComparison.OrdinalIgnoreCase) || + string.Equals(ex.ErrorCode, "OperationTimedOut", StringComparison.OrdinalIgnoreCase); + } + + static TimeSpan ComputeBackoff(TimeSpan baseDelay, int attempt) + { + double factor = Math.Pow(2, Math.Min(attempt, 6)); + int jitterMs; + lock (RandomLock) + { + jitterMs = SharedRandom.Next(0, MaxJitterMs); + } + + return TimeSpan.FromMilliseconds(Math.Min((baseDelay.TotalMilliseconds * factor) + jitterMs, MaxDelayMs)); + } +} diff --git a/src/Extensions/AzureBlobPayloads/Converters/LargePayloadDataConverter.cs b/src/Extensions/AzureBlobPayloads/Converters/LargePayloadDataConverter.cs new file mode 100644 index 000000000..03a2c3e7c --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/Converters/LargePayloadDataConverter.cs @@ -0,0 +1,112 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Diagnostics.CodeAnalysis; +using System.Text; + +namespace Microsoft.DurableTask.Converters; + +/// +/// A DataConverter that wraps another DataConverter and externalizes payloads larger than a configured threshold. +/// It uploads large payloads to an and returns a reference token string. +/// On deserialization, it resolves tokens and feeds the underlying converter the original content. +/// +/// +/// Initializes a new instance of the class. +/// +/// The inner data converter to wrap. +/// The external payload store to use. +/// The options for the externalizing data converter. +/// Thrown when , , or is null. +public sealed class LargePayloadDataConverter( + DataConverter innerConverter, + IPayloadStore payloadStore, + LargePayloadStorageOptions largePayloadStorageOptions) : DataConverter +{ + readonly DataConverter innerConverter = innerConverter ?? throw new ArgumentNullException(nameof(innerConverter)); + readonly IPayloadStore payLoadStore = payloadStore ?? throw new ArgumentNullException(nameof(payloadStore)); + readonly LargePayloadStorageOptions largePayloadStorageOptions = largePayloadStorageOptions ?? throw new ArgumentNullException(nameof(largePayloadStorageOptions)); + + // Use UTF-8 without a BOM (encoderShouldEmitUTF8Identifier=false). JSON in UTF-8 should not include a + // byte order mark per RFC 8259, and omitting it avoids hidden extra bytes that could skew the + // externalization threshold calculation and prevents interop issues with strict JSON parsers. + // A few legacy tools rely on a BOM for encoding detection, but modern JSON tooling assumes BOM-less UTF-8. + readonly Encoding utf8 = new UTF8Encoding(false); + + /// + [return: NotNullIfNotNull("value")] + public override string? Serialize(object? value) + { + throw new NotSupportedException(); + } + + /// + [return: NotNullIfNotNull("data")] + public override object? Deserialize(string? data, Type targetType) + { + throw new NotSupportedException(); + } + + /// + /// Serializes the value to a JSON string and uploads it to the external payload store if it exceeds the configured threshold. + /// + /// The value to serialize. + /// Cancellation token. + /// The serialized value or the token if externalized. + public override async ValueTask SerializeAsync(object? value, CancellationToken cancellationToken = default) + { + string? json = await this.innerConverter.SerializeAsync(value, cancellationToken); + + if (string.IsNullOrEmpty(json)) + { + return null; + } + + int byteCount = this.utf8.GetByteCount(json); + if (byteCount < this.largePayloadStorageOptions.ExternalizeThresholdBytes) + { + return json; + } + + // Upload synchronously in this context by blocking on async. SDK call sites already run on threadpool. + byte[] bytes = this.utf8.GetBytes(json); + return await this.payLoadStore.UploadAsync(bytes, cancellationToken); + } + + /// + /// Deserializes the JSON string or resolves the token to the original value. + /// + /// The JSON string or token. + /// The type to deserialize to. + /// Cancellation token. + /// The deserialized value. + public override async ValueTask DeserializeAsync( + string? data, + Type targetType, + CancellationToken cancellationToken = default) + { + if (data is null) + { + return null; + } + + string toDeserialize = data; + if (this.payLoadStore.IsKnownPayloadToken(data)) + { + toDeserialize = await this.payLoadStore.DownloadAsync(data, CancellationToken.None); + } + + return await this.innerConverter.DeserializeAsync(StripArrayCharacters(toDeserialize), targetType, cancellationToken); + } + + static string? StripArrayCharacters(string? input) + { + if (input != null && input.StartsWith('[') && input.EndsWith(']')) + { + // Strip the outer bracket characters + return input[1..^1]; + } + + return input; + } +} diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs new file mode 100644 index 000000000..19f057726 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -0,0 +1,48 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Converters; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Options; + +namespace Microsoft.DurableTask.Client; + +/// +/// Extension methods to enable externalized payloads using Azure Blob Storage for Durable Task Client. +/// +public static class DurableTaskClientBuilderExtensionsAzureBlobPayloads +{ + /// + /// Enables externalized payload storage using Azure Blob Storage for the specified client builder. + /// + /// The client builder. + /// The configure action. + /// The original builder, for call chaining. + public static IDurableTaskClientBuilder UseExternalizedPayloads( + this IDurableTaskClientBuilder builder, + Action configure) + { + Check.NotNull(builder); + Check.NotNull(configure); + + builder.Services.Configure(builder.Name, configure); + builder.Services.AddSingleton(sp => + { + LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); + return new BlobPayloadStore(opts); + }); + + builder.Services + .AddOptions(builder.Name) + .PostConfigure>((opt, store, monitor) => + { + LargePayloadStorageOptions opts = monitor.Get(builder.Name); + DataConverter inner = opt.DataConverter ?? Converters.JsonDataConverter.Default; + opt.DataConverter = new LargePayloadDataConverter(inner, store, opts); + }); + + return builder; + } +} diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs new file mode 100644 index 000000000..8ae08fffa --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -0,0 +1,46 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Converters; +using Microsoft.DurableTask.Worker; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Microsoft.DurableTask.Worker; + +/// +/// Extension methods to enable externalized payloads using Azure Blob Storage for Durable Task Worker. +/// +public static class DurableTaskWorkerBuilderExtensionsAzureBlobPayloads +{ + /// + /// Enables externalized payload storage using Azure Blob Storage for the specified worker builder. + /// + public static IDurableTaskWorkerBuilder UseExternalizedPayloads( + this IDurableTaskWorkerBuilder builder, + Action configure) + { + Check.NotNull(builder); + Check.NotNull(configure); + + builder.Services.Configure(builder.Name, configure); + builder.Services.AddSingleton(sp => + { + LargePayloadStorageOptions opts = sp.GetRequiredService>().Get(builder.Name); + return new BlobPayloadStore(opts); + }); + + builder.Services + .AddOptions(builder.Name) + .PostConfigure>((opt, store, monitor) => + { + LargePayloadStorageOptions opts = monitor.Get(builder.Name); + DataConverter inner = opt.DataConverter ?? Converters.JsonDataConverter.Default; + opt.DataConverter = new LargePayloadDataConverter(inner, store, opts); + }); + + return builder; + } +} + + diff --git a/src/Shared/Core/Validation/Check.cs b/src/Shared/Core/Validation/Check.cs index 76c154749..dd8b10c8b 100644 --- a/src/Shared/Core/Validation/Check.cs +++ b/src/Shared/Core/Validation/Check.cs @@ -108,6 +108,22 @@ public static void NotEntity(bool entitySupportEnabled, string? instanceId, [Cal } } + /// + /// Checks if the supplied type is a concrete non-abstract type and implements the provided generic type. + /// Throws if the conditions are not met. + /// + /// Whether large payload support is enabled. + /// The name of the operation. + public static void ThrowIfLargePayloadEnabled(bool largePayloadSupportEnabled, string operationName) + { + if (largePayloadSupportEnabled) + { + throw new NotSupportedException( + $"Operation '{operationName}' is not supported when LargePayload is enabled. " + + "Use the async version of this method instead."); + } + } + /// /// Checks if the supplied type is a concrete non-abstract type and implements the provided generic type. /// Throws if the conditions are not met. diff --git a/src/Worker/Core/DurableTaskWorkerOptions.cs b/src/Worker/Core/DurableTaskWorkerOptions.cs index 703bbbd4d..2f3552118 100644 --- a/src/Worker/Core/DurableTaskWorkerOptions.cs +++ b/src/Worker/Core/DurableTaskWorkerOptions.cs @@ -87,6 +87,12 @@ public DataConverter DataConverter /// public bool EnableEntitySupport { get; set; } + /// + /// Gets or sets a value indicating whether this worker should support large payloads using async serialization/deserialization. + /// When enabled, the worker will use async methods for serialization and deserialization to support externalized payloads. + /// + public bool EnableLargePayloadSupport { get; set; } + /// /// Gets or sets the maximum timer interval for the /// method. @@ -162,7 +168,6 @@ public DataConverter DataConverter /// internal bool DataConverterExplicitlySet { get; private set; } - /// /// Applies these option values to another. /// @@ -175,6 +180,7 @@ internal void ApplyTo(DurableTaskWorkerOptions other) other.DataConverter = this.DataConverter; other.MaximumTimerInterval = this.MaximumTimerInterval; other.EnableEntitySupport = this.EnableEntitySupport; + other.EnableLargePayloadSupport = this.EnableLargePayloadSupport; other.Versioning = this.Versioning; other.OrchestrationFilter = this.OrchestrationFilter; } diff --git a/src/Worker/Core/Shims/DurableTaskShimFactory.cs b/src/Worker/Core/Shims/DurableTaskShimFactory.cs index 584b7eeb8..7ff3a73ca 100644 --- a/src/Worker/Core/Shims/DurableTaskShimFactory.cs +++ b/src/Worker/Core/Shims/DurableTaskShimFactory.cs @@ -50,7 +50,12 @@ public TaskActivity CreateActivity(TaskName name, ITaskActivity activity) { Check.NotDefault(name); Check.NotNull(activity); - return new TaskActivityShim(this.loggerFactory, this.options.DataConverter, name, activity); + return new TaskActivityShim( + this.loggerFactory, + this.options.DataConverter, + name, + activity, + this.options.EnableLargePayloadSupport); } /// @@ -151,6 +156,11 @@ public TaskEntity CreateEntity(TaskName name, ITaskEntity entity, EntityId entit // In the future we may consider caching those shims and reusing them, which can reduce // deserialization and allocation overheads. ILogger logger = this.loggerFactory.CreateLogger(entity.GetType()); - return new TaskEntityShim(this.options.DataConverter, entity, entityId, logger); + return new TaskEntityShim( + this.options.DataConverter, + entity, + entityId, + logger, + enableLargePayloadSupport: this.options.EnableLargePayloadSupport); } } diff --git a/src/Worker/Core/Shims/TaskActivityShim.cs b/src/Worker/Core/Shims/TaskActivityShim.cs index ae1f3bb06..f29c7828e 100644 --- a/src/Worker/Core/Shims/TaskActivityShim.cs +++ b/src/Worker/Core/Shims/TaskActivityShim.cs @@ -15,6 +15,7 @@ class TaskActivityShim : TaskActivity readonly ILogger logger; readonly DataConverter dataConverter; readonly TaskName name; + readonly bool enableLargePayloadSupport; /// /// Initializes a new instance of the class. @@ -23,16 +24,19 @@ class TaskActivityShim : TaskActivity /// The data converter. /// The name of the activity. /// The activity implementation to wrap. + /// Whether to use async serialization for large payloads. public TaskActivityShim( ILoggerFactory loggerFactory, DataConverter dataConverter, TaskName name, - ITaskActivity implementation) + ITaskActivity implementation, + bool enableLargePayloadSupport = false) { this.logger = Logs.CreateWorkerLogger(Check.NotNull(loggerFactory), "Activities"); this.dataConverter = Check.NotNull(dataConverter); this.name = Check.NotDefault(name); this.implementation = Check.NotNull(implementation); + this.enableLargePayloadSupport = enableLargePayloadSupport; } /// @@ -40,7 +44,9 @@ public TaskActivityShim( { Check.NotNull(coreContext); string? strippedRawInput = StripArrayCharacters(rawInput); - object? deserializedInput = this.dataConverter.Deserialize(strippedRawInput, this.implementation.InputType); + object? deserializedInput = this.enableLargePayloadSupport + ? await this.dataConverter.DeserializeAsync(strippedRawInput, this.implementation.InputType) + : this.dataConverter.Deserialize(strippedRawInput, this.implementation.InputType); TaskActivityContextWrapper contextWrapper = new(coreContext, this.name); string instanceId = coreContext.OrchestrationInstance.InstanceId; @@ -51,7 +57,9 @@ public TaskActivityShim( object? output = await this.implementation.RunAsync(contextWrapper, deserializedInput); // Return the output (if any) as a serialized string. - string? serializedOutput = this.dataConverter.Serialize(output); + string? serializedOutput = this.enableLargePayloadSupport + ? await this.dataConverter.SerializeAsync(output) + : this.dataConverter.Serialize(output); this.logger.ActivityCompleted(instanceId, this.name); return serializedOutput; diff --git a/src/Worker/Core/Shims/TaskEntityShim.cs b/src/Worker/Core/Shims/TaskEntityShim.cs index bad6c7f08..9d9516174 100644 --- a/src/Worker/Core/Shims/TaskEntityShim.cs +++ b/src/Worker/Core/Shims/TaskEntityShim.cs @@ -19,6 +19,7 @@ class TaskEntityShim : DTCore.Entities.TaskEntity readonly DataConverter dataConverter; readonly ITaskEntity taskEntity; readonly EntityInstanceId entityId; + readonly bool enableLargePayloadSupport; readonly StateShim state; readonly ContextShim context; @@ -32,14 +33,16 @@ class TaskEntityShim : DTCore.Entities.TaskEntity /// The task entity. /// The entity ID. /// The logger. + /// Whether to use async serialization for large payloads. public TaskEntityShim( - DataConverter dataConverter, ITaskEntity taskEntity, EntityId entityId, ILogger logger) + DataConverter dataConverter, ITaskEntity taskEntity, EntityId entityId, ILogger logger, bool enableLargePayloadSupport = false) { this.dataConverter = Check.NotNull(dataConverter); this.taskEntity = Check.NotNull(taskEntity); this.entityId = new EntityInstanceId(entityId.Name, entityId.Key); - this.state = new StateShim(dataConverter); - this.context = new ContextShim(this.entityId, dataConverter); + this.enableLargePayloadSupport = enableLargePayloadSupport; + this.state = new StateShim(dataConverter, enableLargePayloadSupport); + this.context = new ContextShim(this.entityId, dataConverter, enableLargePayloadSupport); this.operation = new OperationShim(this); this.logger = logger; } @@ -58,24 +61,26 @@ public override async Task ExecuteOperationBatchAsync(EntityB List results = new(); foreach (OperationRequest current in operations.Operations!) - { + { var startTime = DateTime.UtcNow; - this.operation.SetNameAndInput(current.Operation!, current.Input); - - // The trace context of the current operation becomes the parent trace context of the TaskEntityContext. - // That way, if processing this operation request leads to the TaskEntityContext signaling another entity or starting an orchestration, - // then the parent trace context of these actions will be set to the trace context of whatever operation request triggered them + this.operation.SetNameAndInput(current.Operation!, current.Input); + + // The trace context of the current operation becomes the parent trace context of the TaskEntityContext. + // That way, if processing this operation request leads to the TaskEntityContext signaling another entity or starting an orchestration, + // then the parent trace context of these actions will be set to the trace context of whatever operation request triggered them this.context.ParentTraceContext = current.TraceContext; try - { + { object? result = await this.taskEntity.RunAsync(this.operation); - string? serializedResult = this.dataConverter.Serialize(result); - results.Add(new OperationResult() - { - Result = serializedResult, - StartTimeUtc = startTime, - EndTimeUtc = DateTime.UtcNow, + string? serializedResult = this.enableLargePayloadSupport + ? await this.dataConverter.SerializeAsync(result) + : this.dataConverter.Serialize(result); + results.Add(new OperationResult() + { + Result = serializedResult, + StartTimeUtc = startTime, + EndTimeUtc = DateTime.UtcNow, }); // the user code completed without exception, so we commit the current state and actions. @@ -87,8 +92,8 @@ public override async Task ExecuteOperationBatchAsync(EntityB this.logger.OperationError(applicationException, this.entityId, current.Operation!); results.Add(new OperationResult() { - FailureDetails = new FailureDetails(applicationException), - StartTimeUtc = startTime, + FailureDetails = new FailureDetails(applicationException), + StartTimeUtc = startTime, EndTimeUtc = DateTime.UtcNow, }); @@ -116,14 +121,16 @@ public override async Task ExecuteOperationBatchAsync(EntityB class StateShim : TaskEntityState { readonly DataConverter dataConverter; + readonly bool enableLargePayloadSupport; string? value; object? cachedValue; string? checkpointValue; - public StateShim(DataConverter dataConverter) + public StateShim(DataConverter dataConverter, bool enableLargePayloadSupport = false) { this.dataConverter = dataConverter; + this.enableLargePayloadSupport = enableLargePayloadSupport; } /// @@ -161,6 +168,7 @@ public void Reset() public override object? GetState(Type type) { + Check.ThrowIfLargePayloadEnabled(this.enableLargePayloadSupport, nameof(this.GetState)); if (this.cachedValue?.GetType() is Type t && t.IsAssignableFrom(type)) { return this.cachedValue; @@ -170,27 +178,51 @@ public void Reset() return this.cachedValue; } + public override async Task GetStateAsync(Type type) + { + if (this.cachedValue?.GetType() is Type t && t.IsAssignableFrom(type)) + { + return this.cachedValue; + } + + this.cachedValue = this.enableLargePayloadSupport + ? await this.dataConverter.DeserializeAsync(this.value, type) + : this.dataConverter.Deserialize(this.value, type); + return this.cachedValue; + } + public override void SetState(object? state) { + Check.ThrowIfLargePayloadEnabled(this.enableLargePayloadSupport, nameof(this.SetState)); this.value = this.dataConverter.Serialize(state); this.cachedValue = state; } + + public override async Task SetStateAsync(object? state) + { + this.value = this.enableLargePayloadSupport + ? await this.dataConverter.SerializeAsync(state) + : this.dataConverter.Serialize(state); + this.cachedValue = state; + } } class ContextShim : TaskEntityContext { readonly EntityInstanceId entityInstanceId; readonly DataConverter dataConverter; + readonly bool enableLargePayloadSupport; List operationActions; - int checkpointPosition; - + int checkpointPosition; + DistributedTraceContext? parentTraceContext; - public ContextShim(EntityInstanceId entityInstanceId, DataConverter dataConverter) + public ContextShim(EntityInstanceId entityInstanceId, DataConverter dataConverter, bool enableLargePayloadSupport = false) { this.entityInstanceId = entityInstanceId; this.dataConverter = dataConverter; + this.enableLargePayloadSupport = enableLargePayloadSupport; this.operationActions = new List(); } @@ -198,12 +230,12 @@ public ContextShim(EntityInstanceId entityInstanceId, DataConverter dataConverte public int CurrentPosition => this.operationActions.Count; - public override EntityInstanceId Id => this.entityInstanceId; - - public DistributedTraceContext? ParentTraceContext - { - get => this.parentTraceContext; - set => this.parentTraceContext = value; + public override EntityInstanceId Id => this.entityInstanceId; + + public DistributedTraceContext? ParentTraceContext + { + get => this.parentTraceContext; + set => this.parentTraceContext = value; } public void Commit() @@ -225,21 +257,46 @@ public void Reset() public override void SignalEntity(EntityInstanceId id, string operationName, object? input = null, SignalEntityOptions? options = null) { Check.NotDefault(id); - + Check.ThrowIfLargePayloadEnabled(this.enableLargePayloadSupport, nameof(this.SignalEntity)); this.operationActions.Add(new SendSignalOperationAction() { InstanceId = id.ToString(), Name = operationName, Input = this.dataConverter.Serialize(input), - ScheduledTime = options?.SignalTime?.UtcDateTime, - RequestTime = DateTimeOffset.UtcNow, + ScheduledTime = options?.SignalTime?.UtcDateTime, + RequestTime = DateTimeOffset.UtcNow, ParentTraceContext = this.parentTraceContext, }); } - public override string ScheduleNewOrchestration(TaskName name, object? input = null, StartOrchestrationOptions? options = null) + public override async ValueTask SignalEntityAsync( + EntityInstanceId id, + string operationName, + object? input = null, + SignalEntityOptions? options = null) + { + Check.NotDefault(id); + + this.operationActions.Add(new SendSignalOperationAction() + { + InstanceId = id.ToString(), + Name = operationName, + Input = this.enableLargePayloadSupport + ? await this.dataConverter.SerializeAsync(input) + : this.dataConverter.Serialize(input), + ScheduledTime = options?.SignalTime?.UtcDateTime, + RequestTime = DateTimeOffset.UtcNow, + ParentTraceContext = this.parentTraceContext, + }); + } + + public override string ScheduleNewOrchestration( + TaskName name, + object? input = null, + StartOrchestrationOptions? options = null) { Check.NotEntity(true, options?.InstanceId); + Check.ThrowIfLargePayloadEnabled(this.enableLargePayloadSupport, nameof(this.ScheduleNewOrchestration)); string instanceId = options?.InstanceId ?? Guid.NewGuid().ToString("N"); this.operationActions.Add(new StartNewOrchestrationOperationAction() @@ -248,8 +305,31 @@ public override string ScheduleNewOrchestration(TaskName name, object? input = n Version = options?.Version ?? string.Empty, InstanceId = instanceId, Input = this.dataConverter.Serialize(input), - ScheduledStartTime = options?.StartAt?.UtcDateTime, - RequestTime = DateTimeOffset.UtcNow, + ScheduledStartTime = options?.StartAt?.UtcDateTime, + RequestTime = DateTimeOffset.UtcNow, + ParentTraceContext = this.parentTraceContext, + }); + return instanceId; + } + + public override async ValueTask ScheduleNewOrchestrationAsync( + TaskName name, + object? input = null, + StartOrchestrationOptions? options = null) + { + Check.NotEntity(true, options?.InstanceId); + + string instanceId = options?.InstanceId ?? Guid.NewGuid().ToString("N"); + this.operationActions.Add(new StartNewOrchestrationOperationAction() + { + Name = name.Name, + Version = options?.Version ?? string.Empty, + InstanceId = instanceId, + Input = this.enableLargePayloadSupport + ? await this.dataConverter.SerializeAsync(input) + : this.dataConverter.Serialize(input), + ScheduledStartTime = options?.StartAt?.UtcDateTime, + RequestTime = DateTimeOffset.UtcNow, ParentTraceContext = this.parentTraceContext, }); return instanceId; @@ -278,9 +358,17 @@ public OperationShim(TaskEntityShim taskEntityShim) public override object? GetInput(Type inputType) { + Check.ThrowIfLargePayloadEnabled(this.taskEntityShim.enableLargePayloadSupport, nameof(this.GetInput)); return this.taskEntityShim.dataConverter.Deserialize(this.input, inputType); } + public override async Task GetInputAsync(Type inputType) + { + return this.taskEntityShim.enableLargePayloadSupport + ? await this.taskEntityShim.dataConverter.DeserializeAsync(this.input, inputType) + : this.taskEntityShim.dataConverter.Deserialize(this.input, inputType); + } + public void SetNameAndInput(string name, string? input) { this.name = name; diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs index 5bf3fbe55..d818541f6 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs @@ -283,7 +283,9 @@ public override Task WaitForExternalEvent(string eventName, CancellationTo // Return immediately if this external event has already arrived. if (this.externalEventBuffer.TryTake(eventName, out string? bufferedEventPayload)) { - return Task.FromResult(this.DataConverter.Deserialize(bufferedEventPayload)); + return this.invocationContext.Options.EnableLargePayloadSupport + ? this.DataConverter.DeserializeAsync(bufferedEventPayload, cancellationToken).AsTask() + : Task.FromResult(this.DataConverter.Deserialize(bufferedEventPayload)); } // Create a task completion source that will be set when the external event arrives. @@ -416,6 +418,9 @@ internal void ExitCriticalSectionIfNeeded() /// The serialized event payload. internal void CompleteExternalEvent(string eventName, string rawEventPayload) { + Check.ThrowIfLargePayloadEnabled( + this.invocationContext.Options.EnableLargePayloadSupport, + nameof(this.CompleteExternalEvent)); if (this.externalEventSources.TryGetValue(eventName, out Queue? waiters)) { object? value; @@ -448,12 +453,69 @@ internal void CompleteExternalEvent(string eventName, string rawEventPayload) } } + /// + /// Completes the external event by name, allowing the orchestration to continue if it is waiting on this event. + /// + /// The name of the event to complete. + /// The serialized event payload. + /// A task representing the asynchronous operation. + internal async Task CompleteExternalEventAsync(string eventName, string rawEventPayload) + { + if (this.externalEventSources.TryGetValue(eventName, out Queue? waiters)) + { + object? value; + + IEventSource waiter = waiters.Dequeue(); + if (waiter.EventType == typeof(OperationResult)) + { + // use the framework-defined deserialization for entity responses, not the application-defined data converter, + // because we are just unwrapping the entity response without yet deserializing any application-defined data. + value = this.entityFeature!.EntityContext.DeserializeEntityResponseEvent(rawEventPayload); + } + else + { + value = this.invocationContext.Options.EnableLargePayloadSupport + ? await this.DataConverter.DeserializeAsync(rawEventPayload, waiter.EventType, CancellationToken.None) + : this.DataConverter.Deserialize(rawEventPayload, waiter.EventType); + } + + // Events are completed in FIFO order. Remove the key if the last event was delivered. + if (waiters.Count == 0) + { + this.externalEventSources.Remove(eventName); + } + + waiter.TrySetResult(value); + } + else + { + // The orchestrator isn't waiting for this event (yet?). Save it in case + // the orchestrator wants it later. + this.externalEventBuffer.Add(eventName, rawEventPayload); + } + } + + /// + /// Gets the serialized custom status. + /// + /// The cancellation token. + /// The custom status serialized to a string, or null if there is not custom status. + internal async ValueTask GetSerializedCustomStatusAsync(CancellationToken cancellationToken = default) + { + return this.invocationContext.Options.EnableLargePayloadSupport + ? await this.DataConverter.SerializeAsync(this.customStatus, cancellationToken) + : this.DataConverter.Serialize(this.customStatus); + } + /// /// Gets the serialized custom status. /// /// The custom status serialized to a string, or null if there is not custom status. internal string? GetSerializedCustomStatus() { + Check.ThrowIfLargePayloadEnabled( + this.invocationContext.Options.EnableLargePayloadSupport, + nameof(this.GetSerializedCustomStatus)); return this.DataConverter.Serialize(this.customStatus); } diff --git a/src/Worker/Core/Shims/TaskOrchestrationEntityContext.cs b/src/Worker/Core/Shims/TaskOrchestrationEntityContext.cs index 330fd1888..a62514318 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationEntityContext.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationEntityContext.cs @@ -98,7 +98,9 @@ public override async Task CallEntityAsync(EntityInstanceId id } else { - return this.wrapper.DataConverter.Deserialize(operationResult.Result!); + return this.wrapper.invocationContext.Options.EnableLargePayloadSupport + ? await this.wrapper.DataConverter.DeserializeAsync(operationResult.Result) + : this.wrapper.DataConverter.Deserialize(operationResult.Result); } } @@ -115,11 +117,10 @@ public override async Task CallEntityAsync(EntityInstanceId id, string operation } /// - public override Task SignalEntityAsync(EntityInstanceId id, string operationName, object? input = null, SignalEntityOptions? options = null) + public override async Task SignalEntityAsync(EntityInstanceId id, string operationName, object? input = null, SignalEntityOptions? options = null) { Check.NotDefault(id); - this.SendOperationMessage(id.ToString(), operationName, input, oneWay: true, scheduledTime: options?.SignalTime); - return Task.CompletedTask; + await this.SendOperationMessage(id.ToString(), operationName, input, oneWay: true, scheduledTime: options?.SignalTime); } /// @@ -173,7 +174,7 @@ static TaskFailureDetails ConvertFailureDetails(FailureDetails failureDetails) async Task CallEntityInternalAsync(EntityInstanceId id, string operationName, object? input) { string instanceId = id.ToString(); - Guid requestId = this.SendOperationMessage(instanceId, operationName, input, oneWay: false, scheduledTime: null); + Guid requestId = await this.SendOperationMessage(instanceId, operationName, input, oneWay: false, scheduledTime: null); OperationResult response = await this.wrapper.WaitForExternalEvent(requestId.ToString()); @@ -186,7 +187,7 @@ async Task CallEntityInternalAsync(EntityInstanceId id, string return response; } - Guid SendOperationMessage(string instanceId, string operationName, object? input, bool oneWay, DateTimeOffset? scheduledTime) + async Task SendOperationMessage(string instanceId, string operationName, object? input, bool oneWay, DateTimeOffset? scheduledTime) { if (!this.EntityContext.ValidateOperationTransition(instanceId, oneWay, out string? errorMessage)) { @@ -194,7 +195,9 @@ Guid SendOperationMessage(string instanceId, string operationName, object? input } Guid guid = this.wrapper.NewGuid(); // deterministically replayable unique id for this request - string? serializedInput = this.wrapper.DataConverter.Serialize(input); + string? serializedInput = this.wrapper.invocationContext.Options.EnableLargePayloadSupport + ? await this.wrapper.DataConverter.SerializeAsync(input, CancellationToken.None) + : this.wrapper.DataConverter.Serialize(input); var target = new OrchestrationInstance() { InstanceId = instanceId }; EntityMessageEvent entityMessageEvent = this.EntityContext.EmitRequestMessage( diff --git a/src/Worker/Core/Shims/TaskOrchestrationShim.cs b/src/Worker/Core/Shims/TaskOrchestrationShim.cs index 127038a38..b1f2e8c2f 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationShim.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationShim.cs @@ -63,7 +63,9 @@ public TaskOrchestrationShim( innerContext.MessageDataConverter = converterShim; innerContext.ErrorDataConverter = converterShim; - object? input = this.DataConverter.Deserialize(rawInput, this.implementation.InputType); + object? input = this.invocationContext.Options.EnableLargePayloadSupport + ? await this.DataConverter.DeserializeAsync(rawInput, this.implementation.InputType) + : this.DataConverter.Deserialize(rawInput, this.implementation.InputType); this.wrapperContext = new(innerContext, this.invocationContext, input, this.properties); string instanceId = innerContext.OrchestrationInstance.InstanceId; @@ -82,7 +84,9 @@ public TaskOrchestrationShim( } // Return the output (if any) as a serialized string. - return this.DataConverter.Serialize(output); + return this.invocationContext.Options.EnableLargePayloadSupport + ? await this.DataConverter.SerializeAsync(output) + : this.DataConverter.Serialize(output); } catch (TaskFailedException e) { @@ -105,9 +109,21 @@ public TaskOrchestrationShim( } } + /// + public override async Task GetStatusAsync() + { + if (this.wrapperContext == null) + { + return null; + } + + return await this.wrapperContext.GetSerializedCustomStatusAsync(); + } + /// public override string? GetStatus() { + Check.ThrowIfLargePayloadEnabled(this.invocationContext.Options.EnableLargePayloadSupport, nameof(this.GetStatus)); return this.wrapperContext?.GetSerializedCustomStatus(); } @@ -116,4 +132,13 @@ public override void RaiseEvent(OrchestrationContext context, string name, strin { this.wrapperContext?.CompleteExternalEvent(name, input); } + + /// + public override async Task RaiseEventAsync(OrchestrationContext context, string name, string input) + { + if (this.wrapperContext != null) + { + await this.wrapperContext.CompleteExternalEventAsync(name, input); + } + } } diff --git a/src/Worker/Core/Worker.csproj b/src/Worker/Core/Worker.csproj index 5791edf73..2351efe56 100644 --- a/src/Worker/Core/Worker.csproj +++ b/src/Worker/Core/Worker.csproj @@ -9,6 +9,7 @@ The worker is responsible for processing durable task work items. + diff --git a/test/Abstractions.Tests/Entities/Mocks/TestEntityState.cs b/test/Abstractions.Tests/Entities/Mocks/TestEntityState.cs index 0477f15ca..a9c096fae 100644 --- a/test/Abstractions.Tests/Entities/Mocks/TestEntityState.cs +++ b/test/Abstractions.Tests/Entities/Mocks/TestEntityState.cs @@ -28,4 +28,10 @@ public override void SetState(object? state) { this.State = state; } + + public override Task SetStateAsync(object? state) + { + this.SetState(state); + return Task.CompletedTask; + } } diff --git a/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj b/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj index e6b0aee76..ba2c8b687 100644 --- a/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj +++ b/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj @@ -7,6 +7,7 @@ + diff --git a/test/Grpc.IntegrationTests/LargePayloadTests.cs b/test/Grpc.IntegrationTests/LargePayloadTests.cs new file mode 100644 index 000000000..a8a092095 --- /dev/null +++ b/test/Grpc.IntegrationTests/LargePayloadTests.cs @@ -0,0 +1,402 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.Entities; +using Microsoft.DurableTask.Converters; +using Microsoft.DurableTask.Entities; +using Microsoft.DurableTask.Worker; +using Microsoft.Extensions.DependencyInjection; +using Xunit.Abstractions; + +namespace Microsoft.DurableTask.Grpc.Tests; + +public class LargePayloadTests(ITestOutputHelper output, GrpcSidecarFixture sidecarFixture) : IntegrationTestBase(output, sidecarFixture) +{ + // Validates client externalizes a large orchestration input and worker resolves it. + [Fact] + public async Task LargeOrchestrationInput() + { + string largeInput = new string('A', 1024 * 1024); // 1MB + TaskName orchestratorName = nameof(LargeOrchestrationInput); + + InMemoryPayloadStore fakeStore = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orchestratorName, + (ctx, input) => Task.FromResult(input))); + + // Enable externalization on the worker + worker.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // small threshold to force externalization for test data + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + + // Override store with in-memory test double + worker.Services.AddSingleton(fakeStore); + }, + client => + { + // Enable externalization on the client + client.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + + // Override store with in-memory test double + client.Services.AddSingleton(fakeStore); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, input: largeInput); + + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + + // Validate that the input made a roundtrip and was resolved on the worker + string? echoed = completed.ReadOutputAs(); + Assert.NotNull(echoed); + Assert.Equal(largeInput.Length, echoed!.Length); + + // Ensure client externalized the input + Assert.True(fakeStore.UploadCount >= 1); + } + + // Validates worker externalizes large activity input and delivers resolved payload to activity. + [Fact] + public async Task LargeActivityInput() + { + string largeParam = new string('P', 700 * 1024); // 700KB + TaskName orchestratorName = nameof(LargeActivityInput); + TaskName activityName = "EchoLength"; + + InMemoryPayloadStore workerStore = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks + .AddOrchestratorFunc( + orchestratorName, + (ctx, _) => ctx.CallActivityAsync(activityName, largeParam)) + .AddActivityFunc(activityName, (ctx, input) => input.Length)); + + worker.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // force externalization for activity input + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + worker.Services.AddSingleton(workerStore); + }, + client => { /* client not needed for externalization path here */ }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal(largeParam.Length, completed.ReadOutputAs()); + + // Worker externalizes when sending activity input; worker resolves when delivering to activity + Assert.True(workerStore.UploadCount >= 1); + Assert.True(workerStore.DownloadCount >= 1); + } + + // Validates worker externalizes large activity output which is resolved by the orchestrator. + [Fact] + public async Task LargeActivityOutput() + { + string largeResult = new string('R', 850 * 1024); // 850KB + TaskName orchestratorName = nameof(LargeActivityOutput); + TaskName activityName = "ProduceLarge"; + + InMemoryPayloadStore workerStore = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks + .AddOrchestratorFunc( + orchestratorName, + async (ctx, _) => (await ctx.CallActivityAsync(activityName)).Length) + .AddActivityFunc(activityName, (ctx) => Task.FromResult(largeResult))); + + worker.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // force externalization for activity result + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + worker.Services.AddSingleton(workerStore); + }, + client => { }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal(largeResult.Length, completed.ReadOutputAs()); + + // Worker externalizes activity output and downloads when the orchestrator reads it + Assert.True(workerStore.UploadCount >= 1); + Assert.True(workerStore.DownloadCount >= 1); + } + + // Ensures querying a completed instance downloads and resolves an externalized output on the client. + [Fact] + public async Task LargeOrchestrationOutput() + { + string largeOutput = new string('Q', 900 * 1024); // 900KB + string smallInput = "input"; + TaskName orchestratorName = nameof(LargeOrchestrationOutput); + + Dictionary shared = new System.Collections.Generic.Dictionary(); + InMemoryPayloadStore workerStore = new InMemoryPayloadStore(shared); + InMemoryPayloadStore clientStore = new InMemoryPayloadStore(shared); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orchestratorName, + (ctx, _) => Task.FromResult(largeOutput))); + + worker.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // force externalization on worker + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + worker.Services.AddSingleton(workerStore); + }, + client => + { + client.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // allow client to resolve on query + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + client.Services.AddSingleton(clientStore); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, input: smallInput); + await server.Client.WaitForInstanceCompletionAsync(instanceId, getInputsAndOutputs: false, this.TimeoutToken); + + OrchestrationMetadata? queried = await server.Client.GetInstanceAsync(instanceId, getInputsAndOutputs: true); + + Assert.NotNull(queried); + Assert.Equal(OrchestrationRuntimeStatus.Completed, queried!.RuntimeStatus); + Assert.Equal(smallInput, queried.ReadInputAs()); + Assert.Equal(largeOutput, queried.ReadOutputAs()); + + Assert.True(workerStore.UploadCount == 0); + Assert.True(clientStore.DownloadCount == 1); + Assert.True(clientStore.UploadCount == 1); + } + + // Ensures payloads below the threshold are not externalized by client or worker. + [Fact] + public async Task NoLargePayloads() + { + string smallPayload = new string('X', 64 * 1024); // 64KB + TaskName orchestratorName = nameof(NoLargePayloads); + + InMemoryPayloadStore workerStore = new InMemoryPayloadStore(); + InMemoryPayloadStore clientStore = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orchestratorName, + (ctx, input) => Task.FromResult(input))); + + worker.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 2 * 1024 * 1024; // 2MB, higher than payload + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + worker.Services.AddSingleton(workerStore); + }, + client => + { + client.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 2 * 1024 * 1024; // 2MB, higher than payload + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + client.Services.AddSingleton(clientStore); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, input: smallPayload); + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal(smallPayload, completed.ReadOutputAs()); + + Assert.Equal(0, workerStore.UploadCount); + Assert.Equal(0, workerStore.DownloadCount); + Assert.Equal(0, clientStore.UploadCount); + Assert.Equal(0, clientStore.DownloadCount); + } + + // Validates client externalizes a large external event payload and worker resolves it. + [Fact] + public async Task LargeExternalEvent() + { + string largeEvent = new string('E', 512 * 1024); // 512KB + TaskName orchestratorName = nameof(LargeExternalEvent); + const string EventName = "LargeEvent"; + + InMemoryPayloadStore fakeStore = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orchestratorName, + async ctx => await ctx.WaitForExternalEvent(EventName))); + + worker.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // force externalization + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + worker.Services.AddSingleton(fakeStore); + }, + client => + { + client.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // force externalization + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + client.Services.AddSingleton(fakeStore); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + await server.Client.WaitForInstanceStartAsync(instanceId, this.TimeoutToken); + + await server.Client.RaiseEventAsync(instanceId, EventName, largeEvent, this.TimeoutToken); + + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + string? output = completed.ReadOutputAs(); + Assert.Equal(largeEvent, output); + Assert.True(fakeStore.UploadCount >= 1); + } + + // Validates worker externalizes both output and custom status; client resolves them on query. + [Fact] + public async Task LargeOutputAndCustomStatus() + { + string largeOutput = new string('O', 768 * 1024); // 768KB + string largeStatus = new string('S', 600 * 1024); // 600KB + TaskName orchestratorName = nameof(LargeOutputAndCustomStatus); + + InMemoryPayloadStore fakeStore = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orchestratorName, + async (ctx, _) => + { + ctx.SetCustomStatus(largeStatus); + await ctx.CreateTimer(TimeSpan.Zero, CancellationToken.None); + return largeOutput; + })); + + worker.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // ensure externalization for status/output + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + worker.Services.AddSingleton(fakeStore); + }, + client => + { + client.UseExternalizedPayloads(opts => + { + opts.ExternalizeThresholdBytes = 1024; // ensure resolution on query + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + client.Services.AddSingleton(fakeStore); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal(largeOutput, completed.ReadOutputAs()); + Assert.Equal(largeStatus, completed.ReadCustomStatusAs()); + + // Worker may externalize both status and output + Assert.True(fakeStore.UploadCount >= 2); + } + + class InMemoryPayloadStore : IPayloadStore + { + const string TokenPrefix = "blob:v1:"; + readonly Dictionary tokenToPayload; + + public InMemoryPayloadStore() + : this(new Dictionary()) + { + } + + public InMemoryPayloadStore(Dictionary shared) + { + this.tokenToPayload = shared; + } + + int uploadCount; + public int UploadCount => this.uploadCount; + int downloadCount; + public int DownloadCount => this.downloadCount; + + public Task UploadAsync(ReadOnlyMemory payloadBytes, CancellationToken cancellationToken) + { + Interlocked.Increment(ref this.uploadCount); + string json = System.Text.Encoding.UTF8.GetString(payloadBytes.Span); + string token = $"blob:v1:test:{Guid.NewGuid():N}"; + this.tokenToPayload[token] = json; + return Task.FromResult(token); + } + + public Task DownloadAsync(string token, CancellationToken cancellationToken) + { + Interlocked.Increment(ref this.downloadCount); + return Task.FromResult(this.tokenToPayload[token]); + } + + public bool IsKnownPayloadToken(string value) + { + return value.StartsWith(TokenPrefix, StringComparison.Ordinal); + } + + } +} diff --git a/test/ScheduledTasks.Tests/ScheduledTasks.Tests.csproj b/test/ScheduledTasks.Tests/ScheduledTasks.Tests.csproj index 33726a624..04f0d7205 100644 --- a/test/ScheduledTasks.Tests/ScheduledTasks.Tests.csproj +++ b/test/ScheduledTasks.Tests/ScheduledTasks.Tests.csproj @@ -9,10 +9,6 @@ true - - - -