Large payload azure blob externalization support

Directory.Packages.props

@@ -24,6 +24,7 @@
   <!-- Azure.* Packages -->
   <ItemGroup>
     <PackageVersion Include="Azure.Identity" Version="1.13.1" />
+    <PackageVersion Include="Azure.Storage.Blobs" Version="12.20.0" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker" Version="1.21.0" />
   </ItemGroup>
 
@@ -34,6 +35,11 @@
   </ItemGroup>
 
   <!-- Grpc / Protobuf Packages -->
+
+  <!-- BCL Packages -->
+  <ItemGroup>
+    <PackageVersion Include="System.ComponentModel.Annotations" Version="5.0.0" />
+  </ItemGroup>
   <ItemGroup>
     <PackageVersion Include="Google.Protobuf" Version="3.21.12" />
     <PackageVersion Include="Grpc.Core" Version="2.46.5" />

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}

samples/LargePayloadConsoleApp/LargePayloadConsoleApp.csproj

@@ -0,0 +1,24 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net8.0</TargetFramework>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Hosting" />
+    <PackageReference Include="Azure.Identity" />
+    <PackageReference Include="Grpc.Net.Client" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <!-- Using p2p references so we can show latest changes in samples. -->
+    <ProjectReference Include="$(SrcRoot)Client/AzureManaged/Client.AzureManaged.csproj" />
+    <ProjectReference Include="$(SrcRoot)Worker/AzureManaged/Worker.AzureManaged.csproj" />
+    <ProjectReference Include="$(SrcRoot)Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj" />
+  </ItemGroup>
+
+</Project>
+
+

samples/LargePayloadConsoleApp/Program.cs

@@ -0,0 +1,198 @@
+// 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<string>("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<string>("DURABLETASK_STORAGE") ?? "UseDevelopmentStorage=true";
+    opts.ContainerName = builder.Configuration.GetValue<string>("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<string, string>("LargeInputEcho", async (ctx, input) =>
+        {
+            string echoed = await ctx.CallActivityAsync<string>("Echo", input);
+            return echoed;
+        });
+
+        // Activity: validate it receives raw input (not token) and return it
+        tasks.AddActivityFunc<string, string>("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<object?, int>(
+            "LargeEntityOperationInput",
+            (ctx, _) => ctx.Entities.CallEntityAsync<int>(
+                new EntityInstanceId(nameof(EchoLengthEntity), "1"),
+                operationName: "EchoLength",
+                input: new string('E', 700 * 1024)));
+        tasks.AddEntity<EchoLengthEntity>(nameof(EchoLengthEntity));
+
+        tasks.AddOrchestratorFunc<object?, int>(
+            "LargeEntityOperationOutput",
+            async (ctx, _) => (await ctx.Entities.CallEntityAsync<string>(
+                new EntityInstanceId(nameof(LargeResultEntity), "1"),
+                operationName: "Produce",
+                input: 850 * 1024)).Length);
+        tasks.AddEntity<LargeResultEntity>(nameof(LargeResultEntity));
+
+        tasks.AddOrchestratorFunc<object?, object?>(
+            "LargeEntityState",
+            async (ctx, _) =>
+            {
+                await ctx.Entities.CallEntityAsync(
+                    new EntityInstanceId(nameof(StateEntity), "1"),
+                    operationName: "Set",
+                    input: new string('S', 900 * 1024));
+                return null;
+            });
+        tasks.AddEntity<StateEntity>(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<DurableTaskClient>();
+
+// 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>() ?? string.Empty;
+string deserializedOutput = result.ReadOutputAs<string>() ?? 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<int>();
+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<int>();
+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<string>? state = await client.Entities.GetEntityAsync<string>(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<int>
+{
+    public int EchoLength(string input)
+    {
+        return input.Length;
+    }
+}
+
+public class LargeResultEntity : TaskEntity<object?>
+{
+    public string Produce(int length)
+    {
+        return new string('R', length);
+    }
+}
+
+public class StateEntity : TaskEntity<string?>
+{
+    protected override string? InitializeState(TaskEntityOperation entityOperation)
+    {
+        // Avoid Activator.CreateInstance<string>() which throws; start as null (no state)
+        return null;
+    }
+
+    public void Set(string value)
+    {
+        this.State = value;
+    }
+}

samples/LargePayloadConsoleApp/Properties/launchSettings.json

@@ -0,0 +1,12 @@
+{
+  "profiles": {
+    "LargePayloadConsoleApp": {
+      "commandName": "Project",
+      "environmentVariables": {
+        "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "",
+        "DURABLETASK_STORAGE": "",
+        "DURABLETASK_PAYLOAD_CONTAINER": ""
+      }
+    }
+  }
+}

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.
+
+

src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj

@@ -0,0 +1,30 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks>netstandard2.0;net6.0</TargetFrameworks>
+    <PackageDescription>Azure Blob Storage externalized payload support for Durable Task.</PackageDescription>
+    <AssemblyName>Microsoft.DurableTask.Extensions.AzureBlobPayloads</AssemblyName>
+    <RootNamespace>Microsoft.DurableTask</RootNamespace>
+    <EnableStyleCop>true</EnableStyleCop>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Azure.Storage.Blobs" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" />
+    <PackageReference Include="Microsoft.Extensions.Options" />
+    <PackageReference Include="Grpc.Net.Client" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\Abstractions\Abstractions.csproj" />
+    <ProjectReference Include="..\..\Client\Core\Client.csproj" />
+    <ProjectReference Include="..\..\Worker\Core\Worker.csproj" />
+    <ProjectReference Include="..\..\Client\Grpc\Client.Grpc.csproj" />
+    <ProjectReference Include="..\..\Worker\Grpc\Worker.Grpc.csproj" />
+    <ProjectReference Include="..\..\Grpc.AzureManagedBackend\Grpc.AzureManagedBackend.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <SharedSection Include="Core" />
+  </ItemGroup>
+</Project>