diff --git a/Directory.Packages.props b/Directory.Packages.props index 7d320515..d5300050 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -24,6 +24,7 @@ + @@ -34,6 +35,11 @@ + + + + + diff --git a/Microsoft.DurableTask.sln b/Microsoft.DurableTask.sln index 26c2e80d..383f2e15 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 00000000..b0f2914c --- /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 00000000..1591d8c0 --- /dev/null +++ b/samples/LargePayloadConsoleApp/Program.cs @@ -0,0 +1,194 @@ +// 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.DurableTask; +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'"); + +// 1) Register shared payload store ONCE +builder.Services.AddExternalizedPayloadStore(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"); +}); + +// 2) Configure Durable Task client +builder.Services.AddDurableTaskClient(b => +{ + b.UseDurableTaskScheduler(schedulerConnectionString); + b.Configure(o => o.EnableEntitySupport = true); + + // Use shared store (no duplication of options) + b.UseExternalizedPayloads(); +}); + +// 3) Configure Durable Task worker +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 (value.StartsWith("blob:v1:", StringComparison.Ordinal)) + { + throw new InvalidOperationException("Activity received a payload token instead of raw input."); + } + + return value; + }); + + // Entity samples + tasks.AddOrchestratorFunc( + "LargeEntityOperationInput", + (ctx, _) => ctx.Entities.CallEntityAsync( + new EntityInstanceId(nameof(EchoLengthEntity), "1"), + operationName: "EchoLength", + input: new string('E', 700 * 1024))); + tasks.AddEntity(nameof(EchoLengthEntity)); + + tasks.AddOrchestratorFunc( + "LargeEntityOperationOutput", + async (ctx, _) => (await ctx.Entities.CallEntityAsync( + new EntityInstanceId(nameof(LargeResultEntity), "1"), + operationName: "Produce", + input: 850 * 1024)).Length); + tasks.AddEntity(nameof(LargeResultEntity)); + + 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)); + }); + + // Use shared store (no duplication of options) + b.UseExternalizedPayloads(); + + b.Configure(o => o.EnableEntitySupport = 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}, Input 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; + } +} \ No newline at end of file diff --git a/samples/LargePayloadConsoleApp/Properties/launchSettings.json b/samples/LargePayloadConsoleApp/Properties/launchSettings.json new file mode 100644 index 00000000..89f6b592 --- /dev/null +++ b/samples/LargePayloadConsoleApp/Properties/launchSettings.json @@ -0,0 +1,12 @@ +{ + "profiles": { + "LargePayloadConsoleApp": { + "commandName": "Project", + "environmentVariables": { + "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "", + "DURABLETASK_STORAGE": "", + "DURABLETASK_PAYLOAD_CONTAINER": "" + } + } + } +} diff --git a/samples/LargePayloadConsoleApp/README.md b/samples/LargePayloadConsoleApp/README.md new file mode 100644 index 00000000..812098ef --- /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/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj new file mode 100644 index 00000000..8462b0f1 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj @@ -0,0 +1,30 @@ + + + + netstandard2.0;net6.0 + Azure Blob Storage externalized payload support for Durable Task. + Microsoft.DurableTask.Extensions.AzureBlobPayloads + Microsoft.DurableTask + true + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs new file mode 100644 index 00000000..0817bcea --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Grpc.Core.Interceptors; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.Grpc; +using Microsoft.DurableTask.Converters; +using Microsoft.DurableTask.Worker.Grpc.Internal; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Microsoft.DurableTask; + +/// +/// Extension methods to enable externalized payloads using Azure Blob Storage for Durable Task Client. +/// +public static class DurableTaskClientBuilderExtensionsAzureBlobPayloads +{ + /// + /// Enables externalized payload storage using a pre-configured shared payload store. + /// This overload helps ensure client and worker use the same configuration. + /// + /// The builder to configure. + /// The original builder, for call chaining. + public static IDurableTaskClientBuilder UseExternalizedPayloads( + this IDurableTaskClientBuilder builder) + { + Check.NotNull(builder); + return UseExternalizedPayloadsCore(builder); + } + + static IDurableTaskClientBuilder UseExternalizedPayloadsCore(IDurableTaskClientBuilder builder) + { + // Wrap the gRPC CallInvoker with our interceptor when using the gRPC client + builder.Services + .AddOptions(builder.Name) + .PostConfigure>((opt, store, monitor) => + { + LargePayloadStorageOptions opts = monitor.Get(builder.Name); + if (opt.Channel is not null) + { + Grpc.Core.CallInvoker invoker = opt.Channel.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts)); + opt.CallInvoker = invoker; + + // Ensure client uses the intercepted invoker path + opt.Channel = null; + } + else if (opt.CallInvoker is not null) + { + opt.CallInvoker = opt.CallInvoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts)); + } + else + { + throw new ArgumentException( + "Channel or CallInvoker must be provided to use Azure Blob Payload Externalization feature"); + } + }); + + 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 00000000..d65e30b8 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -0,0 +1,84 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Grpc.Core.Interceptors; +using Grpc.Net.Client; +using Microsoft.DurableTask.Converters; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.Grpc; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Microsoft.DurableTask; + +/// +/// 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. + /// + /// The builder to configure. + /// The callback to configure the storage options. + /// The original builder, for call chaining. + 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); + }); + + return UseExternalizedPayloadsCore(builder); + } + + /// + /// Enables externalized payload storage using a pre-configured shared payload store. + /// This overload helps ensure client and worker use the same configuration. + /// + /// The builder to configure. + /// The original builder, for call chaining. + public static IDurableTaskWorkerBuilder UseExternalizedPayloads( + this IDurableTaskWorkerBuilder builder) + { + Check.NotNull(builder); + return UseExternalizedPayloadsCore(builder); + } + + static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerBuilder builder) + { + // Wrap the gRPC CallInvoker with our interceptor when using the gRPC worker + builder.Services + .AddOptions(builder.Name) + .PostConfigure>((opt, store, monitor) => + { + LargePayloadStorageOptions opts = monitor.Get(builder.Name); + if (opt.Channel is not null) + { + var invoker = opt.Channel.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts)); + opt.CallInvoker = invoker; + + // Ensure worker uses the intercepted invoker path + opt.Channel = null; + } + else if (opt.CallInvoker is not null) + { + opt.CallInvoker = opt.CallInvoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts)); + } + else + { + throw new ArgumentException( + "Channel or CallInvoker must be provided to use Azure Blob Payload Externalization feature"); + } + }); + + return builder; + } +} diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/ServiceCollectionExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/ServiceCollectionExtensions.AzureBlobPayloads.cs new file mode 100644 index 00000000..787888b6 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/ServiceCollectionExtensions.AzureBlobPayloads.cs @@ -0,0 +1,45 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; + +namespace Microsoft.DurableTask; + +/// +/// DI extensions for configuring a shared Azure Blob payload store used by both client and worker. +/// +public static class ServiceCollectionExtensionsAzureBlobPayloads +{ + /// + /// Registers a shared Azure Blob-based externalized payload store and its options. + /// The provided options apply to all named Durable Task builders (client/worker), + /// so UseExternalizedPayloads() can be called without repeating configuration. + /// + /// The service collection. + /// The configuration callback for the payload store. + /// The original service collection. + public static IServiceCollection AddExternalizedPayloadStore( + this IServiceCollection services, + Action configure) + { + Check.NotNull(services); + Check.NotNull(configure); + + // Apply once to ALL names (IConfigureOptions hits every named options instance), + // so monitor.Get(builder.Name) in the client/worker extensions will see the same config. + services.Configure(configure); + + // Provide a single shared PayloadStore instance built from the default options. + services.AddSingleton(sp => + { + IOptionsMonitor monitor = + sp.GetRequiredService>(); + + LargePayloadStorageOptions opts = monitor.Get(Options.DefaultName); + return new BlobPayloadStore(opts); + }); + + return services; + } +} diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs new file mode 100644 index 00000000..08bda009 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -0,0 +1,363 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Grpc.Core.Interceptors; + +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask; + +/// +/// gRPC interceptor that externalizes large payloads to an on requests +/// and resolves known payload tokens on responses for SideCar. +/// +public sealed class AzureBlobPayloadsSideCarInterceptor(PayloadStore payloadStore, LargePayloadStorageOptions options) + : PayloadInterceptor(payloadStore, options) +{ + /// + protected override async Task ExternalizeRequestPayloadsAsync(TRequest request, CancellationToken cancellation) + { + // Client -> sidecar + switch (request) + { + case P.CreateInstanceRequest r: + r.Input = await this.MaybeExternalizeAsync(r.Input, cancellation); + break; + case P.RaiseEventRequest r: + r.Input = await this.MaybeExternalizeAsync(r.Input, cancellation); + break; + case P.TerminateRequest r: + r.Output = await this.MaybeExternalizeAsync(r.Output, cancellation); + break; + case P.SuspendRequest r: + r.Reason = await this.MaybeExternalizeAsync(r.Reason, cancellation); + break; + case P.ResumeRequest r: + r.Reason = await this.MaybeExternalizeAsync(r.Reason, cancellation); + break; + case P.SignalEntityRequest r: + r.Input = await this.MaybeExternalizeAsync(r.Input, cancellation); + break; + case P.ActivityResponse r: + r.Result = await this.MaybeExternalizeAsync(r.Result, cancellation); + break; + case P.OrchestratorResponse r: + await this.ExternalizeOrchestratorResponseAsync(r, cancellation); + break; + case P.EntityBatchResult r: + await this.ExternalizeEntityBatchResultAsync(r, cancellation); + break; + case P.EntityBatchRequest r: + await this.ExternalizeEntityBatchRequestAsync(r, cancellation); + break; + case P.EntityRequest r: + r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation); + break; + } + } + + /// + protected override async Task ResolveResponsePayloadsAsync(TResponse response, CancellationToken cancellation) + { + // Sidecar -> client/worker + switch (response) + { + case P.GetInstanceResponse r when r.OrchestrationState is { } s: + s.Input = await this.MaybeResolveAsync(s.Input, cancellation); + s.Output = await this.MaybeResolveAsync(s.Output, cancellation); + s.CustomStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation); + break; + case P.HistoryChunk c when c.Events != null: + foreach (P.HistoryEvent e in c.Events) + { + await this.ResolveEventPayloadsAsync(e, cancellation); + } + + break; + case P.QueryInstancesResponse r: + foreach (P.OrchestrationState s in r.OrchestrationState) + { + s.Input = await this.MaybeResolveAsync(s.Input, cancellation); + s.Output = await this.MaybeResolveAsync(s.Output, cancellation); + s.CustomStatus = await this.MaybeResolveAsync(s.CustomStatus, cancellation); + } + + break; + case P.GetEntityResponse r when r.Entity is { } em: + em.SerializedState = await this.MaybeResolveAsync(em.SerializedState, cancellation); + break; + case P.QueryEntitiesResponse r: + foreach (P.EntityMetadata em in r.Entities) + { + em.SerializedState = await this.MaybeResolveAsync(em.SerializedState, cancellation); + } + + break; + case P.WorkItem wi: + // Resolve activity input + if (wi.ActivityRequest is { } ar) + { + ar.Input = await this.MaybeResolveAsync(ar.Input, cancellation); + } + + // Resolve orchestration input embedded in ExecutionStarted event and external events + if (wi.OrchestratorRequest is { } or) + { + foreach (P.HistoryEvent? e in or.PastEvents) + { + await this.ResolveEventPayloadsAsync(e, cancellation); + } + + foreach (P.HistoryEvent? e in or.NewEvents) + { + await this.ResolveEventPayloadsAsync(e, cancellation); + } + } + + // Resolve entity V1 batch request (OperationRequest inputs and entity state) + if (wi.EntityRequest is { } er1) + { + er1.EntityState = await this.MaybeResolveAsync(er1.EntityState, cancellation); + if (er1.Operations != null) + { + foreach (P.OperationRequest op in er1.Operations) + { + op.Input = await this.MaybeResolveAsync(op.Input, cancellation); + } + } + } + + // Resolve entity V2 request (history-based operation requests and entity state) + if (wi.EntityRequestV2 is { } er2) + { + er2.EntityState = await this.MaybeResolveAsync(er2.EntityState, cancellation); + if (er2.OperationRequests != null) + { + foreach (P.HistoryEvent opEvt in er2.OperationRequests) + { + await this.ResolveEventPayloadsAsync(opEvt, cancellation); + } + } + } + + break; + } + } + + async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, CancellationToken cancellation) + { + r.CustomStatus = await this.MaybeExternalizeAsync(r.CustomStatus, cancellation); + foreach (P.OrchestratorAction a in r.Actions) + { + if (a.CompleteOrchestration is { } complete) + { + complete.Result = await this.MaybeExternalizeAsync(complete.Result, cancellation); + complete.Details = await this.MaybeExternalizeAsync(complete.Details, cancellation); + } + + if (a.TerminateOrchestration is { } term) + { + term.Reason = await this.MaybeExternalizeAsync(term.Reason, cancellation); + } + + if (a.ScheduleTask is { } schedule) + { + schedule.Input = await this.MaybeExternalizeAsync(schedule.Input, cancellation); + } + + if (a.CreateSubOrchestration is { } sub) + { + sub.Input = await this.MaybeExternalizeAsync(sub.Input, cancellation); + } + + if (a.SendEvent is { } sendEvt) + { + sendEvt.Data = await this.MaybeExternalizeAsync(sendEvt.Data, cancellation); + } + + if (a.SendEntityMessage is { } entityMsg) + { + if (entityMsg.EntityOperationSignaled is { } sig) + { + sig.Input = await this.MaybeExternalizeAsync(sig.Input, cancellation); + } + + if (entityMsg.EntityOperationCalled is { } called) + { + called.Input = await this.MaybeExternalizeAsync(called.Input, cancellation); + } + } + } + } + + async Task ExternalizeEntityBatchResultAsync(P.EntityBatchResult r, CancellationToken cancellation) + { + r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation); + if (r.Results != null) + { + foreach (P.OperationResult result in r.Results) + { + if (result.Success is { } success) + { + success.Result = await this.MaybeExternalizeAsync(success.Result, cancellation); + } + } + } + + if (r.Actions != null) + { + foreach (P.OperationAction action in r.Actions) + { + if (action.SendSignal is { } sendSig) + { + sendSig.Input = await this.MaybeExternalizeAsync(sendSig.Input, cancellation); + } + + if (action.StartNewOrchestration is { } start) + { + start.Input = await this.MaybeExternalizeAsync(start.Input, cancellation); + } + } + } + } + + async Task ExternalizeEntityBatchRequestAsync(P.EntityBatchRequest r, CancellationToken cancellation) + { + r.EntityState = await this.MaybeExternalizeAsync(r.EntityState, cancellation); + if (r.Operations != null) + { + foreach (P.OperationRequest op in r.Operations) + { + op.Input = await this.MaybeExternalizeAsync(op.Input, cancellation); + } + } + } + + async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancellation) + { + switch (e.EventTypeCase) + { + case P.HistoryEvent.EventTypeOneofCase.ExecutionStarted: + if (e.ExecutionStarted is { } es) + { + es.Input = await this.MaybeResolveAsync(es.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted: + if (e.ExecutionCompleted is { } ec) + { + ec.Result = await this.MaybeResolveAsync(ec.Result, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.EventRaised: + if (e.EventRaised is { } er) + { + er.Input = await this.MaybeResolveAsync(er.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.TaskScheduled: + if (e.TaskScheduled is { } ts) + { + ts.Input = await this.MaybeResolveAsync(ts.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: + if (e.TaskCompleted is { } tc) + { + tc.Result = await this.MaybeResolveAsync(tc.Result, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated: + if (e.SubOrchestrationInstanceCreated is { } soc) + { + soc.Input = await this.MaybeResolveAsync(soc.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: + if (e.SubOrchestrationInstanceCompleted is { } sox) + { + sox.Result = await this.MaybeResolveAsync(sox.Result, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.EventSent: + if (e.EventSent is { } esent) + { + esent.Input = await this.MaybeResolveAsync(esent.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.GenericEvent: + if (e.GenericEvent is { } ge) + { + ge.Data = await this.MaybeResolveAsync(ge.Data, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.ContinueAsNew: + if (e.ContinueAsNew is { } can) + { + can.Input = await this.MaybeResolveAsync(can.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated: + if (e.ExecutionTerminated is { } et) + { + et.Input = await this.MaybeResolveAsync(et.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended: + if (e.ExecutionSuspended is { } esus) + { + esus.Input = await this.MaybeResolveAsync(esus.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionResumed: + if (e.ExecutionResumed is { } eres) + { + eres.Input = await this.MaybeResolveAsync(eres.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled: + if (e.EntityOperationSignaled is { } eos) + { + eos.Input = await this.MaybeResolveAsync(eos.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled: + if (e.EntityOperationCalled is { } eoc) + { + eoc.Input = await this.MaybeResolveAsync(eoc.Input, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted: + if (e.EntityOperationCompleted is { } ecomp) + { + ecomp.Output = await this.MaybeResolveAsync(ecomp.Output, cancellation); + } + + break; + case P.HistoryEvent.EventTypeOneofCase.HistoryState: + if (e.HistoryState is { } hs && hs.OrchestrationState is { } os) + { + os.Input = await this.MaybeResolveAsync(os.Input, cancellation); + os.Output = await this.MaybeResolveAsync(os.Output, cancellation); + os.CustomStatus = await this.MaybeResolveAsync(os.CustomStatus, cancellation); + } + + break; + } + } +} diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs new file mode 100644 index 00000000..921499f2 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs @@ -0,0 +1,230 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Text; +using Grpc.Core; +using Grpc.Core.Interceptors; + +namespace Microsoft.DurableTask; + +/// +/// Base class for gRPC interceptors that externalize large payloads to an on requests +/// and resolves known payload tokens on responses. +/// +/// The namespace for request message types. +/// The namespace for response message types. +/// +/// Initializes a new instance of the class. +/// +/// The payload store. +/// The options. +public abstract class PayloadInterceptor( + PayloadStore payloadStore, + LargePayloadStorageOptions options) : Interceptor + where TRequestNamespace : class + where TResponseNamespace : class +{ + readonly PayloadStore payloadStore = payloadStore; + readonly LargePayloadStorageOptions options = options; + + // Unary: externalize on request, resolve on response + + /// + public override AsyncUnaryCall AsyncUnaryCall( + TRequest request, + ClientInterceptorContext context, + AsyncUnaryCallContinuation continuation) + { + // Build the underlying call lazily after async externalization + Task> startCallTask = Task.Run(async () => + { + // Externalize first; if this fails, do not proceed to send the gRPC call + await this.ExternalizeRequestPayloadsAsync(request, context.Options.CancellationToken); + + // Only if externalization succeeds, proceed with the continuation + return continuation(request, context); + }); + + async Task ResponseAsync() + { + AsyncUnaryCall innerCall = await startCallTask; + TResponse response = await innerCall.ResponseAsync; + await this.ResolveResponsePayloadsAsync(response, context.Options.CancellationToken); + return response; + } + + async Task ResponseHeadersAsync() + { + AsyncUnaryCall innerCall = await startCallTask; + return await innerCall.ResponseHeadersAsync; + } + + Status GetStatus() + { + if (startCallTask.IsCanceled) + { + return new Status(StatusCode.Cancelled, "Call was cancelled."); + } + + if (startCallTask.IsFaulted) + { + return new Status(StatusCode.Internal, startCallTask.Exception?.Message ?? "Unknown error"); + } + + if (startCallTask.Status == TaskStatus.RanToCompletion) + { + return startCallTask.Result.GetStatus(); + } + + // Not started yet; unknown + return new Status(StatusCode.Unknown, string.Empty); + } + + Metadata GetTrailers() + { + return startCallTask.Status == TaskStatus.RanToCompletion ? startCallTask.Result.GetTrailers() : []; + } + + void Dispose() + { + _ = startCallTask.ContinueWith( + t => + { + if (t.Status == TaskStatus.RanToCompletion) + { + t.Result.Dispose(); + } + }, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + + return new AsyncUnaryCall( + ResponseAsync(), + ResponseHeadersAsync(), + GetStatus, + GetTrailers, + Dispose); + } + + /// + /// Asynchronously streams a response from a server. + /// + /// The request type. + /// The response type. + /// The request to process. + /// The context of the call. + /// The continuation of the call. + /// A task representing the async operation. + /// + /// Server streaming: resolve payloads in streamed responses (e.g., GetWorkItems). + /// + public override AsyncServerStreamingCall AsyncServerStreamingCall( + TRequest request, + ClientInterceptorContext context, + AsyncServerStreamingCallContinuation continuation) + { + // For streaming, request externalization is not needed currently + AsyncServerStreamingCall call = continuation(request, context); + + IAsyncStreamReader wrapped = new TransformingStreamReader(call.ResponseStream, async (msg, ct) => + { + await this.ResolveResponsePayloadsAsync(msg, ct); + return msg; + }); + + return new AsyncServerStreamingCall( + wrapped, + call.ResponseHeadersAsync, + call.GetStatus, + call.GetTrailers, + call.Dispose); + } + + /// + /// Externalizes large payloads in request messages. + /// + /// The request type. + /// The request to process. + /// Cancellation token. + /// A task representing the async operation. + protected abstract Task ExternalizeRequestPayloadsAsync(TRequest request, CancellationToken cancellation); + + /// + /// Resolves payload tokens in response messages. + /// + /// The response type. + /// The response to process. + /// Cancellation token. + /// A task representing the async operation. + protected abstract Task ResolveResponsePayloadsAsync(TResponse response, CancellationToken cancellation); + + /// + /// Externalizes a payload if it exceeds the threshold. + /// + /// The value to potentially externalize. + /// Cancellation token. + /// A task that returns the externalized token or the original value. + protected async Task MaybeExternalizeAsync(string? value, CancellationToken cancellation) + { + if (string.IsNullOrEmpty(value)) + { + return value; + } + + int size = Encoding.UTF8.GetByteCount(value); + if (size < this.options.ExternalizeThresholdBytes) + { + return value; + } + + // Enforce a hard cap to prevent unbounded payload sizes + if (size > this.options.MaxExternalizedPayloadBytes) + { + throw new InvalidOperationException( + $"Payload size {size / 1024} kb exceeds the configured maximum of {this.options.MaxExternalizedPayloadBytes / 1024} kb. " + + "Consider reducing the payload or increase MaxExternalizedPayloadBytes setting."); + } + + return await this.payloadStore.UploadAsync(value!, cancellation); + } + + /// + /// Resolves a payload token if it's known to the store. + /// + /// The value to potentially resolve. + /// Cancellation token. + /// The resolved value or the original value if it's not a known payload token. + protected async Task MaybeResolveAsync(string? value, CancellationToken cancellation) + { + if (string.IsNullOrEmpty(value) || !this.payloadStore.IsKnownPayloadToken(value ?? string.Empty)) + { + return value; + } + + return await this.payloadStore.DownloadAsync(value!, cancellation); + } + + sealed class TransformingStreamReader( + IAsyncStreamReader inner, + Func> transform) : IAsyncStreamReader + { + readonly IAsyncStreamReader inner = inner; + readonly Func> transform = transform; + + public T Current { get; private set; } = default!; + + public async Task MoveNext(CancellationToken cancellationToken) + { + bool hasNext = await this.inner.MoveNext(cancellationToken); + if (!hasNext) + { + return false; + } + + this.Current = await this.transform(this.inner.Current, cancellationToken); + return true; + } + } +} diff --git a/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs new file mode 100644 index 00000000..7bd47bfc --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs @@ -0,0 +1,119 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure.Core; + +// Intentionally no DataAnnotations to avoid extra package requirements in minimal hosts. +namespace Microsoft.DurableTask; + +/// +/// Options for externalized payload storage, used by SDKs to store large payloads out-of-band. +/// Supports both connection string and identity-based authentication. +/// +/// +/// Connection string authentication: +/// +/// var options = new LargePayloadStorageOptions("DefaultEndpointsProtocol=https;AccountName=mystorageaccount;AccountKey=..."); +/// +/// +/// Identity-based authentication: +/// +/// var options = new LargePayloadStorageOptions( +/// new Uri("https://mystorageaccount.blob.core.windows.net"), +/// new DefaultAzureCredential()); +/// +/// +/// +public sealed class LargePayloadStorageOptions +{ + int externalizeThresholdBytes = 900_000; + + /// + /// 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; + } + + /// + /// Initializes a new instance of the class. + /// + /// The Azure Storage account URI. + /// The credential to use for authentication. + public LargePayloadStorageOptions(Uri accountUri, TokenCredential credential) + { + Check.NotNull(accountUri, nameof(accountUri)); + Check.NotNull(credential, nameof(credential)); + this.AccountUri = accountUri; + this.Credential = credential; + } + + /// + /// Gets or sets the threshold in bytes at which payloads are externalized. Default is 900_000 bytes. + /// Value must not exceed 1 MiB (1,048,576 bytes). + /// + + public int ExternalizeThresholdBytes + { + get => this.externalizeThresholdBytes; + set + { + const int OneMiB = 1 * 1024 * 1024; + if (value > OneMiB) + { + throw new ArgumentOutOfRangeException( + nameof(this.ExternalizeThresholdBytes), + $"ExternalizeThresholdBytes cannot exceed 1 MiB ({OneMiB} bytes)."); + } + + this.externalizeThresholdBytes = value; + } + } + + /// + /// Gets or sets the maximum allowed size in bytes for any single externalized payload. + /// Defaults to 10MB. Requests exceeding this limit will fail fast + /// with a clear error to prevent unbounded payload growth and excessive storage/network usage. + /// + public int MaxExternalizedPayloadBytes { get; set; } = 10 * 1024 * 1024; + + /// + /// Gets or sets the Azure Storage connection string to the customer's storage account. + /// Either this or and must be set. + /// + public string ConnectionString { get; set; } = string.Empty; + + /// + /// Gets or sets the Azure Storage account URI. + /// Either this and or must be set. + /// + public Uri? AccountUri { get; set; } + + /// + /// Gets or sets the credential to use for authentication. + /// Either this and or must be set. + /// + public TokenCredential? Credential { get; set; } + + /// + /// 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/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs new file mode 100644 index 00000000..06294f5f --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -0,0 +1,165 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.IO.Compression; +using System.Text; +using Azure.Core; +using Azure.Storage.Blobs; +using Azure.Storage.Blobs.Models; + +namespace Microsoft.DurableTask; + +/// +/// Azure Blob Storage implementation of . +/// Stores payloads as blobs and returns opaque tokens in the form "blob:v1:<container>:<blobName>". +/// +public sealed class BlobPayloadStore : PayloadStore +{ + const string TokenPrefix = "blob:v1:"; + const string ContentEncodingGzip = "gzip"; + const int MaxRetryAttempts = 8; + const int BaseDelayMs = 250; + const int MaxDelayMs = 10_000; + const int NetworkTimeoutMinutes = 2; + 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 neither connection string nor account URI/credential are provided. + public BlobPayloadStore(LargePayloadStorageOptions options) + { + this.options = options ?? throw new ArgumentNullException(nameof(options)); + Check.NotNullOrEmpty(options.ContainerName, nameof(options.ContainerName)); + + // Validate that either connection string or account URI/credential are provided + bool hasConnectionString = !string.IsNullOrEmpty(options.ConnectionString); + bool hasIdentityAuth = options.AccountUri != null && options.Credential != null; + + if (!hasConnectionString && !hasIdentityAuth) + { + throw new ArgumentException( + "Either ConnectionString or AccountUri and Credential must be provided.", + nameof(options)); + } + + BlobClientOptions clientOptions = new() + { + Retry = + { + Mode = RetryMode.Exponential, + MaxRetries = MaxRetryAttempts, + Delay = TimeSpan.FromMilliseconds(BaseDelayMs), + MaxDelay = TimeSpan.FromSeconds(MaxDelayMs), + NetworkTimeout = TimeSpan.FromMinutes(NetworkTimeoutMinutes), + }, + }; + + BlobServiceClient serviceClient = hasIdentityAuth + ? new BlobServiceClient(options.AccountUri, options.Credential, clientOptions) + : new BlobServiceClient(options.ConnectionString, clientOptions); + + this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName); + } + + /// + public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken) + { + // One blob per payload using GUID-based name for uniqueness (stable across retries) + string blobName = $"{Guid.NewGuid():N}"; + BlobClient blob = this.containerClient.GetBlobClient(blobName); + + byte[] payloadBuffer = Encoding.UTF8.GetBytes(payLoad); + + // Ensure container exists (idempotent) + await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, default, default, cancellationToken); + + if (this.options.CompressPayloads) + { + BlobOpenWriteOptions writeOptions = new() + { + HttpHeaders = new BlobHttpHeaders { ContentEncoding = ContentEncodingGzip }, + }; + using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, cancellationToken); + using GZipStream compressedBlobStream = new(blobStream, System.IO.Compression.CompressionLevel.Optimal, leaveOpen: true); + + // using MemoryStream payloadStream = new(payloadBuffer, writable: false); + + // await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, cancellationToken); + await compressedBlobStream.WriteAsync(payloadBuffer, 0, payloadBuffer.Length, cancellationToken); + await compressedBlobStream.FlushAsync(cancellationToken); + await blobStream.FlushAsync(cancellationToken); + } + else + { + using Stream blobStream = await blob.OpenWriteAsync(true, default, cancellationToken); + + // using MemoryStream payloadStream = new(payloadBuffer, writable: false); + // await payloadStream.CopyToAsync(blobStream, bufferSize: DefaultCopyBufferSize, cancellationToken); + await blobStream.WriteAsync(payloadBuffer, 0, payloadBuffer.Length, cancellationToken); + await blobStream.FlushAsync(cancellationToken); + } + + return EncodeToken(this.containerClient.Name, blobName); + } + + /// + public override 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); + + using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken); + 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(); + } + + /// + public override 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(TokenPrefix, StringComparison.Ordinal)) + { + throw new ArgumentException("Invalid external payload token.", nameof(token)); + } + + string rest = token.Substring(TokenPrefix.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)); + } +} diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs new file mode 100644 index 00000000..b0fe6f80 --- /dev/null +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs @@ -0,0 +1,34 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask; + +/// +/// Abstraction for storing and retrieving large payloads out-of-band. +/// +public abstract class PayloadStore +{ + /// + /// Uploads a payload and returns an opaque reference token that can be embedded in orchestration messages. + /// + /// The payload. + /// Cancellation token. + /// Opaque reference token. + public abstract Task UploadAsync(string payLoad, CancellationToken cancellationToken); + + /// + /// Downloads the payload referenced by the token. + /// + /// The opaque reference token. + /// Cancellation token. + /// Payload string. + public abstract 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. + public abstract bool IsKnownPayloadToken(string value); +} diff --git a/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj b/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj index e6b0aee7..ba2c8b68 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/GrpcSidecar/Grpc/TaskHubGrpcServer.cs b/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/TaskHubGrpcServer.cs index ecf7975e..aba59317 100644 --- a/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/TaskHubGrpcServer.cs +++ b/test/Grpc.IntegrationTests/GrpcSidecar/Grpc/TaskHubGrpcServer.cs @@ -31,6 +31,9 @@ public class TaskHubGrpcServer : P.TaskHubSidecarService.TaskHubSidecarServiceBa readonly TaskHubDispatcherHost dispatcherHost; readonly IsConnectedSignal isConnectedSignal = new(); readonly SemaphoreSlim sendWorkItemLock = new(initialCount: 1); + readonly ConcurrentDictionary> streamingPastEvents = new(StringComparer.OrdinalIgnoreCase); + + volatile bool supportsHistoryStreaming; // Initialized when a client connects to this service to receive work-item commands. IServerStreamWriter? workerToClientStream; @@ -463,6 +466,8 @@ static P.GetInstanceResponse CreateGetInstanceResponse(OrchestrationState state, public override async Task GetWorkItems(P.GetWorkItemsRequest request, IServerStreamWriter responseStream, ServerCallContext context) { + // Record whether the client supports history streaming + this.supportsHistoryStreaming = request.Capabilities.Contains(P.WorkerCapability.HistoryStreaming); // Use a lock to mitigate the race condition where we signal the dispatch host to start but haven't // yet saved a reference to the client response stream. lock (this.isConnectedSignal) @@ -505,6 +510,35 @@ public override async Task GetWorkItems(P.GetWorkItemsRequest request, IServerSt } } + public override async Task StreamInstanceHistory(P.StreamInstanceHistoryRequest request, IServerStreamWriter responseStream, ServerCallContext context) + { + if (this.streamingPastEvents.TryGetValue(request.InstanceId, out List? pastEvents)) + { + const int MaxChunkBytes = 256 * 1024; // 256KB per chunk to simulate chunked streaming + int currentSize = 0; + P.HistoryChunk chunk = new(); + + foreach (P.HistoryEvent e in pastEvents) + { + int eventSize = e.CalculateSize(); + if (currentSize > 0 && currentSize + eventSize > MaxChunkBytes) + { + await responseStream.WriteAsync(chunk); + chunk = new P.HistoryChunk(); + currentSize = 0; + } + + chunk.Events.Add(e); + currentSize += eventSize; + } + + if (chunk.Events.Count > 0) + { + await responseStream.WriteAsync(chunk); + } + } + } + /// /// Invoked by the when a work item is available, proxies the call to execute an orchestrator over a gRPC channel. /// @@ -531,16 +565,37 @@ async Task ITaskExecutor.ExecuteOrchestrator( try { + var orkRequest = new P.OrchestratorRequest + { + InstanceId = instance.InstanceId, + ExecutionId = instance.ExecutionId, + NewEvents = { newEvents.Select(ProtobufUtils.ToHistoryEventProto) }, + OrchestrationTraceContext = orchestrationTraceContext, + }; + + // Decide whether to stream based on total size of past events (> 1MiB) + List protoPastEvents = pastEvents.Select(ProtobufUtils.ToHistoryEventProto).ToList(); + int totalBytes = 0; + foreach (P.HistoryEvent ev in protoPastEvents) + { + totalBytes += ev.CalculateSize(); + } + + if (this.supportsHistoryStreaming && totalBytes > (1024)) + { + orkRequest.RequiresHistoryStreaming = true; + // Store past events to serve via StreamInstanceHistory + this.streamingPastEvents[instance.InstanceId] = protoPastEvents; + } + else + { + // Embed full history in the work item + orkRequest.PastEvents.AddRange(protoPastEvents); + } + await this.SendWorkItemToClientAsync(new P.WorkItem { - OrchestratorRequest = new P.OrchestratorRequest - { - InstanceId = instance.InstanceId, - ExecutionId = instance.ExecutionId, - NewEvents = { newEvents.Select(ProtobufUtils.ToHistoryEventProto) }, - OrchestrationTraceContext = orchestrationTraceContext, - PastEvents = { pastEvents.Select(ProtobufUtils.ToHistoryEventProto) }, - } + OrchestratorRequest = orkRequest, }); } catch diff --git a/test/Grpc.IntegrationTests/IntegrationTestBase.cs b/test/Grpc.IntegrationTests/IntegrationTestBase.cs index 642107ef..13d92a7a 100644 --- a/test/Grpc.IntegrationTests/IntegrationTestBase.cs +++ b/test/Grpc.IntegrationTests/IntegrationTestBase.cs @@ -45,9 +45,12 @@ void IDisposable.Dispose() GC.SuppressFinalize(this); } - protected async Task StartWorkerAsync(Action workerConfigure, Action? clientConfigure = null) + protected async Task StartWorkerAsync( + Action workerConfigure, + Action? clientConfigure = null, + Action? servicesConfigure = null) { - IHost host = this.CreateHostBuilder(workerConfigure, clientConfigure).Build(); + IHost host = this.CreateHostBuilder(workerConfigure, clientConfigure, servicesConfigure).Build(); await host.StartAsync(this.TimeoutToken); return new HostTestLifetime(host, this.TimeoutToken); } @@ -57,7 +60,10 @@ protected async Task StartWorkerAsync(Action /// Configures the durable task worker builder. /// Configures the durable task client builder. - protected IHostBuilder CreateHostBuilder(Action workerConfigure, Action? clientConfigure) + protected IHostBuilder CreateHostBuilder( + Action workerConfigure, + Action? clientConfigure, + Action? servicesConfigure) { var host = Host.CreateDefaultBuilder() .ConfigureLogging(b => @@ -69,6 +75,7 @@ protected IHostBuilder CreateHostBuilder(Action worke }) .ConfigureServices((context, services) => { + servicesConfigure?.Invoke(services); services.AddDurableTaskWorker(b => { b.UseGrpc(this.sidecarFixture.Channel); diff --git a/test/Grpc.IntegrationTests/LargePayloadTests.cs b/test/Grpc.IntegrationTests/LargePayloadTests.cs new file mode 100644 index 00000000..f87a7b29 --- /dev/null +++ b/test/Grpc.IntegrationTests/LargePayloadTests.cs @@ -0,0 +1,683 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Text.Json; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Converters; +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 LargeOrchestrationInputAndOutputAndCustomStatus() + { + string largeInput = new string('A', 1024 * 1024); // 1MB + TaskName orchestratorName = nameof(LargeOrchestrationInputAndOutputAndCustomStatus); + + InMemoryPayloadStore fakeStore = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orchestratorName, + (ctx, input) => + { + ctx.SetCustomStatus(largeInput); + return Task.FromResult(input + input); + })); + + worker.UseExternalizedPayloads(); + + worker.Services.AddSingleton(fakeStore); + }, + client => + { + client.UseExternalizedPayloads(); + + // Override store with in-memory test double + client.Services.AddSingleton(fakeStore); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + 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 + // validate input + string? input = completed.ReadInputAs(); + Assert.NotNull(input); + Assert.Equal(largeInput.Length, input!.Length); + Assert.Equal(largeInput, input); + + string? echoed = completed.ReadOutputAs(); + Assert.NotNull(echoed); + Assert.Equal(largeInput.Length * 2, echoed!.Length); + Assert.Equal(largeInput + largeInput, echoed); + + string? customStatus = completed.ReadCustomStatusAs(); + Assert.NotNull(customStatus); + Assert.Equal(largeInput.Length, customStatus!.Length); + Assert.Equal(largeInput, customStatus); + + // Ensure client externalized the input + Assert.True(fakeStore.UploadCount >= 1); + Assert.True(fakeStore.DownloadCount >= 1); + Assert.Contains(JsonSerializer.Serialize(largeInput), fakeStore.uploadedPayloads); + Assert.Contains(JsonSerializer.Serialize(largeInput + largeInput), fakeStore.uploadedPayloads); + } + + // Validates history streaming path resolves externalized inputs/outputs in HistoryChunk. + [Fact] + public async Task HistoryStreaming_ResolvesPayloads() + { + // Make payloads large enough so that past events history exceeds 1 MiB to trigger streaming + string largeInput = new string('H', 2 * 1024 * 1024); // 2 MiB + string largeOutput = new string('O', 2 * 1024 * 1024); // 2 MiB + TaskName orch = nameof(HistoryStreaming_ResolvesPayloads); + + InMemoryPayloadStore store = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orch, + async (ctx, input) => + { + // Emit several events so that the serialized history size grows + for (int i = 0; i < 50; i++) + { + await ctx.CreateTimer(TimeSpan.FromMilliseconds(10), CancellationToken.None); + } + return largeOutput; + })); + + worker.UseExternalizedPayloads(); + worker.Services.AddSingleton(store); + }, + client => + { + client.UseExternalizedPayloads(); + client.Services.AddSingleton(store); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + // Start orchestration with large input to exercise history input resolution + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orch, largeInput); + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal(largeInput, completed.ReadInputAs()); + Assert.Equal(largeOutput, completed.ReadOutputAs()); + Assert.True(store.UploadCount >= 2); + Assert.True(store.DownloadCount >= 2); + } + + // Validates client externalizes large suspend and resume reasons. + [Fact] + public async Task SuspendAndResume_Reason_IsExternalizedByClient() + { + string largeReason1 = new string('Z', 700 * 1024); // 700KB + string largeReason2 = new string('Y', 650 * 1024); // 650KB + TaskName orchestratorName = nameof(SuspendAndResume_Reason_IsExternalizedByClient); + + InMemoryPayloadStore clientStore = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + // Long-running orchestrator to give time for suspend/resume + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orchestratorName, + async (ctx, _) => + { + await ctx.CreateTimer(TimeSpan.FromMinutes(5), CancellationToken.None); + return "done"; + })); + }, + client => + { + // Enable externalization on the client and use the in-memory store to track uploads + client.UseExternalizedPayloads(); + client.Services.AddSingleton(clientStore); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + await server.Client.WaitForInstanceStartAsync(instanceId, this.TimeoutToken); + + // Suspend with large reason (should be externalized by client) + await server.Client.SuspendInstanceAsync(instanceId, largeReason1, this.TimeoutToken); + await server.Client.WaitForInstanceStartAsync(instanceId, this.TimeoutToken); + + // poll up to 5 seconds to verify it is suspended + var deadline1 = DateTime.UtcNow.AddSeconds(5); + while (true) + { + OrchestrationMetadata? status1 = await server.Client.GetInstanceAsync(instanceId, getInputsAndOutputs: false, this.TimeoutToken); + if (status1 is not null && status1.RuntimeStatus == OrchestrationRuntimeStatus.Suspended) + { + break; + } + + if (DateTime.UtcNow >= deadline1) + { + Assert.NotNull(status1); + Assert.Equal(OrchestrationRuntimeStatus.Suspended, status1!.RuntimeStatus); + } + } + // Resume with large reason (should be externalized by client) + await server.Client.ResumeInstanceAsync(instanceId, largeReason2, this.TimeoutToken); + + // verify it is resumed (poll up to 5 seconds) + var deadline = DateTime.UtcNow.AddSeconds(5); + while (true) + { + OrchestrationMetadata? status = await server.Client.GetInstanceAsync(instanceId, getInputsAndOutputs: false, this.TimeoutToken); + if (status is not null && status.RuntimeStatus == OrchestrationRuntimeStatus.Running) + { + break; + } + + if (DateTime.UtcNow >= deadline) + { + Assert.NotNull(status); + Assert.Equal(OrchestrationRuntimeStatus.Running, status!.RuntimeStatus); + } + + await Task.Delay(TimeSpan.FromSeconds(1), this.TimeoutToken); + } + + + + Assert.True(clientStore.UploadCount >= 2); + Assert.Contains(largeReason1, clientStore.uploadedPayloads); + Assert.Contains(largeReason2, clientStore.uploadedPayloads); + } + + // Validates terminating an instance with a large output payload is externalized by the client. + [Fact] + public async Task LargeTerminateWithPayload() + { + string largeInput = new string('I', 900 * 1024); + string largeOutput = new string('T', 900 * 1024); + TaskName orch = nameof(LargeTerminateWithPayload); + + InMemoryPayloadStore store = new InMemoryPayloadStore(); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orch, + async (ctx, _) => + { + await ctx.CreateTimer(TimeSpan.FromSeconds(30), CancellationToken.None); + return null; + })); + + worker.UseExternalizedPayloads(); + worker.Services.AddSingleton(store); + }, + client => + { + client.UseExternalizedPayloads(); + client.Services.AddSingleton(store); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + string id = await server.Client.ScheduleNewOrchestrationInstanceAsync(orch, largeInput); + await server.Client.WaitForInstanceStartAsync(id, this.TimeoutToken); + + await server.Client.TerminateInstanceAsync(id, new TerminateInstanceOptions { Output = largeOutput }, this.TimeoutToken); + + await server.Client.WaitForInstanceCompletionAsync(id, this.TimeoutToken); + OrchestrationMetadata? status = await server.Client.GetInstanceAsync(id, getInputsAndOutputs: false); + Assert.NotNull(status); + Assert.Equal(OrchestrationRuntimeStatus.Terminated, status!.RuntimeStatus); + Assert.True(store.UploadCount >= 1); + Assert.True(store.DownloadCount >= 1); + Assert.Contains(JsonSerializer.Serialize(largeOutput), store.uploadedPayloads); + } + // Validates large custom status and ContinueAsNew input are externalized and resolved across iterations. + [Fact] + public async Task LargeContinueAsNewAndCustomStatus() + { + string largeStatus = new string('S', 700 * 1024); + string largeNextInput = new string('N', 800 * 1024); + string largeFinalOutput = new string('F', 750 * 1024); + TaskName orch = nameof(LargeContinueAsNewAndCustomStatus); + + var shared = new Dictionary(); + InMemoryPayloadStore workerStore = new InMemoryPayloadStore(shared); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orch, + async (ctx, input) => + { + if (input == null) + { + ctx.SetCustomStatus(largeStatus); + ctx.ContinueAsNew(largeNextInput); + // unreachable + return ""; + } + else + { + // second iteration returns final + return largeFinalOutput; + } + })); + + worker.UseExternalizedPayloads(); + worker.Services.AddSingleton(workerStore); + }, + client => + { + client.UseExternalizedPayloads(); + client.Services.AddSingleton(workerStore); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orch); + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal(largeFinalOutput, completed.ReadOutputAs()); + Assert.Contains(JsonSerializer.Serialize(largeStatus), workerStore.uploadedPayloads); + Assert.Contains(JsonSerializer.Serialize(largeNextInput), workerStore.uploadedPayloads); + Assert.Contains(JsonSerializer.Serialize(largeFinalOutput), workerStore.uploadedPayloads); + } + + // Validates large sub-orchestration input and an activity large output in one flow. + [Fact] + public async Task LargeSubOrchestrationAndActivityOutput() + { + string largeChildInput = new string('C', 650 * 1024); + string largeActivityOutput = new string('A', 820 * 1024); + TaskName parent = nameof(LargeSubOrchestrationAndActivityOutput) + "_Parent"; + TaskName child = nameof(LargeSubOrchestrationAndActivityOutput) + "_Child"; + TaskName activity = "ProduceBig"; + + var shared = new Dictionary(); + InMemoryPayloadStore workerStore = new InMemoryPayloadStore(shared); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks + .AddOrchestratorFunc( + parent, + async (ctx, _) => + { + string echoed = await ctx.CallSubOrchestratorAsync(child, largeChildInput); + string act = await ctx.CallActivityAsync(activity); + return echoed.Length + act.Length; + }) + .AddOrchestratorFunc(child, (ctx, input) => Task.FromResult(input)) + .AddActivityFunc(activity, (ctx) => Task.FromResult(largeActivityOutput))); + + worker.UseExternalizedPayloads(); + worker.Services.AddSingleton(workerStore); + }, + client => + { + client.UseExternalizedPayloads(); + client.Services.AddSingleton(workerStore); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + string id = await server.Client.ScheduleNewOrchestrationInstanceAsync(parent); + OrchestrationMetadata done = await server.Client.WaitForInstanceCompletionAsync( + id, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, done.RuntimeStatus); + Assert.Equal(largeChildInput.Length + largeActivityOutput.Length, done.ReadOutputAs()); + Assert.True(workerStore.UploadCount >= 1); + Assert.True(workerStore.DownloadCount >= 1); + Assert.Contains(JsonSerializer.Serialize(largeChildInput), workerStore.uploadedPayloads); + Assert.Contains(JsonSerializer.Serialize(largeActivityOutput), workerStore.uploadedPayloads); + } + + // Validates query with fetch I/O resolves large outputs for completed instances. + [Fact] + public async Task LargeQueryFetchInputsAndOutputs() + { + string largeIn = new string('I', 750 * 1024); + string largeOut = new string('Q', 880 * 1024); + TaskName orch = nameof(LargeQueryFetchInputsAndOutputs); + + var shared = new Dictionary(); + InMemoryPayloadStore workerStore = new InMemoryPayloadStore(shared); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orch, + (ctx, input) => Task.FromResult(largeOut))); + + worker.UseExternalizedPayloads(); + worker.Services.AddSingleton(workerStore); + }, + client => + { + client.UseExternalizedPayloads(); + client.Services.AddSingleton(workerStore); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + string id = await server.Client.ScheduleNewOrchestrationInstanceAsync(orch, largeIn); + await server.Client.WaitForInstanceCompletionAsync(id, getInputsAndOutputs: false, this.TimeoutToken); + + var page = server.Client.GetAllInstancesAsync(new OrchestrationQuery { FetchInputsAndOutputs = true, InstanceIdPrefix = id }); + OrchestrationMetadata? found = null; + await foreach (var item in page) + { + if (item.Name == orch.Name) + { + found = item; + break; + } + } + + Assert.NotNull(found); + Assert.Equal(largeOut, found!.ReadOutputAs()); + Assert.True(workerStore.DownloadCount >= 1); + Assert.True(workerStore.UploadCount >= 1); + Assert.Contains(JsonSerializer.Serialize(largeIn), workerStore.uploadedPayloads); + Assert.Contains(JsonSerializer.Serialize(largeOut), workerStore.uploadedPayloads); + } + // Validates worker externalizes large activity input and delivers resolved payload to activity. + [Fact] + public async Task LargeActivityInputAndOutput() + { + string largeParam = new string('P', 700 * 1024); // 700KB + TaskName orchestratorName = nameof(LargeActivityInputAndOutput); + 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 + input)); + + worker.UseExternalizedPayloads(); + worker.Services.AddSingleton(workerStore); + }, + client => { /* client not needed for externalization path here */ }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + OrchestrationMetadata completed = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + + // validate upload and download count + Assert.True(workerStore.UploadCount >= 1); + Assert.True(workerStore.DownloadCount >= 1); + + // validate uploaded payloads include the activity input and output forms + string expectedActivityInputJson = JsonSerializer.Serialize(new[] { largeParam }); + string expectedActivityOutputJson = JsonSerializer.Serialize(largeParam + largeParam); + Assert.Contains(expectedActivityInputJson, workerStore.uploadedPayloads); + Assert.Contains(expectedActivityOutputJson, workerStore.uploadedPayloads); + } + + + // 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(); + worker.Services.AddSingleton(workerStore); + }, + client => + { + client.UseExternalizedPayloads(); + 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 that payloads exceeding max cap are rejected with a clear error. + [Fact] + public async Task ExceedingMaxPayloadIsRejected() + { + string tooLarge = new string('Z', 3044); + TaskName orch = nameof(ExceedingMaxPayloadIsRejected); + + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks.AddOrchestratorFunc( + orch, + (ctx, input) => Task.FromResult("done"))); + worker.UseExternalizedPayloads(); + }, + client => + { + client.UseExternalizedPayloads(); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + // Keep a low threshold to force externalization, but default cap applies + opts.ExternalizeThresholdBytes = 1024; + opts.MaxExternalizedPayloadBytes = 2048; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + // The client will attempt to externalize the input and should fail fast on cap + await Assert.ThrowsAsync(async () => + { + await server.Client.ScheduleNewOrchestrationInstanceAsync(orch, tooLarge); + }); + } + + // 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.Services.AddSingleton(fakeStore); + worker.UseExternalizedPayloads(); + }, + client => + { + client.Services.AddSingleton(fakeStore); + client.UseExternalizedPayloads(); + }, + services => + { + services.AddExternalizedPayloadStore(opts => + { + opts.ExternalizeThresholdBytes = 1024; + opts.ContainerName = "test"; + opts.ConnectionString = "UseDevelopmentStorage=true"; + }); + }); + + 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); + Assert.True(fakeStore.DownloadCount >= 1); + Assert.Contains(JsonSerializer.Serialize(largeEvent), fakeStore.uploadedPayloads); + } + + class InMemoryPayloadStore : PayloadStore + { + const string TokenPrefix = "blob:v1:"; + readonly Dictionary tokenToPayload; + public readonly HashSet uploadedPayloads = new(); + + 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 override Task UploadAsync(string payLoad, CancellationToken cancellationToken) + { + Interlocked.Increment(ref this.uploadCount); + string token = $"blob:v1:test:{Guid.NewGuid():N}"; + this.tokenToPayload[token] = payLoad; + this.uploadedPayloads.Add(payLoad); + return Task.FromResult(token); + } + + public override Task DownloadAsync(string token, CancellationToken cancellationToken) + { + Interlocked.Increment(ref this.downloadCount); + return Task.FromResult(this.tokenToPayload[token]); + } + + public override bool IsKnownPayloadToken(string value) + { + return value.StartsWith(TokenPrefix, StringComparison.Ordinal); + } + } +}